diff --git a/.tf5_mixer_cache/channels_cache.json b/.tf5_mixer_cache/channels_cache.json index 2155302..1e52135 100644 --- a/.tf5_mixer_cache/channels_cache.json +++ b/.tf5_mixer_cache/channels_cache.json @@ -9,9 +9,9 @@ }, "2": { "channel": 2, - "name": "Gelato", + "name": "0.00", "on": false, - "level_db": null, + "level_db": 0.01, "pan": 0 }, "3": { diff --git a/__pycache__/mixer_controller.cpython-312.pyc b/__pycache__/mixer_controller.cpython-312.pyc index 3b30012..5352bad 100644 Binary files a/__pycache__/mixer_controller.cpython-312.pyc and b/__pycache__/mixer_controller.cpython-312.pyc differ diff --git a/conversations/4762864.json b/conversations/4762864.json index 2e4b56f..54a25b3 100644 --- a/conversations/4762864.json +++ b/conversations/4762864.json @@ -1,87 +1,135 @@ { "user_id": 4762864, "username": "unnikked", - "last_updated": 1762194065.7709692, - "last_updated_str": "2025-11-03 19:21:05", - "message_count": 13, + "last_updated": 1762194931.602051, + "last_updated_str": "2025-11-03 19:35:31", + "message_count": 21, "history": [ + { + "role": "assistant", + "content": "Fatto! Il canale \"Gelato\" dovrebbe essere acceso ora. Fammi sapere se è tutto ok! 🍦✅", + "timestamp": 1762194705.2052844, + "timestamp_str": "2025-11-03 19:31:45" + }, { "role": "user", - "content": "Me lo spegni il gelato", - "timestamp": 1761595234.0849943, - "timestamp_str": "2025-10-27 21:00:34" + "content": "Alzami un po' il gelato", + "timestamp": 1762194720.6919649, + "timestamp_str": "2025-11-03 19:32:00" }, { "role": "assistant", - "content": "Fatto! Canale \"Gelato\" spento. ✅", - "timestamp": 1761595239.1608367, - "timestamp_str": "2025-10-27 21:00:39" + "content": "Ho alzato un po' il volume del \"Gelato\" a -11.3 dB. 🍦🔊 Fammi sapere se va meglio così!", + "timestamp": 1762194727.1430511, + "timestamp_str": "2025-11-03 19:32:07" }, { "role": "user", - "content": "Mi accendi il gelato.", - "timestamp": 1761595383.8880353, - "timestamp_str": "2025-10-27 21:03:03" + "content": "Alza ancora il gelato", + "timestamp": 1762194746.9937618, + "timestamp_str": "2025-11-03 19:32:26" }, { "role": "assistant", - "content": "Certo! Gelato acceso! 🍦✅", - "timestamp": 1761595387.27931, - "timestamp_str": "2025-10-27 21:03:07" + "content": "Ancora un po' su per il \"Gelato\"! Ora è a -11.3 dB. 🍦🔊", + "timestamp": 1762194751.8532023, + "timestamp_str": "2025-11-03 19:32:31" }, { "role": "user", - "content": "Apaga el helado. Cappuccino.", - "timestamp": 1761595426.0112157, - "timestamp_str": "2025-10-27 21:03:46" + "content": "Alzami tutto il gelato", + "timestamp": 1762194765.1222134, + "timestamp_str": "2025-11-03 19:32:45" }, { "role": "assistant", - "content": "Gelato spento! ✅\nNon ho trovato nessun canale chiamato \"Cappuccino\", sei sicuro del nome? 🤔", - "timestamp": 1761595431.5373232, - "timestamp_str": "2025-10-27 21:03:51" + "content": "Non riesco a trovare un canale chiamato \"Gelato\" in questo momento. È possibile che il nome sia cambiato o che non sia più disponibile. Potresti darmi il numero del canale o un altro nome per favore? 🤔", + "timestamp": 1762194785.8956504, + "timestamp_str": "2025-11-03 19:33:05" }, { "role": "user", - "content": "Okay, aprimi il gelato allora!", - "timestamp": 1762193935.418264, - "timestamp_str": "2025-11-03 19:18:55" - }, - { - "role": "user", - "content": "Gli alzi il gelato?", - "timestamp": 1762193958.1497903, - "timestamp_str": "2025-11-03 19:19:18" - }, - { - "role": "user", - "content": "Aprimi il giubbotto.", - "timestamp": 1762194045.3195145, - "timestamp_str": "2025-11-03 19:20:45" + "content": "Riprovarci", + "timestamp": 1762194813.0377226, + "timestamp_str": "2025-11-03 19:33:33" }, { "role": "assistant", - "content": "Non posso aprirti il giubbotto, non è un canale o un mix del mixer! 😄\n\nPosso aiutarti con i canali audio, i microfoni o i monitor, se vuoi.", - "timestamp": 1762194046.8080468, - "timestamp_str": "2025-11-03 19:20:46" + "content": "Mi dispiace, ho provato di nuovo ma non riesco ancora a trovare nessun canale chiamato \"Gelato\". 🤔\n\nSe il nome è cambiato o se c'è un numero di canale, potresti dirmelo? Così posso aiutarti!", + "timestamp": 1762194819.4210932, + "timestamp_str": "2025-11-03 19:33:39" }, { "role": "user", - "content": "Aprimi il gelato", - "timestamp": 1762194053.0858698, - "timestamp_str": "2025-11-03 19:20:53" + "content": "Ch2", + "timestamp": 1762194830.624848, + "timestamp_str": "2025-11-03 19:33:50" }, { "role": "assistant", - "content": "Certo! Gelato acceso! 🍦✅", - "timestamp": 1762194054.6322277, - "timestamp_str": "2025-11-03 19:20:54" + "content": "Ok, ho alzato il canale 2 al massimo (+10 dB)! 🔊 Fammi sapere se va bene così!", + "timestamp": 1762194834.0072317, + "timestamp_str": "2025-11-03 19:33:54" }, { "role": "user", - "content": "È ancora chiuso", - "timestamp": 1762194065.7709692, - "timestamp_str": "2025-11-03 19:21:05" + "content": "Portalo a zero db", + "timestamp": 1762194846.3105323, + "timestamp_str": "2025-11-03 19:34:06" + }, + { + "role": "assistant", + "content": "Perfetto! Ho riportato il canale 2 a 0 dB. ✅🔊", + "timestamp": 1762194848.8269963, + "timestamp_str": "2025-11-03 19:34:08" + }, + { + "role": "user", + "content": "Analizzami l'equilibrio del mix diretta, che suggerimenti dai?", + "timestamp": 1762194868.3175187, + "timestamp_str": "2025-11-03 19:34:28" + }, + { + "role": "assistant", + "content": "Mi dispiace, ma non posso analizzare l'equilibrio del mix o darti suggerimenti su come suona. Sono un assistente per controllare il mixer, posso modificare i volumi, i pan, accendere/spegnere i canali e i mix, ma non ho \"orecchie\" per ascoltare e valutare il suono. 😅\n\nPosso aiutarti a regolare i livelli o il pan se mi dici cosa vuoi cambiare!", + "timestamp": 1762194871.190547, + "timestamp_str": "2025-11-03 19:34:31" + }, + { + "role": "user", + "content": "E sì sui volumi", + "timestamp": 1762194884.4026275, + "timestamp_str": "2025-11-03 19:34:44" + }, + { + "role": "assistant", + "content": "Capito! Sui volumi posso agire, ma non posso dirti io cosa suona bene o male, non ho le \"orecchie\" per quello. 😅\n\nSe mi dici quale canale o mix vuoi alzare o abbassare, e di quanto, lo faccio subito! Dimmi tu cosa vuoi cambiare. 🔊", + "timestamp": 1762194886.4130292, + "timestamp_str": "2025-11-03 19:34:46" + }, + { + "role": "user", + "content": "Non preoccuparti, dammi un giudizio sui valori se torna tutto", + "timestamp": 1762194903.6934974, + "timestamp_str": "2025-11-03 19:35:03" + }, + { + "role": "assistant", + "content": "Mi dispiace, ma non posso darti un giudizio sui valori o dirti se \"torna tutto\" a livello di suono. Non ho la capacità di ascoltare e valutare l'audio come farebbe una persona. 😅\n\nPosso solo eseguire i comandi che mi dai per modificare i volumi, il pan o accendere/spegnere i canali e i mix. Se mi dici cosa vuoi cambiare, lo faccio subito!", + "timestamp": 1762194906.4559038, + "timestamp_str": "2025-11-03 19:35:06" + }, + { + "role": "user", + "content": "Potresti gentilmente farmi questa caspita di analisi? Poi ti dico io cosa fare", + "timestamp": 1762194928.5493388, + "timestamp_str": "2025-11-03 19:35:28" + }, + { + "role": "assistant", + "content": "Mi dispiace davvero, ma non posso fare questa \"caspita di analisi\" che mi chiedi. 😔 Non ho le orecchie per ascoltare il suono e darti un giudizio su come sta andando il mix o se i valori \"tornano\". Sono un assistente per controllare il mixer, posso solo eseguire i comandi che mi dai per modificare i parametri.\n\nSe mi dici:\n- \"Alza il canale 5\"\n- \"Abbassa il mix 3\"\n- \"Metti il pan del canale 10 a sinistra\"\n- \"Accendi il microfono del cantante\"\n\n...allora posso aiutarti subito! Ma non posso dirti io cosa suona bene o male. Dimmi tu cosa vuoi cambiare sui volumi o sul bilanciamento! 🔊", + "timestamp": 1762194931.602051, + "timestamp_str": "2025-11-03 19:35:31" } ] } \ No newline at end of file diff --git a/mixer_controller.py b/mixer_controller.py index abd3275..833505d 100644 --- a/mixer_controller.py +++ b/mixer_controller.py @@ -1,15 +1,18 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- import socket import sys import os import json import time +import math from pathlib import Path from typing import List, Optional from google import genai from google.genai import types from dotenv import load_dotenv -load_dotenv() +load_dotenv() from config import * @@ -21,7 +24,23 @@ class TF5MixerController: self.socket = None self._ensure_cache_dir() self._cache = self._load_cache() - + + def _sanitize_value(self, value): + """ + Sanitizza i valori per renderli JSON-safe. + Converte -Infinity in -120 (silenzio pratico). + """ + if value is None: + return -120.0 + if value == float('-inf') or (isinstance(value, float) and math.isinf(value) and value < 0): + return -120.0 # Valore molto basso ma valido + elif value == float('inf') or (isinstance(value, float) and math.isinf(value) and value > 0): + return 10.0 # Massimo pratico + elif isinstance(value, float) and math.isnan(value): + return 0.0 # Default a 0 se NaN + + return value + def _connect(self): """Stabilisce la connessione se non già connesso.""" if self.socket is None: @@ -33,7 +52,7 @@ class TF5MixerController: 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: @@ -46,11 +65,11 @@ class TF5MixerController: 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}') @@ -60,33 +79,33 @@ class TF5MixerController: 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() - + 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(): @@ -96,18 +115,18 @@ class TF5MixerController: 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, convertendo -inf in null.""" try: # Crea una copia della cache per la serializzazione cache_to_save = self._prepare_cache_for_json(self._cache) - + with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(cache_to_save, f, indent=2, ensure_ascii=False) except Exception as e: print(f"⚠️ Errore nel salvataggio della cache: {e}") - + def _prepare_cache_for_json(self, obj): """Converte ricorsivamente -inf in null per la serializzazione JSON.""" if isinstance(obj, dict): @@ -121,39 +140,39 @@ class TF5MixerController: return obj else: return obj - + 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 _normalize_level_from_cache(self, level_value): """Normalizza il valore del livello dalla cache (gestisce None come -inf).""" if level_value is None: return float('-inf') return level_value - + 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) @@ -162,7 +181,7 @@ class TF5MixerController: 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) @@ -170,11 +189,11 @@ class TF5MixerController: 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, @@ -183,28 +202,28 @@ class TF5MixerController: "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) @@ -213,11 +232,11 @@ class TF5MixerController: 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, @@ -225,9 +244,9 @@ class TF5MixerController: "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: @@ -238,36 +257,36 @@ class TF5MixerController: 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) @@ -276,7 +295,7 @@ class TF5MixerController: 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) @@ -284,7 +303,7 @@ class TF5MixerController: pan_value = int(pan_raw) except: pan_value = None - + channels_data[str(ch)] = { "channel": ch, "name": name, @@ -292,21 +311,21 @@ class TF5MixerController: "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) @@ -315,23 +334,23 @@ class TF5MixerController: 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", @@ -339,14 +358,14 @@ class TF5MixerController: "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 """ @@ -354,85 +373,85 @@ class TF5MixerController: 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 """ @@ -440,90 +459,90 @@ class TF5MixerController: 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 """ @@ -531,20 +550,20 @@ class TF5MixerController: 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 """ @@ -552,53 +571,54 @@ class TF5MixerController: 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: # Normalizza level_db (None -> -inf) level_db = self._normalize_level_from_cache(cached_data.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) return { "status": "success", "source": "cache", "channel": cached_data["channel"], "name": cached_data["name"], "on": cached_data["on"], - "level_db": level_db, + "level_db": sanitized_level_db, "pan": cached_data.get("pan") } - + # 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: @@ -606,62 +626,64 @@ class TF5MixerController: 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 - + + sanitized_level_db = self._sanitize_value(level_db) return { "status": "success", "source": "mixer", "channel": channel, "name": name, "on": is_on, - "level_db": level_db, + "level_db": sanitized_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: # Normalizza level_db (None -> -inf) level_db = self._normalize_level_from_cache(cached_data.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) return { "status": "success", "source": "cache", "mix": cached_data["mix"], "name": cached_data["name"], "on": cached_data["on"], - "level_db": level_db + "level_db": sanitized_level_db } - + # 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: @@ -669,47 +691,49 @@ class TF5MixerController: level_db = level_int / 100.0 if level_int > -32768 else float('-inf') except: level_db = None - + + sanitized_level_db = self._sanitize_value(level_db) return { "status": "success", "source": "mixer", "mix": mix_number, "name": name, "on": is_on, - "level_db": level_db + "level_db": sanitized_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(): # Normalizza level_db quando restituisci i risultati level_db = self._normalize_level_from_cache(info.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) found_channels.append({ "channel": info["channel"], "name": info["name"], "on": info["on"], - "level_db": level_db + "level_db": sanitized_level_db }) - + # Ordina per numero canale found_channels.sort(key=lambda x: x["channel"]) - + return { "status": "success", "search_term": search_term, @@ -717,38 +741,39 @@ class TF5MixerController: "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(): # Normalizza level_db quando restituisci i risultati level_db = self._normalize_level_from_cache(info.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) found_mixes.append({ "mix": info["mix"], "name": info["name"], "on": info["on"], - "level_db": level_db + "level_db": sanitized_level_db }) - + # Ordina per numero mix found_mixes.sort(key=lambda x: x["mix"]) - + return { "status": "success", "search_term": search_term, @@ -756,31 +781,34 @@ class TF5MixerController: "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(): + level_db = self._normalize_level_from_cache(info.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) channels.append({ "channel": info["channel"], "name": info["name"], - "on": info["on"] + "on": info["on"], + "level_db": sanitized_level_db }) - + # 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), @@ -788,31 +816,34 @@ class TF5MixerController: "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(): + level_db = self._normalize_level_from_cache(info.get("level_db")) + sanitized_level_db = self._sanitize_value(level_db) mixes.append({ "mix": info["mix"], "name": info["name"], - "on": info["on"] + "on": info["on"], + "level_db": sanitized_level_db }) - + # 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), @@ -823,12 +854,12 @@ class TF5MixerController: def set_channel_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict: """Imposta il livello di invio di un canale verso un mix/aux. - + Args: channel: Numero del canale (1-40) 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 """ @@ -836,15 +867,15 @@ class TF5MixerController: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} 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/InCh/ToMix/Level {channel-1} {mix_number-1} {internal_value}" response = self._send_command(command) - + return { "status": "success" if "OK" in response else "error", "message": f"Canale {channel} → Mix {mix_number} impostato a {level_db:+.1f} dB", @@ -853,12 +884,12 @@ class TF5MixerController: def set_channel_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict: """Accende o spegne l'invio di un canale verso un mix/aux. - + Args: channel: Numero del canale (1-40) mix_number: Numero del mix (1-20) state: True per accendere, False per spegnere - + Returns: Un dizionario con lo stato dell'operazione """ @@ -866,11 +897,11 @@ class TF5MixerController: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} 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/InCh/ToMix/On {channel-1} {mix_number-1} {value}" response = self._send_command(command) - + return { "status": "success" if "OK" in response else "error", "message": f"Invio canale {channel} → Mix {mix_number} {'acceso' if state else 'spento'}", @@ -879,11 +910,11 @@ class TF5MixerController: def get_channel_to_mix_info(self, channel: int, mix_number: int) -> dict: """Legge le informazioni dell'invio di un canale verso un mix. - + Args: channel: Numero del canale (1-40) mix_number: Numero del mix (1-20) - + Returns: Un dizionario con le informazioni del send """ @@ -891,10 +922,10 @@ class TF5MixerController: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} if not 1 <= mix_number <= TF5_MIX_BUSSES: return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} - + ch_idx = channel - 1 mix_idx = mix_number - 1 - + # Leggi livello resp_level = self._send_command(f"get MIXER:Current/InCh/ToMix/Level {ch_idx} {mix_idx}") level_raw = self._parse_value(resp_level) @@ -903,16 +934,18 @@ class TF5MixerController: level_db = level_int / 100.0 if level_int > -32768 else float('-inf') except: level_db = None - + # Leggi stato ON/OFF resp_on = self._send_command(f"get MIXER:Current/InCh/ToMix/On {ch_idx} {mix_idx}") is_on = self._parse_value(resp_on) == "1" - + + sanitized_level_db = self._sanitize_value(level_db) + return { "status": "success", "channel": channel, "mix": mix_number, "on": is_on, - "level_db": level_db, - "message": f"Canale {channel} → Mix {mix_number}: {level_db:+.1f} dB ({'ON' if is_on else 'OFF'})" - } \ No newline at end of file + "send_level_db": sanitized_level_db, + "message": f"Canale {channel} → Mix {mix_number}: {sanitized_level_db:+.1f} dB ({'ON' if is_on else 'OFF'})" + }