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

aggiunti meter

Browse Source
This commit is contained in:
Nick
2026-04-20 20:17:46 +02:00
parent 84c1f0da35
commit c2900bbb2e
10 changed files with 377 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.py
+49 -125
View File
@@ -7,31 +7,43 @@ import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# Configurazione mixer
# ==========================================
# Specifiche Yamaha TF5 (FONDAMENTALI PER LA CACHE E IL CONTROLLER)
# ==========================================
DEFAULT_PORT = 49280
TF5_INPUT_CHANNELS = 40
TF5_MIX_BUSSES = 20
TF5_STEREO_BUSSES = 1
TF5_INPUT_CHANNELS = 40 # CH 1-40
TF5_ST_INPUT_CHANNELS = 2 # ST IN 1-2
TF5_FX_RETURN_CHANNELS = 4 # FX RTN 1-4
TF5_DCA_GROUPS = 8 # DCA 1-8
TF5_MIX_BUSSES = 20 # MIX 1-20
TF5_MATRIX_BUSSES = 4 # MATRIX 1-4
TF5_STEREO_BUSSES = 1 # 1 Bus Stereo (L/R)
TF5_MUTE_GROUPS = 8 # Mute Master 1-8
TF5_FX_SLOTS = 8 # Processori FX (8 slot interni)
# Configurazione cache
CACHE_DIR = Path(".tf5_mixer_cache")
CACHE_FILE = CACHE_DIR / "channels_cache.json"
# ==========================================
# Configurazione Cache (Mixer State & IP)
# ==========================================
CACHE_DIR = Path(__file__).parent / "cache"
CACHE_FILE = CACHE_DIR / "tf5_cache.json"
IP_CACHE_FILE = CACHE_DIR / "mixer_ip.json"
CACHE_DURATION = 3600 # 60 minuti in secondi
IP_CACHE_DURATION = 300 # 5 minuti per l'IP
CACHE_DURATION = 60.0 # Secondi per la validità dello stato del mixer
IP_CACHE_DURATION = 300 # 5 minuti per la validità dell'IP del mixer
# Assicurati che la cartella esista
CACHE_DIR.mkdir(parents=True, exist_ok=True)
# ==========================================
# Funzioni di Utilità per Auto-Discovery
# ==========================================
def log(message: str, level: str = "INFO"):
"""Stampa log formattato con timestamp"""
timestamp = time.strftime("%H:%M:%S")
symbols = {
"INFO": "ℹ️",
"SUCCESS": "✅",
"ERROR": "❌",
"WARNING": "⚠️",
"SEARCH": "🔍",
"NETWORK": "🌐",
"CACHE": "📍",
"TIME": "⏱️"
"INFO": "ℹ️", "SUCCESS": "✅", "ERROR": "❌",
"WARNING": "⚠️", "SEARCH": "🔍", "NETWORK": "🌐",
"CACHE": "📍", "TIME": "⏱️"
}
symbol = symbols.get(level, "•")
print(f"[{timestamp}] {symbol} {message}")
@@ -40,13 +52,11 @@ def get_local_network() -> str:
"""Ottiene la rete locale del sistema"""
log("Rilevamento rete locale in corso...", "NETWORK")
try:
# Crea un socket per ottenere l'IP locale
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
local_ip = s.getsockname()[0]
s.close()
# Converte in network (es: 192.168.1.0/24)
ip_parts = local_ip.split('.')
network = f"{ip_parts[0]}.{ip_parts[1]}.{ip_parts[2]}.0/24"
@@ -67,9 +77,7 @@ def check_port(ip: str, port: int, timeout: float = 0.5) -> bool:
result = sock.connect_ex((ip, port))
sock.close()
return result == 0
except Exception as e:
# Log solo per debug, commentabile se troppo verbose
# log(f"Errore su {ip}: {e}", "ERROR")
except Exception:
return False
def scan_network_for_mixer(port: int = DEFAULT_PORT, max_workers: int = 50) -> Optional[str]:
@@ -78,32 +86,22 @@ def scan_network_for_mixer(port: int = DEFAULT_PORT, max_workers: int = 50) -> O
start_time = time.time()
network = get_local_network()
# Genera lista di IP da scansionare
ip_network = ipaddress.ip_network(network, strict=False)
total_hosts = sum(1 for _ in ip_network.hosts())
log(f"Indirizzi da scansionare: {total_hosts}", "INFO")
log(f"Thread paralleli: {max_workers}", "INFO")
log(f"Indirizzi da scansionare: {total_hosts} (Thread: {max_workers})", "INFO")
found_ip = None
scanned = 0
# Scansione parallela per velocità
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_ip = {
executor.submit(check_port, str(ip), port): str(ip)
for ip in ip_network.hosts()
}
log("Scansione in corso...", "SEARCH")
future_to_ip = {executor.submit(check_port, str(ip), port): str(ip) for ip in ip_network.hosts()}
for future in as_completed(future_to_ip):
ip = future_to_ip[future]
scanned += 1
# Log di progresso ogni 50 IP
if scanned % 50 == 0:
log(f"Progresso: {scanned}/{total_hosts} IP scansionati", "INFO")
log(f"Progresso: {scanned}/{total_hosts} IP", "INFO")
try:
if future.result():
@@ -111,27 +109,19 @@ def scan_network_for_mixer(port: int = DEFAULT_PORT, max_workers: int = 50) -> O
elapsed = time.time() - start_time
log(f"MIXER TROVATO su {ip}:{port}", "SUCCESS")
log(f"Tempo di scansione: {elapsed:.2f}s", "TIME")
log(f"IP scansionati: {scanned}/{total_hosts}", "INFO")
# Cancella i task rimanenti
executor.shutdown(wait=False, cancel_futures=True)
break
except Exception as e:
log(f"Errore durante scansione di {ip}: {e}", "ERROR")
except Exception:
pass
if not found_ip:
elapsed = time.time() - start_time
log(f"Nessun mixer trovato sulla porta {port}", "ERROR")
log(f"Tempo totale di scansione: {elapsed:.2f}s", "TIME")
log(f"IP scansionati: {scanned}/{total_hosts}", "INFO")
log(f"Nessun mixer trovato. Tempo: {time.time() - start_time:.2f}s", "ERROR")
return found_ip
def load_cached_ip() -> Optional[str]:
"""Carica l'IP dalla cache se valido"""
log("Controllo cache IP...", "CACHE")
if not IP_CACHE_FILE.exists():
log("Nessuna cache trovata", "INFO")
return None
try:
@@ -139,115 +129,49 @@ def load_cached_ip() -> Optional[str]:
data = json.load(f)
cached_ip = data.get('ip')
cache_time = data.get('timestamp', 0)
age = time.time() - cache_time
age = time.time() - data.get('timestamp', 0)
log(f"Cache trovata: {cached_ip} (età: {int(age)}s)", "CACHE")
# Controlla se la cache è ancora valida
if age < IP_CACHE_DURATION:
log(f"Verifica raggiungibilità di {cached_ip}...", "CACHE")
# Verifica che l'IP sia ancora valido
log(f"Verifica raggiungibilità IP in cache: {cached_ip}...", "CACHE")
if cached_ip and check_port(cached_ip, DEFAULT_PORT, timeout=1.0):
log(f"IP dalla cache valido: {cached_ip}", "SUCCESS")
return cached_ip
else:
log(f"IP dalla cache non raggiungibile", "WARNING")
log(f"IP dalla cache non raggiungibile.", "WARNING")
else:
log(f"Cache scaduta (max {IP_CACHE_DURATION}s)", "WARNING")
log(f"Cache IP scaduta.", "WARNING")
except Exception as e:
log(f"Errore nel leggere cache IP: {e}", "ERROR")
log(f"Errore lettura cache IP: {e}", "ERROR")
return None
def save_ip_to_cache(ip: str):
"""Salva l'IP nella cache"""
log(f"Salvataggio IP in cache: {ip}", "CACHE")
CACHE_DIR.mkdir(exist_ok=True)
try:
with open(IP_CACHE_FILE, 'w') as f:
json.dump({
'ip': ip,
'timestamp': time.time()
}, f, indent=2)
log(f"Cache salvata: {IP_CACHE_FILE}", "SUCCESS")
json.dump({'ip': ip, 'timestamp': time.time()}, f, indent=2)
except Exception as e:
log(f"Errore nel salvare cache IP: {e}", "ERROR")
def get_mixer_ip() -> str:
"""
Ottiene l'IP del mixer con auto-discovery:
1. Controlla cache
2. Scansiona la rete
3. Fallback su IP predefinito
"""
log("=== INIZIO RICERCA MIXER ===", "INFO")
"""Ottiene l'IP del mixer (Cache -> Discovery -> Fallback)"""
log("=== AVVIO RICERCA YAMAHA TF5 ===", "INFO")
# Prova con la cache
cached_ip = load_cached_ip()
if cached_ip:
log("=== RICERCA COMPLETATA (CACHE) ===", "SUCCESS")
return cached_ip
# Scansiona la rete
log("Avvio scansione di rete...", "SEARCH")
discovered_ip = scan_network_for_mixer()
if discovered_ip:
save_ip_to_cache(discovered_ip)
log("=== RICERCA COMPLETATA (DISCOVERY) ===", "SUCCESS")
return discovered_ip
# Fallback
fallback_ip = "192.168.1.57"
log(f"Uso IP predefinito di fallback: {fallback_ip}", "WARNING")
log("=== RICERCA COMPLETATA (FALLBACK) ===", "WARNING")
return fallback_ip
# Variabile globale per l'host (si aggiorna automaticamente)
DEFAULT_HOST = None
def get_host() -> str:
"""Ottiene l'host del mixer (con lazy loading)"""
global DEFAULT_HOST
if DEFAULT_HOST is None:
log("Host mixer non inizializzato, avvio discovery...", "INFO")
DEFAULT_HOST = get_mixer_ip()
else:
log(f"Host mixer già inizializzato: {DEFAULT_HOST}", "INFO")
return DEFAULT_HOST
def refresh_mixer_ip():
"""Forza un refresh dell'IP del mixer"""
global DEFAULT_HOST
log("=== REFRESH FORZATO IP MIXER ===", "INFO")
log("Cancellazione cache esistente...", "CACHE")
# Rimuovi cache
if IP_CACHE_FILE.exists():
try:
IP_CACHE_FILE.unlink()
log("Cache rimossa", "SUCCESS")
except Exception as e:
log(f"Errore rimozione cache: {e}", "ERROR")
# Nuova scansione
DEFAULT_HOST = scan_network_for_mixer()
if DEFAULT_HOST:
save_ip_to_cache(DEFAULT_HOST)
log(f"Nuovo IP impostato: {DEFAULT_HOST}", "SUCCESS")
else:
log("Refresh fallito, mantengo IP precedente", "ERROR")
log("=== REFRESH COMPLETATO ===", "INFO")
return DEFAULT_HOST
# Test rapido
if __name__ == "__main__":
log("=== TEST AUTO-DISCOVERY MIXER ===", "INFO")
host = get_host()
log(f"Host finale: {host}", "SUCCESS")
# Test refresh
# log("\n=== TEST REFRESH ===", "INFO")
# refresh_mixer_ip()
# ==========================================
# ESECUZIONE AL CARICAMENTO DEL MODULO
# ==========================================
# Viene eseguito nel momento in cui mixer_controller.py fa "from config import *"
DEFAULT_HOST = get_mixer_ip()