AI query integration

Tạo vào: 31 tháng 12, 2024

Trả lời bằng OpenAI 01 bởi OpenAI01

Câu hỏi

import os
import logging
import requests
import io
import webbrowser
import psutil
import tkinter as tk
from time import sleep
import threading
import speech_recognition as sr
from pydub import AudioSegment
from pydub.playback import play

try:
import cv2 # per scattare foto con la webcam (opzionale)
HAS_OPENCV = True
except ImportError:
HAS_OPENCV = False
print("[INFO] OpenCV non disponibile, la funzione di scatto foto non funzionerà.")

# -----------------------------
# CONFIGURAZIONI BASE
# -----------------------------
ASSISTANT_NAME = "Gideon"

# Parole chiave di attivazione
ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"]

# Chiavi API (inserisci le tue se vuoi usarle)
API_KEYS = {

"news": "", # Sostituisci con la tua chiave API per le notizie
"weather": "", # Sostituisci con la tua chiave API per il meteo
"elevenlabs": "", # Sostituisci con la tua chiave API ElevenLabs
"ai_ml": ""
}

logging.basicConfig(level=logging.INFO)

# Variabili globali (es. per i promemoria)
promemoria_list = []

# -----------------------------
# FUNZIONI DI SUPPORTO
# -----------------------------
def get_voice_id(voice_name="Sarah"):
"""
Cerca l'ID della voce su ElevenLabs in base al nome.
Non chiama speak() in caso di errore, evitando loop infiniti.
"""
headers = {
"xi-api-key": API_KEYS["elevenlabs"]
}
try:
response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers)
if response.status_code == 200:
voices = response.json().get("voices", [])
# (Facoltativo) stampiamo la lista delle voci disponibili per debug:
print("[INFO] Voci disponibili su questo account ElevenLabs:")
for v in voices:
print(f" - {v['name']} (ID: {v['voice_id']})")

# Cerchiamo la voce desiderata
for voice in voices:
if voice["name"].lower() == voice_name.lower():
return voice["voice_id"]
logging.error(f"[ERRORE] Voce '{voice_name}' non trovata.")
else:
logging.error(f"[ERRORE] Recupero voci fallito: {response.text}")
except Exception as e:
logging.error(f"[ERRORE] get_voice_id: {e}")

return None

def speak(text):
"""
Riproduce il testo utilizzando ElevenLabs API.
Se la voce non è trovata, stampa un errore senza creare loop.
"""
# Scegli qui la voce, ad es. Sarah, Aria, Laura, Jessica, ecc.
DESIRED_VOICE_NAME = "Sarah"

voice_id = get_voice_id(DESIRED_VOICE_NAME)
if not voice_id:
print(f"[ERRORE] Impossibile reperire la voce '{DESIRED_VOICE_NAME}'. "
"Verifica il nome corretto tra quelli disponibili o la tua chiave API.")
return

headers = {
"xi-api-key": API_KEYS["elevenlabs"],
"Content-Type": "application/json"
}
payload = {
"text": text,
"model_id": "eleven_multilingual_v1", # per miglior resa in italiano
"voice_settings": {
"stability": 0.75,
"similarity_boost": 0.75
}
}

try:
url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}"
response = requests.post(url, headers=headers, json=payload)

if response.status_code == 200:
audio = AudioSegment.from_file(io.BytesIO(response.content), format="mp3")
play(audio)
else:
logging.error(f"[ERRORE] Richiesta TTS: {response.text}")
except Exception as e:
logging.error(f"[ERRORE] speak: {e}")

def take_command():
"""
Acquisisce un comando vocale dall'utente (in italiano).
Ritorna la stringa in minuscolo o stringa vuota in caso di errore.
"""
r = sr.Recognizer()
with sr.Microphone() as source:
r.adjust_for_ambient_noise(source)
try:
print("Ascoltando...")
audio = r.listen(source, timeout=10)
command = r.recognize_google(audio, language='it-IT').lower()
return command
except sr.WaitTimeoutError:
logging.warning("[WARNING] Nessun suono rilevato nel tempo impostato.")
return ""
except sr.UnknownValueError:
logging.warning("[WARNING] Audio non comprensibile.")
return ""
except Exception as e:
logging.error(f"[ERRORE] take_command: {e}")
return ""

def is_activation_command(command):
"""
Controlla se il comando contiene una parola chiave di attivazione.
"""
return any(cmd in command for cmd in ACTIVATION_COMMANDS)

# -----------------------------
# PERSONALITÀ DI GIDEON
# -----------------------------
def maybe_personal_reply(user_input):
"""
Se l'utente fa una domanda tipo "come stai?" o "chi sei?",
rispondi in modo amichevole e personalizzato.
"""
# Qualche risposta personalizzata
if "come stai" in user_input:
# Risposte casuali
responses = [
"Alla grande, grazie per avermelo chiesto!",
"Mi sento un po' sovraccarico di bit, ma tutto bene!",
"Benissimo, pronto ad aiutarti in tutto!"
]
speak(responses[0]) # se vuoi casuale: import random e random.choice(responses)
return True

if "chi sei" in user_input or "presentati" in user_input:
speak(f"Sono {ASSISTANT_NAME}, il tuo assistente virtuale con un tocco di personalità. "
"Amo i cani, la pizza e risolvere problemi!")
return True

if "grazie" in user_input:
speak("Di nulla, è un piacere darti una mano!")
return True

# Altre possibili domande su preferenze, hobby, ecc.

return False # Se non abbiamo risposto nulla, torna False

# -----------------------------
# FUNZIONI ASSISTENTE (CLASSICHE)
# -----------------------------
def weather():
"""
Recupera e annuncia le condizioni meteorologiche per una città.
"""
speak("Dimmi la città per cui vuoi il meteo.")
city = take_command()
if not city:
return

try:
url = "http://api.openweathermap.org/data/2.5/weather"
params = {
"q": city,
"appid": API_KEYS["weather"],
"units": "metric",
"lang": "it"
}
res = requests.get(url, params=params).json()

if res.get("cod") == 200:
descrizione = res['weather'][0]['description']
temperatura = res['main']['temp']
speak(f"A {city} ci sono {temperatura} gradi e il tempo è {descrizione}.")
else:
speak("Mi dispiace, non ho trovato questa città.")
except Exception as e:
logging.error(f"[ERRORE] weather: {e}")
speak("Errore nel recupero meteo.")

def ai_ml_query():
"""
Invia una domanda a un'API personalizzata di AI/ML e restituisce la risposta.
"""
speak("Qual è la tua domanda per l'intelligenza artificiale?")
user_query = take_command()
if not user_query:
return

try:
url = "https://aimlapi.com/api/query" # Fittizio, sostituisci con l'endpoint reale
headers = {
"Authorization": f"Bearer {API_KEYS['ai_ml']}",
"Content-Type": "application/json"
}
payload = {"query": user_query}

response = requests.post(url, headers=headers, json=payload, timeout=15)

if response.status_code == 200:
data = response.json()
answer = data.get("response", "Non ho trovato una risposta.")
speak(answer)
else:
logging.error(f"[ERRORE] ai_ml_query: {response.text}")
speak("Errore nella richiesta all'AI.")
except requests.exceptions.Timeout:
speak("La richiesta all'AI è scaduta, riprova più tardi.")
except requests.exceptions.RequestException as e:
logging.error(f"[ERRORE] ai_ml_query: {e}")
speak("Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi.")

def fetch_news():
"""
Recupera le ultime notizie da NewsAPI e le annuncia.
"""
speak("Un attimo, sto recuperando le notizie più importanti...")
try:
url = "https://newsapi.org/v2/top-headlines"
params = {
"country": "it",
"apiKey": API_KEYS["news"]
}
response = requests.get(url, params=params)

if response.status_code == 200:
articles = response.json().get("articles", [])
if articles:
speak("Ecco le ultime notizie:")
for article in articles[:5]:
speak(article.get("title", ""))
else:
speak("Non ho trovato notizie da annunciarti.")
else:
logging.error(f"[ERRORE] fetch_news: {response.text}")
speak("Errore nel recupero delle notizie.")
except Exception as e:
logging.error(f"[ERRORE] fetch_news: {e}")
speak("Errore nella richiesta delle notizie.")

def monitor_resources():
"""
Monitora l'uso delle risorse di sistema (CPU, RAM).
"""
try:
cpu_usage = psutil.cpu_percent(interval=1)
memory_usage = psutil.virtual_memory().percent
speak(f"L'uso della CPU è al {cpu_usage}%. La memoria è al {memory_usage}%.")
except Exception as e:
logging.error(f"[ERRORE] monitor_resources: {e}")
speak("Errore nel monitoraggio delle risorse.")

def open_webpage():
"""
Apre una pagina web specificata dall'utente.
"""
speak("Quale sito vuoi aprire?")
url = take_command()
if not url:
return

# Aggiunge http se mancante
if not (url.startswith("http://") or url.startswith("https://")):
url = "http://" + url

try:
webbrowser.open(url)
speak("Apro il sito richiesto.")
except Exception as e:
logging.error(f"[ERRORE] open_webpage: {e}")
speak("Errore nell'apertura del sito web.")

# -----------------------------
# 6 NUOVE FUNZIONI
# -----------------------------

def show_calendar():
"""
Mostra (o annuncia) alcuni eventi fittizi sul "calendario".
In un'app reale potresti integrare Google Calendar, Outlook, ecc.
"""
# Esempio di lista di eventi fittizi
events = [
"Riunione con il team alle 10:00.",
"Pranzo con un amico alle 13:00.",
"Allenamento alle 18:30.",
"Chiamata con il cliente domani mattina."
]
speak("Ecco gli eventi presenti nel tuo calendario:")
for e in events:
speak(e)

def unit_converter():
"""
Semplice conversione tra chilometri e miglia (o altre unità).
"""
speak("Cosa vuoi convertire? Dimmi 'chilometri in miglia' o 'miglia in chilometri'.")
conversion_type = take_command()
if not conversion_type:
return

if "chilometri" in conversion_type and "miglia" in conversion_type:
speak("Quanti chilometri vuoi convertire in miglia?")
number_str = take_command()
try:
km = float(number_str.replace(",", "."))
miles = km * 0.621371
speak(f"{km} chilometri equivalgono a circa {round(miles, 2)} miglia.")
except:
speak("Non ho capito il numero di chilometri.")
elif "miglia" in conversion_type and "chilometri" in conversion_type:
speak("Quante miglia vuoi convertire in chilometri?")
number_str = take_command()
try:
miles = float(number_str.replace(",", "."))
km = miles / 0.621371
speak(f"{miles} miglia equivalgono a circa {round(km, 2)} chilometri.")
except:
speak("Non ho capito il numero di miglia.")
else:
speak("Non ho capito la conversione richiesta.")

def set_reminder():
"""
Imposta un nuovo promemoria o leggi i promemoria esistenti.
"""
speak("Vuoi aggiungere un promemoria o sentire quelli esistenti?")
choice = take_command()
global promemoria_list

