1796 lines
100 KiB
Python
1796 lines
100 KiB
Python
#!/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()
|
||
}
|
||
} |