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

web console

Browse Source
This commit is contained in:
Nick
2026-04-13 20:57:13 +02:00
parent deee299af5
commit 84c1f0da35
9 changed files with 1023 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mixer_controller.py
+216 -248
View File
@@ -32,6 +32,8 @@ class TF5MixerController:
self._is_running = threading.Event()
self._is_connected = threading.Event()
self._init_meter_state()
self.start()
def start(self):
@@ -63,15 +65,22 @@ class TF5MixerController:
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()
# Piccola pausa per assicurarsi che il buffer sia pulito
time.sleep(0.2)
# Sottoscrivi DOPO il refresh (cache occupa il socket, meter devono aspettare)
self.subscribe_meters(interval_ms=100)
self._socket_reader() # blocca qui finché connesso
except socket.error as e:
print(f"🔥 Errore di connessione: {e}. Riprovo tra 5 secondi...")
finally:
self.unsubscribe_meters() # tenta di cancellare prima di chiudere
self._is_connected.clear()
if self.socket:
self.socket.close()
@@ -112,6 +121,16 @@ class TF5MixerController:
print(f"🔥 Errore di lettura dal socket: {e}")
break
def _send_raw(self, command: str):
"""Invia un comando senza attendere risposta."""
if not self._is_connected.is_set():
return
try:
print(f"SEND (raw): {command}")
self.socket.sendall((command + '\n').encode('utf-8'))
except socket.error as e:
print(f"⚠️ Errore invio raw: {e}")
def _handle_notify(self, message: str):
"""
Analizza un messaggio NOTIFY e aggiorna la cache in modo robusto.
@@ -120,6 +139,11 @@ class TF5MixerController:
print(f"RECV NOTIFY: {message}")
parts = message.split()
if len(parts) >= 2 and parts[1] == "mtrinfo":
self._handle_meter_notify(message)
return
if len(parts) < 2:
return
@@ -371,6 +395,68 @@ class TF5MixerController:
finally:
print(f" -> LOCK RILASCIATO.")
def _handle_prminfo_notify(self, message: str):
"""
Gestisce: NOTIFY prminfo <meter_id> <val0> <val1> ... <valN>
I valori sono: [ch0_type0, ch0_type1, ..., ch1_type0, ch1_type1, ...]
cioè interleaved per canale.
"""
parts = message.split()
# parts[0]="NOTIFY", parts[1]="prminfo", parts[2]=meter_id, parts[3:]=values
if len(parts) < 4:
return
try:
meter_id = int(parts[2])
except ValueError:
return
if meter_id not in self._METER_SUBSCRIPTIONS:
return
info = self._METER_SUBSCRIPTIONS[meter_id]
key = info["key"]
num_ch = info["channels"]
num_types = info["types"]
try:
values = [int(v) for v in parts[3:]]
except ValueError:
return
expected = num_ch * num_types
if len(values) < expected:
return
with self._meter_lock:
for ch in range(num_ch):
for t in range(num_types):
idx = ch * num_types + t
self._meter_data[key][t][ch] = values[idx]
def _handle_meter_notify(self, message: str):
parts = message.split()
# NOTIFY mtrinfo <id> <val0> <val1> ...
try:
meter_id = int(parts[2])
except (IndexError, ValueError):
return
if meter_id not in self._METER_SUBSCRIPTIONS:
return
info = self._METER_SUBSCRIPTIONS[meter_id]
key = info["key"]
num_ch = info["channels"]
num_types = info["types"]
try:
values = [int(v) for v in parts[3:]]
except ValueError:
return
if len(values) < num_ch * num_types:
return
with self._meter_lock:
for ch in range(num_ch):
for t in range(num_types):
self._meter_data[key][t][ch] = values[ch * num_types + t]
def _send_command(self, command: str) -> str:
"""Invia un comando al mixer e attende la risposta."""
if not self._is_connected.wait(timeout=5):
@@ -788,16 +874,20 @@ class TF5MixerController:
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())
items_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)"}
results = []
for info in items_copy:
ch_num = info.get("channel", 0)
results.append({
"channel": ch_num,
"name": info.get("name") or f"CH {ch_num}",
"on": info.get("on", False),
"level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))
})
results.sort(key=lambda x: x["channel"])
return {"status": "success", "total_channels": len(results), "channels": results, "cache_age_seconds": int(cache_age)}
def mute_multiple_channels(self, channels: List[int]) -> dict:
results = [self.set_channel_on_off(ch, False) for ch in channels]
@@ -963,19 +1053,22 @@ class TF5MixerController:
"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())
items_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)"}
results = []
for info in items_copy:
ch_num = info.get("channel", 0)
results.append({
"channel": ch_num,
"name": info.get("name") or f"ST IN {ch_num}",
"on": info.get("on", False),
"level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))
})
results.sort(key=lambda x: x["channel"])
return {"status": "success", "total_channels": len(results), "channels": results, "cache_age_seconds": int(cache_age)}
# =========================================================================
# FX RETURN CHANNELS (FxRtnCh)
@@ -1090,19 +1183,22 @@ class TF5MixerController:
"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())
items_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)"}
results = []
for info in items_copy:
ch_num = info.get("channel", 0)
results.append({
"channel": ch_num,
"name": info.get("name") or f"FX RTN {ch_num}",
"on": info.get("on", False),
"level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))
})
results.sort(key=lambda x: x["channel"])
return {"status": "success", "total_channels": len(results), "channels": results, "cache_age_seconds": int(cache_age)}
# =========================================================================
# DCA GROUPS
@@ -1158,14 +1254,26 @@ class TF5MixerController:
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"])
dcas = []
for info in dcas_copy:
try:
# Usiamo .get() con valori di default per evitare KeyError
dca_num = info.get("dca", 0)
dcas.append({
"dca": dca_num,
"name": info.get("name", f"DCA {dca_num}"), # Se manca il nome, usa "DCA 1" ecc.
"on": info.get("on", False), # Se manca lo stato, usa False
"level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))
})
except Exception as e:
print(f"⚠️ Errore nel processare DCA info: {e}")
dcas.sort(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)"}
"message": f"Riepilogo di {len(dcas)} DCA"}
# =========================================================================
# MIX BUSES
@@ -1248,16 +1356,20 @@ class TF5MixerController:
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())
items_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)"}
results = []
for info in items_copy:
mix_num = info.get("mix", 0)
results.append({
"mix": mix_num,
"name": info.get("name") or f"MIX {mix_num}",
"on": info.get("on", False),
"level_db": self._sanitize_value(self._normalize_level_from_cache(info.get("level_db")))
})
results.sort(key=lambda x: x["mix"])
return {"status": "success", "total_mixes": len(results), "mixes": results, "cache_age_seconds": int(cache_age)}
def search_mixes_by_name(self, search_term: str) -> dict:
if not self._is_cache_valid(): self.refresh_cache()
@@ -1585,212 +1697,68 @@ class TF5MixerController:
"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.
# METER READING (via prminfo subscription)
# =========================================================================
# 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"]},
_METER_SUBSCRIPTIONS = {
2000: {"key": "InCh", "channels": 40, "types": 3},
2001: {"key": "StInCh", "channels": 4, "types": 3},
2002: {"key": "FxRtnCh", "channels": 4, "types": 2},
2100: {"key": "Mix", "channels": 20, "types": 3},
2101: {"key": "Mtrx", "channels": 4, "types": 3},
2102: {"key": "St", "channels": 2, "types": 3},
2103: {"key": "Mono", "channels": 1, "types": 3},
}
# Dizionario meter: { "InCh": [[ch1t0, ch1t1, ch1t2], [ch2t0,...], ...], ... }
# Oppure più semplice: { "InCh": {type_idx: [val_ch1, val_ch2, ...]}, ... }
_meter_data: dict = {}
_meter_lock: threading.Lock = None # verrà inizializzato in __init__
def _init_meter_state(self):
"""Inizializza la struttura dati per i meter."""
self._meter_lock = threading.Lock()
self._meter_data = {}
for info in self._METER_SUBSCRIPTIONS.values():
key = info["key"]
self._meter_data[key] = {
t: [0] * info["channels"] for t in range(info["types"])
}
def subscribe_meters(self, interval_ms: int = 100):
# Il comando corretto per i meter Yamaha TF è "mtrinfo"
for meter_id in self._METER_SUBSCRIPTIONS:
self._send_raw(f"mtrinfo {meter_id} {interval_ms}")
time.sleep(0.05)
def unsubscribe_meters(self):
for meter_id in self._METER_SUBSCRIPTIONS:
self._send_raw(f"mtrinfo {meter_id} 0")
def get_meter_snapshot(self) -> dict:
"""Restituisce l'ultimo snapshot meter ricevuto."""
with self._meter_lock:
# Costruisce il formato atteso dal frontend
snapshot = {}
for meter_id, info in self._METER_SUBSCRIPTIONS.items():
key = info["key"]
frontend_key = key.lower()
# Il frontend si aspetta { readings: [{channel, raw, level_db}] }
type_idx = info["types"] - 1 # Usiamo l'ultimo tipo (PostOn)
readings = []
for ch_idx, raw in enumerate(self._meter_data[key][type_idx]):
readings.append({
"channel": ch_idx + 1,
"raw": raw,
"level_db": self._meter_raw_to_db(raw)
})
snapshot[frontend_key] = {
"readings": readings,
"type_name": "PostOn"
}
return snapshot
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()
}
}
return round((raw / 127.0) * 90.0 - 72.0, 1)