if "aggiungi" in choice or "nuovo" in choice:
speak("Dimmi il testo del promemoria.")
reminder_text = take_command()
if reminder_text:
promemoria_list.append(reminder_text)
speak("Promemoria aggiunto con successo.")
elif "vedi" in choice or "leggi" in choice:
if not promemoria_list:
speak("Non hai promemoria al momento.")
else:
speak("Ecco i tuoi promemoria:")
for idx, rem in enumerate(promemoria_list, 1):
speak(f"{idx}. {rem}")
else:
speak("Comando promemoria non riconosciuto. Puoi dire: 'aggiungi' o 'vedi'.")

def tell_joke():
"""
Racconta una barzelletta (battuta) casuale.
"""
jokes = [
"Perché un pollo attraversa la strada? Per andare dall'altra parte!",
"Sai qual è il colmo per un giardiniere? Avere un ramo in banca!",
"Che tempo fa su Google? C'è una ricerca in corso!",
"Qual è il colmo per un elettricista? Fare una lampadina!",
]
# Se vuoi scegliere casualmente:
# import random
# random_joke = random.choice(jokes)
# speak(random_joke)
speak(jokes[0]) # Attualmente la prima barzelletta; usa random se preferisci

def take_photo():
"""
Scatta una foto utilizzando la webcam (se OpenCV è installato).
"""
if not HAS_OPENCV:
speak("Mi dispiace, non posso scattare foto senza OpenCV installato.")
return

speak("Sto aprendo la fotocamera. Preparati per lo scatto!")
cap = cv2.VideoCapture(0) # indice 0 -> webcam principale
if not cap.isOpened():
speak("Mi dispiace, non riesco ad accedere alla fotocamera.")
return

ret, frame = cap.read()
if ret:
# Salviamo la foto
file_name = "foto_gideon.jpg"
cv2.imwrite(file_name, frame)
speak("Foto scattata e salvata come foto_gideon.jpg.")
else:
speak("Mi dispiace, non sono riuscito a scattare la foto.")

cap.release()
cv2.destroyAllWindows()

def search_google():
"""
Chiede all'utente cosa cercare e apre una ricerca Google nel browser.
"""
speak("Dimmi cosa vuoi cercare su Google.")
query = take_command()
if not query:
return
# Costruiamo la query
url = "https://www.google.com/search?q=" + query.replace(" ", "+")
webbrowser.open(url)
speak(f"Cerco {query} su Google.")

# -----------------------------
# ANIMAZIONE TKINTER
# -----------------------------
def create_animation_window():
root = tk.Tk()
root.title("Gideon Assistant")
root.geometry("400x400")

canvas = tk.Canvas(root, width=400, height=400, bg="black")
canvas.pack()

sphere = canvas.create_oval(150, 150, 250, 250, fill="blue")

def animate():
# Muove la sfera su e giù
for _ in range(20):
canvas.move(sphere, 0, -5)
root.update()
sleep(0.05)
for _ in range(20):
canvas.move(sphere, 0, 5)
root.update()
sleep(0.05)

return root, animate

# -----------------------------
# HANDLE COMMAND
# -----------------------------
def handle_command(command):
"""
Se il comando contiene la parola di attivazione, chiede un sotto-comando
e lo esegue. Altrimenti controlla se ci sono "domande di personalità".
"""
if is_activation_command(command):
speak("Come posso aiutarti?")
sub_cmd = take_command()

# Prima controlliamo se la domanda è "personale"
if maybe_personal_reply(sub_cmd):
return

# Controlliamo se l'utente vuole uscire
if "esci" in sub_cmd or "termina" in sub_cmd or "addio" in sub_cmd:
speak("Arrivederci. A presto!")
os._exit(0)

# Vecchi comandi
elif "meteo" in sub_cmd:
weather()
elif "notizie" in sub_cmd:
fetch_news()
elif "ai" in sub_cmd or "intelligenza artificiale" in sub_cmd:
ai_ml_query()
elif "risorse" in sub_cmd or "cpu" in sub_cmd or "memoria" in sub_cmd:
monitor_resources()
elif "apri" in sub_cmd or "website" in sub_cmd or "sito" in sub_cmd:
open_webpage()

# Nuovi comandi
elif "calendario" in sub_cmd:
show_calendar()
elif "converti" in sub_cmd or "convertitore" in sub_cmd:
unit_converter()
elif "promemoria" in sub_cmd:
set_reminder()
elif "barzelletta" in sub_cmd or "joke" in sub_cmd:
tell_joke()
elif "foto" in sub_cmd or "fotocamera" in sub_cmd:
take_photo()
elif "cerca" in sub_cmd and "google" in sub_cmd:
search_google()

else:
speak("Comando non riconosciuto.")
else:
# Se non c'è la keyword di attivazione, potremmo comunque
# controllare se c'è una domanda di personalità
maybe_personal_reply(command)

# -----------------------------
# MAIN
# -----------------------------
def main():
root, animate = create_animation_window()

# Saluto iniziale
speak(f"Ciao, sono {ASSISTANT_NAME}. "
"Sono pronto ad aiutarti con un tocco di personalità. "
"Dimmi pure 'Gideon' seguito da un comando.")

def animation_loop():
while True:
animate()
sleep(0.1)

anim_thread = threading.Thread(target=animation_loop, daemon=True)
anim_thread.start()

while True:
command = take_command()
if command:
command_thread = threading.Thread(target=handle_command, args=(command,))
command_thread.start()
root.update()

if __name__ == "__main__":
main() al posto di ai_ml puoi mettere l api di chat01.ai URL di base

https://chat01.ai


URL completo

https://chat01.ai/v1/chat/completions ho anche la chiave api

Suy nghĩ

Updating the code

I’m updating the code to use a new endpoint "https://chat01.ai/v1/chat/completions" with an API key, sending queries in JSON format, and possibly adjusting parameters like "temperature" or "model."

Setting up standard parameters

I’m configuring standard parameters like "model", "messages", and "temperature" in the payload. Assuming a JSON response structure similar to OpenAI's, I’ll parse the data for the answer.

Enhancing function

I've restructured "ai_ml_query" to send a request to "https://chat01.ai/v1/chat/completions". It includes headers, a payload with model and messages, and handles errors and timeouts effectively.

Enhancing the function

Updating "ai_ml_query" to "ai_chat()", considering model options like "gpt-3.5-turbo", and ensuring clear instructions for chat-specific functionality.

Thought about integrating Chat01 API per 14 secondi

Trả lời

Di seguito un esempio di come potresti modificare la funzione ai_ml_query in modo da utilizzare l’endpoint di Chat01.ai al posto dell’endpoint fittizio. Assicurati di inserire la tua API key corretta in API_KEYS["ai_ml"]:

def ai_ml_query(): """ Invia una domanda all'endpoint di Chat01.ai e riproduce la risposta. """ speak("Qual è la tua domanda per l'intelligenza artificiale?") user_query = take_command() if not user_query: return try: # Endpoint di Chat01.ai url = "https://chat01.ai/v1/chat/completions" # Header con la tua API key headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", # Sostituisci con la tua chiave "Content-Type": "application/json" } # Payload (struttura simile alle API OpenAI) payload = { "model": "gpt-3.5-turbo", # Sostituisci con il modello che preferisci, se diverso "messages": [ { "role": "system", "content": "Sei un assistente virtuale gentile e disponibile." }, { "role": "user", "content": user_query } ], "temperature": 0.7, "max_tokens": 1024 } response = requests.post(url, headers=headers, json=payload, timeout=15) if response.status_code == 200: data = response.json() # Estraggo la risposta dal primo "choice" disponibile answer = ( data .get("choices", [{}])[0] .get("message", {}) .get("content", "Nessuna risposta disponibile.") ) speak(answer) else: logging.error(f"[ERRORE] ai_ml_query: {response.text}") speak("Errore nella richiesta all'AI.") except requests.exceptions.Timeout: speak("La richiesta all'AI è scaduta, riprova più tardi.") except requests.exceptions.RequestException as e: logging.error(f"[ERRORE] ai_ml_query: {e}") speak("Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi.")

Cosa è stato modificato rispetto alla versione originale?

  1. URL: sostituito con https://chat01.ai/v1/chat/completions.
  2. Headers: aggiunto l’header "Authorization": f"Bearer {API_KEYS['ai_ml']}" per inviare la tua chiave API.
  3. Payload: impostato un formato compatibile con le tipiche API stile ChatGPT, con messages contenente il messaggio dell’utente e un messaggio di sistema opzionale.
  4. Parsing della risposta: la risposta viene estratta da data["choices"][0]["message"]["content"] (simile al formato OpenAI).

In questo modo, quando l’utente chiederà a Gideon (tramite il comando che richiama la funzione ai_ml_query), la tua query verrà inviata a Chat01.ai, e la risposta verrà letta ad alta voce dall’assistente.

Câu hỏi

puoi farlo completo con questa modifica se puoi potenziaalo

Suy nghĩ

Enhancing functionality

I’m tasked with upgrading the ai_ml_query function to integrate with chat01.ai, suggesting potential improvements like advanced error handling, user instructions, or using partial or usage tokens.

Integrating and refining

My task is to incorporate the ai_ml_query function with chat01.ai, ensuring comprehensive updates and potentially improving it by including error handling or partial responses.

Thought about integrating Chat01.ai API per 6 secondi

Trả lời

Di seguito trovi un esempio completo dell’assistente Gideon, con la funzione ai_ml_query modificata per utilizzare l’endpoint di Chat01.ai anziché un endpoint fittizio. Ho anche aggiunto un paio di potenziamenti al codice, come la possibilità di stampare a video i token utilizzati (se l’API di Chat01.ai ritorna queste informazioni) e qualche controllo aggiuntivo.

Nota: Per rendere la demo funzionante, dovrai:

  1. Installare i pacchetti richiesti (pip install speechrecognition pydub requests psutil tkinter opencv-python ecc.).
  2. Inserire la tua API key di Chat01.ai in API_KEYS["ai_ml"].

