Trixy7/trixy_core/led/engine.py

# -*- 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