AI query integration
बनाया गया: 31 दिसंबर 2024
बनाया गया: 31 दिसंबर 2024
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
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.")
https://chat01.ai/v1/chat/completions
."Authorization": f"Bearer {API_KEYS['ai_ml']}"
per inviare la tua chiave API.messages
contenente il messaggio dell’utente e un messaggio di sistema opzionale.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.
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
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:
- Installare i pacchetti richiesti (
pip install speechrecognition pydub requests psutil tkinter opencv-python
ecc.).- 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()
Funzione ai_ml_query
https://chat01.ai/v1/chat/completions
.Authorization
con la chiave API presente in API_KEYS["ai_ml"]
.Uso dei token
usage
nella risposta (simile a {"usage": {"prompt_tokens": 10, ...}}
), lo stampiamo in console per tracciare quanti token sono stati usati.Messaggio di system
personalizzato
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.Gestione errori
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!
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
Per ridurre i tempi di attesa e impedire il cambio di voce continuo (quando usi ElevenLabs TTS), puoi seguire queste due strategie principali:
get_voice_id()
ad ogni speak()
.Di seguito un esempio di come puoi modificare il tuo codice per mettere in pratica questi accorgimenti.
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()
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.
max_tokens
nella query a Chat01.ai: Se la risposta dell’AI è più breve, di solito arriva un po’ prima.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.
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
Di seguito trovi una versione completa e potenziata del tuo assistente Gideon, con:
speak()
.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()
Recupero voce ElevenLabs una sola volta
main()
viene caricata la voce “Sarah” (o la prima disponibile).Coda per la riproduzione speak_queue
speak_worker
) che recupera i testi da una coda (speech_queue
) e li parla nell’ordine di arrivo.Splitting del testo (chunk)
Messaggio di system personalizzabile in ai_ml_query
{"role": "system", "content": ...}
che puoi personalizzare per definire personalità e stile di default.Fermare in sicurezza
None
nella coda speech_queue
per segnalare al thread worker di chiudersi, e poi effettuiamo un os._exit(0)
.Altre ottimizzazioni
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.
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
Di seguito trovi una versione ulteriormente potenziata del codice, in cui:
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.).
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()
Fallback globale all’AI
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.Integrazione “Smart Home”
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.Domotica realistica
con le tue API di dispositivi (Philips Hue, Shelly, Tapo, ecc.).# e.g. smart_lights.turn_on("camera da letto") # e.g. smart_lights.turn_off("camera da letto")
Qualunque frase non attivi Gideon
maybe_personal_reply
), altrimenti facciamo ancora fallback su AI
.Ordine di priorità
Prima: se c’è attivazione (es. “Gideon, accendi luce”), controlliamo:
control_smart_device
.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.