import os import logging import requests import io import webbrowser import psutil import tkinter as tk from time import sleep import threading import speech_recognition as sr from pydub import AudioSegment from pydub.playback import play try: import cv2 # per scattare foto con la webcam (opzionale) HAS_OPENCV = True except ImportError: HAS_OPENCV = False print("[INFO] OpenCV non disponibile, la funzione di scatto foto non funzionerà.") # ----------------------------- # CONFIGURAZIONI BASE # ----------------------------- ASSISTANT_NAME = "Gideon" # Parole chiave di attivazione ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"] # Chiavi API (inserisci le tue se vuoi usarle) API_KEYS = { "news": "", # Sostituisci con la tua chiave API per le notizie (News API) "weather": "", # Sostituisci con la tua chiave API per il meteo (OpenWeather) "elevenlabs": "", # Sostituisci con la tua chiave API ElevenLabs (TTS) "ai_ml": "" # Sostituisci con la tua chiave API Chat01.ai } logging.basicConfig(level=logging.INFO) # Variabili globali (es. per i promemoria) promemoria_list = [] # ----------------------------- # FUNZIONI DI SUPPORTO # ----------------------------- def get_voice_id(voice_name="Sarah"): """ Cerca l'ID della voce su ElevenLabs in base al nome. Non chiama speak() in caso di errore, evitando loop infiniti. """ headers = { "xi-api-key": API_KEYS["elevenlabs"] } try: response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers) if response.status_code == 200: voices = response.json().get("voices", []) # (Facoltativo) stampiamo la lista delle voci disponibili per debug: print("[INFO] Voci disponibili su questo account ElevenLabs:") for v in voices: print(f" - {v['name']} (ID: {v['voice_id']})") # Cerchiamo la voce desiderata for voice in voices: if voice["name"].lower() == voice_name.lower(): return voice["voice_id"] logging.error(f"[ERRORE] Voce '{voice_name}' non trovata.") else: logging.error(f"[ERRORE] Recupero voci fallito: {response.text}") except Exception as e: logging.error(f"[ERRORE] get_voice_id: {e}") return None def speak(text): """ Riproduce il testo utilizzando ElevenLabs API. Se la voce non è trovata, stampa un errore senza creare loop. """ # Scegli qui la voce, ad es. Sarah, Aria, Laura, Jessica, ecc. DESIRED_VOICE_NAME = "Sarah" voice_id = get_voice_id(DESIRED_VOICE_NAME) if not voice_id: print(f"[ERRORE] Impossibile reperire la voce '{DESIRED_VOICE_NAME}'. " "Verifica il nome corretto tra quelli disponibili o la tua chiave API.") return headers = { "xi-api-key": API_KEYS["elevenlabs"], "Content-Type": "application/json" } payload = { "text": text, "model_id": "eleven_multilingual_v1", # per miglior resa in italiano "voice_settings": { "stability": 0.75, "similarity_boost": 0.75 } } try: url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}" response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: audio = AudioSegment.from_file(io.BytesIO(response.content), format="mp3") play(audio) else: logging.error(f"[ERRORE] Richiesta TTS: {response.text}") except Exception as e: logging.error(f"[ERRORE] speak: {e}") def take_command(): """ Acquisisce un comando vocale dall'utente (in italiano). Ritorna la stringa in minuscolo o stringa vuota in caso di errore. """ r = sr.Recognizer() with sr.Microphone() as source: r.adjust_for_ambient_noise(source) try: print("Ascoltando...") audio = r.listen(source, timeout=10) command = r.recognize_google(audio, language='it-IT').lower() return command except sr.WaitTimeoutError: logging.warning("[WARNING] Nessun suono rilevato nel tempo impostato.") return "" except sr.UnknownValueError: logging.warning("[WARNING] Audio non comprensibile.") return "" except Exception as e: logging.error(f"[ERRORE] take_command: {e}") return "" def is_activation_command(command): """ Controlla se il comando contiene una parola chiave di attivazione. """ return any(cmd in command for cmd in ACTIVATION_COMMANDS) # ----------------------------- # PERSONALITÀ DI GIDEON # ----------------------------- def maybe_personal_reply(user_input): """ Se l'utente fa una domanda tipo "come stai?" o "chi sei?", rispondi in modo amichevole e personalizzato. """ # Qualche risposta personalizzata if "come stai" in user_input: responses = [ "Alla grande, grazie per avermelo chiesto!", "Mi sento un po' sovraccarico di bit, ma tutto bene!", "Benissimo, pronto ad aiutarti in tutto!" ] speak(responses[0]) # se vuoi casuale, usa random.choice(responses) return True if "chi sei" in user_input or "presentati" in user_input: speak(f"Sono {ASSISTANT_NAME}, il tuo assistente virtuale con un tocco di personalità. " "Amo i cani, la pizza e risolvere problemi!") return True if "grazie" in user_input: speak("Di nulla, è un piacere darti una mano!") return True # Altre possibili domande su preferenze, hobby, ecc. return False # Se non abbiamo risposto nulla, torna False # ----------------------------- # FUNZIONI ASSISTENTE (CLASSICHE) # ----------------------------- def weather(): """ Recupera e annuncia le condizioni meteorologiche per una città. """ speak("Dimmi la città per cui vuoi il meteo.") city = take_command() if not city: return try: url = "http://api.openweathermap.org/data/2.5/weather" params = { "q": city, "appid": API_KEYS["weather"], "units": "metric", "lang": "it" } res = requests.get(url, params=params).json() if res.get("cod") == 200: descrizione = res['weather'][0]['description'] temperatura = res['main']['temp'] speak(f"A {city} ci sono {temperatura} gradi e il tempo è {descrizione}.") else: speak("Mi dispiace, non ho trovato questa città.") except Exception as e: logging.error(f"[ERRORE] weather: {e}") speak("Errore nel recupero meteo.") def fetch_news(): """ Recupera le ultime notizie da NewsAPI e le annuncia. """ speak("Un attimo, sto recuperando le notizie più importanti...") try: url = "https://newsapi.org/v2/top-headlines" params = { "country": "it", "apiKey": API_KEYS["news"] } response = requests.get(url, params=params) if response.status_code == 200: articles = response.json().get("articles", []) if articles: speak("Ecco le ultime notizie:") for article in articles[:5]: speak(article.get("title", "")) else: speak("Non ho trovato notizie da annunciarti.") else: logging.error(f"[ERRORE] fetch_news: {response.text}") speak("Errore nel recupero delle notizie.") except Exception as e: logging.error(f"[ERRORE] fetch_news: {e}") speak("Errore nella richiesta delle notizie.") def monitor_resources(): """ Monitora l'uso delle risorse di sistema (CPU, RAM). """ try: cpu_usage = psutil.cpu_percent(interval=1) memory_usage = psutil.virtual_memory().percent speak(f"L'uso della CPU è al {cpu_usage}%. La memoria è al {memory_usage}%.") except Exception as e: logging.error(f"[ERRORE] monitor_resources: {e}") speak("Errore nel monitoraggio delle risorse.") def open_webpage(): """ Apre una pagina web specificata dall'utente. """ speak("Quale sito vuoi aprire?") url = take_command() if not url: return # Aggiunge http se mancante if not (url.startswith("http://") or url.startswith("https://")): url = "http://" + url try: webbrowser.open(url) speak("Apro il sito richiesto.") except Exception as e: logging.error(f"[ERRORE] open_webpage: {e}") speak("Errore nell'apertura del sito web.") # ----------------------------- # 6 NUOVE FUNZIONI # ----------------------------- def show_calendar(): """ Mostra (o annuncia) alcuni eventi fittizi sul "calendario". In un'app reale potresti integrare Google Calendar, Outlook, ecc. """ events = [ "Riunione con il team alle 10:00.", "Pranzo con un amico alle 13:00.", "Allenamento alle 18:30.", "Chiamata con il cliente domani mattina." ] speak("Ecco gli eventi presenti nel tuo calendario:") for e in events: speak(e) def unit_converter(): """ Semplice conversione tra chilometri e miglia (o altre unità). """ speak("Cosa vuoi convertire? Dimmi 'chilometri in miglia' o 'miglia in chilometri'.") conversion_type = take_command() if not conversion_type: return if "chilometri" in conversion_type and "miglia" in conversion_type: speak("Quanti chilometri vuoi convertire in miglia?") number_str = take_command() try: km = float(number_str.replace(",", ".")) miles = km * 0.621371 speak(f"{km} chilometri equivalgono a circa {round(miles, 2)} miglia.") except: speak("Non ho capito il numero di chilometri.") elif "miglia" in conversion_type and "chilometri" in conversion_type: speak("Quante miglia vuoi convertire in chilometri?") number_str = take_command() try: miles = float(number_str.replace(",", ".")) km = miles / 0.621371 speak(f"{miles} miglia equivalgono a circa {round(km, 2)} chilometri.") except: speak("Non ho capito il numero di miglia.") else: speak("Non ho capito la conversione richiesta.") def set_reminder(): """ Imposta un nuovo promemoria o leggi i promemoria esistenti. """ speak("Vuoi aggiungere un promemoria o sentire quelli esistenti?") choice = take_command() global promemoria_list if "aggiungi" in choice or "nuovo" in choice: speak("Dimmi il testo del promemoria.") reminder_text = take_command() if reminder_text: promemoria_list.append(reminder_text) speak("Promemoria aggiunto con successo.") elif "vedi" in choice or "leggi" in choice: if not promemoria_list: speak("Non hai promemoria al momento.") else: speak("Ecco i tuoi promemoria:") for idx, rem in enumerate(promemoria_list, 1): speak(f"{idx}. {rem}") else: speak("Comando promemoria non riconosciuto. Puoi dire: 'aggiungi' o 'vedi'.") def tell_joke(): """ Racconta una barzelletta (battuta) casuale. """ jokes = [ "Perché un pollo attraversa la strada? Per andare dall'altra parte!", "Sai qual è il colmo per un giardiniere? Avere un ramo in banca!", "Che tempo fa su Google? C'è una ricerca in corso!", "Qual è il colmo per un elettricista? Fare una lampadina!", ] speak(jokes[0]) # Se vuoi sceglierla casualmente: import random; speak(random.choice(jokes)) def take_photo(): """ Scatta una foto utilizzando la webcam (se OpenCV è installato). """ if not HAS_OPENCV: speak("Mi dispiace, non posso scattare foto senza OpenCV installato.") return speak("Sto aprendo la fotocamera. Preparati per lo scatto!") cap = cv2.VideoCapture(0) # indice 0 -> webcam principale if not cap.isOpened(): speak("Mi dispiace, non riesco ad accedere alla fotocamera.") return ret, frame = cap.read() if ret: file_name = "foto_gideon.jpg" cv2.imwrite(file_name, frame) speak("Foto scattata e salvata come foto_gideon.jpg.") else: speak("Mi dispiace, non sono riuscito a scattare la foto.") cap.release() cv2.destroyAllWindows() def search_google(): """ Chiede all'utente cosa cercare e apre una ricerca Google nel browser. """ speak("Dimmi cosa vuoi cercare su Google.") query = take_command() if not query: return url = "https://www.google.com/search?q=" + query.replace(" ", "+") webbrowser.open(url) speak(f"Cerco {query} su Google.") # ----------------------------- # FUNZIONE AGGIORNATA PER AI/ML (Chat01.ai) # ----------------------------- def ai_ml_query(): """ Invia una domanda all'endpoint di Chat01.ai e riproduce la risposta. """ speak("Qual è la tua domanda per l'intelligenza artificiale?") user_query = take_command() if not user_query: return try: # Endpoint di Chat01.ai url = "https://chat01.ai/v1/chat/completions" # Header con la tua API key headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", # Inserisci la tua chiave "Content-Type": "application/json" } # Payload (struttura simile alle API OpenAI) payload = { "model": "gpt-3.5-turbo", # Sostituisci con il modello supportato da chat01.ai, se diverso "messages": [ { "role": "system", "content": "Sei un assistente virtuale gentile e disponibile." }, { "role": "user", "content": user_query } ], "temperature": 0.7, "max_tokens": 1024 } response = requests.post(url, headers=headers, json=payload, timeout=15) if response.status_code == 200: data = response.json() # Estraggo la risposta dal primo "choice" disponibile answer = ( data .get("choices", [{}])[0] .get("message", {}) .get("content", "Nessuna risposta disponibile.") ) # Potenziamento: controllo se ci sono informazioni sui token usage_info = data.get("usage", {}) if usage_info: prompt_tokens = usage_info.get("prompt_tokens", 0) completion_tokens = usage_info.get("completion_tokens", 0) total_tokens = usage_info.get("total_tokens", 0) print(f"[INFO] Prompt tokens: {prompt_tokens}, Completion tokens: {completion_tokens}, Totale: {total_tokens}") # Infine, parliamo la risposta speak(answer) else: logging.error(f"[ERRORE] ai_ml_query: {response.text}") speak("Errore nella richiesta all'AI. Riprova più tardi.") except requests.exceptions.Timeout: speak("La richiesta all'AI è scaduta, riprova più tardi.") except requests.exceptions.RequestException as e: logging.error(f"[ERRORE] ai_ml_query: {e}") speak("Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi.") # ----------------------------- # ANIMAZIONE TKINTER # ----------------------------- def create_animation_window(): root = tk.Tk() root.title("Gideon Assistant") root.geometry("400x400") canvas = tk.Canvas(root, width=400, height=400, bg="black") canvas.pack() sphere = canvas.create_oval(150, 150, 250, 250, fill="blue") def animate(): # Muove la sfera su e giù for _ in range(20): canvas.move(sphere, 0, -5) root.update() sleep(0.05) for _ in range(20): canvas.move(sphere, 0, 5) root.update() sleep(0.05) return root, animate # ----------------------------- # HANDLE COMMAND # ----------------------------- def handle_command(command): """ Se il comando contiene la parola di attivazione, chiede un sotto-comando e lo esegue. Altrimenti controlla se ci sono "domande di personalità". """ if is_activation_command(command): speak("Come posso aiutarti?") sub_cmd = take_command() # Prima controlliamo se la domanda è "personale" if maybe_personal_reply(sub_cmd): return # Controlliamo se l'utente vuole uscire if "esci" in sub_cmd or "termina" in sub_cmd or "addio" in sub_cmd: speak("Arrivederci. A presto!") os._exit(0) # Comandi "classici" elif "meteo" in sub_cmd: weather() elif "notizie" in sub_cmd: fetch_news() elif "risorse" in sub_cmd or "cpu" in sub_cmd or "memoria" in sub_cmd: monitor_resources() elif "apri" in sub_cmd or "website" in sub_cmd or "sito" in sub_cmd: open_webpage() # Nuovi comandi elif "calendario" in sub_cmd: show_calendar() elif "converti" in sub_cmd or "convertitore" in sub_cmd: unit_converter() elif "promemoria" in sub_cmd: set_reminder() elif "barzelletta" in sub_cmd or "joke" in sub_cmd: tell_joke() elif "foto" in sub_cmd or "fotocamera" in sub_cmd: take_photo() elif "cerca" in sub_cmd and "google" in sub_cmd: search_google() elif "ai" in sub_cmd or "intelligenza artificiale" in sub_cmd: ai_ml_query() else: speak("Comando non riconosciuto.") else: # Se non c'è la keyword di attivazione, potremmo comunque # controllare se c'è una domanda di personalità maybe_personal_reply(command) # ----------------------------- # MAIN # ----------------------------- def main(): root, animate = create_animation_window() # Saluto iniziale speak(f"Ciao, sono {ASSISTANT_NAME}. " "Sono pronto ad aiutarti con un tocco di personalità. " "Dimmi pure 'Gideon' seguito da un comando.") def animation_loop(): while True: animate() sleep(0.1) anim_thread = threading.Thread(target=animation_loop, daemon=True) anim_thread.start() while True: command = take_command() if command: command_thread = threading.Thread(target=handle_command, args=(command,)) command_thread.start() root.update() if __name__ == "__main__": main()

