Trixy7/trixy_core/led/service.py

# -*- coding: utf-8 -*-
"""
LED-Service fuer Trixy Satellites.

Steuert WS2812/NeoPixel LEDs basierend auf dem aktuellen
System-Zustand. Reagiert auf Events wie Wakeword-Erkennung,
Audio-Streaming, TTS-Ausgabe und Verbindungsstatus.

Laeuft nur auf Geraeten mit GPIO-Unterstuetzung (Raspberry Pi).
Ohne rpi-ws281x deaktiviert sich der Service automatisch.

Farben und Animationen sind pro Zustand individuell
in der client_config.json konfigurierbar.
"""

from __future__ import annotations

import asyncio
from typing import Any, TYPE_CHECKING

from trixy_core.service.iservice import IService
from trixy_core.service.enums import ServicePriority, ServiceGroup
from trixy_core.events.decorators import TrixyEvent
from trixy_core.events.enums import EventPriority
from trixy_core.utils.debug import pinfo, pdebug, perror

from trixy_core.led.driver import WS2812Driver
from trixy_core.led.engine import AnimationEngine
from trixy_core.led.layouts import get_layout

if TYPE_CHECKING:
    from trixy_core.application import BaseApplication


# Standard-Zustandskonfigurationen (Fallback wenn in Config nicht definiert)
_DEFAULT_STATES: dict[str, dict] = {
    "boot":       {"color": {"r": 0, "g": 255, "b": 0},   "animation": "solid",        "speed": 100},
    "standby":    {"color": {"r": 0, "g": 60,  "b": 0},   "animation": "solid",        "speed": 100},
    "wakeword":   {"color": {"r": 0, "g": 0,   "b": 255}, "animation": "double-blink", "speed": 120},
    "listening":  {"color": {"r": 255, "g": 0, "b": 0},   "animation": "solid",        "speed": 100},
    "thinking":   {"color": {"r": 255, "g": 165, "b": 0}, "animation": "pulse",        "speed": 80},
    "speaking":   {"color": {"r": 0, "g": 100, "b": 255}, "animation": "breathe",      "speed": 60},
    "error":      {"color": {"r": 255, "g": 0, "b": 0},   "animation": "triple-blink", "speed": 150},
    "disconnect": {"color": {"r": 255, "g": 0, "b": 0},   "animation": "pulse",        "speed": 40},
}


