commit 49964c20a66ca5a255776e4fca5bd657cb672bff Author: Nick Date: Mon Oct 27 19:26:42 2025 +0100 base script diff --git a/README.md b/README.md new file mode 100644 index 0000000..040ce1b --- /dev/null +++ b/README.md @@ -0,0 +1,136 @@ +Certamente. Ecco un file `README.md` completo e ben formattato che puoi salvare nella stessa cartella del tuo script `yamaha_cli.py`. Questo file spiega a cosa serve lo script, come installarlo e come utilizzarlo, con esempi chiari. + +--- + +# Yamaha Mixer CLI Controller + +Uno script Python per creare un'interfaccia a riga di comando (CLI) per controllare i mixer digitali Yamaha della serie TF (e potenzialmente altri modelli compatibili) tramite connessione Ethernet. + +Questo strumento permette di eseguire operazioni comuni come il richiamo di scene, la regolazione dei fader, l'attivazione/disattivazione dei canali e il panning direttamente dal terminale, senza bisogno di interfacce grafiche o protocolli MIDI. + +## Caratteristiche + +- **Semplice e Intuitivo**: Comandi facili da ricordare (`recall`, `level`, `onoff`, `pan`). +- **Nessuna Dipendenza Esterna**: Utilizza solo le librerie standard di Python. +- **Configurabile**: Permette di specificare l'indirizzo IP e la porta del mixer. +- **Flessibile**: Include un comando `raw` per inviare qualsiasi stringa di comando supportata dal mixer. +- **Feedback Immediato**: Fornisce messaggi di conferma e di errore direttamente nel terminale. + +## Prerequisiti + +1. **Python**: Versione 2.7 o successiva (Python 3.x raccomandato). +2. **Mixer Yamaha**: Un mixer della serie TF (TF1, TF3, TF5, TF-RACK) o un altro modello compatibile con i comandi SCP. +3. **Connessione di Rete**: Il computer che esegue lo script e il mixer devono essere collegati alla stessa rete locale. + +## Configurazione Iniziale + +1. **Imposta l'IP del Mixer**: Assicurati che il tuo mixer Yamaha abbia un indirizzo IP statico. L'indirizzo predefinito utilizzato dallo script è `192.168.0.128`. +2. **Imposta l'IP del Computer**: Configura il tuo computer con un indirizzo IP statico nella stessa subnet del mixer (es. `192.168.0.100` con subnet mask `255.255.255.0`). +3. **Salva lo Script**: Scarica e salva il file `yamaha_cli.py` in una cartella sul tuo computer. +4. **Apri il Terminale**: Apri un terminale (Prompt dei comandi su Windows, Terminale su macOS/Linux) e naviga nella cartella in cui hai salvato lo script. + +## Utilizzo + +La sintassi generale per eseguire un comando è: +```bash +python yamaha_cli.py [opzioni] [argomenti] +``` + +### Ottenere Aiuto + +Per visualizzare la lista completa dei comandi e delle opzioni disponibili: +```bash +python yamaha_cli.py --help +``` + +Per ottenere aiuto su un comando specifico (ad esempio, `level`): +```bash +python yamaha_cli.py level --help +``` + +### Opzioni Globali + +- `--host`: Specifica un indirizzo IP diverso per il mixer. + - Esempio: `--host 192.168.1.50` +- `--port`: Specifica una porta diversa (lo standard è `49280`). + +--- + +### Comandi Disponibili + +#### **`recall`** +Richiama una scena da un banco specifico (A o B). + +- **Sintassi**: `python yamaha_cli.py recall ` +- **Esempio**: Richiama la scena #15 dal banco A. + ```bash + python yamaha_cli.py recall a 15 + ``` + +#### **`level`** +Imposta il livello del fader di un canale di input. + +- **Sintassi**: `python yamaha_cli.py level ` +- **Nota**: I canali sono indicizzati da 0 (Canale 1 = 0, Canale 2 = 1, ecc.). +- **Esempio**: Imposta il fader del Canale 1 (indice 0) a +10.00 dB (valore 1000). + ```bash + python yamaha_cli.py level 0 1000 + ``` + +#### **`onoff`** +Accende (`on`) o spegne (`off`) un canale di input. + +- **Sintassi**: `python yamaha_cli.py onoff ` +- **Esempio**: Spegne il Canale 8 (indice 7). + ```bash + python yamaha_cli.py onoff 7 off + ``` + +#### **`pan`** +Imposta il posizionamento stereo (pan) di un canale di input. + +- **Sintassi**: `python yamaha_cli.py pan ` +- **Valori**: `-63` (tutto a sinistra), `0` (centro), `63` (tutto a destra). +- **Esempio**: Imposta il pan del Canale 3 (indice 2) completamente a destra. + ```bash + python yamaha_cli.py pan 2 63 + ``` + +#### **`raw`** +Invia un comando grezzo non implementato direttamente dalla CLI. + +- **Sintassi**: `python yamaha_cli.py raw ` +- **Esempio**: Invia un comando per impostare un parametro specifico. + ```bash + python yamaha_cli.py raw set MIXER:Current/InCh/Fader/On 3 0 1 + ``` + +## Esempi Pratici + +**1. Richiamare la scena #1 del banco A**```bash +python yamaha_cli.py recall a 1 +``` + +**2. Alzare il fader del Canale 5 (indice 4) a 0 dB (valore 0)** +```bash +python yamaha_cli.py level 4 0 +``` + +**3. Accendere il Canale 12 (indice 11)** +```bash +python yamaha_cli.py onoff 11 on +``` + +**4. Eseguire un comando su un mixer con un IP non standard** +```bash +python yamaha_cli.py --host 192.168.1.128 recall b 42 +``` + +## Note Importanti + +- **Indice dei Canali**: Ricorda sempre che i canali sono **indicizzati a partire da 0**. Per controllare il Canale 1, devi usare l'indice `0`. +- **Compatibilità**: Sebbene testato con la serie TF, lo script potrebbe funzionare anche con mixer delle serie CL/QL, poiché condividono un set di comandi simile. + +## Licenza + +Questo progetto è rilasciato sotto la licenza MIT. Vedi il file `LICENSE` per maggiori dettagli. \ No newline at end of file diff --git a/__pycache__/mixer_controller.cpython-312.pyc b/__pycache__/mixer_controller.cpython-312.pyc new file mode 100644 index 0000000..826c57f Binary files /dev/null and b/__pycache__/mixer_controller.cpython-312.pyc differ diff --git a/__pycache__/yamaha_cli.cpython-312.pyc b/__pycache__/yamaha_cli.cpython-312.pyc new file mode 100644 index 0000000..da9b300 Binary files /dev/null and b/__pycache__/yamaha_cli.cpython-312.pyc differ diff --git a/devices.py b/devices.py new file mode 100644 index 0000000..506667a --- /dev/null +++ b/devices.py @@ -0,0 +1,7 @@ +import sounddevice as sd + +print("--- Elenco delle Host API Disponibili ---") +print(sd.query_hostapis()) + +print("\n--- Elenco Completo dei Dispositivi Audio ---") +print(sd.query_devices()) \ No newline at end of file diff --git a/mixer_agent.py b/mixer_agent.py new file mode 100644 index 0000000..f1df9be --- /dev/null +++ b/mixer_agent.py @@ -0,0 +1,651 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +TF5 Mixer AI Agent - Controllo del mixer Yamaha TF5 tramite linguaggio naturale +usando Google Gemini con function calling. +""" +import socket +import sys +import os +import json +import time +from pathlib import Path +from typing import List, Optional +from google import genai +from google.genai import types + +# Configurazione mixer +DEFAULT_HOST = "192.168.1.62" +DEFAULT_PORT = 49280 +TF5_INPUT_CHANNELS = 40 +TF5_MIX_BUSSES = 20 + +# Configurazione cache +CACHE_DIR = Path.home() / ".tf5_mixer_cache" +CACHE_FILE = CACHE_DIR / "channels_cache.json" +CACHE_DURATION = 3600 # 60 minuti in secondi + + +class TF5MixerController: + """Controller per il mixer Yamaha TF5 con sistema di caching.""" + + def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT): + self.host = host + self.port = port + self._ensure_cache_dir() + self._cache = self._load_cache() + + def _ensure_cache_dir(self): + """Crea la directory di cache se non esiste.""" + CACHE_DIR.mkdir(parents=True, exist_ok=True) + + def _load_cache(self) -> dict: + """Carica la cache dal file.""" + if CACHE_FILE.exists(): + try: + with open(CACHE_FILE, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + print(f"⚠️ Errore nel caricamento della cache: {e}") + return {"channels": {}, "timestamp": 0} + + def _save_cache(self): + """Salva la cache nel file.""" + try: + with open(CACHE_FILE, 'w', encoding='utf-8') as f: + json.dump(self._cache, f, indent=2, ensure_ascii=False) + except Exception as e: + print(f"⚠️ Errore nel salvataggio della cache: {e}") + + def _is_cache_valid(self) -> bool: + """Verifica se la cache è ancora valida.""" + if not self._cache.get("channels"): + return False + cache_age = time.time() - self._cache.get("timestamp", 0) + return cache_age < CACHE_DURATION + + def _send_command(self, command: str) -> str: + """Invia un comando al mixer e restituisce la risposta.""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(5) + s.connect((self.host, self.port)) + s.sendall((command + '\n').encode('utf-8')) + response = s.recv(4096) + s.close() + return response.decode('utf-8', errors='ignore').strip() + except socket.error as e: + return f"Errore di connessione: {e}" + + def _parse_name(self, response: str) -> str: + """Estrae il nome tra virgolette dalla risposta.""" + try: + start = response.find('"') + 1 + end = response.rfind('"') + if start > 0 and end > start: + return response[start:end] + return "Sconosciuto" + except: + return "Errore" + + def _parse_value(self, response: str) -> str: + """Estrae l'ultimo valore da una risposta OK.""" + parts = response.split() + if len(parts) > 0 and parts[0] == "OK": + return parts[-1] + return "N/A" + + def refresh_cache(self) -> dict: + """Aggiorna la cache leggendo tutti i canali dal mixer. + + Returns: + Un dizionario con lo stato dell'operazione + """ + print("🔄 Aggiornamento cache in corso...") + channels_data = {} + + for ch in range(1, TF5_INPUT_CHANNELS + 1): + ch_idx = ch - 1 + + # Leggi nome + resp_name = self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0") + name = self._parse_name(resp_name) + + # Leggi stato ON/OFF + resp_on = self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0") + is_on = self._parse_value(resp_on) == "1" + + # Leggi livello + resp_level = self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0") + level_raw = self._parse_value(resp_level) + try: + level_int = int(level_raw) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + except: + level_db = None + + # Leggi pan + resp_pan = self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0") + pan_raw = self._parse_value(resp_pan) + try: + pan_value = int(pan_raw) + except: + pan_value = None + + channels_data[str(ch)] = { + "channel": ch, + "name": name, + "on": is_on, + "level_db": level_db, + "pan": pan_value + } + + # Piccolo delay per non sovraccaricare il mixer + time.sleep(0.05) + + self._cache = { + "channels": channels_data, + "timestamp": time.time() + } + self._save_cache() + + print(f"✅ Cache aggiornata con {len(channels_data)} canali") + return { + "status": "success", + "message": f"Cache aggiornata con {len(channels_data)} canali", + "channels_count": len(channels_data) + } + + def recall_scene(self, bank: str, scene_number: int) -> dict: + """Richiama una scena dal banco A o B. + + Args: + bank: Il banco della scena ('a' o 'b') + scene_number: Il numero della scena (0-99) + + Returns: + Un dizionario con lo stato dell'operazione + """ + if bank.lower() not in ['a', 'b']: + return {"status": "error", "message": "Il banco deve essere 'a' o 'b'"} + if not 0 <= scene_number <= 99: + return {"status": "error", "message": "Il numero scena deve essere tra 0 e 99"} + + command = f"ssrecall_ex scene_{bank.lower()} {scene_number}" + response = self._send_command(command) + + # Invalida la cache dopo il cambio scena + self._cache["timestamp"] = 0 + + return { + "status": "success" if "OK" in response else "error", + "message": f"Scena {bank.upper()}{scene_number} richiamata. Cache invalidata.", + "response": response + } + + def set_channel_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello del fader di un canale in dB. + + Args: + channel: Numero del canale (1-40) + level_db: Livello in dB (da -inf a +10.0) + + Returns: + Un dizionario con lo stato dell'operazione + """ + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + + # Converti dB in valore interno (moltiplicato per 100) + if level_db <= -138: + internal_value = -32768 # -inf + else: + internal_value = int(level_db * 100) + + command = f"set MIXER:Current/InCh/Fader/Level {channel-1} 0 {internal_value}" + response = self._send_command(command) + + # Aggiorna cache locale + if str(channel) in self._cache.get("channels", {}): + self._cache["channels"][str(channel)]["level_db"] = level_db + + return { + "status": "success" if "OK" in response else "error", + "message": f"Canale {channel} impostato a {level_db:+.1f} dB", + "response": response + } + + def set_channel_on_off(self, channel: int, state: bool) -> dict: + """Accende o spegne un canale. + + Args: + channel: Numero del canale (1-40) + state: True per accendere, False per spegnere + + Returns: + Un dizionario con lo stato dell'operazione + """ + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + + value = 1 if state else 0 + command = f"set MIXER:Current/InCh/Fader/On {channel-1} 0 {value}" + response = self._send_command(command) + + # Aggiorna cache locale + if str(channel) in self._cache.get("channels", {}): + self._cache["channels"][str(channel)]["on"] = state + + return { + "status": "success" if "OK" in response else "error", + "message": f"Canale {channel} {'acceso' if state else 'spento'}", + "response": response + } + + def set_channel_pan(self, channel: int, pan_value: int) -> dict: + """Imposta il pan di un canale. + + Args: + channel: Numero del canale (1-40) + pan_value: Valore pan da -63 (sinistra) a +63 (destra), 0 è centro + + Returns: + Un dizionario con lo stato dell'operazione + """ + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not -63 <= pan_value <= 63: + return {"status": "error", "message": "Il pan deve essere tra -63 e +63"} + + command = f"set MIXER:Current/InCh/ToSt/Pan {channel-1} 0 {pan_value}" + response = self._send_command(command) + + # Aggiorna cache locale + if str(channel) in self._cache.get("channels", {}): + self._cache["channels"][str(channel)]["pan"] = pan_value + + pan_desc = "centro" + if pan_value < 0: + pan_desc = f"sinistra {abs(pan_value)}" + elif pan_value > 0: + pan_desc = f"destra {pan_value}" + + return { + "status": "success" if "OK" in response else "error", + "message": f"Canale {channel} pan impostato a {pan_desc}", + "response": response + } + + def set_mix_level(self, mix_number: int, level_db: float) -> dict: + """Imposta il livello di un mix/aux. + + Args: + mix_number: Numero del mix (1-20) + level_db: Livello in dB (da -inf a +10.0) + + Returns: + Un dizionario con lo stato dell'operazione + """ + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + + if level_db <= -138: + internal_value = -32768 + else: + internal_value = int(level_db * 100) + + command = f"set MIXER:Current/Mix/Fader/Level {mix_number-1} 0 {internal_value}" + response = self._send_command(command) + return { + "status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} impostato a {level_db:+.1f} dB", + "response": response + } + + def mute_multiple_channels(self, channels: List[int]) -> dict: + """Muta più canali contemporaneamente. + + Args: + channels: Lista di numeri di canale da mutare (es: [1, 2, 5, 8]) + + Returns: + Un dizionario con lo stato dell'operazione + """ + results = [] + for ch in channels: + result = self.set_channel_on_off(ch, False) + results.append(result) + + success_count = sum(1 for r in results if r["status"] == "success") + return { + "status": "success" if success_count == len(channels) else "partial", + "message": f"Mutati {success_count}/{len(channels)} canali: {channels}", + "details": results + } + + def unmute_multiple_channels(self, channels: List[int]) -> dict: + """Riattiva più canali contemporaneamente. + + Args: + channels: Lista di numeri di canale da riattivare (es: [1, 2, 5, 8]) + + Returns: + Un dizionario con lo stato dell'operazione + """ + results = [] + for ch in channels: + result = self.set_channel_on_off(ch, True) + results.append(result) + + success_count = sum(1 for r in results if r["status"] == "success") + return { + "status": "success" if success_count == len(channels) else "partial", + "message": f"Riattivati {success_count}/{len(channels)} canali: {channels}", + "details": results + } + + def get_channel_info(self, channel: int, force_refresh: bool = False) -> dict: + """Legge le informazioni di un canale (nome, livello, stato, pan). + Usa la cache se disponibile e valida. + + Args: + channel: Numero del canale (1-40) + force_refresh: Se True, ignora la cache e legge dal mixer + + Returns: + Un dizionario con tutte le informazioni del canale + """ + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + + # Usa cache se valida e non forzato il refresh + if not force_refresh and self._is_cache_valid(): + cached_data = self._cache.get("channels", {}).get(str(channel)) + if cached_data: + return { + "status": "success", + "source": "cache", + **cached_data + } + + # Altrimenti leggi dal mixer + ch_idx = channel - 1 + + resp_name = self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0") + name = self._parse_name(resp_name) + + resp_on = self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0") + is_on = self._parse_value(resp_on) == "1" + + resp_level = self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0") + level_raw = self._parse_value(resp_level) + try: + level_int = int(level_raw) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + except: + level_db = None + + resp_pan = self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0") + pan_raw = self._parse_value(resp_pan) + try: + pan_value = int(pan_raw) + except: + pan_value = None + + return { + "status": "success", + "source": "mixer", + "channel": channel, + "name": name, + "on": is_on, + "level_db": level_db, + "pan": pan_value + } + + def search_channels_by_name(self, search_term: str) -> dict: + """Cerca canali il cui nome contiene un determinato termine. + Usa la cache per velocizzare la ricerca. + + Args: + search_term: Il termine da cercare nei nomi dei canali (case-insensitive) + + Returns: + Un dizionario con la lista dei canali trovati + """ + # Aggiorna cache se non valida + if not self._is_cache_valid(): + self.refresh_cache() + + search_lower = search_term.lower() + found_channels = [] + + for ch_str, info in self._cache.get("channels", {}).items(): + if search_lower in info.get("name", "").lower(): + found_channels.append({ + "channel": info["channel"], + "name": info["name"], + "on": info["on"], + "level_db": info["level_db"] + }) + + # Ordina per numero canale + found_channels.sort(key=lambda x: x["channel"]) + + return { + "status": "success", + "search_term": search_term, + "found_count": len(found_channels), + "channels": found_channels, + "message": f"Trovati {len(found_channels)} canali contenenti '{search_term}'" + } + + def get_all_channels_summary(self) -> dict: + """Ottiene un riepilogo di tutti i canali con nome e stato. + Usa la cache per velocizzare. + + Returns: + Un dizionario con il riepilogo di tutti i canali + """ + # Aggiorna cache se non valida + if not self._is_cache_valid(): + self.refresh_cache() + + channels = [] + for ch_str, info in self._cache.get("channels", {}).items(): + channels.append({ + "channel": info["channel"], + "name": info["name"], + "on": info["on"] + }) + + # Ordina per numero canale + channels.sort(key=lambda x: x["channel"]) + + cache_age = time.time() - self._cache.get("timestamp", 0) + + return { + "status": "success", + "total_channels": len(channels), + "channels": channels, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(channels)} canali (cache: {int(cache_age)}s fa)" + } + + +class TF5AIAgent: + """Agente AI per controllare il mixer TF5 con linguaggio naturale.""" + + def __init__(self, mixer_host=DEFAULT_HOST, mixer_port=DEFAULT_PORT): + self.controller = TF5MixerController(mixer_host, mixer_port) + + # Configura il client Gemini + api_key = 'AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M' + if not api_key: + raise ValueError("GEMINI_API_KEY non trovata nelle variabili d'ambiente") + + self.client = genai.Client(api_key=api_key) + + # Configura gli strumenti con automatic function calling + self.config = types.GenerateContentConfig( + tools=[ + self.controller.recall_scene, + self.controller.set_channel_level, + self.controller.set_channel_on_off, + self.controller.set_channel_pan, + self.controller.set_mix_level, + self.controller.mute_multiple_channels, + self.controller.unmute_multiple_channels, + self.controller.get_channel_info, + self.controller.search_channels_by_name, + self.controller.get_all_channels_summary, + self.controller.refresh_cache, + ], + temperature=0, + ) + + # Messaggio di sistema per dare contesto all'AI + self.system_instruction = """Sei un assistente esperto per il controllo di mixer audio Yamaha TF5. + +Il mixer ha: +- 40 canali di input (numerati da 1 a 40) +- 20 mix/aux bus (numerati da 1 a 20) +- Livelli fader espressi in dB (da -inf a +10.0 dB) +- Pan da -63 (sinistra) a +63 (destra), 0 è centro +- Scene salvate nei banchi A e B (numerate da 0 a 99) + +IMPORTANTE - Sistema di Cache: +- Le informazioni sui canali sono cachate per 5 minuti per evitare di sovraccaricare il mixer +- Usa search_channels_by_name e get_all_channels_summary che usano automaticamente la cache +- La cache viene invalidata automaticamente quando si richiama una scena +- Puoi usare refresh_cache se l'utente chiede esplicitamente dati aggiornati + +Quando l'utente fa una richiesta: +1. Interpreta il linguaggio naturale e identifica l'azione richiesta +2. Usa le funzioni disponibili per eseguire i comandi +3. Conferma all'utente cosa hai fatto in modo chiaro e conciso +4. Se una richiesta non è chiara, chiedi chiarimenti + +Esempi di comandi che puoi gestire: +- "Alza il canale 5 a -10 dB" +- "Spegni i canali dal 10 al 15" +- "Imposta il pan del canale 3 tutto a sinistra" +- "Richiama la scena A5" +- "Muta i canali 1, 3, 5 e 7" +- "Quali canali sono associati ai vox?" (cerca nei nomi usando cache) +- "Mostrami lo stato del canale 12" +- "Dammi la lista di tutti i canali" (usa cache) +- "Aggiorna i dati dal mixer" (refresh_cache) +""" + + def chat(self, user_message: str) -> str: + """Invia un messaggio all'agente e riceve la risposta. + + Args: + user_message: Il messaggio dell'utente in linguaggio naturale + + Returns: + La risposta dell'agente + """ + try: + full_prompt = f"{self.system_instruction}\n\nUtente: {user_message}" + + response = self.client.models.generate_content( + model="gemini-2.5-flash", + contents=full_prompt, + config=self.config, + ) + + return response.text + except Exception as e: + return f"Errore nell'elaborazione della richiesta: {e}" + + def interactive_mode(self): + """Avvia una sessione interattiva con l'agente.""" + print("=" * 70) + print("TF5 Mixer AI Agent - Controllo tramite linguaggio naturale") + print("=" * 70) + + # Mostra stato cache + cache_age = time.time() - self.controller._cache.get("timestamp", 0) + if self.controller._is_cache_valid(): + print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)") + else: + print("\n💾 Cache non disponibile, verrà creata al primo utilizzo") + + print("\nEsempi di comandi:") + print(" - 'Alza il canale 5 a -10 dB'") + print(" - 'Spegni i canali dal 1 al 5'") + print(" - 'Richiama la scena A10'") + print(" - 'Imposta il pan del canale 3 a sinistra'") + print(" - 'Muta i canali 2, 4, 6 e 8'") + print(" - 'Quali canali sono associati ai vox?'") + print(" - 'Mostrami lo stato del canale 12'") + print(" - 'Aggiorna i dati dal mixer'") + print("\nDigita 'esci' o 'quit' per terminare\n") + + while True: + try: + user_input = input("\n🎛️ Tu: ").strip() + + if user_input.lower() in ['esci', 'quit', 'exit', 'q']: + print("\n👋 Arrivederci!") + break + + if not user_input: + continue + + print("\n🤖 Agent: ", end="", flush=True) + response = self.chat(user_input) + print(response) + + except KeyboardInterrupt: + print("\n\n👋 Arrivederci!") + break + except Exception as e: + print(f"\n❌ Errore: {e}") + + +def main(): + """Funzione principale.""" + import argparse + + parser = argparse.ArgumentParser( + description='TF5 Mixer AI Agent - Controllo tramite linguaggio naturale' + ) + parser.add_argument('--host', default=DEFAULT_HOST, help=f'IP del mixer (default: {DEFAULT_HOST})') + parser.add_argument('--port', type=int, default=DEFAULT_PORT, help=f'Porta (default: {DEFAULT_PORT})') + parser.add_argument('--message', '-m', help='Invia un singolo comando invece di avviare la modalità interattiva') + parser.add_argument('--refresh-cache', action='store_true', help='Forza l\'aggiornamento della cache all\'avvio') + + args = parser.parse_args() + + # Verifica che la API key sia impostata + if not 'AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M': + print("❌ Errore: GEMINI_API_KEY non trovata nelle variabili d'ambiente") + print("\nPer impostare la chiave API:") + print(" export GEMINI_API_KEY='la-tua-chiave-api'") + print("\nOttieni una chiave API gratuita su: https://aistudio.google.com/apikey") + sys.exit(1) + + try: + agent = TF5AIAgent(args.host, args.port) + + # Refresh cache se richiesto + if args.refresh_cache: + agent.controller.refresh_cache() + + if args.message: + # Modalità singolo comando + print(f"🎛️ Comando: {args.message}") + print(f"\n🤖 Risposta: {agent.chat(args.message)}") + else: + # Modalità interattiva + agent.interactive_mode() + + except Exception as e: + print(f"❌ Errore fatale: {e}") + sys.exit(1) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/mixer_controller.py b/mixer_controller.py new file mode 100644 index 0000000..b5a29a0 --- /dev/null +++ b/mixer_controller.py @@ -0,0 +1,32 @@ +# mixer_controller.py +import socket +import sys + +# Impostazioni di connessione predefinite +DEFAULT_HOST = "192.168.1.62" # Modifica con l'IP del tuo mixer +DEFAULT_PORT = 49280 + +def send_command(command, host=DEFAULT_HOST, port=DEFAULT_PORT): + """ + Crea una connessione, invia un singolo comando, riceve la risposta e la chiude. + Restituisce la risposta del mixer. + """ + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(5) + s.connect((host, port)) + s.sendall((command + '\n').encode('utf-8')) + print(f"-> Comando inviato al mixer: '{command}'") + response = s.recv(4096).decode('utf-8', errors='ignore').strip() + print(f"<- Risposta dal mixer: '{response}'") + if response.startswith("OK"): + return {"status": "success", "response": response} + else: + return {"status": "error", "response": response} + except socket.error as e: + print(f"Errore critico di connessione a {host}:{port} -> {e}") + error_msg = f"Impossibile connettersi al mixer: {e}" + return {"status": "error", "response": error_msg} + except Exception as e: + print(f"Errore imprevisto: {e}") + return {"status": "error", "response": str(e)} \ No newline at end of file diff --git a/record_from_usb.py b/record_from_usb.py new file mode 100644 index 0000000..e0fc871 --- /dev/null +++ b/record_from_usb.py @@ -0,0 +1,42 @@ +import sounddevice as sd +import numpy as np +from scipy.io.wavfile import write + +# --- 1. Identificazione del Dispositivo --- +# Esegui questo blocco di codice per primo per trovare l'indice del tuo dispositivo. +# print(sd.query_devices()) +# Una volta trovato, inserisci l'indice corretto qui sotto. + +# --- 2. Impostazioni della Registrazione --- +DEVICE_INDEX = 26 # ESEMPIO: Sostituisci con l'indice del tuo dispositivo Yamaha TF5 +CHANNELS = 2 # Numero di canali da registrare (la TF5 ne supporta fino a 34) +SAMPLE_RATE = 48000 # Frequenza di campionamento in Hz (deve corrispondere a quella della console) +DURATION = 10 # Durata della registrazione in secondi +OUTPUT_FILENAME = 'registrazione_multicanale.wav' + +try: + # --- 3. Registrazione dell'Audio --- + print(f"Inizio registrazione di {DURATION} secondi dal dispositivo {DEVICE_INDEX}...") + + # sd.rec() avvia una registrazione non bloccante e restituisce immediatamente + # I dati audio vengono memorizzati in un array NumPy + myrecording = sd.rec(int(DURATION * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=CHANNELS, device=DEVICE_INDEX, dtype='float32') + + # sd.wait() attende che la registrazione sia completata + sd.wait() + + print("Registrazione completata.") + + # --- 4. Salvataggio del File WAV --- + print(f"Salvataggio della registrazione in corso su '{OUTPUT_FILENAME}'...") + + # Scrive l'array NumPy in un file WAV + # La funzione si aspetta dati in un formato specifico (es. int16 o float32) + # È consigliabile scalare i dati se necessario, ma sounddevice con dtype='float32' + # di solito produce un output compatibile. + write(OUTPUT_FILENAME, SAMPLE_RATE, myrecording) + + print("File salvato con successo.") + +except Exception as e: + print(f"Si è verificato un errore: {e}") \ No newline at end of file diff --git a/registrazione_multicanale.wav b/registrazione_multicanale.wav new file mode 100644 index 0000000..a4a99fc Binary files /dev/null and b/registrazione_multicanale.wav differ diff --git a/yamaha_cli.py b/yamaha_cli.py new file mode 100644 index 0000000..d81437f --- /dev/null +++ b/yamaha_cli.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import socket +import argparse +import sys +import time + +# Impostazioni di connessione predefinite +DEFAULT_HOST = "192.168.1.62" +DEFAULT_PORT = 49280 + +# Configurazioni specifiche per Yamaha TF5 +TF5_INPUT_CHANNELS = 40 # 32 locali + 8 altro +TF5_MIX_BUSSES = 20 # 20 Aux/Mix + +def create_connection(host, port): + """Crea e restituisce un socket connesso.""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(5) + s.connect((host, port)) + return s + except socket.error as e: + print(f"Errore critico di connessione a {host}:{port} -> {e}") + sys.exit(1) + +def send_command(host, port, command): + """Invia un comando singolo e stampa la risposta (modalità one-shot).""" + s = create_connection(host, port) + try: + s.sendall((command + '\n').encode('utf-8')) + print(f"Comando inviato: '{command}'") + response = s.recv(4096) + print(f"Risposta: {response.decode('utf-8', errors='ignore').strip()}") + finally: + s.close() + +def get_param(sock, command): + """Funzione helper per ottenere un singolo parametro mantenendo il socket aperto.""" + try: + sock.sendall((command + '\n').encode('utf-8')) + # Riceviamo dati finché non troviamo un newline, per essere sicuri di avere la risposta intera + response = b"" + while b'\n' not in response: + chunk = sock.recv(1024) + if not chunk: break + response += chunk + return response.decode('utf-8', errors='ignore').strip() + except Exception as e: + return f"Error: {e}" + +def parse_val(response): + """Estrae l'ultimo valore numerico da una risposta standard 'OK ... value'.""" + parts = response.split() + if len(parts) > 0 and parts[0] == "OK": + return parts[-1] + return "N/A" + +def parse_name(response): + """Estrae il nome tra virgolette dalla risposta.""" + # Esempio risposta: OK MIXER:Current/InCh/Label/Name 0 0 "Kick Drum" + try: + start = response.find('"') + 1 + end = response.rfind('"') + if start > 0 and end > start: + return response[start:end] + return "Unknown" + except: + return "Error" + +def format_db(val_str): + """Converte il valore interno (es. 1000) in stringa dB (es. +10.0 dB).""" + try: + val = int(val_str) + if val <= -32768: + return "-inf dB" + return f"{val / 100.0:+.1f} dB" + except: + return val_str + +def format_pan(val_str): + """Converte il valore pan (-63 a 63) in formato leggibile (L63, C, R63).""" + try: + val = int(val_str) + if val == 0: return "C" + if val < 0: return f"L{abs(val)}" + return f"R{val}" + except: + return val_str + +def get_full_config(host, port): + """Scarica e stampa la configurazione di tutti i canali e mix.""" + print(f"Connessione a {host}... Inizio download configurazione.") + print("Attendere prego, l'operazione potrebbe richiedere alcuni secondi...\n") + + s = create_connection(host, port) + # Impostiamo un timeout breve per le singole richieste per velocizzare eventuali errori + s.settimeout(2) + + try: + # --- INPUT CHANNELS --- + print("-" * 65) + print(f"{'CH':<4} {'NAME':<16} {'STATUS':<8} {'FADER':<12} {'PAN':<8}") + print("-" * 65) + + for i in range(TF5_INPUT_CHANNELS): + # Recupera Nome + resp_name = get_param(s, f"get MIXER:Current/InCh/Label/Name {i} 0") + name = parse_name(resp_name) + + # Recupera ON/OFF + resp_on = get_param(s, f"get MIXER:Current/InCh/Fader/On {i} 0") + is_on = parse_val(resp_on) + status = " [ON] " if is_on == "1" else " OFF " + + # Recupera Livello Fader + resp_lvl = get_param(s, f"get MIXER:Current/InCh/Fader/Level {i} 0") + level_db = format_db(parse_val(resp_lvl)) + + # Recupera Pan + resp_pan = get_param(s, f"get MIXER:Current/InCh/ToSt/Pan {i} 0") + pan_str = format_pan(parse_val(resp_pan)) + + print(f"CH{i+1:<2} {name[:15]:<16} {status:<8} {level_db:<12} {pan_str:<8}") + # Piccolo sleep per non saturare il buffer del mixer se necessario + # time.sleep(0.01) + + print("\n") + # --- MIX (AUX) BUSSES --- + print("-" * 50) + print(f"{'MIX':<4} {'NAME':<16} {'STATUS':<8} {'FADER':<12}") + print("-" * 50) + + for i in range(TF5_MIX_BUSSES): + # Recupera Nome Mix + resp_name = get_param(s, f"get MIXER:Current/Mix/Label/Name {i} 0") + name = parse_name(resp_name) + + # Recupera ON/OFF Mix + resp_on = get_param(s, f"get MIXER:Current/Mix/Fader/On {i} 0") + is_on = parse_val(resp_on) + status = " [ON] " if is_on == "1" else " OFF " + + # Recupera Livello Mix + resp_lvl = get_param(s, f"get MIXER:Current/Mix/Fader/Level {i} 0") + level_db = format_db(parse_val(resp_lvl)) + + print(f"MX{i+1:<2} {name[:15]:<16} {status:<8} {level_db:<12}") + + print("-" * 50) + print("\nConfigurazione completata.") + + except KeyboardInterrupt: + print("\nOperazione interrotta dall'utente.") + except Exception as e: + print(f"\nErrore durante il recupero della configurazione: {e}") + finally: + s.close() + +def get_full_config_dict(host, port): + """ + Scarica la configurazione di tutti i canali e mix e la restituisce come dizionario. + """ + print(f"Connessione a {host}... Inizio download configurazione.") + print("Attendere prego, l'operazione potrebbe richiedere alcuni secondi...\n") + + s = create_connection(host, port) + # Impostiamo un timeout breve per le singole richieste per velocizzare eventuali errori + s.settimeout(2) + + config = { + "input_channels": [], + "mix_busses": [] + } + + try: + # --- INPUT CHANNELS --- + for i in range(TF5_INPUT_CHANNELS): + channel_data = {} + + # Recupera Nome + resp_name = get_param(s, f"get MIXER:Current/InCh/Label/Name {i} 0") + channel_data['name'] = parse_name(resp_name) + + # Recupera ON/OFF + resp_on = get_param(s, f"get MIXER:Current/InCh/Fader/On {i} 0") + channel_data['status'] = 'ON' if parse_val(resp_on) == "1" else 'OFF' + + # Recupera Livello Fader + resp_lvl = get_param(s, f"get MIXER:Current/InCh/Fader/Level {i} 0") + channel_data['fader_db'] = format_db(parse_val(resp_lvl)) + + # Recupera Pan + resp_pan = get_param(s, f"get MIXER:Current/InCh/ToSt/Pan {i} 0") + channel_data['pan'] = format_pan(parse_val(resp_pan)) + + config["input_channels"].append(channel_data) + + # --- MIX (AUX) BUSSES --- + for i in range(TF5_MIX_BUSSES): + mix_data = {} + + # Recupera Nome Mix + resp_name = get_param(s, f"get MIXER:Current/Mix/Label/Name {i} 0") + mix_data['name'] = parse_name(resp_name) + + # Recupera ON/OFF Mix + resp_on = get_param(s, f"get MIXER:Current/Mix/Fader/On {i} 0") + mix_data['status'] = 'ON' if parse_val(resp_on) == "1" else 'OFF' + + # Recupera Livello Mix + resp_lvl = get_param(s, f"get MIXER:Current/Mix/Fader/Level {i} 0") + mix_data['fader_db'] = format_db(parse_val(resp_lvl)) + + config["mix_busses"].append(mix_data) + + print("Configurazione scaricata con successo.") + return config + + except KeyboardInterrupt: + print("\nOperazione interrotta dall'utente.") + return None + except Exception as e: + print(f"\nErrore durante il recupero della configurazione: {e}") + return None + finally: + s.close() + +def main(): + parser = argparse.ArgumentParser( + description='CLI per controllare un mixer Yamaha TF5.', + formatter_class=argparse.RawTextHelpFormatter + ) + parser.add_argument('--host', default=DEFAULT_HOST, help=f'IP del mixer (default: {DEFAULT_HOST})') + parser.add_argument('--port', type=int, default=DEFAULT_PORT, help=f'Porta (default: {DEFAULT_PORT})') + + subparsers = parser.add_subparsers(dest='command', help='Comandi disponibili') + + # Comandi esistenti + p_recall = subparsers.add_parser('recall', help='Richiama scena (es: recall a 0)') + p_recall.add_argument('banco', choices=['a', 'b']) + p_recall.add_argument('numero', type=int) + + p_level = subparsers.add_parser('level', help='Imposta fader (es: level 0 1000)') + p_level.add_argument('canale', type=int) + p_level.add_argument('valore', type=int) + + p_onoff = subparsers.add_parser('onoff', help='Imposta on/off (es: onoff 0 on)') + p_onoff.add_argument('canale', type=int) + p_onoff.add_argument('stato', choices=['on', 'off']) + + p_pan = subparsers.add_parser('pan', help='Imposta pan (es: pan 0 -63)') + p_pan.add_argument('canale', type=int) + p_pan.add_argument('valore', type=int) + + p_raw = subparsers.add_parser('raw', help='Invia comando grezzo') + p_raw.add_argument('raw_cmd', nargs='+') + + # --- NUOVO COMANDO --- + subparsers.add_parser('config', help='Legge e stampa la configurazione completa del mixer') + # --------------------- + + args = parser.parse_args() + + if args.command == 'config': + get_full_config(args.host, args.port) + elif args.command == 'recall': + send_command(args.host, args.port, f"ssrecall_ex scene_{args.banco} {args.numero}") + elif args.command == 'level': + send_command(args.host, args.port, f"set MIXER:Current/InCh/Fader/Level {args.canale} 0 {args.valore}") + elif args.command == 'onoff': + val = 1 if args.stato == 'on' else 0 + send_command(args.host, args.port, f"set MIXER:Current/InCh/Fader/On {args.canale} 0 {val}") + elif args.command == 'pan': + send_command(args.host, args.port, f"set MIXER:Current/InCh/ToSt/Pan {args.canale} 0 {args.valore}") + elif args.command == 'raw': + send_command(args.host, args.port, ' '.join(args.raw_cmd)) + else: + parser.print_help() + +if __name__ == '__main__': + main() \ No newline at end of file