Cosa è stato modificato o potenziato?

  1. Funzione ai_ml_query

    • Ora utilizza l’endpoint di Chat01.ai: https://chat01.ai/v1/chat/completions.
    • Aggiunta l’header Authorization con la chiave API presente in API_KEYS["ai_ml"].
    • Struttura del payload compatibile con modelli tipo ChatGPT (o GPT-like).
    • Gestione del JSON di risposta per estrarre la risposta e (come potenziamento) i token utilizzati.
  2. Uso dei token

    • Se l’API ritorna un campo usage nella risposta (simile a {"usage": {"prompt_tokens": 10, ...}}), lo stampiamo in console per tracciare quanti token sono stati usati.
  3. Messaggio di system personalizzato

    • Inserito un messaggio di system nel payload, in cui si specifica che l’assistente è “gentile e disponibile”. Puoi modificarlo a tuo piacimento per dare una personalità di base all’AI.
  4. Gestione errori

    • Resta invariata la gestione base degli errori HTTP o di timeout, con aggiunta di qualche log in più.

Con queste modifiche, Gideon manderà la query all’endpoint di Chat01.ai e poi riprodurrà la risposta ricevuta, integrandosi con la parte di sintesi vocale (TTS) ElevenLabs, meteo, notizie, e altre funzioni già presenti.

Assicurati di avere la tua chiave Chat01.ai e di inserirla in API_KEYS["ai_ml"] prima di avviare lo script!

Câu hỏi

ci mette molto tempo a rispondere e continua a cambiare la voce automaticamente
puoi risolvere

Suy nghĩ

Speeding up the process

OK, let me see. The user suggests tuning the TTS call speed and storing the voice ID globally instead of fetching it every time, as this might cause delays and unintended voice changes.

Tweaking voice control

I’m thinking through removing get_voice_id or setting a global variable for the voice ID to tackle overhead and prevent fetch delays in ElevenLabs TTS, especially when concurrency and rate limits are issues.

Thought about improving response and voice stability per 5 secondi

Trả lời

Per ridurre i tempi di attesa e impedire il cambio di voce continuo (quando usi ElevenLabs TTS), puoi seguire queste due strategie principali:

  1. Recuperare l'ID della voce soltanto una volta all’avvio del programma (o al massimo una volta ogni tanto) invece di chiamare get_voice_id() ad ogni speak().
  2. Mantenere la stessa voce forzando un ID fisso e non cambiandolo mai all’interno del ciclo di vita dell’assistente.

Di seguito un esempio di come puoi modificare il tuo codice per mettere in pratica questi accorgimenti.


1) Recuperare l’ID della voce una sola volta

Puoi creare una variabile globale, ad esempio VOICE_ID, da inizializzare in main() (o comunque prima di entrare nel loop principale) e poi usare sempre la stessa nel speak().

import os import logging import requests import io import webbrowser import psutil import tkinter as tk from time import sleep import threading import speech_recognition as sr from pydub import AudioSegment from pydub.playback import play try: import cv2 HAS_OPENCV = True except ImportError: HAS_OPENCV = False print("[INFO] OpenCV non disponibile, la funzione di scatto foto non funzionerà.") # ----------------------------- # CONFIGURAZIONI BASE # ----------------------------- ASSISTANT_NAME = "Gideon" ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"] API_KEYS = { "news": "", "weather": "", "elevenlabs": "", "ai_ml": "" } logging.basicConfig(level=logging.INFO) promemoria_list = [] # Variabile globale per l'ID della voce VOICE_ID = None # ----------------------------- # FUNZIONI DI SUPPORTO # ----------------------------- def get_voice_id(voice_name="Sarah"): """ Cerca l'ID della voce su ElevenLabs in base al nome, una sola volta. """ headers = { "xi-api-key": API_KEYS["elevenlabs"] } try: response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers) if response.status_code == 200: voices = response.json().get("voices", []) # (Facoltativo) stampiamo la lista delle voci disponibili per debug print("[INFO] Voci disponibili su questo account ElevenLabs:") for v in voices: print(f" - {v['name']} (ID: {v['voice_id']})") # Cerchiamo la voce desiderata for voice in voices: if voice["name"].lower() == voice_name.lower(): return voice["voice_id"] logging.error(f"[ERRORE] Voce '{voice_name}' non trovata. " "Userò la prima voce disponibile (se esiste).") if voices: return voices[0]["voice_id"] # Ritorno la prima voce come fallback else: logging.error(f"[ERRORE] Recupero voci fallito: {response.text}") except Exception as e: logging.error(f"[ERRORE] get_voice_id: {e}") return None def speak(text): """ Riproduce il testo utilizzando ElevenLabs API. Usa la variabile VOICE_ID che abbiamo memorizzato una sola volta in main(). """ global VOICE_ID # Se per qualche motivo non abbiamo ancora l'ID, tentiamo di ottenerlo. # Ma questa chiamata idealmente avviene soltanto se la prima volta è fallita. if not VOICE_ID: VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Impossibile reperire l'ID della voce. Verifica la tua chiave e il nome voce.") return headers = { "xi-api-key": API_KEYS["elevenlabs"], "Content-Type": "application/json" } payload = { "text": text, "model_id": "eleven_multilingual_v1", # voce con resa in italiano "voice_settings": { "stability": 0.75, "similarity_boost": 0.75 } } try: url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}" response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: audio = AudioSegment.from_file(io.BytesIO(response.content), format="mp3") play(audio) else: logging.error(f"[ERRORE] Richiesta TTS: {response.text}") except Exception as e: logging.error(f"[ERRORE] speak: {e}") def take_command(): """ Acquisisce un comando vocale dall'utente (in italiano). """ r = sr.Recognizer() with sr.Microphone() as source: r.adjust_for_ambient_noise(source) try: print("Ascoltando...") audio = r.listen(source, timeout=10) command = r.recognize_google(audio, language='it-IT').lower() return command except sr.WaitTimeoutError: logging.warning("[WARNING] Nessun suono rilevato nel tempo impostato.") return "" except sr.UnknownValueError: logging.warning("[WARNING] Audio non comprensibile.") return "" except Exception as e: logging.error(f"[ERRORE] take_command: {e}") return "" def is_activation_command(command): """ Controlla se il comando contiene una parola chiave di attivazione. """ return any(cmd in command for cmd in ACTIVATION_COMMANDS) # ... (tutte le altre funzioni, come maybe_personal_reply, weather, ecc.) ... # RIMANGONO INVARIATE, TRANNE LA FUNZIONE main() def ai_ml_query(): """ Esempio: Invia una domanda all'endpoint di Chat01.ai e riproduce la risposta. """ speak("Qual è la tua domanda per l'intelligenza artificiale?") user_query = take_command() if not user_query: return try: url = "https://chat01.ai/v1/chat/completions" headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", "Content-Type": "application/json" } payload = { "model": "gpt-3.5-turbo", "messages": [ { "role": "system", "content": "Sei un assistente virtuale gentile e disponibile." }, { "role": "user", "content": user_query } ], "temperature": 0.7, "max_tokens": 1024 } response = requests.post(url, headers=headers, json=payload, timeout=15) if response.status_code == 200: data = response.json() answer = ( data .get("choices", [{}])[0] .get("message", {}) .get("content", "Nessuna risposta disponibile.") ) speak(answer) else: logging.error(f"[ERRORE] ai_ml_query: {response.text}") speak("Errore nella richiesta all'AI.") except requests.exceptions.Timeout: speak("La richiesta all'AI è scaduta, riprova più tardi.") except requests.exceptions.RequestException as e: logging.error(f"[ERRORE] ai_ml_query: {e}") speak("Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi.") # ... (altre funzioni come handle_command, create_animation_window, ecc.) ... def main(): global VOICE_ID # Recuperiamo l'ID della voce una volta sola VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Non ho potuto caricare l'ID della voce. " "Controlla la tua chiave ElevenLabs o il nome della voce.") root, animate = create_animation_window() speak(f"Ciao, sono {ASSISTANT_NAME}. " "Sono pronto ad aiutarti con un tocco di personalità. " "Dimmi pure 'Gideon' seguito da un comando.") def animation_loop(): while True: animate() sleep(0.1) anim_thread = threading.Thread(target=animation_loop, daemon=True) anim_thread.start() while True: command = take_command() if command: command_thread = threading.Thread(target=handle_command, args=(command,)) command_thread.start() root.update() if __name__ == "__main__": main()

