1006 lines
36 KiB
Python
1006 lines
36 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
import socket
|
|
import sys
|
|
import os
|
|
import json
|
|
import time
|
|
import math
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
from google import genai
|
|
from google.genai import types
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
from config import *
|
|
|
|
|
|
class TF5MixerController:
|
|
def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT):
|
|
self.host = host
|
|
self.port = port
|
|
self.socket = None
|
|
self._ensure_cache_dir()
|
|
self._cache = self._load_cache()
|
|
|
|
def _sanitize_value(self, value):
|
|
"""
|
|
Sanitizza i valori per renderli JSON-safe.
|
|
Converte -Infinity in -120 (silenzio pratico).
|
|
"""
|
|
if value is None:
|
|
return -120.0
|
|
if value == float('-inf') or (isinstance(value, float) and math.isinf(value) and value < 0):
|
|
return -120.0 # Valore molto basso ma valido
|
|
elif value == float('inf') or (isinstance(value, float) and math.isinf(value) and value > 0):
|
|
return 10.0 # Massimo pratico
|
|
elif isinstance(value, float) and math.isnan(value):
|
|
return 0.0 # Default a 0 se NaN
|
|
|
|
return value
|
|
|
|
def _connect(self):
|
|
"""Stabilisce la connessione se non già connesso."""
|
|
if self.socket is None:
|
|
print('Inizializzazione socket...')
|
|
try:
|
|
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.socket.settimeout(5)
|
|
self.socket.connect((self.host, self.port))
|
|
except socket.error as e:
|
|
self.socket = None
|
|
raise ConnectionError(f"Impossibile connettersi al mixer: {e}")
|
|
|
|
def _disconnect(self):
|
|
"""Chiude la connessione."""
|
|
if self.socket:
|
|
print('Chiusura socket...')
|
|
try:
|
|
self.socket.close()
|
|
print('Socket chiuso!')
|
|
except:
|
|
print('Errore durante chiusura socket!')
|
|
pass
|
|
finally:
|
|
self.socket = None
|
|
|
|
def _send_command(self, command: str) -> str:
|
|
"""Invia un comando al mixer e restituisce la risposta."""
|
|
max_retries = 2
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
print(f'Tentativo di connessione {attempt} per {command}')
|
|
self._connect()
|
|
self.socket.sendall((command + '\n').encode('utf-8'))
|
|
response = self.socket.recv(4096)
|
|
decoded = response.decode('utf-8', errors='ignore').strip()
|
|
print(f'Risposta {decoded}')
|
|
return decoded
|
|
|
|
except socket.error as e:
|
|
print(f'Errore di connessione dopo {max_retries} tentativi: {e}')
|
|
self._disconnect()
|
|
|
|
if attempt < max_retries - 1:
|
|
time.sleep(0.1)
|
|
continue
|
|
else:
|
|
return f"Errore di connessione dopo {max_retries} tentativi: {e}"
|
|
|
|
def close(self):
|
|
"""Chiude la connessione (da chiamare alla fine)."""
|
|
self._disconnect()
|
|
|
|
def __enter__(self):
|
|
"""Context manager entry."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
"""Context manager exit."""
|
|
self.close()
|
|
|
|
def _ensure_cache_dir(self):
|
|
"""Crea la directory di cache se non esiste."""
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _load_cache(self) -> dict:
|
|
"""Carica la cache dal file."""
|
|
if CACHE_FILE.exists():
|
|
try:
|
|
with open(CACHE_FILE, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"⚠️ Errore nel caricamento della cache: {e}")
|
|
return {"channels": {}, "mixes": {}, "timestamp": 0}
|
|
|
|
def _save_cache(self):
|
|
"""Salva la cache nel file, convertendo -inf in null."""
|
|
try:
|
|
# Crea una copia della cache per la serializzazione
|
|
cache_to_save = self._prepare_cache_for_json(self._cache)
|
|
|
|
with open(CACHE_FILE, 'w', encoding='utf-8') as f:
|
|
json.dump(cache_to_save, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
print(f"⚠️ Errore nel salvataggio della cache: {e}")
|
|
|
|
def _prepare_cache_for_json(self, obj):
|
|
"""Converte ricorsivamente -inf in null per la serializzazione JSON."""
|
|
if isinstance(obj, dict):
|
|
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):
|
|
# Converti -inf in null per JSON
|
|
if obj == float('-inf'):
|
|
return None
|
|
return obj
|
|
else:
|
|
return obj
|
|
|
|
def _is_cache_valid(self) -> bool:
|
|
"""Verifica se la cache è ancora valida."""
|
|
if not self._cache.get("channels"):
|
|
return False
|
|
cache_age = time.time() - self._cache.get("timestamp", 0)
|
|
return cache_age < CACHE_DURATION
|
|
|
|
def _normalize_level_from_cache(self, level_value):
|
|
"""Normalizza il valore del livello dalla cache (gestisce None come -inf)."""
|
|
if level_value is None:
|
|
return float('-inf')
|
|
return level_value
|
|
|
|
def _update_channel_cache(self, channel: int):
|
|
"""Aggiorna la cache per un singolo canale leggendo dal mixer.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
"""
|
|
if not 1 <= channel <= TF5_INPUT_CHANNELS:
|
|
return
|
|
|
|
ch_idx = channel - 1
|
|
|
|
# Leggi nome
|
|
resp_name = self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
# Leggi stato ON/OFF
|
|
resp_on = self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
# Leggi livello
|
|
resp_level = self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
# Leggi pan
|
|
resp_pan = self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0")
|
|
pan_raw = self._parse_value(resp_pan)
|
|
try:
|
|
pan_value = int(pan_raw)
|
|
except:
|
|
pan_value = None
|
|
|
|
# Inizializza la struttura channels se non esiste
|
|
if "channels" not in self._cache:
|
|
self._cache["channels"] = {}
|
|
|
|
# Aggiorna la cache
|
|
self._cache["channels"][str(channel)] = {
|
|
"channel": channel,
|
|
"name": name,
|
|
"on": is_on,
|
|
"level_db": level_db,
|
|
"pan": pan_value
|
|
}
|
|
|
|
self._save_cache()
|
|
|
|
def _update_mix_cache(self, mix_number: int):
|
|
"""Aggiorna la cache per un singolo mix/aux leggendo dal mixer.
|
|
|
|
Args:
|
|
mix_number: Numero del mix (1-20)
|
|
"""
|
|
if not 1 <= mix_number <= TF5_MIX_BUSSES:
|
|
return
|
|
|
|
mix_idx = mix_number - 1
|
|
|
|
# Leggi nome
|
|
resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
# Leggi stato ON/OFF
|
|
resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
# Leggi livello
|
|
resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
# Inizializza la struttura mixes se non esiste
|
|
if "mixes" not in self._cache:
|
|
self._cache["mixes"] = {}
|
|
|
|
# Aggiorna la cache
|
|
self._cache["mixes"][str(mix_number)] = {
|
|
"mix": mix_number,
|
|
"name": name,
|
|
"on": is_on,
|
|
"level_db": level_db
|
|
}
|
|
|
|
self._save_cache()
|
|
|
|
def _parse_name(self, response: str) -> str:
|
|
"""Estrae il nome tra virgolette dalla prima riga della risposta."""
|
|
try:
|
|
# Considera solo la prima riga per evitare di includere notifiche successive
|
|
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:
|
|
"""Estrae l'ultimo valore dalla prima riga di una risposta OK."""
|
|
# Considera solo la prima riga per evitare notifiche successive
|
|
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 refresh_cache(self) -> dict:
|
|
"""Aggiorna la cache leggendo tutti i canali e mix dal mixer.
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
print("🔄 Aggiornamento cache completo in corso...")
|
|
channels_data = {}
|
|
mixes_data = {}
|
|
|
|
# Aggiorna canali
|
|
for ch in range(1, TF5_INPUT_CHANNELS + 1):
|
|
ch_idx = ch - 1
|
|
|
|
# Leggi nome
|
|
resp_name = self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
# Leggi stato ON/OFF
|
|
resp_on = self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
# Leggi livello
|
|
resp_level = self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
# Leggi pan
|
|
resp_pan = self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0")
|
|
pan_raw = self._parse_value(resp_pan)
|
|
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.05)
|
|
|
|
# Aggiorna mix
|
|
for mix in range(1, TF5_MIX_BUSSES + 1):
|
|
mix_idx = mix - 1
|
|
|
|
# Leggi nome
|
|
resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
# Leggi stato ON/OFF
|
|
resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
# Leggi livello
|
|
resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
mixes_data[str(mix)] = {
|
|
"mix": mix,
|
|
"name": name,
|
|
"on": is_on,
|
|
"level_db": level_db
|
|
}
|
|
|
|
time.sleep(0.05)
|
|
|
|
self._cache = {
|
|
"channels": channels_data,
|
|
"mixes": mixes_data,
|
|
"timestamp": time.time()
|
|
}
|
|
self._save_cache()
|
|
|
|
print(f"✅ Cache aggiornata con {len(channels_data)} canali e {len(mixes_data)} mix")
|
|
return {
|
|
"status": "success",
|
|
"message": f"Cache aggiornata con {len(channels_data)} canali e {len(mixes_data)} mix",
|
|
"channels_count": len(channels_data),
|
|
"mixes_count": len(mixes_data)
|
|
}
|
|
|
|
def recall_scene(self, bank: str, scene_number: int) -> dict:
|
|
"""Richiama una scena dal banco A o B.
|
|
|
|
Args:
|
|
bank: Il banco della scena ('a' o 'b')
|
|
scene_number: Il numero della scena (0-99)
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
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)
|
|
|
|
# Invalida la cache dopo il cambio scena
|
|
self._cache["timestamp"] = 0
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Scena {bank.upper()}{scene_number} richiamata. Cache invalidata.",
|
|
"response": response
|
|
}
|
|
|
|
def set_channel_level(self, channel: int, level_db: float) -> dict:
|
|
"""Imposta il livello del fader di un canale in dB.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
level_db: Livello in dB (da -inf a +10.0)
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
if not 1 <= channel <= TF5_INPUT_CHANNELS:
|
|
return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"}
|
|
|
|
# Converti dB in valore interno (moltiplicato per 100)
|
|
if level_db <= -138:
|
|
internal_value = -32768 # -inf
|
|
else:
|
|
internal_value = int(level_db * 100)
|
|
|
|
command = f"set MIXER:Current/InCh/Fader/Level {channel-1} 0 {internal_value}"
|
|
response = self._send_command(command)
|
|
|
|
# Aggiorna cache dal mixer per avere dati precisi
|
|
if "OK" in response:
|
|
self._update_channel_cache(channel)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Canale {channel} impostato a {level_db:+.1f} dB",
|
|
"response": response
|
|
}
|
|
|
|
def set_channel_on_off(self, channel: int, state: bool) -> dict:
|
|
"""Accende o spegne un canale.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
state: True per accendere, False per spegnere
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
if not 1 <= channel <= TF5_INPUT_CHANNELS:
|
|
return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"}
|
|
|
|
value = 1 if state else 0
|
|
command = f"set MIXER:Current/InCh/Fader/On {channel-1} 0 {value}"
|
|
response = self._send_command(command)
|
|
|
|
# Aggiorna cache dal mixer per avere dati precisi
|
|
if "OK" in response:
|
|
self._update_channel_cache(channel)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Canale {channel} {'acceso' if state else 'spento'}",
|
|
"response": response
|
|
}
|
|
|
|
def set_channel_pan(self, channel: int, pan_value: int) -> dict:
|
|
"""Imposta il pan di un canale.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
pan_value: Valore pan da -63 (sinistra) a +63 (destra), 0 è centro
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
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)
|
|
|
|
# Aggiorna cache dal mixer per avere dati precisi
|
|
if "OK" in response:
|
|
self._update_channel_cache(channel)
|
|
|
|
pan_desc = "centro"
|
|
if pan_value < 0:
|
|
pan_desc = f"sinistra {abs(pan_value)}"
|
|
elif pan_value > 0:
|
|
pan_desc = f"destra {pan_value}"
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Canale {channel} pan impostato a {pan_desc}",
|
|
"response": response
|
|
}
|
|
|
|
def set_mix_level(self, mix_number: int, level_db: float) -> dict:
|
|
"""Imposta il livello di un mix/aux.
|
|
|
|
Args:
|
|
mix_number: Numero del mix (1-20)
|
|
level_db: Livello in dB (da -inf a +10.0)
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
if not 1 <= mix_number <= TF5_MIX_BUSSES:
|
|
return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"}
|
|
|
|
if level_db <= -138:
|
|
internal_value = -32768
|
|
else:
|
|
internal_value = int(level_db * 100)
|
|
|
|
command = f"set MIXER:Current/Mix/Fader/Level {mix_number-1} 0 {internal_value}"
|
|
response = self._send_command(command)
|
|
|
|
# Aggiorna cache dal mixer per avere dati precisi
|
|
if "OK" in response:
|
|
self._update_mix_cache(mix_number)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Mix {mix_number} impostato a {level_db:+.1f} dB",
|
|
"response": response
|
|
}
|
|
|
|
def set_mix_on_off(self, mix_number: int, state: bool) -> dict:
|
|
"""Accende o spegne un mix/aux.
|
|
|
|
Args:
|
|
mix_number: Numero del mix (1-20)
|
|
state: True per accendere, False per spegnere
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
if not 1 <= mix_number <= TF5_MIX_BUSSES:
|
|
return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"}
|
|
|
|
value = 1 if state else 0
|
|
command = f"set MIXER:Current/Mix/Fader/On {mix_number-1} 0 {value}"
|
|
response = self._send_command(command)
|
|
|
|
# Aggiorna cache dal mixer per avere dati precisi
|
|
if "OK" in response:
|
|
self._update_mix_cache(mix_number)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Mix {mix_number} {'acceso' if state else 'spento'}",
|
|
"response": response
|
|
}
|
|
|
|
def mute_multiple_channels(self, channels: List[int]) -> dict:
|
|
"""Muta più canali contemporaneamente.
|
|
|
|
Args:
|
|
channels: Lista di numeri di canale da mutare (es: [1, 2, 5, 8])
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
results = []
|
|
for ch in channels:
|
|
result = self.set_channel_on_off(ch, False)
|
|
results.append(result)
|
|
|
|
success_count = sum(1 for r in results if r["status"] == "success")
|
|
return {
|
|
"status": "success" if success_count == len(channels) else "partial",
|
|
"message": f"Mutati {success_count}/{len(channels)} canali: {channels}",
|
|
"details": results
|
|
}
|
|
|
|
def unmute_multiple_channels(self, channels: List[int]) -> dict:
|
|
"""Riattiva più canali contemporaneamente.
|
|
|
|
Args:
|
|
channels: Lista di numeri di canale da riattivare (es: [1, 2, 5, 8])
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
results = []
|
|
for ch in channels:
|
|
result = self.set_channel_on_off(ch, True)
|
|
results.append(result)
|
|
|
|
success_count = sum(1 for r in results if r["status"] == "success")
|
|
return {
|
|
"status": "success" if success_count == len(channels) else "partial",
|
|
"message": f"Riattivati {success_count}/{len(channels)} canali: {channels}",
|
|
"details": results
|
|
}
|
|
|
|
def get_channel_info(self, channel: int, force_refresh: bool = False) -> dict:
|
|
"""Legge le informazioni di un canale (nome, livello, stato, pan).
|
|
Usa la cache se disponibile e valida.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
force_refresh: Se True, ignora la cache e legge dal mixer
|
|
|
|
Returns:
|
|
Un dizionario con tutte le informazioni del canale
|
|
"""
|
|
if not 1 <= channel <= TF5_INPUT_CHANNELS:
|
|
return {"status": "error", "message": f"Il canale deve essere tra 1 e {TF5_INPUT_CHANNELS}"}
|
|
|
|
# Usa cache se valida e non forzato il refresh
|
|
if not force_refresh and self._is_cache_valid():
|
|
cached_data = self._cache.get("channels", {}).get(str(channel))
|
|
if cached_data:
|
|
# Normalizza level_db (None -> -inf)
|
|
level_db = self._normalize_level_from_cache(cached_data.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
return {
|
|
"status": "success",
|
|
"source": "cache",
|
|
"channel": cached_data["channel"],
|
|
"name": cached_data["name"],
|
|
"on": cached_data["on"],
|
|
"level_db": sanitized_level_db,
|
|
"pan": cached_data.get("pan")
|
|
}
|
|
|
|
# Altrimenti leggi dal mixer
|
|
ch_idx = channel - 1
|
|
|
|
resp_name = self._send_command(f"get MIXER:Current/InCh/Label/Name {ch_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
resp_on = self._send_command(f"get MIXER:Current/InCh/Fader/On {ch_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
resp_level = self._send_command(f"get MIXER:Current/InCh/Fader/Level {ch_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
resp_pan = self._send_command(f"get MIXER:Current/InCh/ToSt/Pan {ch_idx} 0")
|
|
pan_raw = self._parse_value(resp_pan)
|
|
try:
|
|
pan_value = int(pan_raw)
|
|
except:
|
|
pan_value = None
|
|
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
return {
|
|
"status": "success",
|
|
"source": "mixer",
|
|
"channel": channel,
|
|
"name": name,
|
|
"on": is_on,
|
|
"level_db": sanitized_level_db,
|
|
"pan": pan_value
|
|
}
|
|
|
|
def get_mix_info(self, mix_number: int, force_refresh: bool = False) -> dict:
|
|
"""Legge le informazioni di un mix/aux (nome, livello, stato).
|
|
Usa la cache se disponibile e valida.
|
|
|
|
Args:
|
|
mix_number: Numero del mix (1-20)
|
|
force_refresh: Se True, ignora la cache e legge dal mixer
|
|
|
|
Returns:
|
|
Un dizionario con tutte le informazioni del mix
|
|
"""
|
|
if not 1 <= mix_number <= TF5_MIX_BUSSES:
|
|
return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"}
|
|
|
|
# Usa cache se valida e non forzato il refresh
|
|
if not force_refresh and self._is_cache_valid():
|
|
cached_data = self._cache.get("mixes", {}).get(str(mix_number))
|
|
if cached_data:
|
|
# Normalizza level_db (None -> -inf)
|
|
level_db = self._normalize_level_from_cache(cached_data.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
return {
|
|
"status": "success",
|
|
"source": "cache",
|
|
"mix": cached_data["mix"],
|
|
"name": cached_data["name"],
|
|
"on": cached_data["on"],
|
|
"level_db": sanitized_level_db
|
|
}
|
|
|
|
# Altrimenti leggi dal mixer
|
|
mix_idx = mix_number - 1
|
|
|
|
resp_name = self._send_command(f"get MIXER:Current/Mix/Label/Name {mix_idx} 0")
|
|
name = self._parse_name(resp_name)
|
|
|
|
resp_on = self._send_command(f"get MIXER:Current/Mix/Fader/On {mix_idx} 0")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
resp_level = self._send_command(f"get MIXER:Current/Mix/Fader/Level {mix_idx} 0")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
return {
|
|
"status": "success",
|
|
"source": "mixer",
|
|
"mix": mix_number,
|
|
"name": name,
|
|
"on": is_on,
|
|
"level_db": sanitized_level_db
|
|
}
|
|
|
|
def search_channels_by_name(self, search_term: str) -> dict:
|
|
"""Cerca canali il cui nome contiene un determinato termine.
|
|
Usa la cache per velocizzare la ricerca.
|
|
|
|
Args:
|
|
search_term: Il termine da cercare nei nomi dei canali (case-insensitive)
|
|
|
|
Returns:
|
|
Un dizionario con la lista dei canali trovati
|
|
"""
|
|
# Aggiorna cache se non valida
|
|
if not self._is_cache_valid():
|
|
self.refresh_cache()
|
|
|
|
search_lower = search_term.lower()
|
|
found_channels = []
|
|
|
|
for ch_str, info in self._cache.get("channels", {}).items():
|
|
if search_lower in info.get("name", "").lower():
|
|
# Normalizza level_db quando restituisci i risultati
|
|
level_db = self._normalize_level_from_cache(info.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
found_channels.append({
|
|
"channel": info["channel"],
|
|
"name": info["name"],
|
|
"on": info["on"],
|
|
"level_db": sanitized_level_db
|
|
})
|
|
|
|
# Ordina per numero canale
|
|
found_channels.sort(key=lambda x: x["channel"])
|
|
|
|
return {
|
|
"status": "success",
|
|
"search_term": search_term,
|
|
"found_count": len(found_channels),
|
|
"channels": found_channels,
|
|
"message": f"Trovati {len(found_channels)} canali contenenti '{search_term}'"
|
|
}
|
|
|
|
def search_mixes_by_name(self, search_term: str) -> dict:
|
|
"""Cerca mix/aux il cui nome contiene un determinato termine.
|
|
Usa la cache per velocizzare la ricerca.
|
|
|
|
Args:
|
|
search_term: Il termine da cercare nei nomi dei mix (case-insensitive)
|
|
|
|
Returns:
|
|
Un dizionario con la lista dei mix trovati
|
|
"""
|
|
# Aggiorna cache se non valida
|
|
if not self._is_cache_valid():
|
|
self.refresh_cache()
|
|
|
|
search_lower = search_term.lower()
|
|
found_mixes = []
|
|
|
|
for mix_str, info in self._cache.get("mixes", {}).items():
|
|
if search_lower in info.get("name", "").lower():
|
|
# Normalizza level_db quando restituisci i risultati
|
|
level_db = self._normalize_level_from_cache(info.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
found_mixes.append({
|
|
"mix": info["mix"],
|
|
"name": info["name"],
|
|
"on": info["on"],
|
|
"level_db": sanitized_level_db
|
|
})
|
|
|
|
# Ordina per numero mix
|
|
found_mixes.sort(key=lambda x: x["mix"])
|
|
|
|
return {
|
|
"status": "success",
|
|
"search_term": search_term,
|
|
"found_count": len(found_mixes),
|
|
"mixes": found_mixes,
|
|
"message": f"Trovati {len(found_mixes)} mix contenenti '{search_term}'"
|
|
}
|
|
|
|
def get_all_channels_summary(self) -> dict:
|
|
"""Ottiene un riepilogo di tutti i canali con nome e stato.
|
|
Usa la cache per velocizzare.
|
|
|
|
Returns:
|
|
Un dizionario con il riepilogo di tutti i canali
|
|
"""
|
|
# Aggiorna cache se non valida
|
|
if not self._is_cache_valid():
|
|
self.refresh_cache()
|
|
|
|
channels = []
|
|
for ch_str, info in self._cache.get("channels", {}).items():
|
|
level_db = self._normalize_level_from_cache(info.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
channels.append({
|
|
"channel": info["channel"],
|
|
"name": info["name"],
|
|
"on": info["on"],
|
|
"level_db": sanitized_level_db
|
|
})
|
|
|
|
# Ordina per numero canale
|
|
channels.sort(key=lambda x: x["channel"])
|
|
|
|
cache_age = time.time() - self._cache.get("timestamp", 0)
|
|
|
|
return {
|
|
"status": "success",
|
|
"total_channels": len(channels),
|
|
"channels": channels,
|
|
"cache_age_seconds": int(cache_age),
|
|
"message": f"Riepilogo di {len(channels)} canali (cache: {int(cache_age)}s fa)"
|
|
}
|
|
|
|
def get_all_mixes_summary(self) -> dict:
|
|
"""Ottiene un riepilogo di tutti i mix/aux con nome e stato.
|
|
Usa la cache per velocizzare.
|
|
|
|
Returns:
|
|
Un dizionario con il riepilogo di tutti i mix
|
|
"""
|
|
# Aggiorna cache se non valida
|
|
if not self._is_cache_valid():
|
|
self.refresh_cache()
|
|
|
|
mixes = []
|
|
for mix_str, info in self._cache.get("mixes", {}).items():
|
|
level_db = self._normalize_level_from_cache(info.get("level_db"))
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
mixes.append({
|
|
"mix": info["mix"],
|
|
"name": info["name"],
|
|
"on": info["on"],
|
|
"level_db": sanitized_level_db
|
|
})
|
|
|
|
# Ordina per numero mix
|
|
mixes.sort(key=lambda x: x["mix"])
|
|
|
|
cache_age = time.time() - self._cache.get("timestamp", 0)
|
|
|
|
return {
|
|
"status": "success",
|
|
"total_mixes": len(mixes),
|
|
"mixes": mixes,
|
|
"cache_age_seconds": int(cache_age),
|
|
"message": f"Riepilogo di {len(mixes)} mix (cache: {int(cache_age)}s fa)"
|
|
}
|
|
|
|
def set_channel_to_mix_level(self, channel: int, mix_number: int, level_db: float) -> dict:
|
|
"""Imposta il livello di invio di un canale verso un mix/aux.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
mix_number: Numero del mix (1-20)
|
|
level_db: Livello in dB (da -inf a +10.0)
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
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 level_db <= -138:
|
|
internal_value = -32768
|
|
else:
|
|
internal_value = int(level_db * 100)
|
|
|
|
command = f"set MIXER:Current/InCh/ToMix/Level {channel-1} {mix_number-1} {internal_value}"
|
|
response = self._send_command(command)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Canale {channel} → Mix {mix_number} impostato a {level_db:+.1f} dB",
|
|
"response": response
|
|
}
|
|
|
|
def set_channel_to_mix_on_off(self, channel: int, mix_number: int, state: bool) -> dict:
|
|
"""Accende o spegne l'invio di un canale verso un mix/aux.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
mix_number: Numero del mix (1-20)
|
|
state: True per accendere, False per spegnere
|
|
|
|
Returns:
|
|
Un dizionario con lo stato dell'operazione
|
|
"""
|
|
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}"}
|
|
|
|
value = 1 if state else 0
|
|
command = f"set MIXER:Current/InCh/ToMix/On {channel-1} {mix_number-1} {value}"
|
|
response = self._send_command(command)
|
|
|
|
return {
|
|
"status": "success" if "OK" in response else "error",
|
|
"message": f"Invio canale {channel} → Mix {mix_number} {'acceso' if state else 'spento'}",
|
|
"response": response
|
|
}
|
|
|
|
def get_channel_to_mix_info(self, channel: int, mix_number: int) -> dict:
|
|
"""Legge le informazioni dell'invio di un canale verso un mix.
|
|
|
|
Args:
|
|
channel: Numero del canale (1-40)
|
|
mix_number: Numero del mix (1-20)
|
|
|
|
Returns:
|
|
Un dizionario con le informazioni del send
|
|
"""
|
|
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 = channel - 1
|
|
mix_idx = mix_number - 1
|
|
|
|
# Leggi livello
|
|
resp_level = self._send_command(f"get MIXER:Current/InCh/ToMix/Level {ch_idx} {mix_idx}")
|
|
level_raw = self._parse_value(resp_level)
|
|
try:
|
|
level_int = int(level_raw)
|
|
level_db = level_int / 100.0 if level_int > -32768 else float('-inf')
|
|
except:
|
|
level_db = None
|
|
|
|
# Leggi stato ON/OFF
|
|
resp_on = self._send_command(f"get MIXER:Current/InCh/ToMix/On {ch_idx} {mix_idx}")
|
|
is_on = self._parse_value(resp_on) == "1"
|
|
|
|
sanitized_level_db = self._sanitize_value(level_db)
|
|
|
|
return {
|
|
"status": "success",
|
|
"channel": channel,
|
|
"mix": mix_number,
|
|
"on": is_on,
|
|
"send_level_db": sanitized_level_db,
|
|
"message": f"Canale {channel} → Mix {mix_number}: {sanitized_level_db:+.1f} dB ({'ON' if is_on else 'OFF'})"
|
|
}
|
|
|
|
def get_full_mix_details(self, mix_number: int) -> dict:
|
|
"""
|
|
Raccoglie informazioni complete su un mix, inclusi tutti i canali inviati ad esso.
|
|
Questa funzione è fondamentale per l'analisi del mix.
|
|
|
|
Args:
|
|
mix_number: Numero del mix (1-20)
|
|
|
|
Returns:
|
|
Un dizionario con i dettagli del mix e una lista dei canali inviati.
|
|
"""
|
|
if not 1 <= mix_number <= TF5_MIX_BUSSES:
|
|
return {"status": "error", "message": f"Il mix deve essere tra 1 e {TF5_MIX_BUSSES}"}
|
|
|
|
# Aggiorna la cache se non è valida per avere i nomi dei canali
|
|
if not self._is_cache_valid():
|
|
self.refresh_cache()
|
|
|
|
# Ottieni le info principali del mix
|
|
mix_info = self.get_mix_info(mix_number)
|
|
if mix_info["status"] != "success":
|
|
return mix_info
|
|
|
|
sends = []
|
|
# Itera su tutti i canali di input per vedere cosa mandano a questo mix
|
|
for channel in range(1, TF5_INPUT_CHANNELS + 1):
|
|
send_info = self.get_channel_to_mix_info(channel, mix_number)
|
|
|
|
# Aggiungi solo i canali che sono effettivamente inviati al mix
|
|
if send_info["status"] == "success" and send_info["send_level_db"] > -120.0:
|
|
# Recupera le info principali del canale dalla cache per efficienza
|
|
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"]
|
|
})
|
|
|
|
# Ordina i canali dal più forte al più debole per facilitare l'analisi
|
|
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."
|
|
} |