#!/usr/bin/env python # -*- coding: utf-8 -*- import socket import sys import os import json import time import math import threading import queue from pathlib import Path from typing import List, Optional from config import * class TF5MixerController: """ Controllore per mixer Yamaha TF5 con connessione socket persistente e gestione dei messaggi NOTIFY per l'aggiornamento della cache in tempo reale. """ def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT): self.host = host self.port = port self.socket = None self._ensure_cache_dir() self._cache_lock = threading.Lock() self._cache = self._load_cache() self._command_lock = threading.Lock() self._response_queue = queue.Queue(maxsize=1) self._reader_thread = None self._is_running = threading.Event() self._is_connected = threading.Event() self.start() def start(self): if self._is_running.is_set(): print("Controller giΓ  in esecuzione.") return print("πŸš€ Avvio del Mixer Controller...") self._is_running.set() 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: print("βœ… Controller avviato e connesso.") def _connection_manager(self): while self._is_running.is_set(): try: print(f"πŸ”Œ Tentativo di connessione a {self.host}:{self.port}...") self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(10) self.socket.connect((self.host, self.port)) self.socket.settimeout(1) self._is_connected.set() print(f"πŸ”— Connesso a {self.host}:{self.port}") 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() self._socket_reader() except socket.error as e: print(f"πŸ”₯ Errore di connessione: {e}. Riprovo tra 5 secondi...") finally: self._is_connected.clear() if self.socket: self.socket.close() self.socket = None if self._is_running.is_set(): time.sleep(5) print("πŸ›‘ Gestore connessioni terminato.") def _socket_reader(self): buffer = "" while self._is_running.is_set(): try: data = self.socket.recv(4096) if not data: print("πŸ’” Connessione chiusa dal mixer.") break buffer += data.decode('utf-8', errors='ignore') while '\n' in buffer: line, buffer = buffer.split('\n', 1) line = line.strip() if not line: continue if line.startswith("NOTIFY"): self._handle_notify(line) elif line.startswith("OK") or line.startswith("ERROR"): try: self._response_queue.put_nowait(line) except queue.Full: print(f"⚠️ Coda risposte piena, scartato: {line}") else: print(f"πŸ€” Messaggio non gestito: {line}") except socket.timeout: continue except socket.error as e: print(f"πŸ”₯ Errore di lettura dal socket: {e}") break def _handle_notify(self, message: str): """ Analizza un messaggio NOTIFY e aggiorna la cache in modo robusto. Gestisce InCh, StInCh, FxRtnCh, DCA, Mix, Mtrx, St, Mono, MuteMaster. """ print(f"RECV NOTIFY: {message}") parts = message.split() if len(parts) < 2: return path = None path_index = -1 for i, part in enumerate(parts): if part.startswith("MIXER:"): path = part path_index = i break if path is None: print(f" -> ATTENZIONE: Nessun path valido trovato nel messaggio.") return print(f" -> Tentativo di acquisire il lock per l'aggiornamento (Path: {path})...") with self._cache_lock: print(f" -> LOCK ACQUISITO. Elaborazione...") try: updated = False # --- 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) self._cache.setdefault("channels", {}).setdefault(ch_key, {"channel": int(ch_key)})["name"] = name print(f" -> SUCCESS: InCh {ch_key} nome -> '{name}'") updated = True elif "InCh/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("channels", {}).setdefault(ch_key, {"channel": int(ch_key)})["on"] = state print(f" -> SUCCESS: InCh {ch_key} stato -> {'ON' if state else 'OFF'}") updated = True elif "InCh/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("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: print(f" -> ATTENZIONE: Nessuna condizione di aggiornamento trovata per il path '{path}'") except (ValueError, IndexError) as e: 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): return "ERROR connection - Mixer non connesso." with self._command_lock: try: while not self._response_queue.empty(): self._response_queue.get_nowait() 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 except queue.Empty: return "ERROR timeout - Nessuna risposta dal mixer." except socket.error as e: self._is_connected.clear() return f"ERROR connection - Errore socket: {e}" def close(self): """Ferma il controller e chiude la connessione.""" if not self._is_running.is_set(): return print("πŸ›‘ Chiusura del Mixer Controller...") self._is_running.clear() if self.socket: try: self.socket.shutdown(socket.SHUT_RDWR) self.socket.close() except socket.error: pass if self._reader_thread and self._reader_thread.is_alive(): self._reader_thread.join(timeout=2) print("βœ… Controller fermato.") self._save_cache() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() # ========================================================================= # 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 elif value == float('inf') or (isinstance(value, float) and math.isinf(value) and value > 0): return 10.0 elif isinstance(value, float) and math.isnan(value): return 0.0 return value def _ensure_cache_dir(self): CACHE_DIR.mkdir(parents=True, exist_ok=True) def _load_cache(self) -> dict: with self._cache_lock: if CACHE_FILE.exists(): try: with open(CACHE_FILE, 'r', encoding='utf-8') as f: data = json.load(f) 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": {}, "steinch": {}, "fxrtn": {}, "dcas": {}, "matrices": {}, "stereo": {}, "mono": {}, "mute_masters": {}, "timestamp": 0} def _save_cache(self): with self._cache_lock: try: 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): if isinstance(obj, dict): return {k: self._prepare_cache_for_json(v) for k, v in obj.items()} elif isinstance(obj, list): return [self._prepare_cache_for_json(item) for item in obj] elif isinstance(obj, float): if obj == float('-inf'): return None return obj else: return obj def _is_cache_valid(self) -> bool: with self._cache_lock: 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): 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] start = first_line.find('"') + 1 end = first_line.rfind('"') if start > 0 and end > start: return first_line[start:end] return "Sconosciuto" except: return "Errore" def _parse_value(self, response: str) -> str: first_line = response.split('\n')[0] parts = first_line.split() if len(parts) > 0 and parts[0] == "OK": return parts[-1] return "N/A" 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, 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 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) # 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 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, "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: {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} # ========================================================================= # 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}"} 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} 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}"} 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} 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"} command = f"set MIXER:Current/InCh/ToSt/Pan {channel-1} 0 {pan_value}" response = self._send_command(command) if pan_value < 0: pan_desc = f"sinistra {abs(pan_value)}" elif pan_value > 0: pan_desc = f"destra {pan_value}" 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_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"Canale {channel} β†’ FX {fx_number} impostato a {level_db:+.1f} dB", "response": response} 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"Canale {channel} β†’ FX {fx_number} send {'acceso' if state else 'spento'}", "response": response} 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 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 force_refresh and self._is_cache_valid(): with self._cache_lock: cached_data = self._cache.get("channels", {}).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/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: 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", "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() with self._cache_lock: channels_copy = list(self._cache.get("channels", {}).values()) 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() with self._cache_lock: mixes_copy = list(self._cache.get("mixes", {}).values()) 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 self._is_cache_valid(): self.refresh_cache() mix_info = self.get_mix_info(mix_number) if mix_info["status"] != "success": return mix_info sends = [] for channel in range(1, TF5_INPUT_CHANNELS + 1): send_info = self.get_channel_to_mix_info(channel, mix_number) 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.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."} # ========================================================================= # 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} # ========================================================================= # METER READING # Riferimento prminfo: # 2000 InCh 40ch 3 tipi: 0=PreHPF, 1=PreFader, 2=PostOn # 2001 StInCh 4ch 3 tipi: 0=PreEQ, 1=PreFader, 2=PostOn # 2002 FxRtnCh 4ch 2 tipi: 0=PreFader, 1=PostOn # 2100 Mix 20ch 3 tipi: 0=PreEQ, 1=PreFader, 2=PostOn # 2101 Mtrx 4ch 3 tipi: 0=PreEQ, 1=PreFader, 2=PostOn # 2102 St 2ch 3 tipi: 0=PreEQ, 1=PreFader, 2=PostOn # 2103 Mono 1ch 3 tipi: 0=PreEQ, 1=PreFader, 2=PostOn # # Il valore raw Γ¨ un intero 0–127. # Conversione: 0 = -inf dB, 1–127 lineare su scala ~-72..+18 dBFS. # Formula Yamaha TF: dBFS = (raw / 127) * 90 - 72 (approssimazione lineare) # Oppure si restituisce direttamente il raw per display proporzionale. # ========================================================================= # Mappa dei meter disponibili _METER_DEFS = { "InCh": {"path": "MIXER:Current/Meter/InCh", "channels": 40, "types": ["PreHPF", "PreFader", "PostOn"]}, "StInCh": {"path": "MIXER:Current/Meter/StInCh", "channels": 4, "types": ["PreEQ", "PreFader", "PostOn"]}, "FxRtnCh": {"path": "MIXER:Current/Meter/FxRtnCh", "channels": 4, "types": ["PreFader", "PostOn"]}, "Mix": {"path": "MIXER:Current/Meter/Mix", "channels": 20, "types": ["PreEQ", "PreFader", "PostOn"]}, "Mtrx": {"path": "MIXER:Current/Meter/Mtrx", "channels": 4, "types": ["PreEQ", "PreFader", "PostOn"]}, "St": {"path": "MIXER:Current/Meter/St", "channels": 2, "types": ["PreEQ", "PreFader", "PostOn"]}, "Mono": {"path": "MIXER:Current/Meter/Mono", "channels": 1, "types": ["PreEQ", "PreFader", "PostOn"]}, } def _meter_raw_to_db(self, raw: int) -> float: """ Converte il valore raw del meter (0-127) in dBFS approssimato. Scala Yamaha TF: 127 = +18 dBFS, 76 = 0 dBFS, 1 = -72 dBFS, 0 = -inf. """ if raw <= 0: return float('-inf') # Interpolazione lineare: 0..127 β†’ -72..+18 dBFS (range 90 dB) return round((raw / 127.0) * 90.0 - 72.0, 1) def _parse_meter_response(self, response: str) -> Optional[int]: """Estrae il valore intero raw dal response di un comando get meter.""" try: first_line = response.split('\n')[0].strip() parts = first_line.split() if parts and parts[0] == "OK": return int(parts[-1]) except (ValueError, IndexError): pass return None def get_meter(self, section: str, channel: int, meter_type: int = 2) -> dict: """ Legge il livello meter di un singolo canale. Args: section: Sezione del mixer: "InCh", "StInCh", "FxRtnCh", "Mix", "Mtrx", "St", "Mono" channel: Numero canale (1-based) meter_type: Indice del tipo meter (0-based, vedi _METER_DEFS per i nomi) Returns: dict con raw (0-127), level_db (dBFS), section, channel, type_name """ if section not in self._METER_DEFS: return {"status": "error", "message": f"Sezione '{section}' non valida. Valori: {list(self._METER_DEFS.keys())}"} defn = self._METER_DEFS[section] num_ch = defn["channels"] num_types = len(defn["types"]) if not 1 <= channel <= num_ch: return {"status": "error", "message": f"Canale {section} deve essere tra 1 e {num_ch}"} if not 0 <= meter_type < num_types: return {"status": "error", "message": f"Tipo meter deve essere tra 0 e {num_types - 1} per {section}: {defn['types']}"} ch_idx = channel - 1 command = f"get {defn['path']} {ch_idx} {meter_type}" response = self._send_command(command) raw = self._parse_meter_response(response) if raw is None: return {"status": "error", "message": f"Risposta non valida: {response}"} level_db = self._meter_raw_to_db(raw) type_name = defn["types"][meter_type] return { "status": "success", "section": section, "channel": channel, "type": meter_type, "type_name": type_name, "raw": raw, "level_db": level_db, "message": f"{section} ch{channel} [{type_name}]: {raw}/127 β†’ {level_db} dBFS" } def get_all_meters(self, section: str, meter_type: int = 2) -> dict: """ Legge tutti i canali meter di una sezione. Args: section: "InCh", "StInCh", "FxRtnCh", "Mix", "Mtrx", "St", "Mono" meter_type: Indice tipo meter (0-based) Returns: dict con lista di letture meter per tutti i canali della sezione """ if section not in self._METER_DEFS: return {"status": "error", "message": f"Sezione '{section}' non valida. Valori: {list(self._METER_DEFS.keys())}"} defn = self._METER_DEFS[section] num_ch = defn["channels"] num_types = len(defn["types"]) if not 0 <= meter_type < num_types: return {"status": "error", "message": f"Tipo meter deve essere tra 0 e {num_types - 1} per {section}: {defn['types']}"} type_name = defn["types"][meter_type] readings = [] for ch in range(1, num_ch + 1): ch_idx = ch - 1 command = f"get {defn['path']} {ch_idx} {meter_type}" response = self._send_command(command) raw = self._parse_meter_response(response) if raw is not None: level_db = self._meter_raw_to_db(raw) readings.append({ "channel": ch, "raw": raw, "level_db": level_db }) else: readings.append({ "channel": ch, "raw": None, "level_db": None, "error": response }) return { "status": "success", "section": section, "type": meter_type, "type_name": type_name, "total_channels": num_ch, "readings": readings, "message": f"Meter {section} [{type_name}]: {num_ch} canali letti" } def get_meter_snapshot(self, meter_type: int = 2) -> dict: """ Legge un snapshot dei meter di tutte le sezioni principali. Args: meter_type: Tipo meter da usare per tutte le sezioni. Nota: FxRtnCh ha solo 2 tipi (0-1), se si passa 2 viene automaticamente usato il tipo 1 (PostOn). Returns: dict con un sotto-dict per ogni sezione, contenente le letture. """ snapshot = {} for section, defn in self._METER_DEFS.items(): # Clamp meter_type al massimo disponibile per la sezione safe_type = min(meter_type, len(defn["types"]) - 1) result = self.get_all_meters(section, safe_type) snapshot[section] = result sections_ok = sum(1 for r in snapshot.values() if r["status"] == "success") return { "status": "success", "meter_type_requested": meter_type, "sections": snapshot, "message": f"Snapshot meter: {sections_ok}/{len(self._METER_DEFS)} sezioni lette" } def get_meter_types(self, section: str = None) -> dict: """ Restituisce i tipi di meter disponibili per una sezione (o tutte). Args: section: Nome sezione opzionale. Se None, restituisce tutte. """ if section: if section not in self._METER_DEFS: return {"status": "error", "message": f"Sezione '{section}' non valida. Valori: {list(self._METER_DEFS.keys())}"} defn = self._METER_DEFS[section] return { "status": "success", "section": section, "channels": defn["channels"], "types": {i: name for i, name in enumerate(defn["types"])} } else: return { "status": "success", "sections": { sec: { "channels": d["channels"], "types": {i: name for i, name in enumerate(d["types"])} } for sec, d in self._METER_DEFS.items() } }