gitea/Automixbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
729cb649fcaad4763056eb1b531b56028aaa4c1a
Automixbot/mixer_controller.py
Nick 729cb649fc add extra tools
2026-02-16 21:02:05 +01:00

1585 lines
91 KiB
Python
Raw Blame History

#!/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}
Reference in New Issue View Git Blame Copy Permalink