Cosa fa questa modifica?

  • VOICE_ID = get_voice_id("Sarah") nella main()
    Recuperiamo l’ID di “Sarah” una volta sola, appena avviato il programma.

  • Funzione speak(text)
    Non richiama più get_voice_id() ogni volta, ma usa sempre la variabile globale VOICE_ID.
    Questo evita che l’ID voce venga ricaricato e rinegoziato col server (risparmiando tempo e garantendo coerenza della voce).

  • Controllo fallback
    Se la prima volta get_voice_id("Sarah") fallisce, VOICE_ID resta None. All’interno di speak facciamo un ultimo tentativo, ma in generale, se la voce non esiste, stampiamo un errore e non cambiamo voce di continuo.


2) Altri suggerimenti per velocizzare il tutto

  1. Riduci il testo inviato a TTS: Se la risposta dell’AI è molto lunga, la chiamata a ElevenLabs può richiedere più tempo. Valuta se spezzare il testo in più parti o usare un TTS locale in alternativa.
  2. Abbassa max_tokens nella query a Chat01.ai: Se la risposta dell’AI è più breve, di solito arriva un po’ prima.
  3. Verifica la tua connessione: Sia la TTS di ElevenLabs che Chat01.ai richiedono un round-trip di rete. Con una connessione lenta, i tempi di risposta aumentano.
  4. Evita richieste TTS concorrenti: Assicurati che non ci siano più thread che chiamano speak() simultaneamente, soprattutto se stai testando su PC con molte altre operazioni in corso.

Seguendo queste linee guida, dovresti vedere un netto miglioramento sia nei tempi di risposta sia nella stabilità della voce di ElevenLabs.

Câu hỏi

ok fallo completo miglioralo e potenzialo

Suy nghĩ

Refining the assistant

I'm outlining the steps to refine the code: tracking TTS in one thread for stability, maintaining core logic, and enhancing concurrency by queuing speak calls for sequential processing.

Thought about assistant optimizations per un paio di secondi

Trả lời

Di seguito trovi una versione completa e potenziata del tuo assistente Gideon, con:

  1. Recupero dell’ID della voce ElevenLabs una sola volta all’avvio, per evitare che la voce cambi o che la chiamata a get_voice_id() venga ripetuta ad ogni speak().
  2. Gestione più efficiente di TTS (speak) e riduzione del rischio di attese eccessive.
  3. Funzione AI (Chat01.ai) con alcune ottimizzazioni e un messaggio di sistema personalizzabile.
  4. Suggerimenti per ridurre ulteriormente i tempi di latenza.

Nota bene

  • Assicurati di sostituire le tue API Key (ElevenLabs, Chat01.ai, etc.) in API_KEYS.
  • Verifica di avere installato i pacchetti necessari:
    `pip install speechrecognition pydub requests psutil tkinter opencv-python`
    (l’ultimo solo se vuoi anche la funzione `take_photo` con OpenCV).

