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