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

base script

Browse Source
This commit is contained in:
Nick
2025-10-27 19:26:42 +01:00
commit 49964c20a6
9 changed files with 1151 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
README.md Normal file
View File

@@ -0,0 +1,136 @@
Certamente. Ecco un file `README.md` completo e ben formattato che puoi salvare nella stessa cartella del tuo script `yamaha_cli.py`. Questo file spiega a cosa serve lo script, come installarlo e come utilizzarlo, con esempi chiari.
---
# Yamaha Mixer CLI Controller
Uno script Python per creare un'interfaccia a riga di comando (CLI) per controllare i mixer digitali Yamaha della serie TF (e potenzialmente altri modelli compatibili) tramite connessione Ethernet.
Questo strumento permette di eseguire operazioni comuni come il richiamo di scene, la regolazione dei fader, l'attivazione/disattivazione dei canali e il panning direttamente dal terminale, senza bisogno di interfacce grafiche o protocolli MIDI.
## Caratteristiche
- **Semplice e Intuitivo**: Comandi facili da ricordare (`recall`, `level`, `onoff`, `pan`).
- **Nessuna Dipendenza Esterna**: Utilizza solo le librerie standard di Python.
- **Configurabile**: Permette di specificare l'indirizzo IP e la porta del mixer.
- **Flessibile**: Include un comando `raw` per inviare qualsiasi stringa di comando supportata dal mixer.
- **Feedback Immediato**: Fornisce messaggi di conferma e di errore direttamente nel terminale.
## Prerequisiti
1. **Python**: Versione 2.7 o successiva (Python 3.x raccomandato).
2. **Mixer Yamaha**: Un mixer della serie TF (TF1, TF3, TF5, TF-RACK) o un altro modello compatibile con i comandi SCP.
3. **Connessione di Rete**: Il computer che esegue lo script e il mixer devono essere collegati alla stessa rete locale.
## Configurazione Iniziale
1. **Imposta l'IP del Mixer**: Assicurati che il tuo mixer Yamaha abbia un indirizzo IP statico. L'indirizzo predefinito utilizzato dallo script è `192.168.0.128`.
2. **Imposta l'IP del Computer**: Configura il tuo computer con un indirizzo IP statico nella stessa subnet del mixer (es. `192.168.0.100` con subnet mask `255.255.255.0`).
3. **Salva lo Script**: Scarica e salva il file `yamaha_cli.py` in una cartella sul tuo computer.
4. **Apri il Terminale**: Apri un terminale (Prompt dei comandi su Windows, Terminale su macOS/Linux) e naviga nella cartella in cui hai salvato lo script.
## Utilizzo
La sintassi generale per eseguire un comando è:
```bash
python yamaha_cli.py [opzioni] <comando> [argomenti]
```
### Ottenere Aiuto
Per visualizzare la lista completa dei comandi e delle opzioni disponibili:
```bash
python yamaha_cli.py --help
```
Per ottenere aiuto su un comando specifico (ad esempio, `level`):
```bash
python yamaha_cli.py level --help
```
### Opzioni Globali
- `--host`: Specifica un indirizzo IP diverso per il mixer.
- Esempio: `--host 192.168.1.50`
- `--port`: Specifica una porta diversa (lo standard è `49280`).
---
### Comandi Disponibili
#### **`recall`**
Richiama una scena da un banco specifico (A o B).
- **Sintassi**: `python yamaha_cli.py recall <banco> <numero_scena>`
- **Esempio**: Richiama la scena #15 dal banco A.
```bash
python yamaha_cli.py recall a 15
```
#### **`level`**
Imposta il livello del fader di un canale di input.
- **Sintassi**: `python yamaha_cli.py level <canale> <valore>`
- **Nota**: I canali sono indicizzati da 0 (Canale 1 = 0, Canale 2 = 1, ecc.).
- **Esempio**: Imposta il fader del Canale 1 (indice 0) a +10.00 dB (valore 1000).
```bash
python yamaha_cli.py level 0 1000
```
#### **`onoff`**
Accende (`on`) o spegne (`off`) un canale di input.
- **Sintassi**: `python yamaha_cli.py onoff <canale> <stato>`
- **Esempio**: Spegne il Canale 8 (indice 7).
```bash
python yamaha_cli.py onoff 7 off
```
#### **`pan`**
Imposta il posizionamento stereo (pan) di un canale di input.
- **Sintassi**: `python yamaha_cli.py pan <canale> <valore>`
- **Valori**: `-63` (tutto a sinistra), `0` (centro), `63` (tutto a destra).
- **Esempio**: Imposta il pan del Canale 3 (indice 2) completamente a destra.
```bash
python yamaha_cli.py pan 2 63
```
#### **`raw`**
Invia un comando grezzo non implementato direttamente dalla CLI.
- **Sintassi**: `python yamaha_cli.py raw <stringa_comando>`
- **Esempio**: Invia un comando per impostare un parametro specifico.
```bash
python yamaha_cli.py raw set MIXER:Current/InCh/Fader/On 3 0 1
```
## Esempi Pratici
**1. Richiamare la scena #1 del banco A**```bash
python yamaha_cli.py recall a 1
```
**2. Alzare il fader del Canale 5 (indice 4) a 0 dB (valore 0)**
```bash
python yamaha_cli.py level 4 0
```
**3. Accendere il Canale 12 (indice 11)**
```bash
python yamaha_cli.py onoff 11 on
```
**4. Eseguire un comando su un mixer con un IP non standard**
```bash
python yamaha_cli.py --host 192.168.1.128 recall b 42
```
## Note Importanti
- **Indice dei Canali**: Ricorda sempre che i canali sono **indicizzati a partire da 0**. Per controllare il Canale 1, devi usare l'indice `0`.
- **Compatibilità**: Sebbene testato con la serie TF, lo script potrebbe funzionare anche con mixer delle serie CL/QL, poiché condividono un set di comandi simile.
## Licenza
Questo progetto è rilasciato sotto la licenza MIT. Vedi il file `LICENSE` per maggiori dettagli.

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

Binary file not shown.

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

Binary file not shown.

7
devices.py Normal file
View File

@@ -0,0 +1,7 @@
import sounddevice as sd
print("--- Elenco delle Host API Disponibili ---")
print(sd.query_hostapis())
print("\n--- Elenco Completo dei Dispositivi Audio ---")
print(sd.query_devices())

651
mixer_agent.py Normal file
View File

@@ -0,0 +1,651 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
TF5 Mixer AI Agent - Controllo del mixer Yamaha TF5 tramite linguaggio naturale
usando Google Gemini con function calling.
"""
import socket
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
# Configurazione mixer
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:
"""Controller per il mixer Yamaha TF5 con sistema di caching."""
def __init__(self, host=DEFAULT_HOST, port=DEFAULT_PORT):
self.host = host
self.port = port
self._ensure_cache_dir()
self._cache = self._load_cache()
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 _send_command(self, command: str) -> str:
"""Invia un comando al mixer e restituisce la risposta."""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
s.connect((self.host, self.port))
s.sendall((command + '\n').encode('utf-8'))
response = s.recv(4096)
s.close()
return response.decode('utf-8', errors='ignore').strip()
except socket.error as e:
return f"Errore di connessione: {e}"
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:
"""Agente AI per controllare il mixer TF5 con linguaggio naturale."""
def __init__(self, mixer_host=DEFAULT_HOST, mixer_port=DEFAULT_PORT):
self.controller = TF5MixerController(mixer_host, mixer_port)
# Configura il client Gemini
api_key = 'AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M'
if not api_key:
raise ValueError("GEMINI_API_KEY non trovata nelle variabili d'ambiente")
self.client = genai.Client(api_key=api_key)
# Configura gli strumenti con automatic function calling
self.config = types.GenerateContentConfig(
tools=[
self.controller.recall_scene,
self.controller.set_channel_level,
self.controller.set_channel_on_off,
self.controller.set_channel_pan,
self.controller.set_mix_level,
self.controller.mute_multiple_channels,
self.controller.unmute_multiple_channels,
self.controller.get_channel_info,
self.controller.search_channels_by_name,
self.controller.get_all_channels_summary,
self.controller.refresh_cache,
],
temperature=0,
)
# Messaggio di sistema per dare contesto all'AI
self.system_instruction = """Sei un assistente esperto per il controllo di mixer audio Yamaha TF5.
Il mixer ha:
- 40 canali di input (numerati da 1 a 40)
- 20 mix/aux bus (numerati da 1 a 20)
- Livelli fader espressi in dB (da -inf a +10.0 dB)
- Pan da -63 (sinistra) a +63 (destra), 0 è centro
- Scene salvate nei banchi A e B (numerate da 0 a 99)
IMPORTANTE - Sistema di Cache:
- Le informazioni sui canali sono cachate per 5 minuti per evitare di sovraccaricare il mixer
- Usa search_channels_by_name e get_all_channels_summary che usano automaticamente la cache
- La cache viene invalidata automaticamente quando si richiama una scena
- Puoi usare refresh_cache se l'utente chiede esplicitamente dati aggiornati
Quando l'utente fa una richiesta:
1. Interpreta il linguaggio naturale e identifica l'azione richiesta
2. Usa le funzioni disponibili per eseguire i comandi
3. Conferma all'utente cosa hai fatto in modo chiaro e conciso
4. Se una richiesta non è chiara, chiedi chiarimenti
Esempi di comandi che puoi gestire:
- "Alza il canale 5 a -10 dB"
- "Spegni i canali dal 10 al 15"
- "Imposta il pan del canale 3 tutto a sinistra"
- "Richiama la scena A5"
- "Muta i canali 1, 3, 5 e 7"
- "Quali canali sono associati ai vox?" (cerca nei nomi usando cache)
- "Mostrami lo stato del canale 12"
- "Dammi la lista di tutti i canali" (usa cache)
- "Aggiorna i dati dal mixer" (refresh_cache)
"""
def chat(self, user_message: str) -> str:
"""Invia un messaggio all'agente e riceve la risposta.
Args:
user_message: Il messaggio dell'utente in linguaggio naturale
Returns:
La risposta dell'agente
"""
try:
full_prompt = f"{self.system_instruction}\n\nUtente: {user_message}"
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=full_prompt,
config=self.config,
)
return response.text
except Exception as e:
return f"Errore nell'elaborazione della richiesta: {e}"
def interactive_mode(self):
"""Avvia una sessione interattiva con l'agente."""
print("=" * 70)
print("TF5 Mixer AI Agent - Controllo tramite linguaggio naturale")
print("=" * 70)
# Mostra stato cache
cache_age = time.time() - self.controller._cache.get("timestamp", 0)
if self.controller._is_cache_valid():
print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)")
else:
print("\n💾 Cache non disponibile, verrà creata al primo utilizzo")
print("\nEsempi di comandi:")
print(" - 'Alza il canale 5 a -10 dB'")
print(" - 'Spegni i canali dal 1 al 5'")
print(" - 'Richiama la scena A10'")
print(" - 'Imposta il pan del canale 3 a sinistra'")
print(" - 'Muta i canali 2, 4, 6 e 8'")
print(" - 'Quali canali sono associati ai vox?'")
print(" - 'Mostrami lo stato del canale 12'")
print(" - 'Aggiorna i dati dal mixer'")
print("\nDigita 'esci' o 'quit' per terminare\n")
while True:
try:
user_input = input("\n🎛️ Tu: ").strip()
if user_input.lower() in ['esci', 'quit', 'exit', 'q']:
print("\n👋 Arrivederci!")
break
if not user_input:
continue
print("\n🤖 Agent: ", end="", flush=True)
response = self.chat(user_input)
print(response)
except KeyboardInterrupt:
print("\n\n👋 Arrivederci!")
break
except Exception as e:
print(f"\n❌ Errore: {e}")
def main():
"""Funzione principale."""
import argparse
parser = argparse.ArgumentParser(
description='TF5 Mixer AI Agent - Controllo tramite linguaggio naturale'
)
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})')
parser.add_argument('--message', '-m', help='Invia un singolo comando invece di avviare la modalità interattiva')
parser.add_argument('--refresh-cache', action='store_true', help='Forza l\'aggiornamento della cache all\'avvio')
args = parser.parse_args()
# Verifica che la API key sia impostata
if not 'AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M':
print("❌ Errore: GEMINI_API_KEY non trovata nelle variabili d'ambiente")
print("\nPer impostare la chiave API:")
print(" export GEMINI_API_KEY='la-tua-chiave-api'")
print("\nOttieni una chiave API gratuita su: https://aistudio.google.com/apikey")
sys.exit(1)
try:
agent = TF5AIAgent(args.host, args.port)
# Refresh cache se richiesto
if args.refresh_cache:
agent.controller.refresh_cache()
if args.message:
# Modalità singolo comando
print(f"🎛️ Comando: {args.message}")
print(f"\n🤖 Risposta: {agent.chat(args.message)}")
else:
# Modalità interattiva
agent.interactive_mode()
except Exception as e:
print(f"❌ Errore fatale: {e}")
sys.exit(1)
if __name__ == '__main__':
main()

32
mixer_controller.py Normal file
View File

@@ -0,0 +1,32 @@
# mixer_controller.py
import socket
import sys
# Impostazioni di connessione predefinite
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):
"""
Crea una connessione, invia un singolo comando, riceve la risposta e la chiude.
Restituisce la risposta del mixer.
"""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(5)
s.connect((host, 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:
print(f"Errore critico di connessione a {host}:{port} -> {e}")
error_msg = f"Impossibile connettersi al mixer: {e}"
return {"status": "error", "response": error_msg}
except Exception as e:
print(f"Errore imprevisto: {e}")
return {"status": "error", "response": str(e)}

42
record_from_usb.py Normal file
View File

@@ -0,0 +1,42 @@
import sounddevice as sd
import numpy as np
from scipy.io.wavfile import write
# --- 1. Identificazione del Dispositivo ---
# Esegui questo blocco di codice per primo per trovare l'indice del tuo dispositivo.
# print(sd.query_devices())
# Una volta trovato, inserisci l'indice corretto qui sotto.
# --- 2. Impostazioni della Registrazione ---
DEVICE_INDEX = 26 # ESEMPIO: Sostituisci con l'indice del tuo dispositivo Yamaha TF5
CHANNELS = 2 # Numero di canali da registrare (la TF5 ne supporta fino a 34)
SAMPLE_RATE = 48000 # Frequenza di campionamento in Hz (deve corrispondere a quella della console)
DURATION = 10 # Durata della registrazione in secondi
OUTPUT_FILENAME = 'registrazione_multicanale.wav'
try:
# --- 3. Registrazione dell'Audio ---
print(f"Inizio registrazione di {DURATION} secondi dal dispositivo {DEVICE_INDEX}...")
# sd.rec() avvia una registrazione non bloccante e restituisce immediatamente
# I dati audio vengono memorizzati in un array NumPy
myrecording = sd.rec(int(DURATION * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=CHANNELS, device=DEVICE_INDEX, dtype='float32')
# sd.wait() attende che la registrazione sia completata
sd.wait()
print("Registrazione completata.")
# --- 4. Salvataggio del File WAV ---
print(f"Salvataggio della registrazione in corso su '{OUTPUT_FILENAME}'...")
# Scrive l'array NumPy in un file WAV
# La funzione si aspetta dati in un formato specifico (es. int16 o float32)
# È consigliabile scalare i dati se necessario, ma sounddevice con dtype='float32'
# di solito produce un output compatibile.
write(OUTPUT_FILENAME, SAMPLE_RATE, myrecording)
print("File salvato con successo.")
except Exception as e:
print(f"Si è verificato un errore: {e}")

BIN
registrazione_multicanale.wav Normal file
View File

Binary file not shown.

283
yamaha_cli.py Normal file
View File

@@ -0,0 +1,283 @@
#!/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()