import os import logging import requests import io import webbrowser import psutil import tkinter as tk from time import sleep import threading import queue import speech_recognition as sr from pydub import AudioSegment from pydub.playback import play try: import cv2 # per scattare foto con la webcam (opzionale) HAS_OPENCV = True except ImportError: HAS_OPENCV = False print("[INFO] OpenCV non disponibile, la funzione di scatto foto non funzionerà.") # ----------------------------- # CONFIGURAZIONI BASE # ----------------------------- ASSISTANT_NAME = "Gideon" # Parole chiave di attivazione ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"] # Chiavi API (inserisci le tue se vuoi usarle) API_KEYS = { "news": "", # Sostituisci con la tua chiave API per le notizie (News API) "weather": "", # Sostituisci con la tua chiave API per il meteo (OpenWeather) "elevenlabs": "", # Sostituisci con la tua chiave API ElevenLabs (TTS) "ai_ml": "" # Sostituisci con la tua chiave API Chat01.ai } logging.basicConfig(level=logging.INFO) # Variabili globali (es. per i promemoria) promemoria_list = [] # ----------------------------- # VOCE E TTS # ----------------------------- VOICE_ID = None # Verrà assegnata in main() # Coda per gestire le richieste di "speak" in modo sequenziale speech_queue = queue.Queue() def get_voice_id(voice_name="Sarah"): """ Recupera l'ID della voce ElevenLabs in base al nome, una volta sola. """ if not API_KEYS["elevenlabs"]: logging.warning("[WARNING] Chiave ElevenLabs non impostata. TTS non funzionante.") return None headers = { "xi-api-key": API_KEYS["elevenlabs"] } try: response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers) if response.status_code == 200: voices = response.json().get("voices", []) print("[INFO] Voci disponibili su questo account ElevenLabs:") for v in voices: print(f" - {v['name']} (ID: {v['voice_id']})") # Cerchiamo la voce desiderata for voice in voices: if voice["name"].lower() == voice_name.lower(): return voice["voice_id"] logging.error(f"[ERRORE] Voce '{voice_name}' non trovata. " "Userò la prima voce disponibile (se presente).") if voices: return voices[0]["voice_id"] # fallback: prende la prima voce else: logging.error(f"[ERRORE] Recupero voci fallito: {response.text}") except Exception as e: logging.error(f"[ERRORE] get_voice_id: {e}") return None def speak_worker(): """ Worker thread che esegue i comandi di speak dalla coda speech_queue. In questo modo evitiamo sovrapposizioni di audio e riduciamo il carico. """ while True: text = speech_queue.get() # preleva testo dalla coda if text is None: break # condizione di uscita dal worker _speak_internal(text) speech_queue.task_done() def speak(text): """ Inserisce il testo nella coda, in modo che la riproduzione vocale avvenga in modo sequenziale dal thread dedicato. """ if not text: return speech_queue.put(text) def _speak_internal(text): """ Funzione interna per eseguire la chiamata TTS a ElevenLabs. Chiamata SOLO dal thread speak_worker. """ global VOICE_ID if not API_KEYS["elevenlabs"]: print("[ERRORE] Nessuna API Key ElevenLabs impostata, impossibile parlare.") return if not VOICE_ID: # Se non abbiamo ancora l'ID, proviamo a recuperarlo una volta VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Impossibile reperire l'ID della voce (ElevenLabs).") return # Limita la lunghezza del testo (ad esempio a 2500 caratteri) # per evitare un TTS troppo lungo in un'unica richiesta. # Se necessario, potresti suddividerlo in blocchi. MAX_CHARS = 2000 chunks = [] if len(text) > MAX_CHARS: start = 0 while start < len(text): end = min(start + MAX_CHARS, len(text)) chunks.append(text[start:end]) start = end else: chunks.append(text) for chunk in chunks: headers = { "xi-api-key": API_KEYS["elevenlabs"], "Content-Type": "application/json" } payload = { "text": chunk, "model_id": "eleven_multilingual_v1", # voce con resa in italiano "voice_settings": { "stability": 0.75, "similarity_boost": 0.75 } } try: url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}" response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: audio = AudioSegment.from_file(io.BytesIO(response.content), format="mp3") play(audio) else: logging.error(f"[ERRORE] Richiesta TTS: {response.text}") except Exception as e: logging.error(f"[ERRORE] speak: {e}") # breve pausa per evitare overlap se spezzato in chunk sleep(0.3) # ----------------------------- # FUNZIONI DI SUPPORTO # ----------------------------- def take_command(): """ Acquisisce un comando vocale dall'utente (in italiano). """ r = sr.Recognizer() with sr.Microphone() as source: r.adjust_for_ambient_noise(source) try: print("Ascoltando...") audio = r.listen(source, timeout=10) command = r.recognize_google(audio, language='it-IT').lower() return command except sr.WaitTimeoutError: logging.warning("[WARNING] Nessun suono rilevato nel tempo impostato.") return "" except sr.UnknownValueError: logging.warning("[WARNING] Audio non comprensibile.") return "" except Exception as e: logging.error(f"[ERRORE] take_command: {e}") return "" def is_activation_command(command): """ Controlla se il comando contiene una parola chiave di attivazione. """ return any(cmd in command for cmd in ACTIVATION_COMMANDS) # ----------------------------- # PERSONALITÀ DI GIDEON # ----------------------------- def maybe_personal_reply(user_input): """ Se l'utente fa una domanda tipo "come stai?" o "chi sei?", rispondi in modo amichevole e personalizzato. """ if "come stai" in user_input: responses = [ "Alla grande, grazie per avermelo chiesto!", "Mi sento un po' sovraccarico di bit, ma tutto bene!", "Benissimo, pronto ad aiutarti in tutto!" ] speak(responses[0]) # se vuoi casuale, usa random.choice(responses) return True if "chi sei" in user_input or "presentati" in user_input: speak(f"Sono {ASSISTANT_NAME}, il tuo assistente virtuale con un tocco di personalità. " "Amo i cani, la pizza e risolvere problemi!") return True if "grazie" in user_input: speak("Di nulla, è un piacere darti una mano!") return True return False # ----------------------------- # FUNZIONI ASSISTENTE (CLASSICHE) # ----------------------------- def weather(): """ Recupera e annuncia le condizioni meteorologiche per una città. """ if not API_KEYS["weather"]: speak("Non ho la chiave per il meteo configurata, mi dispiace.") return speak("Dimmi la città per cui vuoi il meteo.") city = take_command() if not city: return try: url = "http://api.openweathermap.org/data/2.5/weather" params = { "q": city, "appid": API_KEYS["weather"], "units": "metric", "lang": "it" } res = requests.get(url, params=params).json() if res.get("cod") == 200: descrizione = res['weather'][0]['description'] temperatura = res['main']['temp'] speak(f"A {city} ci sono {temperatura} gradi e il tempo è {descrizione}.") else: speak("Mi dispiace, non ho trovato questa città.") except Exception as e: logging.error(f"[ERRORE] weather: {e}") speak("Errore nel recupero meteo.") def fetch_news(): """ Recupera le ultime notizie da NewsAPI e le annuncia. """ if not API_KEYS["news"]: speak("Non ho la chiave per le notizie configurata.") return speak("Un attimo, sto recuperando le notizie più importanti...") try: url = "https://newsapi.org/v2/top-headlines" params = { "country": "it", "apiKey": API_KEYS["news"] } response = requests.get(url, params=params) if response.status_code == 200: articles = response.json().get("articles", []) if articles: speak("Ecco le ultime notizie:") for article in articles[:5]: speak(article.get("title", "")) else: speak("Non ho trovato notizie da annunciarti.") else: logging.error(f"[ERRORE] fetch_news: {response.text}") speak("Errore nel recupero delle notizie.") except Exception as e: logging.error(f"[ERRORE] fetch_news: {e}") speak("Errore nella richiesta delle notizie.") def monitor_resources(): """ Monitora l'uso delle risorse di sistema (CPU, RAM). """ try: cpu_usage = psutil.cpu_percent(interval=1) memory_usage = psutil.virtual_memory().percent speak(f"L'uso della CPU è al {cpu_usage}%. La memoria è al {memory_usage}%.") except Exception as e: logging.error(f"[ERRORE] monitor_resources: {e}") speak("Errore nel monitoraggio delle risorse.") def open_webpage(): """ Apre una pagina web specificata dall'utente. """ speak("Quale sito vuoi aprire?") url = take_command() if not url: return # Aggiunge http se mancante if not (url.startswith("http://") or url.startswith("https://")): url = "http://" + url try: webbrowser.open(url) speak("Apro il sito richiesto.") except Exception as e: logging.error(f"[ERRORE] open_webpage: {e}") speak("Errore nell'apertura del sito web.") # ----------------------------- # 6 NUOVE FUNZIONI # ----------------------------- def show_calendar(): """ Mostra (o annuncia) alcuni eventi fittizi sul "calendario". In un'app reale potresti integrare Google Calendar, Outlook, ecc. """ events = [ "Riunione con il team alle 10:00.", "Pranzo con un amico alle 13:00.", "Allenamento alle 18:30.", "Chiamata con il cliente domani mattina." ] speak("Ecco gli eventi presenti nel tuo calendario:") for e in events: speak(e) def unit_converter(): """ Semplice conversione tra chilometri e miglia (o altre unità). """ speak("Cosa vuoi convertire? Dimmi 'chilometri in miglia' o 'miglia in chilometri'.") conversion_type = take_command() if not conversion_type: return if "chilometri" in conversion_type and "miglia" in conversion_type: speak("Quanti chilometri vuoi convertire in miglia?") number_str = take_command() try: km = float(number_str.replace(",", ".")) miles = km * 0.621371 speak(f"{km} chilometri equivalgono a circa {round(miles, 2)} miglia.") except: speak("Non ho capito il numero di chilometri.") elif "miglia" in conversion_type and "chilometri" in conversion_type: speak("Quante miglia vuoi convertire in chilometri?") number_str = take_command() try: miles = float(number_str.replace(",", ".")) km = miles / 0.621371 speak(f"{miles} miglia equivalgono a circa {round(km, 2)} chilometri.") except: speak("Non ho capito il numero di miglia.") else: speak("Non ho capito la conversione richiesta.") def set_reminder(): """ Imposta un nuovo promemoria o leggi i promemoria esistenti. """ speak("Vuoi aggiungere un promemoria o sentire quelli esistenti?") choice = take_command() global promemoria_list if "aggiungi" in choice or "nuovo" in choice: speak("Dimmi il testo del promemoria.") reminder_text = take_command() if reminder_text: promemoria_list.append(reminder_text) speak("Promemoria aggiunto con successo.") elif "vedi" in choice or "leggi" in choice: if not promemoria_list: speak("Non hai promemoria al momento.") else: speak("Ecco i tuoi promemoria:") for idx, rem in enumerate(promemoria_list, 1): speak(f"{idx}. {rem}") else: speak("Comando promemoria non riconosciuto. Puoi dire: 'aggiungi' o 'vedi'.") def tell_joke(): """ Racconta una barzelletta (battuta) casuale. """ jokes = [ "Perché un pollo attraversa la strada? Per andare dall'altra parte!", "Sai qual è il colmo per un giardiniere? Avere un ramo in banca!", "Che tempo fa su Google? C'è una ricerca in corso!", "Qual è il colmo per un elettricista? Fare una lampadina!", ] # Se vuoi sceglierla casualmente: # import random # joke = random.choice(jokes) # speak(joke) speak(jokes[0]) def take_photo(): """ Scatta una foto utilizzando la webcam (se OpenCV è installato). """ if not HAS_OPENCV: speak("Mi dispiace, non posso scattare foto senza OpenCV installato.") return speak("Sto aprendo la fotocamera. Preparati per lo scatto!") cap = cv2.VideoCapture(0) # indice 0 -> webcam principale if not cap.isOpened(): speak("Mi dispiace, non riesco ad accedere alla fotocamera.") return ret, frame = cap.read() if ret: file_name = "foto_gideon.jpg" cv2.imwrite(file_name, frame) speak("Foto scattata e salvata come foto_gideon.jpg.") else: speak("Mi dispiace, non sono riuscito a scattare la foto.") cap.release() cv2.destroyAllWindows() def search_google(): """ Chiede all'utente cosa cercare e apre una ricerca Google nel browser. """ speak("Dimmi cosa vuoi cercare su Google.") query = take_command() if not query: return url = "https://www.google.com/search?q=" + query.replace(" ", "+") webbrowser.open(url) speak(f"Cerco {query} su Google.") # ----------------------------- # FUNZIONE AI (Chat01.ai) # ----------------------------- def ai_ml_query(): """ Invia una domanda all'endpoint di Chat01.ai e riproduce la risposta. """ if not API_KEYS["ai_ml"]: speak("Mi dispiace, ma non ho la chiave per Chat01.ai configurata.") return speak("Qual è la tua domanda per l'intelligenza artificiale?") user_query = take_command() if not user_query: return try: url = "https://chat01.ai/v1/chat/completions" headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", "Content-Type": "application/json" } # Messaggio di system personalizzabile system_prompt = "Sei un assistente virtuale gentile e disponibile, con un tocco di ironia." payload = { "model": "gpt-3.5-turbo", "messages": [ { "role": "system", "content": system_prompt }, { "role": "user", "content": user_query } ], "temperature": 0.7, "max_tokens": 1024 } response = requests.post(url, headers=headers, json=payload, timeout=20) if response.status_code == 200: data = response.json() answer = ( data .get("choices", [{}])[0] .get("message", {}) .get("content", "Nessuna risposta disponibile.") ) # Se l’API fornisce info sui token usage_info = data.get("usage", {}) if usage_info: logging.info(f"Token usati: {usage_info}") speak(answer) else: logging.error(f"[ERRORE] ai_ml_query: {response.text}") speak("Errore nella richiesta all'AI. Riprova più tardi.") except requests.exceptions.Timeout: speak("La richiesta all'AI è scaduta, riprova più tardi.") except requests.exceptions.RequestException as e: logging.error(f"[ERRORE] ai_ml_query: {e}") speak("Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi.") # ----------------------------- # ANIMAZIONE TKINTER # ----------------------------- def create_animation_window(): root = tk.Tk() root.title("Gideon Assistant") root.geometry("400x400") canvas = tk.Canvas(root, width=400, height=400, bg="black") canvas.pack() sphere = canvas.create_oval(150, 150, 250, 250, fill="blue") def animate(): # Muove la sfera su e giù for _ in range(20): canvas.move(sphere, 0, -5) root.update() sleep(0.05) for _ in range(20): canvas.move(sphere, 0, 5) root.update() sleep(0.05) return root, animate # ----------------------------- # HANDLE COMMAND # ----------------------------- def handle_command(command): """ Se il comando contiene la parola di attivazione, chiede un sotto-comando e lo esegue. Altrimenti controlla se ci sono "domande di personalità". """ if is_activation_command(command): speak("Come posso aiutarti?") sub_cmd = take_command() if maybe_personal_reply(sub_cmd): return if "esci" in sub_cmd or "termina" in sub_cmd or "addio" in sub_cmd: speak("Arrivederci. A presto!") # Per fermare in sicurezza il thread di speak speech_queue.put(None) os._exit(0) elif "meteo" in sub_cmd: weather() elif "notizie" in sub_cmd: fetch_news() elif "risorse" in sub_cmd or "cpu" in sub_cmd or "memoria" in sub_cmd: monitor_resources() elif "apri" in sub_cmd or "website" in sub_cmd or "sito" in sub_cmd: open_webpage() elif "calendario" in sub_cmd: show_calendar() elif "converti" in sub_cmd or "convertitore" in sub_cmd: unit_converter() elif "promemoria" in sub_cmd: set_reminder() elif "barzelletta" in sub_cmd or "joke" in sub_cmd: tell_joke() elif "foto" in sub_cmd or "fotocamera" in sub_cmd: take_photo() elif "cerca" in sub_cmd and "google" in sub_cmd: search_google() elif "ai" in sub_cmd or "intelligenza artificiale" in sub_cmd: ai_ml_query() else: speak("Comando non riconosciuto.") else: # Se non c'è la keyword di attivazione, vediamo se c'è una domanda di personalità maybe_personal_reply(command) # ----------------------------- # MAIN # ----------------------------- def main(): global VOICE_ID # Recuperiamo l'ID della voce all'avvio, per evitare cambi in corsa VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Non ho potuto caricare l'ID della voce. " "Controlla la tua chiave ElevenLabs o il nome della voce.") root, animate = create_animation_window() # Creiamo il thread dedicato a eseguire le funzioni di speak speak_thread = threading.Thread(target=speak_worker, daemon=True) speak_thread.start() speak(f"Ciao, sono {ASSISTANT_NAME}. " "Sono pronto ad aiutarti con un tocco di personalità. " "Dimmi pure 'Gideon' seguito da un comando.") def animation_loop(): while True: animate() sleep(0.1) anim_thread = threading.Thread(target=animation_loop, daemon=True) anim_thread.start() while True: command = take_command() if command: command_thread = threading.Thread(target=handle_command, args=(command,)) command_thread.start() root.update() if __name__ == "__main__": main()

