283 lines
10 KiB
Python
283 lines
10 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import socket
|
|
import argparse
|
|
import sys
|
|
import time
|
|
|
|
# Impostazioni di connessione predefinite
|
|
DEFAULT_HOST = "192.168.1.62"
|
|
DEFAULT_PORT = 49280
|
|
|
|
# Configurazioni specifiche per Yamaha TF5
|
|
TF5_INPUT_CHANNELS = 40 # 32 locali + 8 altro
|
|
TF5_MIX_BUSSES = 20 # 20 Aux/Mix
|
|
|
|
def create_connection(host, port):
|
|
"""Crea e restituisce un socket connesso."""
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(5)
|
|
s.connect((host, port))
|
|
return s
|
|
except socket.error as e:
|
|
print(f"Errore critico di connessione a {host}:{port} -> {e}")
|
|
sys.exit(1)
|
|
|
|
def send_command(host, port, command):
|
|
"""Invia un comando singolo e stampa la risposta (modalità one-shot)."""
|
|
s = create_connection(host, port)
|
|
try:
|
|
s.sendall((command + '\n').encode('utf-8'))
|
|
print(f"Comando inviato: '{command}'")
|
|
response = s.recv(4096)
|
|
print(f"Risposta: {response.decode('utf-8', errors='ignore').strip()}")
|
|
finally:
|
|
s.close()
|
|
|
|
def get_param(sock, command):
|
|
"""Funzione helper per ottenere un singolo parametro mantenendo il socket aperto."""
|
|
try:
|
|
sock.sendall((command + '\n').encode('utf-8'))
|
|
# Riceviamo dati finché non troviamo un newline, per essere sicuri di avere la risposta intera
|
|
response = b""
|
|
while b'\n' not in response:
|
|
chunk = sock.recv(1024)
|
|
if not chunk: break
|
|
response += chunk
|
|
return response.decode('utf-8', errors='ignore').strip()
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
def parse_val(response):
|
|
"""Estrae l'ultimo valore numerico da una risposta standard 'OK ... value'."""
|
|
parts = response.split()
|
|
if len(parts) > 0 and parts[0] == "OK":
|
|
return parts[-1]
|
|
return "N/A"
|
|
|
|
def parse_name(response):
|
|
"""Estrae il nome tra virgolette dalla risposta."""
|
|
# Esempio risposta: OK MIXER:Current/InCh/Label/Name 0 0 "Kick Drum"
|
|
try:
|
|
start = response.find('"') + 1
|
|
end = response.rfind('"')
|
|
if start > 0 and end > start:
|
|
return response[start:end]
|
|
return "Unknown"
|
|
except:
|
|
return "Error"
|
|
|
|
def format_db(val_str):
|
|
"""Converte il valore interno (es. 1000) in stringa dB (es. +10.0 dB)."""
|
|
try:
|
|
val = int(val_str)
|
|
if val <= -32768:
|
|
return "-inf dB"
|
|
return f"{val / 100.0:+.1f} dB"
|
|
except:
|
|
return val_str
|
|
|
|
def format_pan(val_str):
|
|
"""Converte il valore pan (-63 a 63) in formato leggibile (L63, C, R63)."""
|
|
try:
|
|
val = int(val_str)
|
|
if val == 0: return "C"
|
|
if val < 0: return f"L{abs(val)}"
|
|
return f"R{val}"
|
|
except:
|
|
return val_str
|
|
|
|
def get_full_config(host, port):
|
|
"""Scarica e stampa la configurazione di tutti i canali e mix."""
|
|
print(f"Connessione a {host}... Inizio download configurazione.")
|
|
print("Attendere prego, l'operazione potrebbe richiedere alcuni secondi...\n")
|
|
|
|
s = create_connection(host, port)
|
|
# Impostiamo un timeout breve per le singole richieste per velocizzare eventuali errori
|
|
s.settimeout(2)
|
|
|
|
try:
|
|
# --- INPUT CHANNELS ---
|
|
print("-" * 65)
|
|
print(f"{'CH':<4} {'NAME':<16} {'STATUS':<8} {'FADER':<12} {'PAN':<8}")
|
|
print("-" * 65)
|
|
|
|
for i in range(TF5_INPUT_CHANNELS):
|
|
# Recupera Nome
|
|
resp_name = get_param(s, f"get MIXER:Current/InCh/Label/Name {i} 0")
|
|
name = parse_name(resp_name)
|
|
|
|
# Recupera ON/OFF
|
|
resp_on = get_param(s, f"get MIXER:Current/InCh/Fader/On {i} 0")
|
|
is_on = parse_val(resp_on)
|
|
status = " [ON] " if is_on == "1" else " OFF "
|
|
|
|
# Recupera Livello Fader
|
|
resp_lvl = get_param(s, f"get MIXER:Current/InCh/Fader/Level {i} 0")
|
|
level_db = format_db(parse_val(resp_lvl))
|
|
|
|
# Recupera Pan
|
|
resp_pan = get_param(s, f"get MIXER:Current/InCh/ToSt/Pan {i} 0")
|
|
pan_str = format_pan(parse_val(resp_pan))
|
|
|
|
print(f"CH{i+1:<2} {name[:15]:<16} {status:<8} {level_db:<12} {pan_str:<8}")
|
|
# Piccolo sleep per non saturare il buffer del mixer se necessario
|
|
# time.sleep(0.01)
|
|
|
|
print("\n")
|
|
# --- MIX (AUX) BUSSES ---
|
|
print("-" * 50)
|
|
print(f"{'MIX':<4} {'NAME':<16} {'STATUS':<8} {'FADER':<12}")
|
|
print("-" * 50)
|
|
|
|
for i in range(TF5_MIX_BUSSES):
|
|
# Recupera Nome Mix
|
|
resp_name = get_param(s, f"get MIXER:Current/Mix/Label/Name {i} 0")
|
|
name = parse_name(resp_name)
|
|
|
|
# Recupera ON/OFF Mix
|
|
resp_on = get_param(s, f"get MIXER:Current/Mix/Fader/On {i} 0")
|
|
is_on = parse_val(resp_on)
|
|
status = " [ON] " if is_on == "1" else " OFF "
|
|
|
|
# Recupera Livello Mix
|
|
resp_lvl = get_param(s, f"get MIXER:Current/Mix/Fader/Level {i} 0")
|
|
level_db = format_db(parse_val(resp_lvl))
|
|
|
|
print(f"MX{i+1:<2} {name[:15]:<16} {status:<8} {level_db:<12}")
|
|
|
|
print("-" * 50)
|
|
print("\nConfigurazione completata.")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nOperazione interrotta dall'utente.")
|
|
except Exception as e:
|
|
print(f"\nErrore durante il recupero della configurazione: {e}")
|
|
finally:
|
|
s.close()
|
|
|
|
def get_full_config_dict(host, port):
|
|
"""
|
|
Scarica la configurazione di tutti i canali e mix e la restituisce come dizionario.
|
|
"""
|
|
print(f"Connessione a {host}... Inizio download configurazione.")
|
|
print("Attendere prego, l'operazione potrebbe richiedere alcuni secondi...\n")
|
|
|
|
s = create_connection(host, port)
|
|
# Impostiamo un timeout breve per le singole richieste per velocizzare eventuali errori
|
|
s.settimeout(2)
|
|
|
|
config = {
|
|
"input_channels": [],
|
|
"mix_busses": []
|
|
}
|
|
|
|
try:
|
|
# --- INPUT CHANNELS ---
|
|
for i in range(TF5_INPUT_CHANNELS):
|
|
channel_data = {}
|
|
|
|
# Recupera Nome
|
|
resp_name = get_param(s, f"get MIXER:Current/InCh/Label/Name {i} 0")
|
|
channel_data['name'] = parse_name(resp_name)
|
|
|
|
# Recupera ON/OFF
|
|
resp_on = get_param(s, f"get MIXER:Current/InCh/Fader/On {i} 0")
|
|
channel_data['status'] = 'ON' if parse_val(resp_on) == "1" else 'OFF'
|
|
|
|
# Recupera Livello Fader
|
|
resp_lvl = get_param(s, f"get MIXER:Current/InCh/Fader/Level {i} 0")
|
|
channel_data['fader_db'] = format_db(parse_val(resp_lvl))
|
|
|
|
# Recupera Pan
|
|
resp_pan = get_param(s, f"get MIXER:Current/InCh/ToSt/Pan {i} 0")
|
|
channel_data['pan'] = format_pan(parse_val(resp_pan))
|
|
|
|
config["input_channels"].append(channel_data)
|
|
|
|
# --- MIX (AUX) BUSSES ---
|
|
for i in range(TF5_MIX_BUSSES):
|
|
mix_data = {}
|
|
|
|
# Recupera Nome Mix
|
|
resp_name = get_param(s, f"get MIXER:Current/Mix/Label/Name {i} 0")
|
|
mix_data['name'] = parse_name(resp_name)
|
|
|
|
# Recupera ON/OFF Mix
|
|
resp_on = get_param(s, f"get MIXER:Current/Mix/Fader/On {i} 0")
|
|
mix_data['status'] = 'ON' if parse_val(resp_on) == "1" else 'OFF'
|
|
|
|
# Recupera Livello Mix
|
|
resp_lvl = get_param(s, f"get MIXER:Current/Mix/Fader/Level {i} 0")
|
|
mix_data['fader_db'] = format_db(parse_val(resp_lvl))
|
|
|
|
config["mix_busses"].append(mix_data)
|
|
|
|
print("Configurazione scaricata con successo.")
|
|
return config
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nOperazione interrotta dall'utente.")
|
|
return None
|
|
except Exception as e:
|
|
print(f"\nErrore durante il recupero della configurazione: {e}")
|
|
return None
|
|
finally:
|
|
s.close()
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description='CLI per controllare un mixer Yamaha TF5.',
|
|
formatter_class=argparse.RawTextHelpFormatter
|
|
)
|
|
parser.add_argument('--host', default=DEFAULT_HOST, help=f'IP del mixer (default: {DEFAULT_HOST})')
|
|
parser.add_argument('--port', type=int, default=DEFAULT_PORT, help=f'Porta (default: {DEFAULT_PORT})')
|
|
|
|
subparsers = parser.add_subparsers(dest='command', help='Comandi disponibili')
|
|
|
|
# Comandi esistenti
|
|
p_recall = subparsers.add_parser('recall', help='Richiama scena (es: recall a 0)')
|
|
p_recall.add_argument('banco', choices=['a', 'b'])
|
|
p_recall.add_argument('numero', type=int)
|
|
|
|
p_level = subparsers.add_parser('level', help='Imposta fader (es: level 0 1000)')
|
|
p_level.add_argument('canale', type=int)
|
|
p_level.add_argument('valore', type=int)
|
|
|
|
p_onoff = subparsers.add_parser('onoff', help='Imposta on/off (es: onoff 0 on)')
|
|
p_onoff.add_argument('canale', type=int)
|
|
p_onoff.add_argument('stato', choices=['on', 'off'])
|
|
|
|
p_pan = subparsers.add_parser('pan', help='Imposta pan (es: pan 0 -63)')
|
|
p_pan.add_argument('canale', type=int)
|
|
p_pan.add_argument('valore', type=int)
|
|
|
|
p_raw = subparsers.add_parser('raw', help='Invia comando grezzo')
|
|
p_raw.add_argument('raw_cmd', nargs='+')
|
|
|
|
# --- NUOVO COMANDO ---
|
|
subparsers.add_parser('config', help='Legge e stampa la configurazione completa del mixer')
|
|
# ---------------------
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == 'config':
|
|
get_full_config(args.host, args.port)
|
|
elif args.command == 'recall':
|
|
send_command(args.host, args.port, f"ssrecall_ex scene_{args.banco} {args.numero}")
|
|
elif args.command == 'level':
|
|
send_command(args.host, args.port, f"set MIXER:Current/InCh/Fader/Level {args.canale} 0 {args.valore}")
|
|
elif args.command == 'onoff':
|
|
val = 1 if args.stato == 'on' else 0
|
|
send_command(args.host, args.port, f"set MIXER:Current/InCh/Fader/On {args.canale} 0 {val}")
|
|
elif args.command == 'pan':
|
|
send_command(args.host, args.port, f"set MIXER:Current/InCh/ToSt/Pan {args.canale} 0 {args.valore}")
|
|
elif args.command == 'raw':
|
|
send_command(args.host, args.port, ' '.join(args.raw_cmd))
|
|
else:
|
|
parser.print_help()
|
|
|
|
if __name__ == '__main__':
|
|
main() |