# -*- coding: utf-8 -*- """ LED Animations-Engine. Spielt LED-Animationen ab und verwaltet Zustandswechsel. Sorgt fuer sauberes Abbrechen laufender Animationen bei neuen Zustaenden. """ from __future__ import annotations import asyncio from typing import Callable from trixy_core.led.animations import AnimationRegistry, IAnimation, RGB from trixy_core.led.driver import WS2812Driver from trixy_core.led.layouts import LEDLayout from trixy_core.utils.debug import pdebug, perror, pwarn class AnimationEngine: """Spielt LED-Animationen ab. Verwaltet den aktuellen Animations-Task und sorgt fuer sauberes Abbrechen bei Zustandswechseln. Unterstuetzt sowohl endlose (CONTINUOUS) als auch einmalige Animationen. Attributes: _driver: WS2812-Hardware-Treiber. _layout: Aktives LED-Layout. _registry: Registry mit verfuegbaren Animationen. _current_task: Aktuell laufender asyncio-Task. _current_state: Name des aktuellen Zustands. """ def __init__(self, driver: WS2812Driver, layout: LEDLayout) -> None: """Erstellt eine neue AnimationEngine. Args: driver: WS2812-Hardware-Treiber (kann im No-Op-Modus sein). layout: LED-Layout fuer die Animationen. """ self._driver = driver self._layout = layout self._registry = AnimationRegistry() self._registry.register_builtins() self._current_task: asyncio.Task[None] | None = None self._current_state: str = "" @property def current_state(self) -> str: """Name des aktuell aktiven Zustands.""" return self._current_state @property def available_animations(self) -> list[str]: """Liste aller registrierten Animations-Namen.""" return self._registry.list_names() async def play( self, state_name: str, state_config: dict[str, object], on_complete: Callable[[], None] | None = None, ) -> None: """Startet eine Animation fuer einen Zustand. Bricht eine eventuell laufende Animation ab und startet die neue. Bei einmaligen Animationen (CONTINUOUS=False) wird on_complete nach Abschluss aufgerufen. Args: state_name: Name des Zustands (z.B. "wakeword", "listening"). state_config: Konfigurations-Dict mit Feldern: - color: Dict mit r, g, b (0-255) - animation: Name der Animation (z.B. "breathe") - speed: Frame-Geschwindigkeit in ms (optional, Standard: 50) - center_color: Dict mit r, g, b fuer Center-LED (optional) - Weitere animations-spezifische Parameter on_complete: Callback wenn eine einmalige Animation fertig ist. Wird bei endlosen Animationen ignoriert. """ # Laufende Animation abbrechen self._cancel_current() # Zustand aktualisieren self._current_state = state_name # Konfiguration parsen color = self._parse_color(state_config.get("color", {})) # type: ignore[arg-type] animation_name: str = str(state_config.get("animation", "solid")) speed: int = int(state_config.get("speed", 50)) # type: ignore[arg-type] # Zusaetzliche Optionen extrahieren (alles ausser den Standardfeldern) reserved_keys = {"color", "animation", "speed"} opts: dict[str, object] = { k: v for k, v in state_config.items() if k not in reserved_keys } # Center-Color separat parsen wenn vorhanden if "center_color" in opts and isinstance(opts["center_color"], dict): opts["center_color"] = self._parse_color(opts["center_color"]) # type: ignore[arg-type] # Animation aus Registry holen anim_cls = self._registry.get(animation_name) if anim_cls is None: pwarn( f"Animation '{animation_name}' nicht gefunden — " f"verfuegbar: {', '.join(self.available_animations)}" ) return anim = anim_cls() pdebug( f"LED-Zustand '{state_name}': Animation '{animation_name}', " f"Farbe {color}, Speed {speed}ms" ) # Animation starten if anim.CONTINUOUS: self._current_task = asyncio.create_task( self._run_continuous(anim, color, speed, opts), name=f"led-anim-{state_name}", ) else: self._current_task = asyncio.create_task( self._run_once_with_callback(anim, color, speed, opts, on_complete), name=f"led-anim-{state_name}", ) async def stop(self) -> None: """Stoppt die aktuelle Animation und schaltet alle LEDs aus.""" self._cancel_current() self._current_state = "" self._driver.clear() pdebug("LED-Engine gestoppt — alle LEDs aus") async def _run_continuous( self, anim: IAnimation, color: RGB, speed: int, opts: dict[str, object], ) -> None: """Spielt eine endlose Animation ab bis sie abgebrochen wird. Args: anim: Animations-Instanz. color: Hauptfarbe als RGB-Tupel. speed: Frame-Dauer in Millisekunden. opts: Zusaetzliche Animations-Parameter. """ try: frame_delay = speed / 1000.0 for frame in anim.frames( self._driver.led_count, color, speed, self._layout, **opts ): self._driver.set_pixels(frame) self._driver.show() await asyncio.sleep(frame_delay) except asyncio.CancelledError: pdebug("Endlos-Animation abgebrochen") except Exception as e: perror(f"Fehler in Endlos-Animation: {e}") async def _run_once( self, anim: IAnimation, color: RGB, speed: int, opts: dict[str, object], ) -> None: """Spielt eine einmalige Animation ab. Args: anim: Animations-Instanz. color: Hauptfarbe als RGB-Tupel. speed: Frame-Dauer in Millisekunden. opts: Zusaetzliche Animations-Parameter. """ try: frame_delay = speed / 1000.0 for frame in anim.frames( self._driver.led_count, color, speed, self._layout, **opts ): self._driver.set_pixels(frame) self._driver.show() await asyncio.sleep(frame_delay) except asyncio.CancelledError: pdebug("Einmal-Animation abgebrochen") except Exception as e: perror(f"Fehler in Einmal-Animation: {e}") async def _run_once_with_callback( self, anim: IAnimation, color: RGB, speed: int, opts: dict[str, object], on_complete: Callable[[], None] | None, ) -> None: """Spielt einmalige Animation ab und ruft Callback auf. Args: anim: Animations-Instanz. color: Hauptfarbe als RGB-Tupel. speed: Frame-Dauer in Millisekunden. opts: Zusaetzliche Animations-Parameter. on_complete: Callback nach Abschluss der Animation. """ await self._run_once(anim, color, speed, opts) # Callback nur ausfuehren wenn nicht abgebrochen if on_complete is not None and self._current_task is not None: if not self._current_task.cancelled(): try: on_complete() except Exception as e: perror(f"Fehler im on_complete-Callback: {e}") def _parse_color(self, color_dict: dict[str, int]) -> RGB: """Konvertiert ein Farb-Dict in ein RGB-Tupel. Args: color_dict: Dict mit 'r', 'g', 'b' Schluesseln (0-255). Fehlende Werte werden als 0 behandelt. Returns: RGB-Tupel (r, g, b) mit Werten von 0-255. """ r = min(255, max(0, int(color_dict.get("r", 0)))) g = min(255, max(0, int(color_dict.get("g", 0)))) b = min(255, max(0, int(color_dict.get("b", 0)))) return (r, g, b) def _cancel_current(self) -> None: """Bricht den aktuellen Animations-Task ab. Wartet nicht auf das Ende des Tasks — der Cancel wird asynchron verarbeitet. """ if self._current_task is not None and not self._current_task.done(): self._current_task.cancel() pdebug(f"Animation fuer Zustand '{self._current_state}' abgebrochen") self._current_task = None