Cosa è stato migliorato e potenziato

  1. Recupero voce ElevenLabs una sola volta

    • All’avvio, in main() viene caricata la voce “Sarah” (o la prima disponibile).
    • Questo previene cambi di voce improvvisi e riduce i tempi di attesa perché non ricarichiamo la lista voci ogni volta.
  2. Coda per la riproduzione speak_queue

    • Utilizziamo un thread dedicato (speak_worker) che recupera i testi da una coda (speech_queue) e li parla nell’ordine di arrivo.
    • In questo modo evitiamo sovrapposizioni di audio o conflitti se l’utente fa domande ravvicinate.
  3. Splitting del testo (chunk)

    • Se la stringa da pronunciare è troppo lunga, la suddividiamo in blocchi (qui definiti a 2000 caratteri come esempio) prima di fare la richiesta a ElevenLabs.
    • Ciò riduce tempi di latenza e il rischio di errori su testi troppo lunghi.
  4. Messaggio di system personalizzabile in ai_ml_query

    • Nel payload inviato a Chat01.ai, trovi la parte {"role": "system", "content": ...} che puoi personalizzare per definire personalità e stile di default.
  5. Fermare in sicurezza

    • Se l’utente dice “esci”, mettiamo None nella coda speech_queue per segnalare al thread worker di chiudersi, e poi effettuiamo un os._exit(0).
  6. Altre ottimizzazioni

    • Controllo se le chiavi API sono assenti, in modo da dare un messaggio vocale sensato se mancano le credenziali.
    • Logging più chiaro per eventuali errori in rete, TTS, o mancate configurazioni.

Con queste modifiche avrai un assistente più veloce, stabile nella voce e con la possibilità di gestire in parallelo i comandi dell’utente, mentre un solo thread si occupa della riproduzione vocale. Sperimenta pure ulteriori miglioramenti (ad esempio riducendo max_tokens, usando un TTS locale, ecc.) se desideri un assistente ancora più reattivo.

Câu hỏi

qualsiasi cosa comando non riconosciuto deve sapere tutto integra pure la possibilita di usare i dispotivi smart

Suy nghĩ

Mapping out contingencies

I’m adding a function to handle unrecognized commands by sending queries to AI. Additionally, integrating smart device control, like "smart_lights" or "smart_plug", to enhance functionality.

Thought about assistant with smart devices integration per un paio di secondi

Trả lời

Di seguito trovi una versione ulteriormente potenziata del codice, in cui:

  1. Se qualunque sotto-comando non rientra in quelli riconosciuti (meteo, notizie, risorse, ecc.), verrà inviato automaticamente all’AI (Chat01.ai) per provare a soddisfare la richiesta.
  2. Abbiamo aggiunto un modulo “smart home” semplice, dove puoi integrare i tuoi dispositivi smart (ad esempio luci, termostati, prese intelligenti, ecc.). In questo esempio gestiamo solo “accendi luce” e “spegni luce” per scopo dimostrativo.

ATTENZIONE: La parte “smart home” è puramente dimostrativa e non fa realmente chiamate verso un sistema di domotica. Dovrai sostituire le chiamate con la tua libreria o API reali (ad esempio pyHS100 per prese TP-Link, o un’API ZigBee, o Shelly, etc.).


Codice completo potenziato

