diff --git a/.tf5_mixer_cache/channels_cache.json b/.tf5_mixer_cache/channels_cache.json new file mode 100644 index 0000000..be269e5 --- /dev/null +++ b/.tf5_mixer_cache/channels_cache.json @@ -0,0 +1,407 @@ +{ + "channels": { + "1": { + "channel": 1, + "name": "Sconosciuto", + "on": false, + "level_db": 0.65, + "pan": 0 + }, + "2": { + "channel": 2, + "name": "Gelato", + "on": true, + "level_db": -6.25, + "pan": 0 + }, + "3": { + "channel": 3, + "name": "Talkback", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "4": { + "channel": 4, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "5": { + "channel": 5, + "name": "Vox1", + "on": true, + "level_db": 0.65, + "pan": 0 + }, + "6": { + "channel": 6, + "name": "Vox2", + "on": true, + "level_db": -1.1, + "pan": 0 + }, + "7": { + "channel": 7, + "name": "Basso", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "8": { + "channel": 8, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "9": { + "channel": 9, + "name": "Kick", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "10": { + "channel": 10, + "name": "Snare", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "11": { + "channel": 11, + "name": "Tom 1", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "12": { + "channel": 12, + "name": "Tom 2", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "13": { + "channel": 13, + "name": "Tom3", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "14": { + "channel": 14, + "name": "Pan SX", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "15": { + "channel": 15, + "name": "Pan dx", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "16": { + "channel": 16, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "17": { + "channel": 17, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "18": { + "channel": 18, + "name": "Archetto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "19": { + "channel": 19, + "name": "Vox 5", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "20": { + "channel": 20, + "name": "Tast", + "on": true, + "level_db": 1.0, + "pan": 0 + }, + "21": { + "channel": 21, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "22": { + "channel": 22, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "23": { + "channel": 23, + "name": " Vox3", + "on": true, + "level_db": 3.8, + "pan": 0 + }, + "24": { + "channel": 24, + "name": "Chit cnt", + "on": true, + "level_db": -2.1, + "pan": 0 + }, + "25": { + "channel": 25, + "name": "Chit dx", + "on": true, + "level_db": -7.7, + "pan": 0 + }, + "26": { + "channel": 26, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "27": { + "channel": 27, + "name": "Vox 4", + "on": false, + "level_db": -132.0, + "pan": 0 + }, + "28": { + "channel": 28, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "29": { + "channel": 29, + "name": "Pad", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "30": { + "channel": 30, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "31": { + "channel": 31, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "32": { + "channel": 32, + "name": "Sconosciuto", + "on": false, + "level_db": -Infinity, + "pan": 0 + }, + "33": { + "channel": 33, + "name": "PC", + "on": true, + "level_db": -19.8, + "pan": -63 + }, + "34": { + "channel": 34, + "name": "PC", + "on": true, + "level_db": -19.8, + "pan": 63 + }, + "35": { + "channel": 35, + "name": "ch35", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "36": { + "channel": 36, + "name": "ch36", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "37": { + "channel": 37, + "name": "ch37", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "38": { + "channel": 38, + "name": "ch38", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "39": { + "channel": 39, + "name": "ch39", + "on": true, + "level_db": -Infinity, + "pan": 0 + }, + "40": { + "channel": 40, + "name": "ch40", + "on": true, + "level_db": -Infinity, + "pan": 0 + } + }, + "mixes": { + "1": { + "mix": 1, + "name": "Sinistro", + "on": true, + "level_db": -Infinity + }, + "2": { + "mix": 2, + "name": "Aux 2", + "on": true, + "level_db": -0.05 + }, + "3": { + "mix": 3, + "name": "Aux 3", + "on": true, + "level_db": -0.5 + }, + "4": { + "mix": 4, + "name": "Aux 4", + "on": true, + "level_db": -0.6 + }, + "5": { + "mix": 5, + "name": "batteria", + "on": true, + "level_db": 1.35 + }, + "6": { + "mix": 6, + "name": "Aux 6", + "on": true, + "level_db": 0.9 + }, + "7": { + "mix": 7, + "name": "Destro", + "on": true, + "level_db": -2.7 + }, + "8": { + "mix": 8, + "name": "Aux 8", + "on": true, + "level_db": -0.15 + }, + "9": { + "mix": 9, + "name": "Aux 9/10", + "on": true, + "level_db": -8.1 + }, + "10": { + "mix": 10, + "name": "Aux 9/10", + "on": true, + "level_db": -8.1 + }, + "11": { + "mix": 11, + "name": "Aux Pulp", + "on": true, + "level_db": -10.3 + }, + "12": { + "mix": 12, + "name": "Aux Pulp", + "on": true, + "level_db": -10.3 + }, + "13": { + "mix": 13, + "name": "Aux13/14", + "on": false, + "level_db": -1.6 + }, + "14": { + "mix": 14, + "name": "Aux13/14", + "on": false, + "level_db": -1.6 + }, + "15": { + "mix": 15, + "name": "Aux15/16", + "on": true, + "level_db": 0.1 + }, + "16": { + "mix": 16, + "name": "Aux15/16", + "on": true, + "level_db": 0.1 + }, + "17": { + "mix": 17, + "name": "Diretta", + "on": true, + "level_db": 0.0 + }, + "18": { + "mix": 18, + "name": "Diretta", + "on": true, + "level_db": 0.0 + }, + "19": { + "mix": 19, + "name": "Traduzio", + "on": true, + "level_db": 6.0 + }, + "20": { + "mix": 20, + "name": "Traduzio", + "on": true, + "level_db": 6.0 + } + }, + "timestamp": 1761592356.5132928 +} \ No newline at end of file diff --git a/__pycache__/config.cpython-312.pyc b/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000..674be36 Binary files /dev/null and b/__pycache__/config.cpython-312.pyc differ diff --git a/__pycache__/mixer_controller.cpython-312.pyc b/__pycache__/mixer_controller.cpython-312.pyc index 826c57f..21741ab 100644 Binary files a/__pycache__/mixer_controller.cpython-312.pyc and b/__pycache__/mixer_controller.cpython-312.pyc differ diff --git a/config.py b/config.py new file mode 100644 index 0000000..2271daa --- /dev/null +++ b/config.py @@ -0,0 +1,13 @@ +# Configurazione mixer +DEFAULT_HOST = "192.168.1.62" +DEFAULT_PORT = 49280 +TF5_INPUT_CHANNELS = 40 +TF5_MIX_BUSSES = 20 + +from pathlib import Path + + +# Configurazione cache +CACHE_DIR = Path(".tf5_mixer_cache") +CACHE_FILE = CACHE_DIR / "channels_cache.json" +CACHE_DURATION = 3600 # 60 minuti in secondi \ No newline at end of file diff --git a/mixer_agent.py b/mixer_agent.py index b467eb5..7fd84b6 100644 --- a/mixer_agent.py +++ b/mixer_agent.py @@ -13,513 +13,12 @@ from pathlib import Path from typing import List, Optional from google import genai from google.genai import types - from dotenv import load_dotenv +from mixer_controller import TF5MixerController + load_dotenv() -# 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: - def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT): - self.host = host - self.port = port - self.socket = None - self._ensure_cache_dir() - self._cache = self._load_cache() - - def _connect(self): - """Stabilisce la connessione se non già connesso.""" - if self.socket is None: - print('Inizializzazione socket...') - try: - self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.socket.settimeout(5) - self.socket.connect((self.host, self.port)) - except socket.error as e: - self.socket = None - raise ConnectionError(f"Impossibile connettersi al mixer: {e}") - - def _disconnect(self): - """Chiude la connessione.""" - if self.socket: - print('Chiusura socket...') - try: - self.socket.close() - print('Socket chiuso!') - except: - print('Errore durante chiusura socket!') - pass - finally: - self.socket = None - - def _send_command(self, command: str) -> str: - """Invia un comando al mixer e restituisce la risposta.""" - max_retries = 2 - - for attempt in range(max_retries): - try: - print(f'Tentativo di connessione {attempt} per {command}') - self._connect() - self.socket.sendall((command + '\n').encode('utf-8')) - response = self.socket.recv(4096) - decoded = response.decode('utf-8', errors='ignore').strip() - print(f'Risposta {decoded}') - return decoded - - except socket.error as e: - print(f'Errore di connessione dopo {max_retries} tentativi: {e}') - self._disconnect() # Forza riconnessione al prossimo tentativo - - if attempt < max_retries - 1: - time.sleep(0.1) - continue - else: - return f"Errore di connessione dopo {max_retries} tentativi: {e}" - - def close(self): - """Chiude la connessione (da chiamare alla fine).""" - self._disconnect() - - def __enter__(self): - """Context manager entry.""" - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """Context manager exit.""" - self.close() - - 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 _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)" - } +from config import * class TF5AIAgent: @@ -543,11 +42,15 @@ class TF5AIAgent: self.controller.set_channel_on_off, self.controller.set_channel_pan, self.controller.set_mix_level, + self.controller.set_mix_on_off, self.controller.mute_multiple_channels, self.controller.unmute_multiple_channels, self.controller.get_channel_info, + self.controller.get_mix_info, self.controller.search_channels_by_name, + self.controller.search_mixes_by_name, self.controller.get_all_channels_summary, + self.controller.get_all_mixes_summary, self.controller.refresh_cache, ], temperature=0, @@ -564,9 +67,11 @@ Il mixer ha: - Scene memorizzate nei banchi A e B (da 0 a 99) IMPORTANTE - Sistema di Cache: -- Le info sui canali sono salvate per 5 minuti per non sovraccaricare il mixer -- Usa search_channels_by_name e get_all_channels_summary per cercare velocemente -- Quando si carica una scena, i dati vengono aggiornati automaticamente +- Le info sui canali e mix sono salvate per 60 minuti per non sovraccaricare il mixer +- Usa search_channels_by_name e search_mixes_by_name per cercare velocemente +- Usa get_all_channels_summary e get_all_mixes_summary per vedere tutto +- Quando modifichi un canale/mix, la cache viene aggiornata automaticamente +- Quando si carica una scena, i dati vengono invalidati e aggiornati alla prossima richiesta - Puoi fare refresh_cache solo se l'utente lo chiede esplicitamente Come interpretare le richieste: @@ -582,8 +87,8 @@ VOLUME/LIVELLO: - "silenzio/muto" → spegni il canale ON/OFF: -- "accendi/attiva/apri" → canale ON -- "spegni/muta/chiudi/stacca" → canale OFF +- "accendi/attiva/apri" → canale/mix ON +- "spegni/muta/chiudi/stacca" → canale/mix OFF - "muto" può significare sia spegnere che abbassare molto BILANCIAMENTO (PAN): @@ -592,12 +97,13 @@ BILANCIAMENTO (PAN): - "al centro" → pan 0 - "un po' a sinistra" → pan -30 circa -IDENTIFICAZIONE CANALI: -- Accetta sia numeri ("canale 5") che nomi ("il microfono del cantante") -- Se non trovi un canale per nome, cerca usando search_channels_by_name +IDENTIFICAZIONE CANALI E MIX: +- Accetta sia numeri ("canale 5", "mix 3") che nomi ("il microfono del cantante", "monitor palco") +- Se non trovi un canale/mix per nome, cerca usando search_channels_by_name o search_mixes_by_name - "il mio mic/microfono" → cerca tra i canali chi è sul palco - "le chitarre/i vox/le tastiere" → cerca per strumento - "tutti i mic/tutte le chitarre" → cerca e gestisci multipli +- "il monitor" / "l'aux 2" → cerca tra i mix SCENE: - "carica/richiama/vai alla scena X" → recall_scene @@ -610,7 +116,7 @@ GRUPPI DI CANALI: CASI PARTICOLARI: - Se la richiesta è ambigua, chiedi chiarimenti in modo colloquiale -- Se serve cercare un canale, usa prima la cache (search_channels_by_name) +- Se serve cercare un canale/mix, usa prima la cache (search_channels_by_name / search_mixes_by_name) - Conferma sempre cosa hai fatto con un messaggio breve e chiaro - Usa emoji occasionalmente per rendere le risposte più amichevoli (✅ ❌ 🎤 🎸 🔊) - Se qualcosa non funziona, spiega il problema in modo semplice @@ -620,7 +126,7 @@ ESEMPI DI INTERPRETAZIONE: "abbassa un po' le chitarre" → cerca canali chitarra, riduci di 3-5 dB "muto tutto" → spegni tutti i 40 canali "solo voce" → cerca canali voce, accendi quelli e spegni gli altri -"mettimi più forte nel monitor" → NON puoi (sono gli aux), spiega che serve il tecnico +"alza il monitor 2" → cerca mix 2, aumenta volume "carica la scena del soundcheck" → cerca nel nome o chiedi numero scena "troppo forte, abbassa" → riduci di 5-8 dB "spegni questo canale" → se non specifica numero, chiedi quale @@ -646,7 +152,7 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. def close(self): """Chiude le connessioni.""" self.controller.close() - + def chat(self, user_message: str) -> str: """Invia un messaggio all'agente e riceve la risposta. @@ -678,7 +184,10 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. # Mostra stato cache cache_age = time.time() - self.controller._cache.get("timestamp", 0) if self.controller._is_cache_valid(): + channels_count = len(self.controller._cache.get("channels", {})) + mixes_count = len(self.controller._cache.get("mixes", {})) print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)") + print(f" 📊 {channels_count} canali, {mixes_count} mix") else: print("\n💾 Cache non disponibile, verrà creata al primo utilizzo") @@ -688,8 +197,10 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. print(" - 'Richiama la scena A10'") print(" - 'Imposta il pan del canale 3 a sinistra'") print(" - 'Muta i canali 2, 4, 6 e 8'") + print(" - 'Alza il mix 3 di 5 dB'") print(" - 'Quali canali sono associati ai vox?'") print(" - 'Mostrami lo stato del canale 12'") + print(" - 'Cerca i monitor'") print(" - 'Aggiorna i dati dal mixer'") print("\nDigita 'esci' o 'quit' per terminare\n") diff --git a/mixer_controller.py b/mixer_controller.py index b5a29a0..8c1b4b5 100644 --- a/mixer_controller.py +++ b/mixer_controller.py @@ -1,32 +1,785 @@ -# mixer_controller.py 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 +from dotenv import load_dotenv -# Impostazioni di connessione predefinite -DEFAULT_HOST = "192.168.1.62" # Modifica con l'IP del tuo mixer -DEFAULT_PORT = 49280 +load_dotenv() + +from config import * + + +class TF5MixerController: + def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT): + self.host = host + self.port = port + self.socket = None + self._ensure_cache_dir() + self._cache = self._load_cache() + + def _connect(self): + """Stabilisce la connessione se non già connesso.""" + if self.socket is None: + print('Inizializzazione socket...') + try: + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.settimeout(5) + self.socket.connect((self.host, self.port)) + except socket.error as e: + self.socket = None + raise ConnectionError(f"Impossibile connettersi al mixer: {e}") + + def _disconnect(self): + """Chiude la connessione.""" + if self.socket: + print('Chiusura socket...') + try: + self.socket.close() + print('Socket chiuso!') + except: + print('Errore durante chiusura socket!') + pass + finally: + self.socket = None + + def _send_command(self, command: str) -> str: + """Invia un comando al mixer e restituisce la risposta.""" + max_retries = 2 + + for attempt in range(max_retries): + try: + print(f'Tentativo di connessione {attempt} per {command}') + self._connect() + self.socket.sendall((command + '\n').encode('utf-8')) + response = self.socket.recv(4096) + decoded = response.decode('utf-8', errors='ignore').strip() + print(f'Risposta {decoded}') + return decoded + + except socket.error as e: + print(f'Errore di connessione dopo {max_retries} tentativi: {e}') + self._disconnect() # Forza riconnessione al prossimo tentativo + + if attempt < max_retries - 1: + time.sleep(0.1) + continue + else: + return f"Errore di connessione dopo {max_retries} tentativi: {e}" + + def close(self): + """Chiude la connessione (da chiamare alla fine).""" + self._disconnect() + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.close() + + 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": {}, "mixes": {}, "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 _update_channel_cache(self, channel: int): + """Aggiorna la cache per un singolo canale leggendo dal mixer. + + Args: + channel: Numero del canale (1-40) + """ + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return + + ch_idx = channel - 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 + + # Inizializza la struttura channels se non esiste + if "channels" not in self._cache: + self._cache["channels"] = {} + + # Aggiorna la cache + self._cache["channels"][str(channel)] = { + "channel": channel, + "name": name, + "on": is_on, + "level_db": level_db, + "pan": pan_value + } + + self._save_cache() + + def _update_mix_cache(self, mix_number: int): + """Aggiorna la cache per un singolo mix/aux leggendo dal mixer. + + Args: + mix_number: Numero del mix (1-20) + """ + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return + + mix_idx = mix_number - 1 + + # Leggi nome + resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0") + name = self._parse_name(resp_name) + + # Leggi stato ON/OFF + resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0") + is_on = self._parse_value(resp_on) == "1" + + # Leggi livello + resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_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 + + # Inizializza la struttura mixes se non esiste + if "mixes" not in self._cache: + self._cache["mixes"] = {} + + # Aggiorna la cache + self._cache["mixes"][str(mix_number)] = { + "mix": mix_number, + "name": name, + "on": is_on, + "level_db": level_db + } + + self._save_cache() + + 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 e mix dal mixer. + + Returns: + Un dizionario con lo stato dell'operazione + """ + print("🔄 Aggiornamento cache completo in corso...") + channels_data = {} + mixes_data = {} + + # Aggiorna canali + 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 + } + + time.sleep(0.05) + + # Aggiorna mix + for mix in range(1, TF5_MIX_BUSSES + 1): + mix_idx = mix - 1 + + # Leggi nome + resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0") + name = self._parse_name(resp_name) + + # Leggi stato ON/OFF + resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0") + is_on = self._parse_value(resp_on) == "1" + + # Leggi livello + resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_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 + + mixes_data[str(mix)] = { + "mix": mix, + "name": name, + "on": is_on, + "level_db": level_db + } + + time.sleep(0.05) + + self._cache = { + "channels": channels_data, + "mixes": mixes_data, + "timestamp": time.time() + } + self._save_cache() + + print(f"✅ Cache aggiornata con {len(channels_data)} canali e {len(mixes_data)} mix") + return { + "status": "success", + "message": f"Cache aggiornata con {len(channels_data)} canali e {len(mixes_data)} mix", + "channels_count": len(channels_data), + "mixes_count": len(mixes_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 dal mixer per avere dati precisi + if "OK" in response: + self._update_channel_cache(channel) + + 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 dal mixer per avere dati precisi + if "OK" in response: + self._update_channel_cache(channel) + + 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 dal mixer per avere dati precisi + if "OK" in response: + self._update_channel_cache(channel) + + 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) + + # Aggiorna cache dal mixer per avere dati precisi + if "OK" in response: + self._update_mix_cache(mix_number) + + return { + "status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} impostato a {level_db:+.1f} dB", + "response": response + } + + def set_mix_on_off(self, mix_number: int, state: bool) -> dict: + """Accende o spegne un mix/aux. + + Args: + mix_number: Numero del mix (1-20) + state: True per accendere, False per spegnere + + 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}"} + + value = 1 if state else 0 + command = f"set MIXER:Current/Mix/Fader/On {mix_number-1} 0 {value}" + response = self._send_command(command) + + # Aggiorna cache dal mixer per avere dati precisi + if "OK" in response: + self._update_mix_cache(mix_number) + + return { + "status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} {'acceso' if state else 'spento'}", + "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 get_mix_info(self, mix_number: int, force_refresh: bool = False) -> dict: + """Legge le informazioni di un mix/aux (nome, livello, stato). + Usa la cache se disponibile e valida. + + Args: + mix_number: Numero del mix (1-20) + force_refresh: Se True, ignora la cache e legge dal mixer + + Returns: + Un dizionario con tutte le informazioni del mix + """ + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + + # Usa cache se valida e non forzato il refresh + if not force_refresh and self._is_cache_valid(): + cached_data = self._cache.get("mixes", {}).get(str(mix_number)) + if cached_data: + return { + "status": "success", + "source": "cache", + **cached_data + } + + # Altrimenti leggi dal mixer + mix_idx = mix_number - 1 + + resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0") + name = self._parse_name(resp_name) + + resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0") + is_on = self._parse_value(resp_on) == "1" + + resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_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 + + return { + "status": "success", + "source": "mixer", + "mix": mix_number, + "name": name, + "on": is_on, + "level_db": level_db + } + + 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 search_mixes_by_name(self, search_term: str) -> dict: + """Cerca mix/aux il cui nome contiene un determinato termine. + Usa la cache per velocizzare la ricerca. + + Args: + search_term: Il termine da cercare nei nomi dei mix (case-insensitive) + + Returns: + Un dizionario con la lista dei mix trovati + """ + # Aggiorna cache se non valida + if not self._is_cache_valid(): + self.refresh_cache() + + search_lower = search_term.lower() + found_mixes = [] + + for mix_str, info in self._cache.get("mixes", {}).items(): + if search_lower in info.get("name", "").lower(): + found_mixes.append({ + "mix": info["mix"], + "name": info["name"], + "on": info["on"], + "level_db": info["level_db"] + }) + + # Ordina per numero mix + found_mixes.sort(key=lambda x: x["mix"]) + + return { + "status": "success", + "search_term": search_term, + "found_count": len(found_mixes), + "mixes": found_mixes, + "message": f"Trovati {len(found_mixes)} mix 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)" + } + + def get_all_mixes_summary(self) -> dict: + """Ottiene un riepilogo di tutti i mix/aux con nome e stato. + Usa la cache per velocizzare. + + Returns: + Un dizionario con il riepilogo di tutti i mix + """ + # Aggiorna cache se non valida + if not self._is_cache_valid(): + self.refresh_cache() + + mixes = [] + for mix_str, info in self._cache.get("mixes", {}).items(): + mixes.append({ + "mix": info["mix"], + "name": info["name"], + "on": info["on"] + }) + + # Ordina per numero mix + mixes.sort(key=lambda x: x["mix"]) + + cache_age = time.time() - self._cache.get("timestamp", 0) + + return { + "status": "success", + "total_mixes": len(mixes), + "mixes": mixes, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(mixes)} mix (cache: {int(cache_age)}s fa)" + } -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