diff --git a/.tf5_mixer_cache/channels_cache.json b/.tf5_mixer_cache/channels_cache.json index ef6a8da..3697599 100644 --- a/.tf5_mixer_cache/channels_cache.json +++ b/.tf5_mixer_cache/channels_cache.json @@ -403,5 +403,12 @@ "level_db": 6.0 } }, - "timestamp": 1770577942.7806144 + "timestamp": 1770577942.7806144, + "steinch": {}, + "fxrtn": {}, + "dcas": {}, + "matrices": {}, + "stereo": {}, + "mono": {}, + "mute_masters": {} } \ No newline at end of file diff --git a/.tf5_mixer_cache/mixer_ip.json b/.tf5_mixer_cache/mixer_ip.json index e39c941..51b1087 100644 --- a/.tf5_mixer_cache/mixer_ip.json +++ b/.tf5_mixer_cache/mixer_ip.json @@ -1,4 +1,4 @@ { "ip": "192.168.1.59", - "timestamp": 1771271281.9053261 + "timestamp": 1771271930.0475464 } \ No newline at end of file diff --git a/__pycache__/mixer_controller.cpython-312.pyc b/__pycache__/mixer_controller.cpython-312.pyc index 9f5e7d3..4e91b65 100644 Binary files a/__pycache__/mixer_controller.cpython-312.pyc and b/__pycache__/mixer_controller.cpython-312.pyc differ diff --git a/mixer_controller.py b/mixer_controller.py index 275ec24..58c21a0 100644 --- a/mixer_controller.py +++ b/mixer_controller.py @@ -1,5 +1,3 @@ -# mixer_controller.py - #!/usr/bin/env python # -*- coding: utf-8 -*- import socket @@ -46,7 +44,7 @@ class TF5MixerController: self._reader_thread = threading.Thread(target=self._connection_manager) self._reader_thread.daemon = True self._reader_thread.start() - + if not self._is_connected.wait(timeout=5): print("⚠️ Impossibile connettersi al mixer all'avvio.") else: @@ -65,7 +63,7 @@ class TF5MixerController: with self._cache_lock: is_cache_empty = not self._cache.get("channels") - + if is_cache_empty: print("Cache iniziale vuota. Avvio refresh completo...") self.refresh_cache() @@ -117,12 +115,13 @@ class TF5MixerController: def _handle_notify(self, message: str): """ Analizza un messaggio NOTIFY e aggiorna la cache in modo robusto. - Versione che gestisce il formato esatto 'NOTIFY set ...' con parametri extra. + Gestisce InCh, StInCh, FxRtnCh, DCA, Mix, Mtrx, St, Mono, MuteMaster. """ print(f"RECV NOTIFY: {message}") - + parts = message.split() - if len(parts) < 2: return + if len(parts) < 2: + return path = None path_index = -1 @@ -131,9 +130,9 @@ class TF5MixerController: path = part path_index = i break - + if path is None: - print(f" -> ATTENZIONE: Nessun path valido (che inizia con 'MIXER:') trovato nel messaggio.") + print(f" -> ATTENZIONE: Nessun path valido trovato nel messaggio.") return print(f" -> Tentativo di acquisire il lock per l'aggiornamento (Path: {path})...") @@ -141,51 +140,227 @@ class TF5MixerController: print(f" -> LOCK ACQUISITO. Elaborazione...") try: updated = False - - # --- LOGICA DI PARSING SPECIFICA E ROBUSTA --- - # Per 'set' commands, i parametri sono: ch_idx, 0, value, [optional_string] - # Indice ch: path_index + 1 - # Indice valore: path_index + 3 - # Gestione NOME Canale (usa un parsing diverso) + # --- InCh --- if "InCh/Label/Name" in path: params = parts[path_index + 1:] if len(params) >= 2: ch_idx, name = int(params[0]), " ".join(params[1:]).strip('"') ch_key = str(ch_idx + 1) - - channel_data = self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)}) - channel_data["name"] = name - print(f" -> SUCCESS: Ch {ch_key} nome -> '{name}'") + self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)})["name"] = name + print(f" -> SUCCESS: InCh {ch_key} nome -> '{name}'") updated = True - # Gestione ON/OFF Canale elif "InCh/Fader/On" in path: if len(parts) > path_index + 3: ch_idx = int(parts[path_index + 1]) - state_val = parts[path_index + 3] - state = state_val == "1" + state = parts[path_index + 3] == "1" ch_key = str(ch_idx + 1) - - channel_data = self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)}) - channel_data["on"] = state - print(f" -> SUCCESS: Ch {ch_key} stato -> {'ON' if state else 'OFF'}") + self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)})["on"] = state + print(f" -> SUCCESS: InCh {ch_key} stato -> {'ON' if state else 'OFF'}") updated = True - # Gestione LIVELLO Canale elif "InCh/Fader/Level" in path: if len(parts) > path_index + 3: ch_idx = int(parts[path_index + 1]) - level_int_str = parts[path_index + 3] - level_int = int(level_int_str) + level_int = int(parts[path_index + 3]) ch_key = str(ch_idx + 1) level_db = level_int / 100.0 if level_int > -32768 else float('-inf') - - channel_data = self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)}) - channel_data["level_db"] = level_db - print(f" -> SUCCESS: Ch {ch_key} level -> {level_db:.2f} dB") + self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)})["level_db"] = level_db + print(f" -> SUCCESS: InCh {ch_key} level -> {level_db:.2f} dB") updated = True - + + # --- StInCh --- + elif "StInCh/Label/Name" in path: + params = parts[path_index + 1:] + if len(params) >= 2: + ch_idx, name = int(params[0]), " ".join(params[1:]).strip('"') + ch_key = str(ch_idx + 1) + self._cache.setdefault("steinch", {}).setdefault(ch_key, {"channel": int(ch_key)})["name"] = name + print(f" -> SUCCESS: StInCh {ch_key} nome -> '{name}'") + updated = True + + elif "StInCh/Fader/On" in path: + if len(parts) > path_index + 3: + ch_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + ch_key = str(ch_idx + 1) + self._cache.setdefault("steinch", {}).setdefault(ch_key, {"channel": int(ch_key)})["on"] = state + print(f" -> SUCCESS: StInCh {ch_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "StInCh/Fader/Level" in path: + if len(parts) > path_index + 3: + ch_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + ch_key = str(ch_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("steinch", {}).setdefault(ch_key, {"channel": int(ch_key)})["level_db"] = level_db + print(f" -> SUCCESS: StInCh {ch_key} level -> {level_db:.2f} dB") + updated = True + + # --- FxRtnCh --- + elif "FxRtnCh/Label/Name" in path: + params = parts[path_index + 1:] + if len(params) >= 2: + ch_idx, name = int(params[0]), " ".join(params[1:]).strip('"') + ch_key = str(ch_idx + 1) + self._cache.setdefault("fxrtn", {}).setdefault(ch_key, {"channel": int(ch_key)})["name"] = name + print(f" -> SUCCESS: FxRtnCh {ch_key} nome -> '{name}'") + updated = True + + elif "FxRtnCh/Fader/On" in path: + if len(parts) > path_index + 3: + ch_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + ch_key = str(ch_idx + 1) + self._cache.setdefault("fxrtn", {}).setdefault(ch_key, {"channel": int(ch_key)})["on"] = state + print(f" -> SUCCESS: FxRtnCh {ch_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "FxRtnCh/Fader/Level" in path: + if len(parts) > path_index + 3: + ch_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + ch_key = str(ch_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("fxrtn", {}).setdefault(ch_key, {"channel": int(ch_key)})["level_db"] = level_db + print(f" -> SUCCESS: FxRtnCh {ch_key} level -> {level_db:.2f} dB") + updated = True + + # --- DCA --- + elif "DCA/Label/Name" in path: + params = parts[path_index + 1:] + if len(params) >= 2: + dca_idx, name = int(params[0]), " ".join(params[1:]).strip('"') + dca_key = str(dca_idx + 1) + self._cache.setdefault("dcas", {}).setdefault(dca_key, {"dca": int(dca_key)})["name"] = name + print(f" -> SUCCESS: DCA {dca_key} nome -> '{name}'") + updated = True + + elif "DCA/Fader/On" in path: + if len(parts) > path_index + 3: + dca_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + dca_key = str(dca_idx + 1) + self._cache.setdefault("dcas", {}).setdefault(dca_key, {"dca": int(dca_key)})["on"] = state + print(f" -> SUCCESS: DCA {dca_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "DCA/Fader/Level" in path: + if len(parts) > path_index + 3: + dca_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + dca_key = str(dca_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("dcas", {}).setdefault(dca_key, {"dca": int(dca_key)})["level_db"] = level_db + print(f" -> SUCCESS: DCA {dca_key} level -> {level_db:.2f} dB") + updated = True + + # --- Mix --- + elif "Mix/Label/Name" in path: + params = parts[path_index + 1:] + if len(params) >= 2: + mix_idx, name = int(params[0]), " ".join(params[1:]).strip('"') + mix_key = str(mix_idx + 1) + self._cache.setdefault("mixes", {}).setdefault(mix_key, {"mix": int(mix_key)})["name"] = name + print(f" -> SUCCESS: Mix {mix_key} nome -> '{name}'") + updated = True + + elif "Mix/Fader/On" in path: + if len(parts) > path_index + 3: + mix_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + mix_key = str(mix_idx + 1) + self._cache.setdefault("mixes", {}).setdefault(mix_key, {"mix": int(mix_key)})["on"] = state + print(f" -> SUCCESS: Mix {mix_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "Mix/Fader/Level" in path: + if len(parts) > path_index + 3: + mix_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + mix_key = str(mix_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("mixes", {}).setdefault(mix_key, {"mix": int(mix_key)})["level_db"] = level_db + print(f" -> SUCCESS: Mix {mix_key} level -> {level_db:.2f} dB") + updated = True + + # --- Mtrx --- + elif "Mtrx/Label/Name" in path: + params = parts[path_index + 1:] + if len(params) >= 2: + mtrx_idx, name = int(params[0]), " ".join(params[1:]).strip('"') + mtrx_key = str(mtrx_idx + 1) + self._cache.setdefault("matrices", {}).setdefault(mtrx_key, {"matrix": int(mtrx_key)})["name"] = name + print(f" -> SUCCESS: Mtrx {mtrx_key} nome -> '{name}'") + updated = True + + elif "Mtrx/Fader/On" in path: + if len(parts) > path_index + 3: + mtrx_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + mtrx_key = str(mtrx_idx + 1) + self._cache.setdefault("matrices", {}).setdefault(mtrx_key, {"matrix": int(mtrx_key)})["on"] = state + print(f" -> SUCCESS: Mtrx {mtrx_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "Mtrx/Fader/Level" in path: + if len(parts) > path_index + 3: + mtrx_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + mtrx_key = str(mtrx_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("matrices", {}).setdefault(mtrx_key, {"matrix": int(mtrx_key)})["level_db"] = level_db + print(f" -> SUCCESS: Mtrx {mtrx_key} level -> {level_db:.2f} dB") + updated = True + + # --- Stereo Bus --- + elif "St/Fader/On" in path: + if len(parts) > path_index + 3: + st_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + st_key = str(st_idx + 1) + self._cache.setdefault("stereo", {}).setdefault(st_key, {"bus": int(st_key)})["on"] = state + print(f" -> SUCCESS: St {st_key} stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "St/Fader/Level" in path: + if len(parts) > path_index + 3: + st_idx = int(parts[path_index + 1]) + level_int = int(parts[path_index + 3]) + st_key = str(st_idx + 1) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("stereo", {}).setdefault(st_key, {"bus": int(st_key)})["level_db"] = level_db + print(f" -> SUCCESS: St {st_key} level -> {level_db:.2f} dB") + updated = True + + # --- Mono Bus --- + elif "Mono/Fader/On" in path: + if len(parts) > path_index + 3: + state = parts[path_index + 3] == "1" + self._cache.setdefault("mono", {})["on"] = state + print(f" -> SUCCESS: Mono stato -> {'ON' if state else 'OFF'}") + updated = True + + elif "Mono/Fader/Level" in path: + if len(parts) > path_index + 3: + level_int = int(parts[path_index + 3]) + level_db = level_int / 100.0 if level_int > -32768 else float('-inf') + self._cache.setdefault("mono", {})["level_db"] = level_db + print(f" -> SUCCESS: Mono level -> {level_db:.2f} dB") + updated = True + + # --- MuteMaster --- + elif "MuteMaster/On" in path: + if len(parts) > path_index + 3: + group_idx = int(parts[path_index + 1]) + state = parts[path_index + 3] == "1" + group_key = str(group_idx + 1) + self._cache.setdefault("mute_masters", {}).setdefault(group_key, {"group": int(group_key)})["on"] = state + print(f" -> SUCCESS: MuteMaster {group_key} stato -> {'ON' if state else 'OFF'}") + updated = True + if updated: self._cache['timestamp'] = time.time() else: @@ -195,7 +370,7 @@ class TF5MixerController: print(f" -> ERRORE CRITICO durante l'elaborazione: {e}") finally: print(f" -> LOCK RILASCIATO.") - + def _send_command(self, command: str) -> str: """Invia un comando al mixer e attende la risposta.""" if not self._is_connected.wait(timeout=5): @@ -208,7 +383,7 @@ class TF5MixerController: print(f"SEND: {command}") self.socket.sendall((command + '\n').encode('utf-8')) - + response = self._response_queue.get(timeout=5) print(f"RECV RESP: {response}") return response @@ -242,7 +417,10 @@ class TF5MixerController: def __exit__(self, exc_type, exc_val, exc_tb): self.close() - # --- Metodi di gestione della cache e di utilità (invariati) --- + # ========================================================================= + # UTILITY E CACHE + # ========================================================================= + def _sanitize_value(self, value): 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 @@ -259,13 +437,14 @@ class TF5MixerController: try: with open(CACHE_FILE, 'r', encoding='utf-8') as f: data = json.load(f) - # Assicura che le chiavi principali esistano - if "channels" not in data: data["channels"] = {} - if "mixes" not in data: data["mixes"] = {} + for key in ("channels", "mixes", "steinch", "fxrtn", "dcas", "matrices", "stereo", "mono", "mute_masters"): + if key not in data: + data[key] = {} if key != "mono" else {} return data except Exception as e: print(f"⚠️ Errore nel caricamento della cache: {e}") - return {"channels": {}, "mixes": {}, "timestamp": 0} + return {"channels": {}, "mixes": {}, "steinch": {}, "fxrtn": {}, "dcas": {}, + "matrices": {}, "stereo": {}, "mono": {}, "mute_masters": {}, "timestamp": 0} def _save_cache(self): with self._cache_lock: @@ -293,7 +472,7 @@ class TF5MixerController: def _normalize_level_from_cache(self, level_value): if level_value is None: return float('-inf') return level_value - + def _parse_name(self, response: str) -> str: try: first_line = response.split('\n')[0] @@ -308,248 +487,794 @@ class TF5MixerController: parts = first_line.split() if len(parts) > 0 and parts[0] == "OK": return parts[-1] return "N/A" - - # --- API Pubblica del Controller (invariata) --- - # ... tutti gli altri metodi da refresh_cache in poi rimangono identici ... + + def _level_to_internal(self, level_db: float) -> int: + """Converte dB in valore interno intero.""" + return -32768 if level_db <= -138 else int(level_db * 100) + + def _internal_to_level(self, raw: str): + """Converte stringa valore interno in dB float.""" + try: + v = int(raw) + return v / 100.0 if v > -32768 else float('-inf') + except: + return None + + # ========================================================================= + # REFRESH CACHE COMPLETO + # ========================================================================= + def refresh_cache(self) -> dict: print("🔄 Aggiornamento cache completo in corso...") - channels_data = {} - mixes_data = {} + channels_data, mixes_data, steinch_data = {}, {}, {} + fxrtn_data, dcas_data, matrices_data = {}, {}, {} + stereo_data, mute_masters_data = {}, {} + mono_data = {} + # Input Channels for ch in range(1, TF5_INPUT_CHANNELS + 1): ch_idx = ch - 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_db = int(level_raw) / 100.0 if int(level_raw) > -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) + name = self._parse_name(self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0"))) + pan_raw = self._parse_value(self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0")) 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.01) # Riduci il delay, la connessione persistente è veloce + time.sleep(0.01) + # Stereo Input Channels + for ch in range(1, TF5_ST_INPUT_CHANNELS + 1): + ch_idx = ch - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/StInCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/StInCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/StInCh/Fader/Level {ch_idx} 0"))) + pan_raw = self._parse_value(self._send_command(f"get MIXER:Current/StInCh/ToSt/Pan {ch_idx} 0")) + try: pan_value = int(pan_raw) + except: pan_value = None + steinch_data[str(ch)] = {"channel": ch, "name": name, "on": is_on, "level_db": level_db, "pan": pan_value} + time.sleep(0.01) + + # FX Return Channels + for ch in range(1, TF5_FX_RETURN_CHANNELS + 1): + ch_idx = ch - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/FxRtnCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/FxRtnCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/FxRtnCh/Fader/Level {ch_idx} 0"))) + fxrtn_data[str(ch)] = {"channel": ch, "name": name, "on": is_on, "level_db": level_db} + time.sleep(0.01) + + # DCA Groups + for dca in range(1, TF5_DCA_GROUPS + 1): + dca_idx = dca - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/DCA/Label/Name {dca_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/DCA/Fader/On {dca_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/DCA/Fader/Level {dca_idx} 0"))) + dcas_data[str(dca)] = {"dca": dca, "name": name, "on": is_on, "level_db": level_db} + time.sleep(0.01) + + # Mix Buses for mix in range(1, TF5_MIX_BUSSES + 1): mix_idx = mix - 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_db = int(level_raw) / 100.0 if int(level_raw) > -32768 else float('-inf') - except: level_db = None + name = self._parse_name(self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_idx} 0"))) mixes_data[str(mix)] = {"mix": mix, "name": name, "on": is_on, "level_db": level_db} time.sleep(0.01) + # Matrix Buses + for mtrx in range(1, TF5_MATRIX_BUSSES + 1): + mtrx_idx = mtrx - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/Mtrx/Label/Name {mtrx_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/Mtrx/Fader/On {mtrx_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/Mtrx/Fader/Level {mtrx_idx} 0"))) + matrices_data[str(mtrx)] = {"matrix": mtrx, "name": name, "on": is_on, "level_db": level_db} + time.sleep(0.01) + + # Stereo Bus (max 2: L/R) + for st in range(1, TF5_STEREO_BUSSES + 1): + st_idx = st - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/St/Label/Name {st_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/St/Fader/On {st_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/St/Fader/Level {st_idx} 0"))) + stereo_data[str(st)] = {"bus": st, "name": name, "on": is_on, "level_db": level_db} + time.sleep(0.01) + + # Mono Bus + name = self._parse_name(self._send_command("get MIXER:Current/Mono/Label/Name 0 0")) + is_on = self._parse_value(self._send_command("get MIXER:Current/Mono/Fader/On 0 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command("get MIXER:Current/Mono/Fader/Level 0 0"))) + mono_data = {"name": name, "on": is_on, "level_db": level_db} + + # Mute Masters + for group in range(1, TF5_MUTE_GROUPS + 1): + group_idx = group - 1 + is_on = self._parse_value(self._send_command(f"get MIXER:Current/MuteMaster/On {group_idx} 0")) == "1" + name_raw = self._parse_name(self._send_command(f"get MIXER:Current/MuteMaster/Label/Name {group_idx} 0")) + mute_masters_data[str(group)] = {"group": group, "name": name_raw, "on": is_on} + time.sleep(0.01) + with self._cache_lock: - self._cache = {"channels": channels_data, "mixes": mixes_data, "timestamp": time.time()} + self._cache = { + "channels": channels_data, + "mixes": mixes_data, + "steinch": steinch_data, + "fxrtn": fxrtn_data, + "dcas": dcas_data, + "matrices": matrices_data, + "stereo": stereo_data, + "mono": mono_data, + "mute_masters": mute_masters_data, + "timestamp": time.time() + } self._save_cache() - - msg = f"Cache aggiornata con {len(channels_data)} canali e {len(mixes_data)} mix" + + msg = (f"Cache aggiornata: {len(channels_data)} InCh, {len(steinch_data)} StInCh, " + f"{len(fxrtn_data)} FxRtn, {len(dcas_data)} DCA, {len(mixes_data)} Mix, " + f"{len(matrices_data)} Mtrx, {len(stereo_data)} St, Mono, {len(mute_masters_data)} MuteMaster") print(f"✅ {msg}") - return {"status": "success", "message": msg, "channels_count": len(channels_data), "mixes_count": len(mixes_data)} + return {"status": "success", "message": msg} - # ... TUTTI GLI ALTRI METODI PUBBLICI RESTANO UGUALI ... - def recall_scene(self, bank: str, scene_number: int) -> dict: - 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) - - if "OK" in response: - print("Scene richiamata. La cache si aggiornerà tramite NOTIFY. Schedulato un refresh completo in 5s per sicurezza.") - threading.Timer(5.0, self.refresh_cache).start() - - return {"status": "success" if "OK" in response else "error", "message": f"Scena {bank.upper()}{scene_number} richiamata.", "response": response} + # ========================================================================= + # INPUT CHANNELS (InCh) + # ========================================================================= def set_channel_level(self, channel: int, level_db: float) -> dict: - if not 1 <= channel <= TF5_INPUT_CHANNELS: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} - internal_value = -32768 if level_db <= -138 else int(level_db * 100) - command = f"set MIXER:Current/InCh/Fader/Level {channel-1} 0 {internal_value}" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + command = f"set MIXER:Current/InCh/Fader/Level {channel-1} 0 {self._level_to_internal(level_db)}" response = self._send_command(command) - return {"status": "success" if "OK" in response else "error", "message": f"Canale {channel} impostato a {level_db:+.1f} dB", "response": response} + 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: - if not 1 <= channel <= TF5_INPUT_CHANNELS: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} command = f"set MIXER:Current/InCh/Fader/On {channel-1} 0 {1 if state else 0}" response = self._send_command(command) - return {"status": "success" if "OK" in response else "error", "message": f"Canale {channel} {'acceso' if state else 'spento'}", "response": response} + 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: - 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"} + 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) - 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} + else: pan_desc = "centro" + 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: - if not 1 <= mix_number <= TF5_MIX_BUSSES: return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} - internal_value = -32768 if level_db <= -138 else int(level_db * 100) - command = f"set MIXER:Current/Mix/Fader/Level {mix_number-1} 0 {internal_value}" + def set_channel_to_fx_level(self, channel: int, fx_number: int, level_db: float) -> dict: + """Imposta il livello di send di un canale di ingresso verso un processore FX.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not 1 <= fx_number <= TF5_FX_SLOTS: + return {"status": "error", "message": f"Il numero FX deve essere tra 1 e {TF5_FX_SLOTS}"} + command = f"set MIXER:Current/InCh/ToFx/Level {channel-1} {fx_number-1} {self._level_to_internal(level_db)}" 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} + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → FX {fx_number} impostato a {level_db:+.1f} dB", "response": response} - def set_mix_on_off(self, mix_number: int, state: bool) -> dict: - if not 1 <= mix_number <= TF5_MIX_BUSSES: return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} - command = f"set MIXER:Current/Mix/Fader/On {mix_number-1} 0 {1 if state else 0}" + def set_channel_to_fx_on_off(self, channel: int, fx_number: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale di ingresso verso un processore FX.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not 1 <= fx_number <= TF5_FX_SLOTS: + return {"status": "error", "message": f"Il numero FX deve essere tra 1 e {TF5_FX_SLOTS}"} + command = f"set MIXER:Current/InCh/ToFx/On {channel-1} {fx_number-1} {1 if state else 0}" response = self._send_command(command) - return {"status": "success" if "OK" in response else "error", "message": f"Mix {mix_number} {'acceso' if state else 'spento'}", "response": response} + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → FX {fx_number} send {'acceso' if state else 'spento'}", "response": response} - def mute_multiple_channels(self, channels: List[int]) -> dict: - results = [self.set_channel_on_off(ch, False) for ch in channels] - 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 set_channel_to_fx_prepost(self, channel: int, fx_number: int, post_fader: bool) -> dict: + """Imposta pre/post fader il send di un canale di ingresso verso un processore FX.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not 1 <= fx_number <= TF5_FX_SLOTS: + return {"status": "error", "message": f"Il numero FX deve essere tra 1 e {TF5_FX_SLOTS}"} + command = f"set MIXER:Current/InCh/ToFx/PrePost {channel-1} {fx_number-1} {1 if post_fader else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → FX {fx_number} impostato a {'post' if post_fader else 'pre'} fader", "response": response} - def unmute_multiple_channels(self, channels: List[int]) -> dict: - results = [self.set_channel_on_off(ch, True) for ch in channels] - 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 set_channel_to_mono_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello di send di un canale di ingresso verso il bus Mono.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + command = f"set MIXER:Current/InCh/ToMono/Level {channel-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → Mono impostato a {level_db:+.1f} dB", "response": response} + + def set_channel_to_mono_on_off(self, channel: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale di ingresso verso il bus Mono.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + command = f"set MIXER:Current/InCh/ToMono/On {channel-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → Mono send {'acceso' if state else 'spento'}", "response": response} + + def set_channel_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict: + if not 1 <= channel <= TF5_INPUT_CHANNELS: + 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}"} + command = f"set MIXER:Current/InCh/ToMix/Level {channel-1} {mix_number-1} {self._level_to_internal(level_db)}" + 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", "response": response} + + def set_channel_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict: + if not 1 <= channel <= TF5_INPUT_CHANNELS: + 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}"} + command = f"set MIXER:Current/InCh/ToMix/On {channel-1} {mix_number-1} {1 if state else 0}" + 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'}", "response": response} + + def set_channel_to_mix_pan(self, channel: int, mix_number: int, pan_value: int) -> dict: + """Imposta il pan del send di un canale verso un Mix bus.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + 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 not -63 <= pan_value <= 63: + return {"status": "error", "message": "Il pan deve essere tra -63 e +63"} + command = f"set MIXER:Current/InCh/ToMix/Pan {channel-1} {mix_number-1} {pan_value}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → Mix {mix_number} pan impostato a {pan_value}", "response": response} + + def set_channel_to_mix_prepost(self, channel: int, mix_number: int, post_fader: bool) -> dict: + """Imposta pre/post fader il send di un canale verso un Mix bus.""" + if not 1 <= channel <= TF5_INPUT_CHANNELS: + 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}"} + command = f"set MIXER:Current/InCh/ToMix/PrePost {channel-1} {mix_number-1} {1 if post_fader else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Canale {channel} → Mix {mix_number} impostato a {'post' if post_fader else 'pre'} fader", "response": response} def get_channel_info(self, channel: int, force_refresh: bool = False) -> dict: - if not 1 <= channel <= TF5_INPUT_CHANNELS: return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} + if not 1 <= channel <= TF5_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"} if not force_refresh and self._is_cache_valid(): with self._cache_lock: cached_data = self._cache.get("channels", {}).get(str(channel)) if cached_data: - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db"))) - return {"status": "success", "source": "cache", "channel": cached_data["channel"], "name": cached_data["name"], "on": cached_data["on"], "level_db": sanitized_level_db, "pan": cached_data.get("pan")} - + return {"status": "success", "source": "cache", + "channel": cached_data["channel"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db"))), + "pan": cached_data.get("pan")} + 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_db = int(level_raw) / 100.0 if int(level_raw) > -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) + name = self._parse_name(self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0"))) + pan_raw = self._parse_value(self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0")) try: pan_value = int(pan_raw) except: pan_value = None - - with self._cache_lock: - if "channels" not in self._cache: self._cache["channels"] = {} - self._cache["channels"][str(channel)] = {"channel": channel, "name": name, "on": is_on, "level_db": level_db, "pan": pan_value} - self._cache['timestamp'] = time.time() - - return {"status": "success", "source": "mixer", "channel": channel, "name": name, "on": is_on, "level_db": self._sanitize_value(level_db), "pan": pan_value} - def get_mix_info(self, mix_number: int, force_refresh: bool = False) -> dict: - if not 1 <= mix_number <= TF5_MIX_BUSSES: return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} - if not force_refresh and self._is_cache_valid(): - with self._cache_lock: - cached_data = self._cache.get("mixes", {}).get(str(mix_number)) - if cached_data: - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db"))) - return {"status": "success", "source": "cache", "mix": cached_data["mix"], "name": cached_data["name"], "on": cached_data["on"], "level_db": sanitized_level_db} - - 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_db = int(level_raw) / 100.0 if int(level_raw) > -32768 else float('-inf') - except: level_db = None - with self._cache_lock: - if "mixes" not in self._cache: self._cache["mixes"] = {} - self._cache["mixes"][str(mix_number)] = {"mix": mix_number, "name": name, "on": is_on, "level_db": level_db} + self._cache.setdefault("channels", {})[str(channel)] = { + "channel": channel, "name": name, "on": is_on, "level_db": level_db, "pan": pan_value} self._cache['timestamp'] = time.time() - return {"status": "success", "source": "mixer", "mix": mix_number, "name": name, "on": is_on, "level_db": self._sanitize_value(level_db)} + return {"status": "success", "source": "mixer", "channel": channel, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db), "pan": pan_value} + + def get_channel_to_mix_info(self, channel: int, mix_number: int) -> dict: + if not 1 <= channel <= TF5_INPUT_CHANNELS: + 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, mix_idx = channel - 1, mix_number - 1 + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/InCh/ToMix/Level {ch_idx} {mix_idx}"))) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/InCh/ToMix/On {ch_idx} {mix_idx}")) == "1" + sanitized = self._sanitize_value(level_db) + return {"status": "success", "channel": channel, "mix": mix_number, "on": is_on, + "send_level_db": sanitized, + "message": f"Canale {channel} → Mix {mix_number}: {sanitized:+.1f} dB ({'ON' if is_on else 'OFF'})"} + + def get_all_channels_summary(self) -> dict: + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + channels_copy = list(self._cache.get("channels", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + channels = sorted([ + {"channel": info["channel"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in channels_copy + ], key=lambda x: x["channel"]) + 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 mute_multiple_channels(self, channels: List[int]) -> dict: + results = [self.set_channel_on_off(ch, False) for ch in channels] + 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: + results = [self.set_channel_on_off(ch, True) for ch in channels] + 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 search_channels_by_name(self, search_term: str) -> dict: if not self._is_cache_valid(): self.refresh_cache() search_lower = search_term.lower() - found_channels = [] with self._cache_lock: channels_copy = list(self._cache.get("channels", {}).values()) - for info in channels_copy: - if search_lower in info.get("name", "").lower(): - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(info.get("level_db"))) - found_channels.append({"channel": info["channel"], "name": info["name"], "on": info["on"], "level_db": sanitized_level_db}) - 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}'"} + found = sorted([ + {"channel": info["channel"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in channels_copy if search_lower in info.get("name", "").lower() + ], key=lambda x: x["channel"]) + return {"status": "success", "search_term": search_term, "found_count": len(found), "channels": found, + "message": f"Trovati {len(found)} canali contenenti '{search_term}'"} + + # ========================================================================= + # STEREO INPUT CHANNELS (StInCh) + # ========================================================================= + + def get_steinch_info(self, channel: int, force_refresh: bool = False) -> dict: + """Restituisce le informazioni di un canale stereo di ingresso.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("steinch", {}).get(str(channel)) + if cached_data: + return {"status": "success", "source": "cache", + "channel": cached_data["channel"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db"))), + "pan": cached_data.get("pan")} + + ch_idx = channel - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/StInCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/StInCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/StInCh/Fader/Level {ch_idx} 0"))) + pan_raw = self._parse_value(self._send_command(f"get MIXER:Current/StInCh/ToSt/Pan {ch_idx} 0")) + try: pan_value = int(pan_raw) + except: pan_value = None + + with self._cache_lock: + self._cache.setdefault("steinch", {})[str(channel)] = { + "channel": channel, "name": name, "on": is_on, "level_db": level_db, "pan": pan_value} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "channel": channel, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db), "pan": pan_value} + + def set_steinch_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello fader di un canale stereo di ingresso.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + command = f"set MIXER:Current/StInCh/Fader/Level {channel-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} impostato a {level_db:+.1f} dB", "response": response} + + def set_steinch_on_off(self, channel: int, state: bool) -> dict: + """Abilita/disabilita un canale stereo di ingresso.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + command = f"set MIXER:Current/StInCh/Fader/On {channel-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} {'acceso' if state else 'spento'}", "response": response} + + def set_steinch_pan(self, channel: int, pan_value: int) -> dict: + """Imposta il pan di un canale stereo di ingresso verso il bus Stereo.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_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/StInCh/ToSt/Pan {channel-1} 0 {pan_value}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} pan impostato a {pan_value}", "response": response} + + def set_steinch_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict: + """Imposta il livello di send di un canale stereo verso un Mix bus.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_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}"} + command = f"set MIXER:Current/StInCh/ToMix/Level {channel-1} {mix_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → Mix {mix_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_steinch_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale stereo verso un Mix bus.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_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}"} + command = f"set MIXER:Current/StInCh/ToMix/On {channel-1} {mix_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → Mix {mix_number} send {'acceso' if state else 'spento'}", "response": response} + + def set_steinch_to_mix_pan(self, channel: int, mix_number: int, pan_value: int) -> dict: + """Imposta il pan del send di un canale stereo verso un Mix bus.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_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 not -63 <= pan_value <= 63: + return {"status": "error", "message": "Il pan deve essere tra -63 e +63"} + command = f"set MIXER:Current/StInCh/ToMix/Pan {channel-1} {mix_number-1} {pan_value}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → Mix {mix_number} pan impostato a {pan_value}", "response": response} + + def set_steinch_to_fx_level(self, channel: int, fx_number: int, level_db: float) -> dict: + """Imposta il livello di send di un canale stereo verso un processore FX.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + if not 1 <= fx_number <= TF5_FX_SLOTS: + return {"status": "error", "message": f"Il numero FX deve essere tra 1 e {TF5_FX_SLOTS}"} + command = f"set MIXER:Current/StInCh/ToFx/Level {channel-1} {fx_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → FX {fx_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_steinch_to_fx_on_off(self, channel: int, fx_number: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale stereo verso un processore FX.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + if not 1 <= fx_number <= TF5_FX_SLOTS: + return {"status": "error", "message": f"Il numero FX deve essere tra 1 e {TF5_FX_SLOTS}"} + command = f"set MIXER:Current/StInCh/ToFx/On {channel-1} {fx_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → FX {fx_number} send {'acceso' if state else 'spento'}", "response": response} + + def set_steinch_to_mono_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello di send di un canale stereo verso il bus Mono.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + command = f"set MIXER:Current/StInCh/ToMono/Level {channel-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → Mono impostato a {level_db:+.1f} dB", "response": response} + + def set_steinch_to_mono_on_off(self, channel: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale stereo verso il bus Mono.""" + if not 1 <= channel <= TF5_ST_INPUT_CHANNELS: + return {"status": "error", "message": f"Il canale stereo deve essere tra 1 e {TF5_ST_INPUT_CHANNELS}"} + command = f"set MIXER:Current/StInCh/ToMono/On {channel-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"StInCh {channel} → Mono send {'acceso' if state else 'spento'}", "response": response} + + def get_all_steinch_summary(self) -> dict: + """Restituisce il riepilogo di tutti i canali stereo di ingresso dalla cache.""" + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + steinch_copy = list(self._cache.get("steinch", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + channels = sorted([ + {"channel": info["channel"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in steinch_copy + ], key=lambda x: x["channel"]) + return {"status": "success", "total_channels": len(channels), "channels": channels, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(channels)} canali stereo (cache: {int(cache_age)}s fa)"} + + # ========================================================================= + # FX RETURN CHANNELS (FxRtnCh) + # ========================================================================= + + def get_fxrtn_info(self, channel: int, force_refresh: bool = False) -> dict: + """Restituisce le informazioni di un canale di ritorno FX.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("fxrtn", {}).get(str(channel)) + if cached_data: + return {"status": "success", "source": "cache", + "channel": cached_data["channel"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + ch_idx = channel - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/FxRtnCh/Label/Name {ch_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/FxRtnCh/Fader/On {ch_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/FxRtnCh/Fader/Level {ch_idx} 0"))) + + with self._cache_lock: + self._cache.setdefault("fxrtn", {})[str(channel)] = { + "channel": channel, "name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "channel": channel, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_fxrtn_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello fader di un canale FX return.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + command = f"set MIXER:Current/FxRtnCh/Fader/Level {channel-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} impostato a {level_db:+.1f} dB", "response": response} + + def set_fxrtn_on_off(self, channel: int, state: bool) -> dict: + """Abilita/disabilita un canale FX return.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + command = f"set MIXER:Current/FxRtnCh/Fader/On {channel-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} {'acceso' if state else 'spento'}", "response": response} + + def set_fxrtn_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict: + """Imposta il livello di send di un canale FX return verso un Mix bus.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + command = f"set MIXER:Current/FxRtnCh/ToMix/Level {channel-1} {mix_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → Mix {mix_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_fxrtn_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale FX return verso un Mix bus.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + command = f"set MIXER:Current/FxRtnCh/ToMix/On {channel-1} {mix_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → Mix {mix_number} send {'acceso' if state else 'spento'}", "response": response} + + def set_fxrtn_to_mix_pan(self, channel: int, mix_number: int, pan_value: int) -> dict: + """Imposta il pan del send di un canale FX return verso un Mix bus.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_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 not -63 <= pan_value <= 63: + return {"status": "error", "message": "Il pan deve essere tra -63 e +63"} + command = f"set MIXER:Current/FxRtnCh/ToMix/Pan {channel-1} {mix_number-1} {pan_value}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → Mix {mix_number} pan impostato a {pan_value}", "response": response} + + def set_fxrtn_to_mono_level(self, channel: int, level_db: float) -> dict: + """Imposta il livello di send di un canale FX return verso il bus Mono.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + command = f"set MIXER:Current/FxRtnCh/ToMono/Level {channel-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → Mono impostato a {level_db:+.1f} dB", "response": response} + + def set_fxrtn_to_mono_on_off(self, channel: int, state: bool) -> dict: + """Abilita/disabilita il send di un canale FX return verso il bus Mono.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + command = f"set MIXER:Current/FxRtnCh/ToMono/On {channel-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → Mono send {'acceso' if state else 'spento'}", "response": response} + + def set_fxrtn_to_st_pan(self, channel: int, pan_value: int) -> dict: + """Imposta il pan di un canale FX return verso il bus Stereo.""" + if not 1 <= channel <= TF5_FX_RETURN_CHANNELS: + return {"status": "error", "message": f"Il canale FX return deve essere tra 1 e {TF5_FX_RETURN_CHANNELS}"} + if not -63 <= pan_value <= 63: + return {"status": "error", "message": "Il pan deve essere tra -63 e +63"} + command = f"set MIXER:Current/FxRtnCh/ToSt/Pan {channel-1} 0 {pan_value}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"FxRtnCh {channel} → St pan impostato a {pan_value}", "response": response} + + def get_all_fxrtn_summary(self) -> dict: + """Restituisce il riepilogo di tutti i canali FX return dalla cache.""" + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + fxrtn_copy = list(self._cache.get("fxrtn", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + channels = sorted([ + {"channel": info["channel"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in fxrtn_copy + ], key=lambda x: x["channel"]) + return {"status": "success", "total_channels": len(channels), "channels": channels, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(channels)} canali FX return (cache: {int(cache_age)}s fa)"} + + # ========================================================================= + # DCA GROUPS + # ========================================================================= + + def get_dca_info(self, dca: int, force_refresh: bool = False) -> dict: + """Restituisce le informazioni di un gruppo DCA.""" + if not 1 <= dca <= TF5_DCA_GROUPS: + return {"status": "error", "message": f"Il DCA deve essere tra 1 e {TF5_DCA_GROUPS}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("dcas", {}).get(str(dca)) + if cached_data: + return {"status": "success", "source": "cache", + "dca": cached_data["dca"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + dca_idx = dca - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/DCA/Label/Name {dca_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/DCA/Fader/On {dca_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/DCA/Fader/Level {dca_idx} 0"))) + + with self._cache_lock: + self._cache.setdefault("dcas", {})[str(dca)] = { + "dca": dca, "name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "dca": dca, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_dca_level(self, dca: int, level_db: float) -> dict: + """Imposta il livello fader di un gruppo DCA.""" + if not 1 <= dca <= TF5_DCA_GROUPS: + return {"status": "error", "message": f"Il DCA deve essere tra 1 e {TF5_DCA_GROUPS}"} + command = f"set MIXER:Current/DCA/Fader/Level {dca-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"DCA {dca} impostato a {level_db:+.1f} dB", "response": response} + + def set_dca_on_off(self, dca: int, state: bool) -> dict: + """Abilita/disabilita un gruppo DCA.""" + if not 1 <= dca <= TF5_DCA_GROUPS: + return {"status": "error", "message": f"Il DCA deve essere tra 1 e {TF5_DCA_GROUPS}"} + command = f"set MIXER:Current/DCA/Fader/On {dca-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"DCA {dca} {'acceso' if state else 'spento'}", "response": response} + + def get_all_dca_summary(self) -> dict: + """Restituisce il riepilogo di tutti i gruppi DCA dalla cache.""" + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + dcas_copy = list(self._cache.get("dcas", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + dcas = sorted([ + {"dca": info["dca"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in dcas_copy + ], key=lambda x: x["dca"]) + return {"status": "success", "total_dcas": len(dcas), "dcas": dcas, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(dcas)} DCA (cache: {int(cache_age)}s fa)"} + + # ========================================================================= + # MIX BUSES + # ========================================================================= + + def get_mix_info(self, mix_number: int, force_refresh: bool = False) -> dict: + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("mixes", {}).get(str(mix_number)) + if cached_data: + return {"status": "success", "source": "cache", + "mix": cached_data["mix"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + mix_idx = mix_number - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_idx} 0"))) + + with self._cache_lock: + self._cache.setdefault("mixes", {})[str(mix_number)] = { + "mix": mix_number, "name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "mix": mix_number, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_mix_level(self, mix_number: int, level_db: float) -> dict: + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + command = f"set MIXER:Current/Mix/Fader/Level {mix_number-1} 0 {self._level_to_internal(level_db)}" + 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 set_mix_on_off(self, mix_number: int, state: bool) -> dict: + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + command = f"set MIXER:Current/Mix/Fader/On {mix_number-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} {'acceso' if state else 'spento'}", "response": response} + + def set_mix_out_balance(self, mix_number: int, balance: int) -> dict: + """Imposta il balance di uscita di un Mix bus stereo.""" + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + if not -63 <= balance <= 63: + return {"status": "error", "message": "Il balance deve essere tra -63 e +63"} + command = f"set MIXER:Current/Mix/Out/Balance {mix_number-1} 0 {balance}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} balance impostato a {balance}", "response": response} + + def set_mix_to_mtrx_level(self, mix_number: int, matrix_number: int, level_db: float) -> dict: + """Imposta il livello di send da un Mix bus verso un bus Matrix.""" + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mix/ToMtrx/Level {mix_number-1} {matrix_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} → Matrix {matrix_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_mix_to_mtrx_on_off(self, mix_number: int, matrix_number: int, state: bool) -> dict: + """Abilita/disabilita il send da un Mix bus verso un bus Matrix.""" + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mix/ToMtrx/On {mix_number-1} {matrix_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mix {mix_number} → Matrix {matrix_number} send {'acceso' if state else 'spento'}", "response": response} + + def get_all_mixes_summary(self) -> dict: + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + mixes_copy = list(self._cache.get("mixes", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + mixes = sorted([ + {"mix": info["mix"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in mixes_copy + ], key=lambda x: x["mix"]) + 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 search_mixes_by_name(self, search_term: str) -> dict: if not self._is_cache_valid(): self.refresh_cache() search_lower = search_term.lower() - found_mixes = [] with self._cache_lock: mixes_copy = list(self._cache.get("mixes", {}).values()) - for info in mixes_copy: - if search_lower in info.get("name", "").lower(): - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(info.get("level_db"))) - found_mixes.append({"mix": info["mix"], "name": info["name"], "on": info["on"], "level_db": sanitized_level_db}) - 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: - if not self._is_cache_valid(): self.refresh_cache() - channels = [] - with self._cache_lock: - channels_copy = list(self._cache.get("channels", {}).values()) - cache_age = time.time() - self._cache.get("timestamp", 0) - for info in channels_copy: - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(info.get("level_db"))) - channels.append({"channel": info["channel"], "name": info["name"], "on": info["on"], "level_db": sanitized_level_db}) - channels.sort(key=lambda x: x["channel"]) - 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: - if not self._is_cache_valid(): self.refresh_cache() - mixes = [] - with self._cache_lock: - mixes_copy = list(self._cache.get("mixes", {}).values()) - cache_age = time.time() - self._cache.get("timestamp", 0) - for info in mixes_copy: - sanitized_level_db = self._sanitize_value(self._normalize_level_from_cache(info.get("level_db"))) - mixes.append({"mix": info["mix"], "name": info["name"], "on": info["on"], "level_db": sanitized_level_db}) - mixes.sort(key=lambda x: x["mix"]) - 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 set_channel_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict: - if not 1 <= channel <= TF5_INPUT_CHANNELS: 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}"} - internal_value = -32768 if level_db <= -138 else 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", "response": response} - - def set_channel_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict: - if not 1 <= channel <= TF5_INPUT_CHANNELS: 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}"} - command = f"set MIXER:Current/InCh/ToMix/On {channel-1} {mix_number-1} {1 if state else 0}" - 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'}", "response": response} - - def get_channel_to_mix_info(self, channel: int, mix_number: int) -> dict: - if not 1 <= channel <= TF5_INPUT_CHANNELS: 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, mix_idx = channel - 1, mix_number - 1 - resp_level = self._send_command(f"get MIXER:Current/InCh/ToMix/Level {ch_idx} {mix_idx}") - level_raw = self._parse_value(resp_level) - try: level_db = int(level_raw) / 100.0 if int(level_raw) > -32768 else float('-inf') - except: level_db = None - 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, "send_level_db": sanitized_level_db, "message": f"Canale {channel} → Mix {mix_number}: {sanitized_level_db:+.1f} dB ({'ON' if is_on else 'OFF'})"} + found = sorted([ + {"mix": info["mix"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in mixes_copy if search_lower in info.get("name", "").lower() + ], key=lambda x: x["mix"]) + return {"status": "success", "search_term": search_term, "found_count": len(found), "mixes": found, + "message": f"Trovati {len(found)} mix contenenti '{search_term}'"} def get_full_mix_details(self, mix_number: int) -> dict: - if not 1 <= mix_number <= TF5_MIX_BUSSES: return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} + if not 1 <= mix_number <= TF5_MIX_BUSSES: + return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"} if not self._is_cache_valid(): self.refresh_cache() mix_info = self.get_mix_info(mix_number) if mix_info["status"] != "success": return mix_info @@ -559,6 +1284,302 @@ class TF5MixerController: if send_info["status"] == "success" and send_info["send_level_db"] > -120.0: with self._cache_lock: channel_cache = self._cache.get("channels", {}).get(str(channel), {}) - sends.append({"channel": channel, "channel_name": channel_cache.get("name", "Sconosciuto"), "channel_is_on": channel_cache.get("on", False), "send_level_db": send_info["send_level_db"], "send_is_on": send_info["on"]}) + sends.append({"channel": channel, + "channel_name": channel_cache.get("name", "Sconosciuto"), + "channel_is_on": channel_cache.get("on", False), + "send_level_db": send_info["send_level_db"], + "send_is_on": send_info["on"]}) sends.sort(key=lambda x: x["send_level_db"], reverse=True) - return {"status": "success", "mix_details": mix_info, "active_sends": sends, "message": f"Dettagli completi per il Mix {mix_number} recuperati."} \ No newline at end of file + return {"status": "success", "mix_details": mix_info, "active_sends": sends, + "message": f"Dettagli completi per il Mix {mix_number} recuperati."} + + # ========================================================================= + # MATRIX BUSES (Mtrx) + # ========================================================================= + + def get_mtrx_info(self, matrix: int, force_refresh: bool = False) -> dict: + """Restituisce le informazioni di un bus Matrix.""" + if not 1 <= matrix <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("matrices", {}).get(str(matrix)) + if cached_data: + return {"status": "success", "source": "cache", + "matrix": cached_data["matrix"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + mtrx_idx = matrix - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/Mtrx/Label/Name {mtrx_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/Mtrx/Fader/On {mtrx_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/Mtrx/Fader/Level {mtrx_idx} 0"))) + + with self._cache_lock: + self._cache.setdefault("matrices", {})[str(matrix)] = { + "matrix": matrix, "name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "matrix": matrix, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_mtrx_level(self, matrix: int, level_db: float) -> dict: + """Imposta il livello fader di un bus Matrix.""" + if not 1 <= matrix <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mtrx/Fader/Level {matrix-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Matrix {matrix} impostata a {level_db:+.1f} dB", "response": response} + + def set_mtrx_on_off(self, matrix: int, state: bool) -> dict: + """Abilita/disabilita un bus Matrix.""" + if not 1 <= matrix <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mtrx/Fader/On {matrix-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Matrix {matrix} {'accesa' if state else 'spenta'}", "response": response} + + def get_all_mtrx_summary(self) -> dict: + """Restituisce il riepilogo di tutti i bus Matrix dalla cache.""" + if not self._is_cache_valid(): self.refresh_cache() + with self._cache_lock: + matrices_copy = list(self._cache.get("matrices", {}).values()) + cache_age = time.time() - self._cache.get("timestamp", 0) + matrices = sorted([ + {"matrix": info["matrix"], "name": info["name"], "on": info["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))} + for info in matrices_copy + ], key=lambda x: x["matrix"]) + return {"status": "success", "total_matrices": len(matrices), "matrices": matrices, + "cache_age_seconds": int(cache_age), + "message": f"Riepilogo di {len(matrices)} Matrix bus (cache: {int(cache_age)}s fa)"} + + # ========================================================================= + # STEREO BUS (St) + # ========================================================================= + + def get_stereo_info(self, bus: int = 1, force_refresh: bool = False) -> dict: + """Restituisce le informazioni di un bus Stereo (1=L, 2=R).""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("stereo", {}).get(str(bus)) + if cached_data: + return {"status": "success", "source": "cache", + "bus": cached_data["bus"], "name": cached_data["name"], + "on": cached_data["on"], + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + bus_idx = bus - 1 + name = self._parse_name(self._send_command(f"get MIXER:Current/St/Label/Name {bus_idx} 0")) + is_on = self._parse_value(self._send_command(f"get MIXER:Current/St/Fader/On {bus_idx} 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command(f"get MIXER:Current/St/Fader/Level {bus_idx} 0"))) + + with self._cache_lock: + self._cache.setdefault("stereo", {})[str(bus)] = { + "bus": bus, "name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "bus": bus, "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_stereo_level(self, bus: int, level_db: float) -> dict: + """Imposta il livello fader del bus Stereo.""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + command = f"set MIXER:Current/St/Fader/Level {bus-1} 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"St bus {bus} impostato a {level_db:+.1f} dB", "response": response} + + def set_stereo_on_off(self, bus: int, state: bool) -> dict: + """Abilita/disabilita il bus Stereo.""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + command = f"set MIXER:Current/St/Fader/On {bus-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"St bus {bus} {'acceso' if state else 'spento'}", "response": response} + + def set_stereo_balance(self, bus: int, balance: int) -> dict: + """Imposta il balance di uscita del bus Stereo.""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + if not -63 <= balance <= 63: + return {"status": "error", "message": "Il balance deve essere tra -63 e +63"} + command = f"set MIXER:Current/St/Out/Balance {bus-1} 0 {balance}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"St bus {bus} balance impostato a {balance}", "response": response} + + def set_stereo_to_mtrx_level(self, bus: int, matrix_number: int, level_db: float) -> dict: + """Imposta il livello di send dal bus Stereo verso un bus Matrix.""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/St/ToMtrx/Level {bus-1} {matrix_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"St bus {bus} → Matrix {matrix_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_stereo_to_mtrx_on_off(self, bus: int, matrix_number: int, state: bool) -> dict: + """Abilita/disabilita il send dal bus Stereo verso un bus Matrix.""" + if not 1 <= bus <= TF5_STEREO_BUSSES: + return {"status": "error", "message": f"Il bus stereo deve essere tra 1 e {TF5_STEREO_BUSSES}"} + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/St/ToMtrx/On {bus-1} {matrix_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"St bus {bus} → Matrix {matrix_number} send {'acceso' if state else 'spento'}", "response": response} + + # ========================================================================= + # MONO BUS + # ========================================================================= + + def get_mono_info(self, force_refresh: bool = False) -> dict: + """Restituisce le informazioni del bus Mono.""" + if not force_refresh and self._is_cache_valid(): + with self._cache_lock: + cached_data = self._cache.get("mono", {}) + if cached_data: + return {"status": "success", "source": "cache", + "name": cached_data.get("name", "MONO"), + "on": cached_data.get("on", False), + "level_db": self._sanitize_value(self._normalize_level_from_cache(cached_data.get("level_db")))} + + name = self._parse_name(self._send_command("get MIXER:Current/Mono/Label/Name 0 0")) + is_on = self._parse_value(self._send_command("get MIXER:Current/Mono/Fader/On 0 0")) == "1" + level_db = self._internal_to_level(self._parse_value(self._send_command("get MIXER:Current/Mono/Fader/Level 0 0"))) + + with self._cache_lock: + self._cache["mono"] = {"name": name, "on": is_on, "level_db": level_db} + self._cache['timestamp'] = time.time() + + return {"status": "success", "source": "mixer", "name": name, "on": is_on, + "level_db": self._sanitize_value(level_db)} + + def set_mono_level(self, level_db: float) -> dict: + """Imposta il livello fader del bus Mono.""" + command = f"set MIXER:Current/Mono/Fader/Level 0 0 {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mono impostato a {level_db:+.1f} dB", "response": response} + + def set_mono_on_off(self, state: bool) -> dict: + """Abilita/disabilita il bus Mono.""" + command = f"set MIXER:Current/Mono/Fader/On 0 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mono {'acceso' if state else 'spento'}", "response": response} + + def set_mono_to_mtrx_level(self, matrix_number: int, level_db: float) -> dict: + """Imposta il livello di send dal bus Mono verso un bus Matrix.""" + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mono/ToMtrx/Level 0 {matrix_number-1} {self._level_to_internal(level_db)}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mono → Matrix {matrix_number} impostato a {level_db:+.1f} dB", "response": response} + + def set_mono_to_mtrx_on_off(self, matrix_number: int, state: bool) -> dict: + """Abilita/disabilita il send dal bus Mono verso un bus Matrix.""" + if not 1 <= matrix_number <= TF5_MATRIX_BUSSES: + return {"status": "error", "message": f"La matrix deve essere tra 1 e {TF5_MATRIX_BUSSES}"} + command = f"set MIXER:Current/Mono/ToMtrx/On 0 {matrix_number-1} {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mono → Matrix {matrix_number} send {'acceso' if state else 'spento'}", "response": response} + + # ========================================================================= + # MUTE MASTERS + # ========================================================================= + + def get_mute_master_state(self, group: int) -> dict: + """Restituisce lo stato di un Mute Master group.""" + if not 1 <= group <= TF5_MUTE_GROUPS: + return {"status": "error", "message": f"Il gruppo mute deve essere tra 1 e {TF5_MUTE_GROUPS}"} + with self._cache_lock: + cached = self._cache.get("mute_masters", {}).get(str(group)) + if cached: + return {"status": "success", "source": "cache", + "group": cached["group"], "name": cached.get("name", f"MUTE {group}"), + "on": cached["on"]} + + is_on = self._parse_value(self._send_command(f"get MIXER:Current/MuteMaster/On {group-1} 0")) == "1" + name = self._parse_name(self._send_command(f"get MIXER:Current/MuteMaster/Label/Name {group-1} 0")) + with self._cache_lock: + self._cache.setdefault("mute_masters", {})[str(group)] = {"group": group, "name": name, "on": is_on} + + return {"status": "success", "source": "mixer", "group": group, "name": name, "on": is_on} + + def set_mute_master(self, group: int, state: bool) -> dict: + """Attiva/disattiva un Mute Master group.""" + if not 1 <= group <= TF5_MUTE_GROUPS: + return {"status": "error", "message": f"Il gruppo mute deve essere tra 1 e {TF5_MUTE_GROUPS}"} + command = f"set MIXER:Current/MuteMaster/On {group-1} 0 {1 if state else 0}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Mute Master {group} {'attivato' if state else 'disattivato'}", "response": response} + + def get_all_mute_masters_summary(self) -> dict: + """Restituisce lo stato di tutti i Mute Master groups dalla cache.""" + groups = [] + for g in range(1, TF5_MUTE_GROUPS + 1): + info = self.get_mute_master_state(g) + if info["status"] == "success": + groups.append({"group": info["group"], "name": info["name"], "on": info["on"]}) + return {"status": "success", "total_groups": len(groups), "groups": groups, + "message": f"Riepilogo di {len(groups)} Mute Master groups"} + + # ========================================================================= + # SCENE MANAGEMENT + # ========================================================================= + + def recall_scene(self, bank: str, scene_number: int) -> dict: + """Richiama una scena da un banco (a o b).""" + 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) + + if "OK" in response: + print("Scene richiamata. La cache si aggiornerà tramite NOTIFY. Schedulato un refresh completo in 5s per sicurezza.") + threading.Timer(5.0, self.refresh_cache).start() + + return {"status": "success" if "OK" in response else "error", + "message": f"Scena {bank.upper()}{scene_number} richiamata.", "response": response} + + def recall_scene_inc(self) -> dict: + """Richiama la scena successiva nella sequenza.""" + response = self._send_command("set MIXER:Lib/Scene/RecallInc 0 0") + if "OK" in response: + threading.Timer(5.0, self.refresh_cache).start() + return {"status": "success" if "OK" in response else "error", + "message": "Scena successiva richiamata.", "response": response} + + def recall_scene_dec(self) -> dict: + """Richiama la scena precedente nella sequenza.""" + response = self._send_command("set MIXER:Lib/Scene/RecallDec 0 0") + if "OK" in response: + threading.Timer(5.0, self.refresh_cache).start() + return {"status": "success" if "OK" in response else "error", + "message": "Scena precedente richiamata.", "response": response} + + def store_scene(self, bank: str, scene_number: int) -> dict: + """Salva la scena corrente in un banco (a o b).""" + 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"set MIXER:Lib/Bank/Scene/Store {1 if bank.lower() == 'b' else 0} {scene_number}" + response = self._send_command(command) + return {"status": "success" if "OK" in response else "error", + "message": f"Scena salvata nel banco {bank.upper()}{scene_number}.", "response": response} \ No newline at end of file