247 lines
9.6 KiB
Python
247 lines
9.6 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
task_runner.py
|
|
==============
|
|
Esegue file di task YAML sul Yamaha TF5 tramite TF5MixerController.
|
|
|
|
Uso da CLI:
|
|
python task_runner.py tasks_concerto_live.yaml # lista le cue disponibili
|
|
python task_runner.py tasks_concerto_live.yaml CUE_01 # esegue una cue specifica
|
|
python task_runner.py tasks_concerto_live.yaml --all # esegue tutte le cue in sequenza
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from mixer_controller import TF5MixerController
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# DISPATCHER: mappa action → metodo del controller
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
class TaskRunner:
|
|
def __init__(self, controller: TF5MixerController):
|
|
self.ctrl = controller
|
|
|
|
# Mappa nome azione → callable
|
|
self._action_map = {
|
|
"set_channel_level": self._set_channel_level,
|
|
"set_channel_on_off": self._set_channel_on_off,
|
|
"set_channel_pan": self._set_channel_pan,
|
|
"mute_channels": self._mute_channels,
|
|
"unmute_channels": self._unmute_channels,
|
|
"set_mix_level": self._set_mix_level,
|
|
"set_mix_on_off": self._set_mix_on_off,
|
|
"set_channel_to_mix_level": self._set_channel_to_mix_level,
|
|
"set_channel_to_mix_on_off":self._set_channel_to_mix_on_off,
|
|
"recall_scene": self._recall_scene,
|
|
"wait": self._wait,
|
|
"refresh_cache": self._refresh_cache,
|
|
}
|
|
|
|
# ── Handlers specifici ──────────────────────────────────
|
|
|
|
def _set_channel_level(self, step: dict) -> dict:
|
|
return self.ctrl.set_channel_level(step["channel"], float(step["level_db"]))
|
|
|
|
def _set_channel_on_off(self, step: dict) -> dict:
|
|
return self.ctrl.set_channel_on_off(step["channel"], bool(step["state"]))
|
|
|
|
def _set_channel_pan(self, step: dict) -> dict:
|
|
return self.ctrl.set_channel_pan(step["channel"], int(step["pan"]))
|
|
|
|
def _mute_channels(self, step: dict) -> dict:
|
|
return self.ctrl.mute_multiple_channels(list(step["channels"]))
|
|
|
|
def _unmute_channels(self, step: dict) -> dict:
|
|
return self.ctrl.unmute_multiple_channels(list(step["channels"]))
|
|
|
|
def _set_mix_level(self, step: dict) -> dict:
|
|
return self.ctrl.set_mix_level(step["mix"], float(step["level_db"]))
|
|
|
|
def _set_mix_on_off(self, step: dict) -> dict:
|
|
return self.ctrl.set_mix_on_off(step["mix"], bool(step["state"]))
|
|
|
|
def _set_channel_to_mix_level(self, step: dict) -> dict:
|
|
return self.ctrl.set_channel_to_mix_level(step["channel"], step["mix"], float(step["level_db"]))
|
|
|
|
def _set_channel_to_mix_on_off(self, step: dict) -> dict:
|
|
return self.ctrl.set_channel_to_mix_on_off(step["channel"], step["mix"], bool(step["state"]))
|
|
|
|
def _recall_scene(self, step: dict) -> dict:
|
|
return self.ctrl.recall_scene(str(step["bank"]), int(step["scene"]))
|
|
|
|
def _wait(self, step: dict) -> dict:
|
|
ms = int(step.get("ms", 0))
|
|
print(f" ⏱ Wait puro: {ms} ms")
|
|
time.sleep(ms / 1000.0)
|
|
return {"status": "success", "message": f"Atteso {ms} ms"}
|
|
|
|
def _refresh_cache(self, step: dict) -> dict:
|
|
return self.ctrl.refresh_cache()
|
|
|
|
# ── Esecuzione di un singolo step ───────────────────────
|
|
|
|
def _run_step(self, step_num: int, step: dict) -> bool:
|
|
"""Esegue un singolo step. Ritorna True se OK, False se errore."""
|
|
action = step.get("action")
|
|
delay_ms = int(step.get("delay_ms", 0))
|
|
wait_ms = int(step.get("wait_ms", 0))
|
|
|
|
if delay_ms > 0:
|
|
print(f" ⏳ delay_ms={delay_ms} prima dell'azione...")
|
|
time.sleep(delay_ms / 1000.0)
|
|
|
|
handler = self._action_map.get(action)
|
|
if handler is None:
|
|
print(f" ⚠️ Step {step_num}: azione '{action}' non riconosciuta — saltato.")
|
|
return False
|
|
|
|
print(f" ▶ Step {step_num}: {action} | params: { {k:v for k,v in step.items() if k not in ('action','delay_ms','wait_ms')} }")
|
|
try:
|
|
result = handler(step)
|
|
status = result.get("status", "unknown")
|
|
msg = result.get("message", "")
|
|
icon = "✅" if status in ("success", "partial") else "❌"
|
|
print(f" {icon} [{status}] {msg}")
|
|
|
|
if wait_ms > 0:
|
|
time.sleep(wait_ms / 1000.0)
|
|
|
|
return status in ("success", "partial")
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Eccezione durante '{action}': {e}")
|
|
return False
|
|
|
|
# ── Esecuzione di una cue ────────────────────────────────
|
|
|
|
def run_cue(self, cue: dict) -> bool:
|
|
"""
|
|
Esegue tutti gli step di una cue.
|
|
Ritorna True se tutti gli step sono andati a buon fine.
|
|
"""
|
|
cue_id = cue.get("id", "???")
|
|
name = cue.get("name", "")
|
|
on_error = cue.get("on_error", "stop")
|
|
steps = cue.get("steps", [])
|
|
|
|
print(f"\n{'═'*60}")
|
|
print(f"🎬 CUE: {cue_id} — {name}")
|
|
if cue.get("description"):
|
|
print(f" 📋 {cue['description']}")
|
|
print(f" Steps: {len(steps)} | on_error: {on_error}")
|
|
print(f"{'─'*60}")
|
|
|
|
all_ok = True
|
|
for i, step in enumerate(steps, start=1):
|
|
ok = self._run_step(i, step)
|
|
if not ok:
|
|
all_ok = False
|
|
if on_error == "stop":
|
|
print(f"\n 🛑 on_error=stop: cue {cue_id} interrotta allo step {i}.")
|
|
return False
|
|
# on_error == "continue": vai avanti
|
|
|
|
result_icon = "✅" if all_ok else "⚠️ (con errori)"
|
|
print(f"\n {result_icon} Cue {cue_id} completata.")
|
|
return all_ok
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# LOADER YAML
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def load_task_file(path: str) -> dict:
|
|
"""Carica e valida un file di task YAML."""
|
|
p = Path(path)
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"File non trovato: {path}")
|
|
with open(p, "r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
if "cues" not in data or not isinstance(data["cues"], list):
|
|
raise ValueError("Il file YAML deve contenere una lista 'cues'.")
|
|
return data
|
|
|
|
|
|
def print_event_info(data: dict):
|
|
event = data.get("event", {})
|
|
if event:
|
|
print(f"\n{'═'*60}")
|
|
print(f"📅 Evento : {event.get('name', 'N/D')}")
|
|
print(f" Data : {event.get('date', 'N/D')}")
|
|
print(f" Venue : {event.get('venue', 'N/D')}")
|
|
if event.get("notes"):
|
|
print(f" Note : {event['notes']}")
|
|
|
|
|
|
def list_cues(data: dict):
|
|
print_event_info(data)
|
|
print(f"\n{'═'*60}")
|
|
print("📋 CUE DISPONIBILI:")
|
|
print(f"{'─'*60}")
|
|
for cue in data["cues"]:
|
|
steps_count = len(cue.get("steps", []))
|
|
print(f" {cue.get('id'):12s} │ {cue.get('name','')} ({steps_count} steps)")
|
|
print(f"{'═'*60}\n")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# ENTRY POINT
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
sys.exit(1)
|
|
|
|
task_file = sys.argv[1]
|
|
cue_arg = sys.argv[2] if len(sys.argv) > 2 else None
|
|
|
|
try:
|
|
data = load_task_file(task_file)
|
|
except (FileNotFoundError, ValueError) as e:
|
|
print(f"❌ Errore nel caricamento del file: {e}")
|
|
sys.exit(1)
|
|
|
|
# Solo lista cue senza mixer
|
|
if cue_arg is None:
|
|
list_cues(data)
|
|
sys.exit(0)
|
|
|
|
# Connessione al mixer ed esecuzione
|
|
print("🔌 Connessione al mixer...")
|
|
import config
|
|
|
|
|
|
with TF5MixerController(host=config.get_host()) as ctrl:
|
|
runner = TaskRunner(ctrl)
|
|
|
|
if cue_arg == "--all":
|
|
print_event_info(data)
|
|
print("\n▶▶ Esecuzione di TUTTE le cue in sequenza...")
|
|
for cue in data["cues"]:
|
|
ok = runner.run_cue(cue)
|
|
if not ok and cue.get("on_error", "stop") == "stop":
|
|
print(f"\n🛑 Esecuzione interrotta a causa di un errore in {cue.get('id')}.")
|
|
sys.exit(1)
|
|
else:
|
|
# Cerca la cue per ID
|
|
cue_found = next((c for c in data["cues"] if c.get("id") == cue_arg), None)
|
|
if cue_found is None:
|
|
print(f"❌ Cue '{cue_arg}' non trovata nel file.")
|
|
list_cues(data)
|
|
sys.exit(1)
|
|
runner.run_cue(cue_found)
|
|
|
|
print("\n✅ Task runner completato.\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|