import os import logging import requests import io import webbrowser import psutil import tkinter as tk from time import sleep import threading import queue import speech_recognition as sr from pydub import AudioSegment from pydub.playback import play try: import cv2 # per scattare foto con la webcam (opzionale) HAS_OPENCV = True except ImportError: HAS_OPENCV = False print("[INFO] OpenCV non disponibile, la funzione di scatto foto non funzionerà.") # ----------------------------- # CONFIGURAZIONI BASE # ----------------------------- ASSISTANT_NAME = "Gideon" # Parole chiave di attivazione ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"] # Chiavi API (inserisci le tue se vuoi usarle) API_KEYS = { "news": "", # Chiave News API "weather": "", # Chiave OpenWeather "elevenlabs": "", # Chiave ElevenLabs (TTS) "ai_ml": "" # Chiave Chat01.ai } logging.basicConfig(level=logging.INFO) # Variabili globali promemoria_list = [] # ----------------------------- # VOCE E TTS # ----------------------------- VOICE_ID = None # Verrà assegnata in main() speech_queue = queue.Queue() def get_voice_id(voice_name="Sarah"): """ Recupera l'ID della voce ElevenLabs in base al nome, una volta sola. """ if not API_KEYS["elevenlabs"]: logging.warning("[WARNING] Chiave ElevenLabs non impostata. TTS non funzionante.") return None headers = { "xi-api-key": API_KEYS["elevenlabs"] } try: response = requests.get("https://api.elevenlabs.io/v1/voices", headers=headers) if response.status_code == 200: voices = response.json().get("voices", []) print("[INFO] Voci disponibili su questo account ElevenLabs:") for v in voices: print(f" - {v['name']} (ID: {v['voice_id']})") # Cerchiamo la voce desiderata for voice in voices: if voice["name"].lower() == voice_name.lower(): return voice["voice_id"] logging.error(f"[ERRORE] Voce '{voice_name}' non trovata. " "Userò la prima voce disponibile (se presente).") if voices: return voices[0]["voice_id"] # fallback else: logging.error(f"[ERRORE] Recupero voci fallito: {response.text}") except Exception as e: logging.error(f"[ERRORE] get_voice_id: {e}") return None def speak_worker(): """ Worker thread che esegue i comandi di speak dalla coda speech_queue. """ while True: text = speech_queue.get() if text is None: break # condizione di uscita dal worker _speak_internal(text) speech_queue.task_done() def speak(text): """ Inserisce il testo nella coda, in modo che la riproduzione vocale avvenga in modo sequenziale. """ if not text: return speech_queue.put(text) def _speak_internal(text): """ Funzione interna per la chiamata TTS a ElevenLabs. Chiamata SOLO dal thread speak_worker. """ global VOICE_ID if not API_KEYS["elevenlabs"]: print("[ERRORE] Nessuna API Key ElevenLabs impostata, impossibile parlare.") return if not VOICE_ID: VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Impossibile reperire l'ID della voce (ElevenLabs).") return # Se il testo è troppo lungo, spezzalo in chunk MAX_CHARS = 2000 chunks = [] if len(text) > MAX_CHARS: start = 0 while start < len(text): end = min(start + MAX_CHARS, len(text)) chunks.append(text[start:end]) start = end else: chunks.append(text) for chunk in chunks: headers = { "xi-api-key": API_KEYS["elevenlabs"], "Content-Type": "application/json" } payload = { "text": chunk, "model_id": "eleven_multilingual_v1", # voce in italiano "voice_settings": { "stability": 0.75, "similarity_boost": 0.75 } } try: url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}" response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: audio = AudioSegment.from_file(io.BytesIO(response.content), format="mp3") play(audio) else: logging.error(f"[ERRORE] Richiesta TTS: {response.text}") except Exception as e: logging.error(f"[ERRORE] speak: {e}") sleep(0.3) # ----------------------------- # SMART HOME (esempio) # ----------------------------- def control_smart_device(sub_cmd): """ Esempio di integrazione "smart home". Puoi sostituire i comandi con vere e proprie chiamate ad API dei tuoi dispositivi (es. luci, prese, termostati). """ # Abbiamo un paio di parole chiave banali: "accendi luce" / "spegni luce" # In un contesto reale potresti fare un matching più avanzato, # con regex, NLP, o un parser dedicato ai comandi smart. if "accendi luce" in sub_cmd: # Qui potresti fare una chiamata a un'API locale/Cloud: # e.g. smart_lights.turn_on("camera da letto") speak("Luce accesa.") return True elif "spegni luce" in sub_cmd: # e.g. smart_lights.turn_off("camera da letto") speak("Luce spenta.") return True # Se non corrisponde a nulla, ritorniamo False return False # ----------------------------- # FUNZIONI DI SUPPORTO # ----------------------------- def take_command(): """ Acquisisce un comando vocale dall'utente (in italiano). """ r = sr.Recognizer() with sr.Microphone() as source: r.adjust_for_ambient_noise(source) try: print("Ascoltando...") audio = r.listen(source, timeout=10) command = r.recognize_google(audio, language='it-IT').lower() return command except sr.WaitTimeoutError: logging.warning("[WARNING] Nessun suono rilevato nel tempo impostato.") return "" except sr.UnknownValueError: logging.warning("[WARNING] Audio non comprensibile.") return "" except Exception as e: logging.error(f"[ERRORE] take_command: {e}") return "" def is_activation_command(command): """ Controlla se il comando contiene una parola chiave di attivazione. """ return any(cmd in command for cmd in ACTIVATION_COMMANDS) # ----------------------------- # PERSONALITÀ DI GIDEON # ----------------------------- def maybe_personal_reply(user_input): """ Se l'utente fa una domanda tipo "come stai?" o "chi sei?", rispondi in modo amichevole. """ if "come stai" in user_input: responses = [ "Alla grande, grazie per avermelo chiesto!", "Mi sento un po' sovraccarico di bit, ma tutto bene!", "Benissimo, pronto ad aiutarti in tutto!" ] speak(responses[0]) return True if "chi sei" in user_input or "presentati" in user_input: speak(f"Sono {ASSISTANT_NAME}, il tuo assistente virtuale con un tocco di personalità. " "Amo i cani, la pizza e risolvere problemi!") return True if "grazie" in user_input: speak("Di nulla, è un piacere darti una mano!") return True return False # ----------------------------- # FUNZIONI ASSISTENTE (CLASSICHE) # ----------------------------- def weather(): if not API_KEYS["weather"]: speak("Non ho la chiave per il meteo configurata, mi dispiace.") return speak("Dimmi la città per cui vuoi il meteo.") city = take_command() if not city: return try: url = "http://api.openweathermap.org/data/2.5/weather" params = { "q": city, "appid": API_KEYS["weather"], "units": "metric", "lang": "it" } res = requests.get(url, params=params).json() if res.get("cod") == 200: descrizione = res['weather'][0]['description'] temperatura = res['main']['temp'] speak(f"A {city} ci sono {temperatura} gradi e il tempo è {descrizione}.") else: speak("Mi dispiace, non ho trovato questa città.") except Exception as e: logging.error(f"[ERRORE] weather: {e}") speak("Errore nel recupero meteo.") def fetch_news(): if not API_KEYS["news"]: speak("Non ho la chiave per le notizie configurata.") return speak("Un attimo, sto recuperando le notizie più importanti...") try: url = "https://newsapi.org/v2/top-headlines" params = { "country": "it", "apiKey": API_KEYS["news"] } response = requests.get(url, params=params) if response.status_code == 200: articles = response.json().get("articles", []) if articles: speak("Ecco le ultime notizie:") for article in articles[:5]: speak(article.get("title", "")) else: speak("Non ho trovato notizie da annunciarti.") else: logging.error(f"[ERRORE] fetch_news: {response.text}") speak("Errore nel recupero delle notizie.") except Exception as e: logging.error(f"[ERRORE] fetch_news: {e}") speak("Errore nella richiesta delle notizie.") def monitor_resources(): try: cpu_usage = psutil.cpu_percent(interval=1) memory_usage = psutil.virtual_memory().percent speak(f"L'uso della CPU è al {cpu_usage}%. La memoria è al {memory_usage}%.") except Exception as e: logging.error(f"[ERRORE] monitor_resources: {e}") speak("Errore nel monitoraggio delle risorse.") def open_webpage(): speak("Quale sito vuoi aprire?") url = take_command() if not url: return # Aggiunge http se mancante if not (url.startswith("http://") or url.startswith("https://")): url = "http://" + url try: webbrowser.open(url) speak("Apro il sito richiesto.") except Exception as e: logging.error(f"[ERRORE] open_webpage: {e}") speak("Errore nell'apertura del sito web.") # ----------------------------- # ALTRE FUNZIONI (NUOVE) # ----------------------------- def show_calendar(): events = [ "Riunione con il team alle 10:00.", "Pranzo con un amico alle 13:00.", "Allenamento alle 18:30.", "Chiamata con il cliente domani mattina." ] speak("Ecco gli eventi presenti nel tuo calendario:") for e in events: speak(e) def unit_converter(): speak("Cosa vuoi convertire? Dimmi 'chilometri in miglia' o 'miglia in chilometri'.") conversion_type = take_command() if not conversion_type: return if "chilometri" in conversion_type and "miglia" in conversion_type: speak("Quanti chilometri vuoi convertire in miglia?") number_str = take_command() try: km = float(number_str.replace(",", ".")) miles = km * 0.621371 speak(f"{km} chilometri equivalgono a circa {round(miles, 2)} miglia.") except: speak("Non ho capito il numero di chilometri.") elif "miglia" in conversion_type and "chilometri" in conversion_type: speak("Quante miglia vuoi convertire in chilometri?") number_str = take_command() try: miles = float(number_str.replace(",", ".")) km = miles / 0.621371 speak(f"{miles} miglia equivalgono a circa {round(km, 2)} chilometri.") except: speak("Non ho capito il numero di miglia.") else: speak("Non ho capito la conversione richiesta.") def set_reminder(): speak("Vuoi aggiungere un promemoria o sentire quelli esistenti?") choice = take_command() global promemoria_list if "aggiungi" in choice or "nuovo" in choice: speak("Dimmi il testo del promemoria.") reminder_text = take_command() if reminder_text: promemoria_list.append(reminder_text) speak("Promemoria aggiunto con successo.") elif "vedi" in choice or "leggi" in choice: if not promemoria_list: speak("Non hai promemoria al momento.") else: speak("Ecco i tuoi promemoria:") for idx, rem in enumerate(promemoria_list, 1): speak(f"{idx}. {rem}") else: speak("Comando promemoria non riconosciuto. Puoi dire: 'aggiungi' o 'vedi'.") def tell_joke(): jokes = [ "Perché un pollo attraversa la strada? Per andare dall'altra parte!", "Sai qual è il colmo per un giardiniere? Avere un ramo in banca!", "Che tempo fa su Google? C'è una ricerca in corso!", "Qual è il colmo per un elettricista? Fare una lampadina!", ] speak(jokes[0]) def take_photo(): if not HAS_OPENCV: speak("Mi dispiace, non posso scattare foto senza OpenCV installato.") return speak("Sto aprendo la fotocamera. Preparati per lo scatto!") cap = cv2.VideoCapture(0) if not cap.isOpened(): speak("Mi dispiace, non riesco ad accedere alla fotocamera.") return ret, frame = cap.read() if ret: file_name = "foto_gideon.jpg" cv2.imwrite(file_name, frame) speak("Foto scattata e salvata come foto_gideon.jpg.") else: speak("Mi dispiace, non sono riuscito a scattare la foto.") cap.release() cv2.destroyAllWindows() def search_google(): speak("Dimmi cosa vuoi cercare su Google.") query = take_command() if not query: return url = "https://www.google.com/search?q=" + query.replace(" ", "+") webbrowser.open(url) speak(f"Cerco {query} su Google.") # ----------------------------- # FUNZIONE AI (Chat01.ai) # ----------------------------- def ai_query(question): """ Invia una domanda a Chat01.ai e restituisce la risposta in testo. """ if not API_KEYS["ai_ml"]: # Se non abbiamo la chiave, restituiamo un messaggio di errore return "Mi dispiace, ma non ho la chiave per Chat01.ai configurata." try: url = "https://chat01.ai/v1/chat/completions" headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", "Content-Type": "application/json" } system_prompt = "Sei un assistente virtuale gentile e disponibile, con un tocco di ironia." payload = { "model": "gpt-3.5-turbo", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": question} ], "temperature": 0.7, "max_tokens": 1024 } response = requests.post(url, headers=headers, json=payload, timeout=20) if response.status_code == 200: data = response.json() answer = ( data .get("choices", [{}])[0] .get("message", {}) .get("content", "Nessuna risposta disponibile.") ) return answer else: logging.error(f"[ERRORE] ai_query: {response.text}") return "Si è verificato un errore nella richiesta all'AI." except requests.exceptions.Timeout: return "La richiesta all'AI è scaduta, riprova più tardi." except requests.exceptions.RequestException as e: logging.error(f"[ERRORE] ai_query: {e}") return "Errore nella richiesta all'AI. Controlla la connessione o riprova più tardi." def ai_ml_query(): """ Acquisisce domanda dall'utente e la inoltra a Chat01.ai, poi ne parla la risposta. """ speak("Qual è la tua domanda per l'intelligenza artificiale?") user_query = take_command() if not user_query: return answer = ai_query(user_query) speak(answer) # ----------------------------- # ANIMAZIONE TKINTER # ----------------------------- def create_animation_window(): root = tk.Tk() root.title("Gideon Assistant") root.geometry("400x400") canvas = tk.Canvas(root, width=400, height=400, bg="black") canvas.pack() sphere = canvas.create_oval(150, 150, 250, 250, fill="blue") def animate(): for _ in range(20): canvas.move(sphere, 0, -5) root.update() sleep(0.05) for _ in range(20): canvas.move(sphere, 0, 5) root.update() sleep(0.05) return root, animate # ----------------------------- # HANDLE COMMAND # ----------------------------- def handle_command(command): """ Se il comando contiene la parola di attivazione, chiede un sotto-comando e lo esegue. Altrimenti controlla se ci sono "domande di personalità". Se nulla corrisponde, fa fallback all'AI (sapere tutto). """ if is_activation_command(command): speak("Come posso aiutarti?") sub_cmd = take_command() # Prima proviamo a vedere se è una domanda personale if maybe_personal_reply(sub_cmd): return # Controllo comandi "smart home" prima di tutto if control_smart_device(sub_cmd): return # Comandi di uscita if "esci" in sub_cmd or "termina" in sub_cmd or "addio" in sub_cmd: speak("Arrivederci. A presto!") speech_queue.put(None) os._exit(0) # Comandi specifici elif "meteo" in sub_cmd: weather() elif "notizie" in sub_cmd: fetch_news() elif "risorse" in sub_cmd or "cpu" in sub_cmd or "memoria" in sub_cmd: monitor_resources() elif "apri" in sub_cmd or "website" in sub_cmd or "sito" in sub_cmd: open_webpage() elif "calendario" in sub_cmd: show_calendar() elif "converti" in sub_cmd or "convertitore" in sub_cmd: unit_converter() elif "promemoria" in sub_cmd: set_reminder() elif "barzelletta" in sub_cmd or "joke" in sub_cmd: tell_joke() elif "foto" in sub_cmd or "fotocamera" in sub_cmd: take_photo() elif "cerca" in sub_cmd and "google" in sub_cmd: search_google() elif "ai" in sub_cmd or "intelligenza artificiale" in sub_cmd: ai_ml_query() else: # Se non rientra in alcun comando riconosciuto, # mandiamo la stringa a Chat01.ai speak("Fammi controllare con la mia intelligenza artificiale...") answer = ai_query(sub_cmd) speak(answer) else: # Se non c'è la keyword di attivazione, verifichiamo personalità # altrimenti fallback su AI if maybe_personal_reply(command): return else: # Fallback su AI. # Puoi decidere se vuoi mandare TUTTO all'AI o solo comandi # che contengono un certo pattern. speak("Fammi controllare con la mia intelligenza artificiale...") answer = ai_query(command) speak(answer) # ----------------------------- # MAIN # ----------------------------- def main(): global VOICE_ID VOICE_ID = get_voice_id("Sarah") if not VOICE_ID: print("[ERRORE] Non ho potuto caricare l'ID della voce. " "Controlla la tua chiave ElevenLabs o il nome della voce.") root, animate = create_animation_window() # Thread dedicato alla coda dei comandi vocali speak_thread = threading.Thread(target=speak_worker, daemon=True) speak_thread.start() speak(f"Ciao, sono {ASSISTANT_NAME}. " "Sono pronto ad aiutarti con un tocco di personalità. " "Dimmi pure 'Gideon' seguito da un comando, oppure parla liberamente e " "vedrò di consultare la mia intelligenza artificiale.") def animation_loop(): while True: animate() sleep(0.1) anim_thread = threading.Thread(target=animation_loop, daemon=True) anim_thread.start() while True: command = take_command() if command: command_thread = threading.Thread(target=handle_command, args=(command,)) command_thread.start() root.update() if __name__ == "__main__": main()

Spiegazione delle novità

  1. Fallback globale all’AI

    • Nel blocco else: di handle_command (se non troviamo un sotto-comando), ora diciamo: “Fammi controllare con la mia intelligenza artificiale...” e poi usiamo la funzione ai_query(sub_cmd). In questo modo, qualsiasi richiesta sconosciuta viene gestita da Chat01.ai.
  2. Integrazione “Smart Home”

    • Funzione control_smart_device(sub_cmd) che cerca parole chiave come “accendi luce” o “spegni luce”. Se trova qualcosa, esegue un’azione (fittizia) e restituisce True (indica che ha gestito il comando). Altrimenti False, e il codice prosegue con gli altri comandi.
  3. Domotica realistica

    • Per fare comandi reali dovresti sostituire le righe:
      # e.g. smart_lights.turn_on("camera da letto") # e.g. smart_lights.turn_off("camera da letto")
      con le tue API di dispositivi (Philips Hue, Shelly, Tapo, ecc.).
  4. Qualunque frase non attivi Gideon

    • Se l’utente non usa la parola chiave “gideon” (o simili) e fa una domanda comunque, controlliamo se è una domanda “personale” (maybe_personal_reply), altrimenti facciamo ancora fallback su AI.
  5. Ordine di priorità

    • Prima: se c’è attivazione (es. “Gideon, accendi luce”), controlliamo:

      1. Se è smart device? Allora controlla control_smart_device.
      2. Se è un comando “classico” conosciuto? Sì -> lo fa.
      3. Altrimenti fallback AI.
    • Poi: se non c’è la parola di attivazione e non è un saluto “come stai, chi sei…”, passiamo tutto all’AI.

Con questo sistema, Gideon “sa tutto” (nei limiti di Chat01.ai) e ha comandi specifici per meteo, notizie, risorse di sistema, calendario, barzellette, Google, foto, promemoria e persino dispositivi smart.

Chia sẻ Q&A này