gitea/Automixbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
Nick
2025-10-27 20:13:23 +01:00
parent d60858ae67
commit 0ed2435536
6 changed files with 1227 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files

407
.tf5_mixer_cache/channels_cache.json Normal file
View File

@@ -0,0 +1,407 @@
{
"channels": {
"1": {
"channel": 1,
"name": "Sconosciuto",
"on": false,
"level_db": 0.65,
"pan": 0
},
"2": {
"channel": 2,
"name": "Gelato",
"on": true,
"level_db": -6.25,
"pan": 0
},
"3": {
"channel": 3,
"name": "Talkback",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"4": {
"channel": 4,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"5": {
"channel": 5,
"name": "Vox1",
"on": true,
"level_db": 0.65,
"pan": 0
},
"6": {
"channel": 6,
"name": "Vox2",
"on": true,
"level_db": -1.1,
"pan": 0
},
"7": {
"channel": 7,
"name": "Basso",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"8": {
"channel": 8,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"9": {
"channel": 9,
"name": "Kick",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"10": {
"channel": 10,
"name": "Snare",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"11": {
"channel": 11,
"name": "Tom 1",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"12": {
"channel": 12,
"name": "Tom 2",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"13": {
"channel": 13,
"name": "Tom3",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"14": {
"channel": 14,
"name": "Pan SX",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"15": {
"channel": 15,
"name": "Pan dx",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"16": {
"channel": 16,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"17": {
"channel": 17,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"18": {
"channel": 18,
"name": "Archetto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"19": {
"channel": 19,
"name": "Vox 5",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"20": {
"channel": 20,
"name": "Tast",
"on": true,
"level_db": 1.0,
"pan": 0
},
"21": {
"channel": 21,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"22": {
"channel": 22,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"23": {
"channel": 23,
"name": " Vox3",
"on": true,
"level_db": 3.8,
"pan": 0
},
"24": {
"channel": 24,
"name": "Chit cnt",
"on": true,
"level_db": -2.1,
"pan": 0
},
"25": {
"channel": 25,
"name": "Chit dx",
"on": true,
"level_db": -7.7,
"pan": 0
},
"26": {
"channel": 26,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"27": {
"channel": 27,
"name": "Vox 4",
"on": false,
"level_db": -132.0,
"pan": 0
},
"28": {
"channel": 28,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"29": {
"channel": 29,
"name": "Pad",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"30": {
"channel": 30,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"31": {
"channel": 31,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"32": {
"channel": 32,
"name": "Sconosciuto",
"on": false,
"level_db": -Infinity,
"pan": 0
},
"33": {
"channel": 33,
"name": "PC",
"on": true,
"level_db": -19.8,
"pan": -63
},
"34": {
"channel": 34,
"name": "PC",
"on": true,
"level_db": -19.8,
"pan": 63
},
"35": {
"channel": 35,
"name": "ch35",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"36": {
"channel": 36,
"name": "ch36",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"37": {
"channel": 37,
"name": "ch37",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"38": {
"channel": 38,
"name": "ch38",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"39": {
"channel": 39,
"name": "ch39",
"on": true,
"level_db": -Infinity,
"pan": 0
},
"40": {
"channel": 40,
"name": "ch40",
"on": true,
"level_db": -Infinity,
"pan": 0
}
},
"mixes": {
"1": {
"mix": 1,
"name": "Sinistro",
"on": true,
"level_db": -Infinity
},
"2": {
"mix": 2,
"name": "Aux 2",
"on": true,
"level_db": -0.05
},
"3": {
"mix": 3,
"name": "Aux 3",
"on": true,
"level_db": -0.5
},
"4": {
"mix": 4,
"name": "Aux 4",
"on": true,
"level_db": -0.6
},
"5": {
"mix": 5,
"name": "batteria",
"on": true,
"level_db": 1.35
},
"6": {
"mix": 6,
"name": "Aux 6",
"on": true,
"level_db": 0.9
},
"7": {
"mix": 7,
"name": "Destro",
"on": true,
"level_db": -2.7
},
"8": {
"mix": 8,
"name": "Aux 8",
"on": true,
"level_db": -0.15
},
"9": {
"mix": 9,
"name": "Aux 9/10",
"on": true,
"level_db": -8.1
},
"10": {
"mix": 10,
"name": "Aux 9/10",
"on": true,
"level_db": -8.1
},
"11": {
"mix": 11,
"name": "Aux Pulp",
"on": true,
"level_db": -10.3
},
"12": {
"mix": 12,
"name": "Aux Pulp",
"on": true,
"level_db": -10.3
},
"13": {
"mix": 13,
"name": "Aux13/14",
"on": false,
"level_db": -1.6
},
"14": {
"mix": 14,
"name": "Aux13/14",
"on": false,
"level_db": -1.6
},
"15": {
"mix": 15,
"name": "Aux15/16",
"on": true,
"level_db": 0.1
},
"16": {
"mix": 16,
"name": "Aux15/16",
"on": true,
"level_db": 0.1
},
"17": {
"mix": 17,
"name": "Diretta",
"on": true,
"level_db": 0.0
},
"18": {
"mix": 18,
"name": "Diretta",
"on": true,
"level_db": 0.0
},
"19": {
"mix": 19,
"name": "Traduzio",
"on": true,
"level_db": 6.0
},
"20": {
"mix": 20,
"name": "Traduzio",
"on": true,
"level_db": 6.0
}
},
"timestamp": 1761592356.5132928
}

BIN
__pycache__/config.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
__pycache__/mixer_controller.cpython-312.pyc
View File

Binary file not shown.

13
config.py Normal file
View File

@@ -0,0 +1,13 @@
# Configurazione mixer
DEFAULT_HOST = "192.168.1.62"
DEFAULT_PORT = 49280
TF5_INPUT_CHANNELS = 40
TF5_MIX_BUSSES = 20
from pathlib import Path
# Configurazione cache
CACHE_DIR = Path(".tf5_mixer_cache")
CACHE_FILE = CACHE_DIR / "channels_cache.json"
CACHE_DURATION = 3600 # 60 minuti in secondi

539
mixer_agent.py
View File

@@ -13,513 +13,12 @@ from pathlib import Path
from typing import List, Optional from typing import List, Optional
from google import genai from google import genai
from google.genai import types from google.genai import types
from dotenv import load_dotenv from dotenv import load_dotenv
from mixer_controller import TF5MixerController
load_dotenv() load_dotenv()
# Configurazione mixer from config import *
DEFAULT_HOST = "192.168.1.62"
DEFAULT_PORT = 49280
TF5_INPUT_CHANNELS = 40
TF5_MIX_BUSSES = 20
# Configurazione cache
CACHE_DIR = Path.home() / ".tf5_mixer_cache"
CACHE_FILE = CACHE_DIR / "channels_cache.json"
CACHE_DURATION = 3600 # 60 minuti in secondi
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 _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() # Forza riconnessione al prossimo tentativo
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": {}, "timestamp": 0}
def _save_cache(self):
"""Salva la cache nel file."""
try:
with open(CACHE_FILE, 'w', encoding='utf-8') as f:
json.dump(self._cache, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"⚠️ Errore nel salvataggio della cache: {e}")
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 _parse_name(self, response: str) -> str:
"""Estrae il nome tra virgolette dalla risposta."""
try:
start = response.find('"') + 1
end = response.rfind('"')
if start > 0 and end > start:
return response[start:end]
return "Sconosciuto"
except:
return "Errore"
def _parse_value(self, response: str) -> str:
"""Estrae l'ultimo valore da una risposta OK."""
parts = response.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 dal mixer.
Returns:
Un dizionario con lo stato dell'operazione
"""
print("🔄 Aggiornamento cache in corso...")
channels_data = {}
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
}
# Piccolo delay per non sovraccaricare il mixer
time.sleep(0.05)
self._cache = {
"channels": channels_data,
"timestamp": time.time()
}
self._save_cache()
print(f"✅ Cache aggiornata con {len(channels_data)} canali")
return {
"status": "success",
"message": f"Cache aggiornata con {len(channels_data)} canali",
"channels_count": len(channels_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 locale
if str(channel) in self._cache.get("channels", {}):
self._cache["channels"][str(channel)]["level_db"] = level_db
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 locale
if str(channel) in self._cache.get("channels", {}):
self._cache["channels"][str(channel)]["on"] = state
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 locale
if str(channel) in self._cache.get("channels", {}):
self._cache["channels"][str(channel)]["pan"] = pan_value
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)
return {
"status": "success" if "OK" in response else "error",
"message": f"Mix {mix_number} impostato a {level_db:+.1f} dB",
"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:
return {
"status": "success",
"source": "cache",
**cached_data
}
# 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
return {
"status": "success",
"source": "mixer",
"channel": channel,
"name": name,
"on": is_on,
"level_db": level_db,
"pan": pan_value
}
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():
found_channels.append({
"channel": info["channel"],
"name": info["name"],
"on": info["on"],
"level_db": info["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 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():
channels.append({
"channel": info["channel"],
"name": info["name"],
"on": info["on"]
})
# 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)"
}
class TF5AIAgent: class TF5AIAgent:
@@ -543,11 +42,15 @@ class TF5AIAgent:
self.controller.set_channel_on_off, self.controller.set_channel_on_off,
self.controller.set_channel_pan, self.controller.set_channel_pan,
self.controller.set_mix_level, self.controller.set_mix_level,
self.controller.set_mix_on_off,
self.controller.mute_multiple_channels, self.controller.mute_multiple_channels,
self.controller.unmute_multiple_channels, self.controller.unmute_multiple_channels,
self.controller.get_channel_info, self.controller.get_channel_info,
self.controller.get_mix_info,
self.controller.search_channels_by_name, self.controller.search_channels_by_name,
self.controller.search_mixes_by_name,
self.controller.get_all_channels_summary, self.controller.get_all_channels_summary,
self.controller.get_all_mixes_summary,
self.controller.refresh_cache, self.controller.refresh_cache,
], ],
temperature=0, temperature=0,
@@ -564,9 +67,11 @@ Il mixer ha:
- Scene memorizzate nei banchi A e B (da 0 a 99) - Scene memorizzate nei banchi A e B (da 0 a 99)
IMPORTANTE - Sistema di Cache: IMPORTANTE - Sistema di Cache:
- Le info sui canali sono salvate per 5 minuti per non sovraccaricare il mixer - Le info sui canali e mix sono salvate per 60 minuti per non sovraccaricare il mixer
- Usa search_channels_by_name e get_all_channels_summary per cercare velocemente - Usa search_channels_by_name e search_mixes_by_name per cercare velocemente
- Quando si carica una scena, i dati vengono aggiornati automaticamente - Usa get_all_channels_summary e get_all_mixes_summary per vedere tutto
- Quando modifichi un canale/mix, la cache viene aggiornata automaticamente
- Quando si carica una scena, i dati vengono invalidati e aggiornati alla prossima richiesta
- Puoi fare refresh_cache solo se l'utente lo chiede esplicitamente - Puoi fare refresh_cache solo se l'utente lo chiede esplicitamente
Come interpretare le richieste: Come interpretare le richieste:
@@ -582,8 +87,8 @@ VOLUME/LIVELLO:
- "silenzio/muto" → spegni il canale - "silenzio/muto" → spegni il canale
ON/OFF: ON/OFF:
- "accendi/attiva/apri" → canale ON - "accendi/attiva/apri" → canale/mix ON
- "spegni/muta/chiudi/stacca" → canale OFF - "spegni/muta/chiudi/stacca" → canale/mix OFF
- "muto" può significare sia spegnere che abbassare molto - "muto" può significare sia spegnere che abbassare molto
BILANCIAMENTO (PAN): BILANCIAMENTO (PAN):
@@ -592,12 +97,13 @@ BILANCIAMENTO (PAN):
- "al centro" → pan 0 - "al centro" → pan 0
- "un po' a sinistra" → pan -30 circa - "un po' a sinistra" → pan -30 circa
IDENTIFICAZIONE CANALI: IDENTIFICAZIONE CANALI E MIX:
- Accetta sia numeri ("canale 5") che nomi ("il microfono del cantante") - Accetta sia numeri ("canale 5", "mix 3") che nomi ("il microfono del cantante", "monitor palco")
- Se non trovi un canale per nome, cerca usando search_channels_by_name - Se non trovi un canale/mix per nome, cerca usando search_channels_by_name o search_mixes_by_name
- "il mio mic/microfono" → cerca tra i canali chi è sul palco - "il mio mic/microfono" → cerca tra i canali chi è sul palco
- "le chitarre/i vox/le tastiere" → cerca per strumento - "le chitarre/i vox/le tastiere" → cerca per strumento
- "tutti i mic/tutte le chitarre" → cerca e gestisci multipli - "tutti i mic/tutte le chitarre" → cerca e gestisci multipli
- "il monitor" / "l'aux 2" → cerca tra i mix
SCENE: SCENE:
- "carica/richiama/vai alla scena X" → recall_scene - "carica/richiama/vai alla scena X" → recall_scene
@@ -610,7 +116,7 @@ GRUPPI DI CANALI:
CASI PARTICOLARI: CASI PARTICOLARI:
- Se la richiesta è ambigua, chiedi chiarimenti in modo colloquiale - Se la richiesta è ambigua, chiedi chiarimenti in modo colloquiale
- Se serve cercare un canale, usa prima la cache (search_channels_by_name) - Se serve cercare un canale/mix, usa prima la cache (search_channels_by_name / search_mixes_by_name)
- Conferma sempre cosa hai fatto con un messaggio breve e chiaro - Conferma sempre cosa hai fatto con un messaggio breve e chiaro
- Usa emoji occasionalmente per rendere le risposte più amichevoli (✅ ❌ 🎤 🎸 🔊) - Usa emoji occasionalmente per rendere le risposte più amichevoli (✅ ❌ 🎤 🎸 🔊)
- Se qualcosa non funziona, spiega il problema in modo semplice - Se qualcosa non funziona, spiega il problema in modo semplice
@@ -620,7 +126,7 @@ ESEMPI DI INTERPRETAZIONE:
"abbassa un po' le chitarre" → cerca canali chitarra, riduci di 3-5 dB "abbassa un po' le chitarre" → cerca canali chitarra, riduci di 3-5 dB
"muto tutto" → spegni tutti i 40 canali "muto tutto" → spegni tutti i 40 canali
"solo voce" → cerca canali voce, accendi quelli e spegni gli altri "solo voce" → cerca canali voce, accendi quelli e spegni gli altri
"mettimi più forte nel monitor" → NON puoi (sono gli aux), spiega che serve il tecnico "alza il monitor 2" → cerca mix 2, aumenta volume
"carica la scena del soundcheck" → cerca nel nome o chiedi numero scena "carica la scena del soundcheck" → cerca nel nome o chiedi numero scena
"troppo forte, abbassa" → riduci di 5-8 dB "troppo forte, abbassa" → riduci di 5-8 dB
"spegni questo canale" → se non specifica numero, chiedi quale "spegni questo canale" → se non specifica numero, chiedi quale
@@ -678,7 +184,10 @@ Devi essere veloce, chiaro e capire anche richieste approssimative.
# Mostra stato cache # Mostra stato cache
cache_age = time.time() - self.controller._cache.get("timestamp", 0) cache_age = time.time() - self.controller._cache.get("timestamp", 0)
if self.controller._is_cache_valid(): if self.controller._is_cache_valid():
channels_count = len(self.controller._cache.get("channels", {}))
mixes_count = len(self.controller._cache.get("mixes", {}))
print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)") print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)")
print(f" 📊 {channels_count} canali, {mixes_count} mix")
else: else:
print("\n💾 Cache non disponibile, verrà creata al primo utilizzo") print("\n💾 Cache non disponibile, verrà creata al primo utilizzo")
@@ -688,8 +197,10 @@ Devi essere veloce, chiaro e capire anche richieste approssimative.
print(" - 'Richiama la scena A10'") print(" - 'Richiama la scena A10'")
print(" - 'Imposta il pan del canale 3 a sinistra'") print(" - 'Imposta il pan del canale 3 a sinistra'")
print(" - 'Muta i canali 2, 4, 6 e 8'") print(" - 'Muta i canali 2, 4, 6 e 8'")
print(" - 'Alza il mix 3 di 5 dB'")
print(" - 'Quali canali sono associati ai vox?'") print(" - 'Quali canali sono associati ai vox?'")
print(" - 'Mostrami lo stato del canale 12'") print(" - 'Mostrami lo stato del canale 12'")
print(" - 'Cerca i monitor'")
print(" - 'Aggiorna i dati dal mixer'") print(" - 'Aggiorna i dati dal mixer'")
print("\nDigita 'esci' o 'quit' per terminare\n") print("\nDigita 'esci' o 'quit' per terminare\n")

803
mixer_controller.py
View File

@@ -1,32 +1,785 @@
# mixer_controller.py
import socket import socket
import sys import sys
import os
import json
import time
from pathlib import Path
from typing import List, Optional
from google import genai
from google.genai import types
from dotenv import load_dotenv
# Impostazioni di connessione predefinite load_dotenv()
DEFAULT_HOST = "192.168.1.62" # Modifica con l'IP del tuo mixer
DEFAULT_PORT = 49280
def send_command(command, host=DEFAULT_HOST, port=DEFAULT_PORT): from config import *
"""
Crea una connessione, invia un singolo comando, riceve la risposta e la chiude.
Restituisce la risposta del mixer. 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 _connect(self):
"""Stabilisce la connessione se non già connesso."""
if self.socket is None:
print('Inizializzazione socket...')
try: try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5) self.socket.settimeout(5)
s.connect((host, port)) self.socket.connect((self.host, self.port))
s.sendall((command + '\n').encode('utf-8'))
print(f"-> Comando inviato al mixer: '{command}'")
response = s.recv(4096).decode('utf-8', errors='ignore').strip()
print(f"<- Risposta dal mixer: '{response}'")
if response.startswith("OK"):
return {"status": "success", "response": response}
else:
return {"status": "error", "response": response}
except socket.error as e: except socket.error as e:
print(f"Errore critico di connessione a {host}:{port} -> {e}") self.socket = None
error_msg = f"Impossibile connettersi al mixer: {e}" raise ConnectionError(f"Impossibile connettersi al mixer: {e}")
return {"status": "error", "response": error_msg}
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() # Forza riconnessione al prossimo tentativo
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: except Exception as e:
print(f"Errore imprevisto: {e}") print(f"⚠️ Errore nel caricamento della cache: {e}")
return {"status": "error", "response": str(e)} return {"channels": {}, "mixes": {}, "timestamp": 0}
def _save_cache(self):
"""Salva la cache nel file."""
try:
with open(CACHE_FILE, 'w', encoding='utf-8') as f:
json.dump(self._cache, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"⚠️ Errore nel salvataggio della cache: {e}")
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 _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 risposta."""
try:
start = response.find('"') + 1
end = response.rfind('"')
if start > 0 and end > start:
return response[start:end]
return "Sconosciuto"
except:
return "Errore"
def _parse_value(self, response: str) -> str:
"""Estrae l'ultimo valore da una risposta OK."""
parts = response.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:
return {
"status": "success",
"source": "cache",
**cached_data
}
# 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
return {
"status": "success",
"source": "mixer",
"channel": channel,
"name": name,
"on": is_on,
"level_db": 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:
return {
"status": "success",
"source": "cache",
**cached_data
}
# 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
return {
"status": "success",
"source": "mixer",
"mix": mix_number,
"name": name,
"on": is_on,
"level_db": 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():
found_channels.append({
"channel": info["channel"],
"name": info["name"],
"on": info["on"],
"level_db": info["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():
found_mixes.append({
"mix": info["mix"],
"name": info["name"],
"on": info["on"],
"level_db": info["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():
channels.append({
"channel": info["channel"],
"name": info["name"],
"on": info["on"]
})
# 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():
mixes.append({
"mix": info["mix"],
"name": info["name"],
"on": info["on"]
})
# 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)"
}