AI query integration
생성일: 2024년 12월 31일
생성일: 2024년 12월 31일
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à.")
ASSISTANT_NAME = "Gideon"
ACTIVATION_COMMANDS = ["gideon", "gideo", "gideoun", "gigget", "g", "gide", "giddy"]
API_KEYS = {
text"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)
promemoria_list = []
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']})")
text# 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"
textvoice_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)
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
textif "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
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
texttry: 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
texttry: 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)
textif 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
text# 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.")
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
textif "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
textif "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
textspeak("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.")
def create_animation_window():
root = tk.Tk()
root.title("Gideon Assistant")
root.geometry("400x400")
textcanvas = 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
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()
text# 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)
def main():
root, animate = create_animation_window()
text# 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
URL completo
https://chat01.ai/v1/chat/completions ho anche la chiave api
Ecco un esempio aggiornato della funzione ai_ml_query
che utilizza l'endpoint di chat01.ai invece di quello fittizio, con un payload in stile ChatGPT:
pythondef ai_ml_query(): """ Invia una domanda a chat01.ai 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://chat01.ai/v1/chat/completions" headers = { "Authorization": f"Bearer {API_KEYS['ai_ml']}", # Inserisci la tua chiave API nel dizionario "Content-Type": "application/json" } payload = { "model": "gpt-3.5-turbo", "messages": [ {"role": "user", "content": user_query} ], "temperature": 0.7 } response = requests.post(url, headers=headers, json=payload, timeout=15) if response.status_code == 200: data = response.json() # Estraggo la risposta dalla prima 'choice' disponibile answer = data["choices"][0]["message"]["content"] 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.")
Assicurati di:
API_KEYS["ai_ml"]
.model
(es. "gpt-3.5-turbo"
) in base ai modelli supportati dall’endpoint specifico (se differiscono).max_tokens
, top_p
, etc.) se richiesti o desiderati.