class LEDService(IService):
    """
    LED-Steuerung fuer Trixy Satellites.

    Reagiert auf System-Events und zeigt den aktuellen Zustand
    ueber WS2812/NeoPixel LEDs an. Unterstuetzt verschiedene
    LED-Layouts (Single, Jewel 7, Ring 12) mit konfigurierbaren
    Animationen pro Zustand.
    """

    PRIORITY = ServicePriority.OPTIONAL
    GROUP = ServiceGroup.UTILITY
    DEPENDENCIES = ["EventManager"]
    NAME = "LEDService"

    def __init__(self, application: "BaseApplication", led_config: dict) -> None:
        super().__init__(application)
        self._config = led_config
        self._driver: WS2812Driver | None = None
        self._engine: AnimationEngine | None = None
        self._current_state: str = ""
        self._states: dict[str, dict] = {}

    @property
    def current_state(self) -> str:
        """Aktueller LED-Zustand."""
        return self._current_state

    @property
    def available(self) -> bool:
        """True wenn LED-Hardware verfuegbar ist."""
        return self._driver is not None and self._driver.available

    async def start(self) -> None:
        """Initialisiert LED-Hardware und zeigt Boot-Animation."""
        gpio_pin = self._config.get("gpio_pin", 18)
        led_count = self._config.get("led_count", 1)
        brightness = self._config.get("brightness", 80)
        led_type = self._config.get("type", "single")

        # Zustandskonfigurationen laden (Config → Defaults Fallback)
        config_states = self._config.get("states", {})
        self._states = {}
        for state_name, default_cfg in _DEFAULT_STATES.items():
            if state_name in config_states:
                # Config-Werte mit Defaults mergen
                merged = {**default_cfg, **config_states[state_name]}
                self._states[state_name] = merged
            else:
                self._states[state_name] = dict(default_cfg)

        # Hardware initialisieren
        self._driver = WS2812Driver(
            gpio_pin=gpio_pin,
            led_count=led_count,
            brightness=brightness,
        )

        if not self._driver.init():
            pdebug("LEDService: Keine LED-Hardware verfuegbar — Service deaktiviert")
            self._driver = None
            return

        # Layout und Engine erstellen
        layout = get_layout(led_type, led_count)
        self._engine = AnimationEngine(self._driver, layout)

        # Boot-Animation abspielen
        await self._set_state("boot")

        pinfo(
            f"LEDService: Gestartet — {led_count} LEDs auf GPIO {gpio_pin} "
            f"(Typ: {led_type}, Helligkeit: {brightness})"
        )

    async def stop(self) -> None:
        """Stoppt LED-Service und schaltet LEDs aus."""
        if self._engine:
            await self._engine.stop()
        if self._driver:
            self._driver.cleanup()
            self._driver = None
        self._engine = None
        pdebug("LEDService: Gestoppt")

    async def health_check(self) -> bool:
        """Prueft ob der LED-Service funktioniert."""
        return self.available

    # === Zustandssteuerung ===

    async def _set_state(self, state_name: str) -> None:
        """
        Wechselt den LED-Zustand und startet die passende Animation.

        Args:
            state_name: Name des Zustands (z.B. "standby", "wakeword")
        """
        if not self._engine:
            return

        state_config = self._states.get(state_name)
        if not state_config:
            pdebug(f"LEDService: Unbekannter Zustand '{state_name}'")
            return

        self._current_state = state_name
        pdebug(f"LEDService: Zustand → {state_name} ({state_config.get('animation', 'solid')})")

        await self._engine.play(state_name, state_config)

    async def _set_state_with_followup(
        self, state_name: str, followup_state: str,
    ) -> None:
        """
        Spielt eine Einmal-Animation und wechselt danach zum Folgezustand.

        Wird fuer Wakeword verwendet: 2× blau blinken, dann → listening.
        """
        if not self._engine:
            return

        state_config = self._states.get(state_name)
        if not state_config:
            return

        self._current_state = state_name
        pdebug(f"LEDService: Zustand → {state_name} → {followup_state}")

        async def on_complete():
            await self._set_state(followup_state)

        await self._engine.play(
            state_name, state_config,
            on_complete=on_complete,
        )

    # === Event-Handler ===

    @TrixyEvent(["application_ready"], priority=EventPriority.LOW)
    async def _on_ready(self, event_name: str, event_data: Any) -> None:
        """System bereit → Standby."""
        await self._set_state("standby")

    @TrixyEvent(["wakeword_detected"], priority=EventPriority.HIGHEST)
    async def _on_wakeword(self, event_name: str, event_data: Any) -> None:
        """Wakeword erkannt → Blinken, dann Listening."""
        await self._set_state_with_followup("wakeword", "listening")

    @TrixyEvent(["recording_started"], priority=EventPriority.HIGH)
    async def _on_recording_started(self, event_name: str, event_data: Any) -> None:
        """Aufnahme gestartet → Listening (rot)."""
        # Nur setzen wenn nicht schon durch Wakeword-Handler gesetzt
        if self._current_state not in ("wakeword", "listening"):
            await self._set_state("listening")

    @TrixyEvent(["recording_stopped"], priority=EventPriority.NORMAL)
    async def _on_recording_stopped(self, event_name: str, event_data: Any) -> None:
        """Aufnahme beendet → Thinking (verarbeiten)."""
        await self._set_state("thinking")

    @TrixyEvent(["tts_started"], priority=EventPriority.NORMAL)
    async def _on_tts_started(self, event_name: str, event_data: Any) -> None:
        """TTS-Ausgabe gestartet → Speaking."""
        await self._set_state("speaking")

    @TrixyEvent(["tts_completed", "conversation_ended"], priority=EventPriority.LOW)
    async def _on_done(self, event_name: str, event_data: Any) -> None:
        """TTS fertig oder Konversation beendet → Standby."""
        await self._set_state("standby")

    @TrixyEvent(["client_connected"], priority=EventPriority.LOW)
    async def _on_connected(self, event_name: str, event_data: Any) -> None:
        """Server-Verbindung hergestellt → Standby."""
        await self._set_state("standby")

    @TrixyEvent(["client_disconnected"], priority=EventPriority.HIGH)
    async def _on_disconnected(self, event_name: str, event_data: Any) -> None:
        """Server-Verbindung verloren → Disconnect."""
        await self._set_state("disconnect")

    # === Oeffentliche API ===

    async def set_color(self, r: int, g: int, b: int) -> None:
        """Setzt alle LEDs auf eine feste Farbe (z.B. fuer Config-Tool)."""
        if not self._engine:
            return
        await self._engine.play("custom", {
            "color": {"r": r, "g": g, "b": b},
            "animation": "solid",
        })

    async def set_animation(self, animation: str, color: dict, speed: int = 100, **opts) -> None:
        """Spielt eine beliebige Animation ab (z.B. fuer Config-Tool)."""
        if not self._engine:
            return
        config = {"color": color, "animation": animation, "speed": speed, **opts}
        await self._engine.play("custom", config)

    async def turn_off(self) -> None:
        """Schaltet alle LEDs aus."""
        if self._engine:
            await self._engine.stop()
        self._current_state = "off"