tasks yaml
This commit is contained in:
Nick
246
task_runner.py
Normal file
246
task_runner.py
Normal file
@@ -0,0 +1,246 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user