930 lines
41 KiB
Python
930 lines
41 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
╔══════════════════════════════════════════════════════════╗
|
||
║ TF5 MIXER — CLI INTERATTIVO DI TEST ║
|
||
║ Yamaha TF5 · Socket Protocol Explorer ║
|
||
╚══════════════════════════════════════════════════════════╝
|
||
|
||
Script per esplorare e testare le funzionalità del controller TF5.
|
||
Modalità: SIMULAZIONE (mock) o REALE (connessione socket).
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
import time
|
||
import math
|
||
import socket
|
||
import threading
|
||
import queue
|
||
try:
|
||
import readline # history navigazione CLI (Linux/macOS)
|
||
except ImportError:
|
||
pass # Windows: readline non disponibile, nessun problema
|
||
from typing import Optional, List
|
||
|
||
# ─────────────────────────────────────────────
|
||
# COLORI ANSI
|
||
# ─────────────────────────────────────────────
|
||
class C:
|
||
RESET = "\033[0m"
|
||
BOLD = "\033[1m"
|
||
DIM = "\033[2m"
|
||
RED = "\033[91m"
|
||
GREEN = "\033[92m"
|
||
YELLOW = "\033[93m"
|
||
BLUE = "\033[94m"
|
||
MAGENTA = "\033[95m"
|
||
CYAN = "\033[96m"
|
||
WHITE = "\033[97m"
|
||
GRAY = "\033[90m"
|
||
|
||
def cprint(text, color=C.WHITE, bold=False):
|
||
prefix = C.BOLD if bold else ""
|
||
print(f"{prefix}{color}{text}{C.RESET}")
|
||
|
||
def header(text):
|
||
w = 58
|
||
print()
|
||
cprint("╔" + "═" * w + "╗", C.CYAN)
|
||
cprint("║" + f" {text} ".center(w) + "║", C.CYAN, bold=True)
|
||
cprint("╚" + "═" * w + "╝", C.CYAN)
|
||
|
||
def section(text):
|
||
print()
|
||
cprint(f" ┌─ {text} {'─' * max(0, 50 - len(text))}", C.BLUE)
|
||
|
||
def ok(msg): cprint(f" ✓ {msg}", C.GREEN)
|
||
def err(msg): cprint(f" ✗ {msg}", C.RED)
|
||
def info(msg): cprint(f" · {msg}", C.GRAY)
|
||
def warn(msg): cprint(f" ⚠ {msg}", C.YELLOW)
|
||
def result(d):
|
||
"""Stampa un dict risultato in modo leggibile."""
|
||
if isinstance(d, dict):
|
||
status = d.get("status", "?")
|
||
color = C.GREEN if status == "success" else C.RED if status == "error" else C.YELLOW
|
||
cprint(f"\n STATUS: {status}", color, bold=True)
|
||
for k, v in d.items():
|
||
if k == "status": continue
|
||
if isinstance(v, list):
|
||
cprint(f" {k}:", C.CYAN)
|
||
for item in v[:20]: # max 20 righe
|
||
print(f" {item}")
|
||
if len(v) > 20:
|
||
cprint(f" ... e altri {len(v)-20} elementi", C.GRAY)
|
||
elif isinstance(v, dict):
|
||
cprint(f" {k}:", C.CYAN)
|
||
for sk, sv in v.items():
|
||
print(f" {sk}: {sv}")
|
||
else:
|
||
cprint(f" {k}: ", C.CYAN, bold=False)
|
||
print(f" {v}")
|
||
else:
|
||
print(f" {d}")
|
||
|
||
def meter_bar(raw, width=30):
|
||
"""Barra visuale per i meter (0-127)."""
|
||
if raw is None:
|
||
return f"[{'?':^{width}}]"
|
||
filled = int((raw / 127.0) * width)
|
||
# Colore: verde → giallo → rosso
|
||
if raw < 80: col = C.GREEN
|
||
elif raw < 110: col = C.YELLOW
|
||
else: col = C.RED
|
||
bar = "█" * filled + "░" * (width - filled)
|
||
return f"{col}[{bar}]{C.RESET} {raw:3d}/127"
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MOCK CONTROLLER (per test senza mixer fisico)
|
||
# ─────────────────────────────────────────────
|
||
class MockTF5Controller:
|
||
"""
|
||
Simulazione del TF5MixerController per testing offline.
|
||
Risponde con dati fake realistici.
|
||
"""
|
||
METER_DEFS = {
|
||
"InCh": {"channels": 40, "types": ["PreHPF", "PreFader", "PostOn"]},
|
||
"StInCh": {"channels": 4, "types": ["PreEQ", "PreFader", "PostOn"]},
|
||
"FxRtnCh": {"channels": 4, "types": ["PreFader", "PostOn"]},
|
||
"Mix": {"channels": 20, "types": ["PreEQ", "PreFader", "PostOn"]},
|
||
"Mtrx": {"channels": 4, "types": ["PreEQ", "PreFader", "PostOn"]},
|
||
"St": {"channels": 2, "types": ["PreEQ", "PreFader", "PostOn"]},
|
||
"Mono": {"channels": 1, "types": ["PreEQ", "PreFader", "PostOn"]},
|
||
}
|
||
|
||
CH_NAMES = ["Kick", "Snare", "HiHat", "OH L", "OH R", "Tom1", "Tom2", "Tom3",
|
||
"Bass", "Gtr L", "Gtr R", "Keys L", "Keys R", "Vox 1", "Vox 2",
|
||
"Vox 3", "BGV1", "BGV2", "BGV3", "Sax", "Trumpet", "Trombone",
|
||
"Perc", "Click", "Playback"] + [f"Ch {i}" for i in range(26, 41)]
|
||
|
||
def __init__(self):
|
||
import random
|
||
self._r = random
|
||
self._state = {
|
||
"channels": {str(i): {"channel": i, "name": self.CH_NAMES[i-1],
|
||
"on": True, "level_db": -10.0 + self._r.uniform(-20, 10),
|
||
"pan": self._r.randint(-63, 63)} for i in range(1, 41)},
|
||
"mixes": {str(i): {"mix": i, "name": f"MX {i}",
|
||
"on": True, "level_db": 0.0} for i in range(1, 21)},
|
||
"steinch": {str(i): {"channel": i, "name": f"Rt{i}L",
|
||
"on": True, "level_db": -12.0} for i in range(1, 5)},
|
||
"fxrtn": {str(i): {"channel": i, "name": f"Fx{i}L",
|
||
"on": True, "level_db": -6.0} for i in range(1, 5)},
|
||
"dcas": {str(i): {"dca": i, "name": f"DCA {i}",
|
||
"on": True, "level_db": 0.0} for i in range(1, 9)},
|
||
"matrices": {str(i): {"matrix": i, "name": f"MT {i}",
|
||
"on": True, "level_db": 0.0} for i in range(1, 5)},
|
||
"stereo": {"1": {"bus": 1, "name": "ST L", "on": True, "level_db": 0.0},
|
||
"2": {"bus": 2, "name": "ST R", "on": True, "level_db": 0.0}},
|
||
"mono": {"name": "MONO", "on": True, "level_db": -6.0},
|
||
"mute_masters": {str(i): {"group": i, "name": f"MUTE {i}", "on": False} for i in range(1, 7)},
|
||
}
|
||
warn("Modalità SIMULAZIONE attiva (nessun mixer fisico collegato)")
|
||
|
||
def _fake_raw(self):
|
||
import random
|
||
return random.randint(0, 127)
|
||
|
||
def _meter_raw_to_db(self, raw):
|
||
if raw <= 0: return float('-inf')
|
||
return round((raw / 127.0) * 90.0 - 72.0, 1)
|
||
|
||
def get_meter(self, section, channel, meter_type=2):
|
||
if section not in self.METER_DEFS:
|
||
return {"status": "error", "message": f"Sezione '{section}' non valida"}
|
||
defn = self.METER_DEFS[section]
|
||
if not 1 <= channel <= defn["channels"]:
|
||
return {"status": "error", "message": f"Canale fuori range (1-{defn['channels']})"}
|
||
if not 0 <= meter_type < len(defn["types"]):
|
||
return {"status": "error", "message": f"Tipo fuori range (0-{len(defn['types'])-1})"}
|
||
raw = self._fake_raw()
|
||
db = self._meter_raw_to_db(raw)
|
||
return {"status": "success", "section": section, "channel": channel,
|
||
"type": meter_type, "type_name": defn["types"][meter_type],
|
||
"raw": raw, "level_db": db,
|
||
"message": f"{section} ch{channel} [{defn['types'][meter_type]}]: {raw}/127 → {db} dBFS"}
|
||
|
||
def get_all_meters(self, section, meter_type=2):
|
||
if section not in self.METER_DEFS:
|
||
return {"status": "error", "message": f"Sezione '{section}' non valida"}
|
||
defn = self.METER_DEFS[section]
|
||
safe_type = min(meter_type, len(defn["types"]) - 1)
|
||
readings = []
|
||
for ch in range(1, defn["channels"] + 1):
|
||
raw = self._fake_raw()
|
||
readings.append({"channel": ch, "raw": raw, "level_db": self._meter_raw_to_db(raw)})
|
||
return {"status": "success", "section": section, "type": safe_type,
|
||
"type_name": defn["types"][safe_type], "total_channels": defn["channels"],
|
||
"readings": readings, "message": f"Meter {section}: {defn['channels']} canali"}
|
||
|
||
def get_meter_snapshot(self, meter_type=2):
|
||
snapshot = {}
|
||
for section in self.METER_DEFS:
|
||
snapshot[section] = self.get_all_meters(section, meter_type)
|
||
return {"status": "success", "sections": snapshot,
|
||
"message": f"Snapshot meter completo ({len(self.METER_DEFS)} sezioni)"}
|
||
|
||
def get_meter_types(self, section=None):
|
||
if section:
|
||
if section not in self.METER_DEFS:
|
||
return {"status": "error", "message": f"Sezione '{section}' non valida"}
|
||
d = self.METER_DEFS[section]
|
||
return {"status": "success", "section": section,
|
||
"channels": d["channels"], "types": {i: n for i, n in enumerate(d["types"])}}
|
||
return {"status": "success", "sections": {
|
||
s: {"channels": d["channels"], "types": {i: n for i, n in enumerate(d["types"])}}
|
||
for s, d in self.METER_DEFS.items()}}
|
||
|
||
def get_channel_info(self, channel, force_refresh=False):
|
||
if not 1 <= channel <= 40:
|
||
return {"status": "error", "message": "Canale deve essere tra 1 e 40"}
|
||
return {"status": "success", "source": "mock", **self._state["channels"][str(channel)]}
|
||
|
||
def get_all_channels_summary(self):
|
||
return {"status": "success", "total_channels": 40,
|
||
"channels": list(self._state["channels"].values()),
|
||
"cache_age_seconds": 0, "message": "40 canali (mock)"}
|
||
|
||
def set_channel_level(self, channel, level_db):
|
||
if not 1 <= channel <= 40:
|
||
return {"status": "error", "message": "Canale deve essere tra 1 e 40"}
|
||
self._state["channels"][str(channel)]["level_db"] = level_db
|
||
return {"status": "success", "message": f"Canale {channel} → {level_db:+.1f} dB", "response": "OK (mock)"}
|
||
|
||
def set_channel_on_off(self, channel, state):
|
||
if not 1 <= channel <= 40:
|
||
return {"status": "error", "message": "Canale deve essere tra 1 e 40"}
|
||
self._state["channels"][str(channel)]["on"] = state
|
||
return {"status": "success", "message": f"Canale {channel} {'ON' if state else 'OFF'}", "response": "OK (mock)"}
|
||
|
||
def set_channel_pan(self, channel, pan_value):
|
||
if not 1 <= channel <= 40:
|
||
return {"status": "error", "message": "Canale deve essere tra 1 e 40"}
|
||
self._state["channels"][str(channel)]["pan"] = pan_value
|
||
return {"status": "success", "message": f"Canale {channel} pan → {pan_value}", "response": "OK (mock)"}
|
||
|
||
def get_mix_info(self, mix_number, force_refresh=False):
|
||
if not 1 <= mix_number <= 20:
|
||
return {"status": "error", "message": "Mix deve essere tra 1 e 20"}
|
||
return {"status": "success", "source": "mock", **self._state["mixes"][str(mix_number)]}
|
||
|
||
def get_all_mixes_summary(self):
|
||
return {"status": "success", "total_mixes": 20,
|
||
"mixes": list(self._state["mixes"].values()),
|
||
"cache_age_seconds": 0, "message": "20 mix (mock)"}
|
||
|
||
def set_mix_level(self, mix_number, level_db):
|
||
if not 1 <= mix_number <= 20:
|
||
return {"status": "error", "message": "Mix deve essere tra 1 e 20"}
|
||
self._state["mixes"][str(mix_number)]["level_db"] = level_db
|
||
return {"status": "success", "message": f"Mix {mix_number} → {level_db:+.1f} dB", "response": "OK (mock)"}
|
||
|
||
def set_mix_on_off(self, mix_number, state):
|
||
if not 1 <= mix_number <= 20:
|
||
return {"status": "error", "message": "Mix deve essere tra 1 e 20"}
|
||
self._state["mixes"][str(mix_number)]["on"] = state
|
||
return {"status": "success", "message": f"Mix {mix_number} {'ON' if state else 'OFF'}", "response": "OK (mock)"}
|
||
|
||
def get_dca_info(self, dca, force_refresh=False):
|
||
if not 1 <= dca <= 8:
|
||
return {"status": "error", "message": "DCA deve essere tra 1 e 8"}
|
||
return {"status": "success", "source": "mock", **self._state["dcas"][str(dca)]}
|
||
|
||
def get_all_dca_summary(self):
|
||
return {"status": "success", "total_dcas": 8,
|
||
"dcas": list(self._state["dcas"].values()),
|
||
"cache_age_seconds": 0, "message": "8 DCA (mock)"}
|
||
|
||
def set_dca_level(self, dca, level_db):
|
||
if not 1 <= dca <= 8:
|
||
return {"status": "error", "message": "DCA deve essere tra 1 e 8"}
|
||
self._state["dcas"][str(dca)]["level_db"] = level_db
|
||
return {"status": "success", "message": f"DCA {dca} → {level_db:+.1f} dB", "response": "OK (mock)"}
|
||
|
||
def set_dca_on_off(self, dca, state):
|
||
if not 1 <= dca <= 8:
|
||
return {"status": "error", "message": "DCA deve essere tra 1 e 8"}
|
||
self._state["dcas"][str(dca)]["on"] = state
|
||
return {"status": "success", "message": f"DCA {dca} {'ON' if state else 'OFF'}", "response": "OK (mock)"}
|
||
|
||
def get_mono_info(self, force_refresh=False):
|
||
return {"status": "success", "source": "mock", **self._state["mono"]}
|
||
|
||
def set_mono_level(self, level_db):
|
||
self._state["mono"]["level_db"] = level_db
|
||
return {"status": "success", "message": f"Mono → {level_db:+.1f} dB", "response": "OK (mock)"}
|
||
|
||
def set_mono_on_off(self, state):
|
||
self._state["mono"]["on"] = state
|
||
return {"status": "success", "message": f"Mono {'ON' if state else 'OFF'}", "response": "OK (mock)"}
|
||
|
||
def get_stereo_info(self, bus=1, force_refresh=False):
|
||
if not 1 <= bus <= 2:
|
||
return {"status": "error", "message": "Bus stereo deve essere 1 o 2"}
|
||
return {"status": "success", "source": "mock", **self._state["stereo"][str(bus)]}
|
||
|
||
def set_stereo_level(self, bus, level_db):
|
||
if not 1 <= bus <= 2:
|
||
return {"status": "error", "message": "Bus stereo deve essere 1 o 2"}
|
||
self._state["stereo"][str(bus)]["level_db"] = level_db
|
||
return {"status": "success", "message": f"St {bus} → {level_db:+.1f} dB", "response": "OK (mock)"}
|
||
|
||
def get_steinch_info(self, channel, force_refresh=False):
|
||
if not 1 <= channel <= 4:
|
||
return {"status": "error", "message": "StInCh deve essere tra 1 e 4"}
|
||
return {"status": "success", "source": "mock", **self._state["steinch"][str(channel)]}
|
||
|
||
def get_fxrtn_info(self, channel, force_refresh=False):
|
||
if not 1 <= channel <= 4:
|
||
return {"status": "error", "message": "FxRtnCh deve essere tra 1 e 4"}
|
||
return {"status": "success", "source": "mock", **self._state["fxrtn"][str(channel)]}
|
||
|
||
def get_mute_master_state(self, group):
|
||
if not 1 <= group <= 6:
|
||
return {"status": "error", "message": "Mute group deve essere tra 1 e 6"}
|
||
return {"status": "success", "source": "mock", **self._state["mute_masters"][str(group)]}
|
||
|
||
def set_mute_master(self, group, state):
|
||
if not 1 <= group <= 6:
|
||
return {"status": "error", "message": "Mute group deve essere tra 1 e 6"}
|
||
self._state["mute_masters"][str(group)]["on"] = state
|
||
return {"status": "success", "message": f"MuteMaster {group} {'ON' if state else 'OFF'}", "response": "OK (mock)"}
|
||
|
||
def get_all_mute_masters_summary(self):
|
||
return {"status": "success", "total_groups": 6,
|
||
"groups": list(self._state["mute_masters"].values()),
|
||
"message": "6 Mute Master (mock)"}
|
||
|
||
def search_channels_by_name(self, term):
|
||
found = [v for v in self._state["channels"].values()
|
||
if term.lower() in v.get("name", "").lower()]
|
||
return {"status": "success", "found_count": len(found), "channels": found}
|
||
|
||
def refresh_cache(self):
|
||
return {"status": "success", "message": "Cache refresh (mock) completato"}
|
||
|
||
def recall_scene(self, bank, scene_number):
|
||
return {"status": "success", "message": f"Scena {bank.upper()}{scene_number} richiamata (mock)", "response": "OK (mock)"}
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# CARICAMENTO CONTROLLER REALE
|
||
# ─────────────────────────────────────────────
|
||
def try_load_real_controller(host, port):
|
||
"""Prova a importare e istanziare il TF5MixerController reale."""
|
||
try:
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from mixer_controller import TF5MixerController
|
||
ctrl = TF5MixerController(host=host, port=port)
|
||
return ctrl, None
|
||
except ImportError as e:
|
||
return None, f"Import fallito: {e}"
|
||
except Exception as e:
|
||
return None, f"Connessione fallita: {e}"
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# HELPERS INPUT
|
||
# ─────────────────────────────────────────────
|
||
def ask(prompt, default=None):
|
||
try:
|
||
dp = f" [{default}]" if default is not None else ""
|
||
val = input(f"{C.CYAN} → {prompt}{dp}: {C.RESET}").strip()
|
||
return val if val else str(default) if default is not None else ""
|
||
except (EOFError, KeyboardInterrupt):
|
||
return str(default) if default is not None else ""
|
||
|
||
def ask_int(prompt, default=None, min_val=None, max_val=None):
|
||
while True:
|
||
raw = ask(prompt, default)
|
||
try:
|
||
v = int(raw)
|
||
if min_val is not None and v < min_val:
|
||
err(f"Valore minimo: {min_val}")
|
||
continue
|
||
if max_val is not None and v > max_val:
|
||
err(f"Valore massimo: {max_val}")
|
||
continue
|
||
return v
|
||
except ValueError:
|
||
err("Inserisci un numero intero valido")
|
||
|
||
def ask_float(prompt, default=None):
|
||
while True:
|
||
raw = ask(prompt, default)
|
||
try:
|
||
return float(raw)
|
||
except ValueError:
|
||
err("Inserisci un numero decimale valido (es. -10.5)")
|
||
|
||
def ask_bool(prompt, default=True):
|
||
d = "s/N" if not default else "S/n"
|
||
raw = ask(f"{prompt} ({d})", "s" if default else "n").lower()
|
||
return raw in ("s", "si", "sì", "y", "yes", "1")
|
||
|
||
def pause():
|
||
try:
|
||
input(f"\n{C.GRAY} [Invio per continuare...]{C.RESET}")
|
||
except (EOFError, KeyboardInterrupt):
|
||
pass
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU METER
|
||
# ─────────────────────────────────────────────
|
||
def menu_meter(ctrl):
|
||
SECTIONS = ["InCh", "StInCh", "FxRtnCh", "Mix", "Mtrx", "St", "Mono"]
|
||
|
||
while True:
|
||
header("METER READING")
|
||
cprint(" 1) Leggi un singolo canale meter", C.WHITE)
|
||
cprint(" 2) Leggi tutti i canali di una sezione", C.WHITE)
|
||
cprint(" 3) Snapshot completo (tutte le sezioni)", C.WHITE)
|
||
cprint(" 4) Visualizza tipi meter disponibili", C.WHITE)
|
||
cprint(" 5) Monitor live (aggiornamento continuo)", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
|
||
elif cmd == "1":
|
||
section("Meter singolo canale")
|
||
cprint(f" Sezioni: {', '.join(SECTIONS)}", C.GRAY)
|
||
sec = ask("Sezione", "InCh")
|
||
if sec not in SECTIONS:
|
||
err(f"Sezione non valida"); continue
|
||
defs = ctrl.METER_DEFS if hasattr(ctrl, 'METER_DEFS') else MockTF5Controller.METER_DEFS
|
||
defn = defs.get(sec, {})
|
||
max_ch = defn.get("channels", 40)
|
||
types = defn.get("types", [])
|
||
cprint(f" Tipi disponibili: {', '.join(f'{i}={t}' for i,t in enumerate(types))}", C.GRAY)
|
||
ch = ask_int("Canale", 1, 1, max_ch)
|
||
mt = ask_int("Tipo meter", len(types)-1, 0, len(types)-1)
|
||
r = ctrl.get_meter(sec, ch, mt)
|
||
result(r)
|
||
if r.get("status") == "success":
|
||
raw = r.get("raw", 0)
|
||
print(f"\n {meter_bar(raw)}")
|
||
pause()
|
||
|
||
elif cmd == "2":
|
||
section("Meter sezione completa")
|
||
cprint(f" Sezioni: {', '.join(SECTIONS)}", C.GRAY)
|
||
sec = ask("Sezione", "InCh")
|
||
if sec not in SECTIONS:
|
||
err(f"Sezione non valida"); continue
|
||
defs = ctrl.METER_DEFS if hasattr(ctrl, 'METER_DEFS') else MockTF5Controller.METER_DEFS
|
||
defn = defs.get(sec, {})
|
||
types = defn.get("types", [])
|
||
cprint(f" Tipi disponibili: {', '.join(f'{i}={t}' for i,t in enumerate(types))}", C.GRAY)
|
||
mt = ask_int("Tipo meter", len(types)-1, 0, len(types)-1)
|
||
r = ctrl.get_all_meters(sec, mt)
|
||
if r.get("status") == "success":
|
||
print()
|
||
cprint(f" {'CH':>4} {'RAW':>5} {'dBFS':>8} LIVELLO", C.CYAN, bold=True)
|
||
cprint(" " + "─" * 52, C.GRAY)
|
||
for rd in r.get("readings", []):
|
||
ch = rd["channel"]
|
||
raw = rd.get("raw")
|
||
db = rd.get("level_db")
|
||
db_str = f"{db:+6.1f}" if db is not None and db != float('-inf') else " -inf"
|
||
bar = meter_bar(raw, 20)
|
||
print(f" {ch:>4} {(raw or 0):>5} {db_str} dB {bar}")
|
||
else:
|
||
result(r)
|
||
pause()
|
||
|
||
elif cmd == "3":
|
||
section("Snapshot completo")
|
||
defs = ctrl.METER_DEFS if hasattr(ctrl, 'METER_DEFS') else MockTF5Controller.METER_DEFS
|
||
mt = ask_int("Tipo meter preferito (verrà clampato per sezione)", 2, 0, 2)
|
||
r = ctrl.get_meter_snapshot(mt)
|
||
if r.get("status") == "success":
|
||
for sec_name, sec_data in r.get("sections", {}).items():
|
||
readings = sec_data.get("readings", [])
|
||
type_name = sec_data.get("type_name", "?")
|
||
if readings:
|
||
avg_raw = sum(rd.get("raw") or 0 for rd in readings) / len(readings)
|
||
peak_raw = max((rd.get("raw") or 0 for rd in readings), default=0)
|
||
print(f"\n {C.CYAN}{C.BOLD}{sec_name:10}{C.RESET} [{type_name}] "
|
||
f"ch:{len(readings)} avg:{avg_raw:5.1f} peak:{meter_bar(peak_raw, 15)}")
|
||
else:
|
||
result(r)
|
||
pause()
|
||
|
||
elif cmd == "4":
|
||
section("Tipi meter disponibili")
|
||
defs = ctrl.METER_DEFS if hasattr(ctrl, 'METER_DEFS') else MockTF5Controller.METER_DEFS
|
||
r = ctrl.get_meter_types()
|
||
print()
|
||
cprint(f" {'SEZIONE':12} {'CH':>5} TIPI DISPONIBILI", C.CYAN, bold=True)
|
||
cprint(" " + "─" * 50, C.GRAY)
|
||
if r.get("status") == "success":
|
||
for sec_name, data in r.get("sections", {}).items():
|
||
types_str = " ".join(f"{i}={t}" for i, t in data.get("types", {}).items())
|
||
print(f" {sec_name:12} {data.get('channels', '?'):>5} {types_str}")
|
||
pause()
|
||
|
||
elif cmd == "5":
|
||
section("Monitor live meter")
|
||
cprint(" Premi CTRL+C per fermare", C.GRAY)
|
||
defs = ctrl.METER_DEFS if hasattr(ctrl, 'METER_DEFS') else MockTF5Controller.METER_DEFS
|
||
sec = ask("Sezione da monitorare", "InCh")
|
||
if sec not in SECTIONS:
|
||
err("Sezione non valida"); continue
|
||
mt = len(defs.get(sec, {}).get("types", [])) - 1
|
||
interval = ask_float("Intervallo secondi", 1.0)
|
||
try:
|
||
iteration = 0
|
||
while True:
|
||
r = ctrl.get_all_meters(sec, mt)
|
||
os.system("clear" if os.name != "nt" else "cls")
|
||
header(f"LIVE METER — {sec} [{r.get('type_name','?')}]")
|
||
cprint(f" Aggiornamento #{iteration+1} · {time.strftime('%H:%M:%S')}", C.GRAY)
|
||
print()
|
||
if r.get("status") == "success":
|
||
for rd in r.get("readings", []):
|
||
ch = rd["channel"]
|
||
raw = rd.get("raw")
|
||
db = rd.get("level_db")
|
||
db_str = f"{db:+6.1f}" if db is not None and db != float('-inf') else " -inf"
|
||
bar = meter_bar(raw, 25)
|
||
print(f" ch{ch:02d} {db_str} dBFS {bar}")
|
||
time.sleep(interval)
|
||
iteration += 1
|
||
except KeyboardInterrupt:
|
||
ok("Monitor fermato")
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU CANALI
|
||
# ─────────────────────────────────────────────
|
||
def menu_channels(ctrl):
|
||
while True:
|
||
header("CANALI INPUT (InCh)")
|
||
cprint(" 1) Info canale", C.WHITE)
|
||
cprint(" 2) Imposta livello fader", C.WHITE)
|
||
cprint(" 3) Abilita / Disabilita canale", C.WHITE)
|
||
cprint(" 4) Imposta pan", C.WHITE)
|
||
cprint(" 5) Riepilogo tutti i canali", C.WHITE)
|
||
cprint(" 6) Cerca canale per nome", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
ch = ask_int("Numero canale", 1, 1, 40)
|
||
result(ctrl.get_channel_info(ch))
|
||
pause()
|
||
elif cmd == "2":
|
||
ch = ask_int("Numero canale", 1, 1, 40)
|
||
db = ask_float("Livello dB (es. -10, 0, +5)", -10)
|
||
result(ctrl.set_channel_level(ch, db))
|
||
pause()
|
||
elif cmd == "3":
|
||
ch = ask_int("Numero canale", 1, 1, 40)
|
||
state = ask_bool("Abilitare il canale?", True)
|
||
result(ctrl.set_channel_on_off(ch, state))
|
||
pause()
|
||
elif cmd == "4":
|
||
ch = ask_int("Numero canale", 1, 1, 40)
|
||
pan = ask_int("Pan (-63=L, 0=C, +63=R)", 0, -63, 63)
|
||
result(ctrl.set_channel_pan(ch, pan))
|
||
pause()
|
||
elif cmd == "5":
|
||
r = ctrl.get_all_channels_summary()
|
||
if r.get("status") == "success":
|
||
print()
|
||
cprint(f" {'CH':>4} {'NOME':15} {'ON':>4} {'dB':>8}", C.CYAN, bold=True)
|
||
cprint(" " + "─" * 40, C.GRAY)
|
||
for ch in r.get("channels", []):
|
||
on_str = f"{C.GREEN}ON {C.RESET}" if ch["on"] else f"{C.RED}OFF{C.RESET}"
|
||
db = ch.get("level_db", -120)
|
||
db_str = f"{db:+6.1f}" if db > -120 else " -inf"
|
||
print(f" {ch['channel']:>4} {ch['name']:15} {on_str} {db_str} dB")
|
||
pause()
|
||
elif cmd == "6":
|
||
term = ask("Nome da cercare")
|
||
result(ctrl.search_channels_by_name(term))
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU MIX BUS
|
||
# ─────────────────────────────────────────────
|
||
def menu_mixes(ctrl):
|
||
while True:
|
||
header("MIX BUSES")
|
||
cprint(" 1) Info mix bus", C.WHITE)
|
||
cprint(" 2) Imposta livello fader", C.WHITE)
|
||
cprint(" 3) Abilita / Disabilita mix bus", C.WHITE)
|
||
cprint(" 4) Riepilogo tutti i mix", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
m = ask_int("Numero mix", 1, 1, 20)
|
||
result(ctrl.get_mix_info(m))
|
||
pause()
|
||
elif cmd == "2":
|
||
m = ask_int("Numero mix", 1, 1, 20)
|
||
db = ask_float("Livello dB", 0)
|
||
result(ctrl.set_mix_level(m, db))
|
||
pause()
|
||
elif cmd == "3":
|
||
m = ask_int("Numero mix", 1, 1, 20)
|
||
state = ask_bool("Abilitare?", True)
|
||
result(ctrl.set_mix_on_off(m, state))
|
||
pause()
|
||
elif cmd == "4":
|
||
r = ctrl.get_all_mixes_summary()
|
||
if r.get("status") == "success":
|
||
print()
|
||
cprint(f" {'MIX':>5} {'NOME':15} {'ON':>4} {'dB':>8}", C.CYAN, bold=True)
|
||
cprint(" " + "─" * 40, C.GRAY)
|
||
for m in r.get("mixes", []):
|
||
on_str = f"{C.GREEN}ON {C.RESET}" if m["on"] else f"{C.RED}OFF{C.RESET}"
|
||
db = m.get("level_db", -120)
|
||
db_str = f"{db:+6.1f}" if db > -120 else " -inf"
|
||
print(f" {m['mix']:>5} {m['name']:15} {on_str} {db_str} dB")
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU DCA
|
||
# ─────────────────────────────────────────────
|
||
def menu_dca(ctrl):
|
||
while True:
|
||
header("DCA GROUPS")
|
||
cprint(" 1) Info DCA", C.WHITE)
|
||
cprint(" 2) Imposta livello", C.WHITE)
|
||
cprint(" 3) Abilita / Disabilita DCA", C.WHITE)
|
||
cprint(" 4) Riepilogo tutti i DCA", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
d = ask_int("Numero DCA", 1, 1, 8)
|
||
result(ctrl.get_dca_info(d))
|
||
pause()
|
||
elif cmd == "2":
|
||
d = ask_int("Numero DCA", 1, 1, 8)
|
||
db = ask_float("Livello dB", 0)
|
||
result(ctrl.set_dca_level(d, db))
|
||
pause()
|
||
elif cmd == "3":
|
||
d = ask_int("Numero DCA", 1, 1, 8)
|
||
state = ask_bool("Abilitare?", True)
|
||
result(ctrl.set_dca_on_off(d, state))
|
||
pause()
|
||
elif cmd == "4":
|
||
r = ctrl.get_all_dca_summary()
|
||
if r.get("status") == "success":
|
||
print()
|
||
cprint(f" {'DCA':>5} {'NOME':15} {'ON':>4} {'dB':>8}", C.CYAN, bold=True)
|
||
cprint(" " + "─" * 40, C.GRAY)
|
||
for d in r.get("dcas", []):
|
||
on_str = f"{C.GREEN}ON {C.RESET}" if d["on"] else f"{C.RED}OFF{C.RESET}"
|
||
db = d.get("level_db", 0)
|
||
print(f" {d['dca']:>5} {d['name']:15} {on_str} {db:+6.1f} dB")
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU BUS MASTER
|
||
# ─────────────────────────────────────────────
|
||
def menu_masters(ctrl):
|
||
while True:
|
||
header("BUS MASTER (Stereo / Mono)")
|
||
cprint(" 1) Info Stereo bus", C.WHITE)
|
||
cprint(" 2) Imposta livello Stereo", C.WHITE)
|
||
cprint(" 3) Info Mono bus", C.WHITE)
|
||
cprint(" 4) Imposta livello Mono", C.WHITE)
|
||
cprint(" 5) Mute Masters — stato", C.WHITE)
|
||
cprint(" 6) Attiva / Disattiva Mute Master group", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
bus = ask_int("Bus stereo (1=L, 2=R)", 1, 1, 2)
|
||
result(ctrl.get_stereo_info(bus))
|
||
pause()
|
||
elif cmd == "2":
|
||
bus = ask_int("Bus stereo (1=L, 2=R)", 1, 1, 2)
|
||
db = ask_float("Livello dB", 0)
|
||
result(ctrl.set_stereo_level(bus, db))
|
||
pause()
|
||
elif cmd == "3":
|
||
result(ctrl.get_mono_info())
|
||
pause()
|
||
elif cmd == "4":
|
||
db = ask_float("Livello dB", -6)
|
||
result(ctrl.set_mono_level(db))
|
||
pause()
|
||
elif cmd == "5":
|
||
result(ctrl.get_all_mute_masters_summary())
|
||
pause()
|
||
elif cmd == "6":
|
||
g = ask_int("Numero gruppo (1-6)", 1, 1, 6)
|
||
state = ask_bool("Attivare il mute?", False)
|
||
result(ctrl.set_mute_master(g, state))
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU COMANDI RAW
|
||
# ─────────────────────────────────────────────
|
||
def menu_raw(ctrl):
|
||
"""Invia comandi raw al mixer (solo modalità reale)."""
|
||
if isinstance(ctrl, MockTF5Controller):
|
||
warn("I comandi raw non sono disponibili in modalità simulazione.")
|
||
pause()
|
||
return
|
||
|
||
header("COMANDI RAW")
|
||
cprint(" Invia comandi testuali diretti al mixer.", C.GRAY)
|
||
cprint(" Esempi:", C.GRAY)
|
||
cprint(" get MIXER:Current/InCh/Fader/Level 0 0", C.GRAY)
|
||
cprint(" set MIXER:Current/InCh/Fader/On 0 0 0", C.GRAY)
|
||
cprint(" get MIXER:Current/Meter/InCh 0 2", C.GRAY)
|
||
cprint(" Digita 'exit' per uscire.", C.GRAY)
|
||
print()
|
||
|
||
while True:
|
||
try:
|
||
cmd = input(f"{C.MAGENTA} RAW> {C.RESET}").strip()
|
||
if not cmd:
|
||
continue
|
||
if cmd.lower() in ("exit", "quit", "0"):
|
||
break
|
||
response = ctrl._send_command(cmd)
|
||
cprint(f" ← {response}", C.YELLOW)
|
||
except KeyboardInterrupt:
|
||
break
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU SCENE
|
||
# ─────────────────────────────────────────────
|
||
def menu_scenes(ctrl):
|
||
while True:
|
||
header("SCENE")
|
||
cprint(" 1) Richiama scena (banco A o B)", C.WHITE)
|
||
cprint(" 2) Refresh cache", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
bank = ask("Banco (a/b)", "a").lower()
|
||
num = ask_int("Numero scena (0-99)", 0, 0, 99)
|
||
result(ctrl.recall_scene(bank, num))
|
||
pause()
|
||
elif cmd == "2":
|
||
ok("Avvio refresh cache...")
|
||
result(ctrl.refresh_cache())
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MENU ALTRI CANALI
|
||
# ─────────────────────────────────────────────
|
||
def menu_other_channels(ctrl):
|
||
while True:
|
||
header("ALTRI CANALI")
|
||
cprint(" 1) Stereo Input (StInCh) — info", C.WHITE)
|
||
cprint(" 2) FX Return (FxRtnCh) — info", C.WHITE)
|
||
cprint(" 0) ← Torna al menu principale", C.GRAY)
|
||
|
||
cmd = ask("\nScelta")
|
||
|
||
if cmd == "0":
|
||
break
|
||
elif cmd == "1":
|
||
ch = ask_int("Canale StInCh (1-4)", 1, 1, 4)
|
||
result(ctrl.get_steinch_info(ch))
|
||
pause()
|
||
elif cmd == "2":
|
||
ch = ask_int("Canale FxRtnCh (1-4)", 1, 1, 4)
|
||
result(ctrl.get_fxrtn_info(ch))
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# HELP / REFERENCE
|
||
# ─────────────────────────────────────────────
|
||
def show_help():
|
||
header("REFERENCE — PROTOCOLLO TF5")
|
||
section("Struttura comandi")
|
||
print("""
|
||
get <PATH> <IDX1> <IDX2> → legge un parametro
|
||
set <PATH> <IDX1> <IDX2> <V> → scrive un parametro
|
||
ssrecall_ex scene_a <N> → richiama scena banco A
|
||
|
||
Risposta OK: OK <valore>
|
||
Risposta ERR: ERROR <codice> <messaggio>
|
||
Notify: NOTIFY <PATH> <IDX1> <IDX2> <V>
|
||
""")
|
||
section("Livelli fader")
|
||
print("""
|
||
Valore interno: intero × 100 (es. -1000 = -10.0 dB)
|
||
Minimo: -32768 = -∞ (fader giù)
|
||
Massimo: +1000 = +10.0 dB
|
||
Unity gain: 0 = 0.0 dB
|
||
""")
|
||
section("Meter")
|
||
print("""
|
||
Valore raw: 0–127 (intero)
|
||
Conversione: dBFS = (raw/127) × 90 - 72
|
||
0 → -inf dBFS
|
||
76 → ~ 0 dBFS (approssimato)
|
||
127→ +18 dBFS
|
||
|
||
SEZIONI METER:
|
||
┌──────────┬─────┬────────────────────────────────────┐
|
||
│ InCh │ 40 │ 0=PreHPF 1=PreFader 2=PostOn │
|
||
│ StInCh │ 4 │ 0=PreEQ 1=PreFader 2=PostOn │
|
||
│ FxRtnCh │ 4 │ 0=PreFader 1=PostOn │
|
||
│ Mix │ 20 │ 0=PreEQ 1=PreFader 2=PostOn │
|
||
│ Mtrx │ 4 │ 0=PreEQ 1=PreFader 2=PostOn │
|
||
│ St │ 2 │ 0=PreEQ 1=PreFader 2=PostOn │
|
||
│ Mono │ 1 │ 0=PreEQ 1=PreFader 2=PostOn │
|
||
└──────────┴─────┴────────────────────────────────────┘
|
||
""")
|
||
section("Pan")
|
||
print("""
|
||
Valore: -63 (full left) → 0 (center) → +63 (full right)
|
||
""")
|
||
pause()
|
||
|
||
|
||
# ─────────────────────────────────────────────
|
||
# MAIN / SETUP
|
||
# ─────────────────────────────────────────────
|
||
def startup_screen():
|
||
os.system("clear" if os.name != "nt" else "cls")
|
||
print()
|
||
cprint("╔══════════════════════════════════════════════════════════╗", C.CYAN)
|
||
cprint("║ ║", C.CYAN)
|
||
cprint("║ YAMAHA TF5 — CLI INTERATTIVO DI TEST ║", C.CYAN, bold=True)
|
||
cprint("║ Socket Protocol Explorer ║", C.CYAN)
|
||
cprint("║ ║", C.CYAN)
|
||
cprint("╚══════════════════════════════════════════════════════════╝", C.CYAN)
|
||
print()
|
||
|
||
def choose_mode():
|
||
cprint(" Modalità di connessione:", C.WHITE, bold=True)
|
||
cprint(" 1) SIMULAZIONE (mock — nessun mixer necessario)", C.GREEN)
|
||
cprint(" 2) REALE (connessione socket al TF5)", C.YELLOW)
|
||
print()
|
||
choice = ask("Scelta", "1")
|
||
|
||
if choice == "2":
|
||
host = ask("Indirizzo IP del mixer", "192.168.0.128")
|
||
port = ask_int("Porta", 49280, 1, 65535)
|
||
info(f"Connessione a {host}:{port}...")
|
||
ctrl, error = try_load_real_controller(host, port)
|
||
if ctrl is None:
|
||
err(f"Impossibile connettersi: {error}")
|
||
warn("Avvio in modalità simulazione come fallback.")
|
||
return MockTF5Controller()
|
||
ok(f"Connesso a {host}:{port}")
|
||
return ctrl
|
||
else:
|
||
return MockTF5Controller()
|
||
|
||
|
||
def main():
|
||
startup_screen()
|
||
ctrl = choose_mode()
|
||
|
||
while True:
|
||
header("MENU PRINCIPALE")
|
||
cprint(f" Modalità: {'SIMULAZIONE (mock)' if isinstance(ctrl, MockTF5Controller) else 'REALE'}", C.GRAY)
|
||
print()
|
||
cprint(" M) Meter reading ──────────── leggi livelli meter", C.MAGENTA, bold=True)
|
||
cprint(" C) Canali input ────────────── InCh 1-40", C.WHITE)
|
||
cprint(" X) Mix buses ───────────────── MX 1-20", C.WHITE)
|
||
cprint(" D) DCA groups ──────────────── DCA 1-8", C.WHITE)
|
||
cprint(" B) Bus master ──────────────── Stereo / Mono / Mute", C.WHITE)
|
||
cprint(" A) Altri canali ────────────── StInCh / FxRtnCh", C.WHITE)
|
||
cprint(" S) Scene ───────────────────── recall / store", C.WHITE)
|
||
cprint(" R) Comandi RAW ─────────────── protocollo diretto", C.WHITE)
|
||
cprint(" H) Help / Reference ────────── documentazione", C.GRAY)
|
||
cprint(" Q) Esci", C.RED)
|
||
|
||
cmd = ask("\nScelta").upper()
|
||
|
||
if cmd == "Q":
|
||
cprint("\n Chiusura in corso...", C.GRAY)
|
||
ctrl.close()
|
||
cprint(" Arrivederci! 🎚️\n", C.CYAN)
|
||
break
|
||
elif cmd == "M":
|
||
menu_meter(ctrl)
|
||
elif cmd == "C":
|
||
menu_channels(ctrl)
|
||
elif cmd == "X":
|
||
menu_mixes(ctrl)
|
||
elif cmd == "D":
|
||
menu_dca(ctrl)
|
||
elif cmd == "B":
|
||
menu_masters(ctrl)
|
||
elif cmd == "A":
|
||
menu_other_channels(ctrl)
|
||
elif cmd == "S":
|
||
menu_scenes(ctrl)
|
||
elif cmd == "R":
|
||
menu_raw(ctrl)
|
||
elif cmd == "H":
|
||
show_help()
|
||
else:
|
||
warn("Scelta non valida")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print(f"\n\n{C.GRAY} Interrotto dall'utente.{C.RESET}\n")
|
||
sys.exit(0) |