# -*- coding: utf-8 -*- """ Timer-Plugin fuer Trixy. Ermoeglicht das Setzen, Stoppen und Auflisten von Timern per Sprache. Bei Ablauf wird ein Notification-Sound und eine TTS-Meldung abgespielt. """ import asyncio import time from dataclasses import dataclass, field from pathlib import Path from typing import Any from trixy_core.plugins import TrixyPlugin from trixy_core.nlp.intent_registry import IntentRegistry, IntentDefinition, IntentSlot from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn @dataclass class TimerEntry: """Ein aktiver Timer.""" name: str duration: int # Sekunden satellite_id: str room_id: str created_at: float # time.time() task: asyncio.Task | None = None @property def remaining(self) -> int: """Verbleibende Sekunden.""" elapsed = time.time() - self.created_at return max(0, int(self.duration - elapsed)) class TimerPlugin(TrixyPlugin): """ Timer-Plugin. Bietet: - Timer setzen mit optionalem Namen - Timer stoppen/entfernen - Aktive Timer auflisten - Notification-Sound und TTS bei Ablauf """ NAME = "timer" VERSION = "1.0.0" DESCRIPTION = "Timer per Sprachbefehl setzen und verwalten" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None): super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() self._timers: dict[str, TimerEntry] = {} self._notification_pcm: bytes | None = None # ========================================================================= # Lifecycle # ========================================================================= async def on_load(self) -> None: """Initialisiert das Plugin und registriert Intents.""" pinfo(f"TimerPlugin: Lade Plugin v{self.VERSION}") # Notification-Sound vorladen self._load_notification_sound() # Intents werden automatisch vom PluginManager registriert # (via @intent/@pattern/@example Dekoratoren auf den Handler-Methoden) pinfo("TimerPlugin: Geladen") async def on_unload(self) -> None: """Entlaedt das Plugin, stoppt alle Timer.""" # Alle laufenden Timer abbrechen for entry in self._timers.values(): if entry.task and not entry.task.done(): entry.task.cancel() self._timers.clear() # Intents deregistrieren self._registry.unregister_plugin(self.NAME) self._notification_pcm = None pdebug("TimerPlugin: Entladen") async def on_enable(self) -> None: """Aktiviert das Plugin.""" pinfo("TimerPlugin: Aktiviert") async def on_disable(self) -> None: """Deaktiviert das Plugin.""" pinfo("TimerPlugin: Deaktiviert") # ========================================================================= # Notification-Sound # ========================================================================= def _load_notification_sound(self) -> None: """Laedt den Notification-Sound und entfernt den WAV-Header.""" sound_name = self.get_config_value("notification_sound", "notification.wav") asset_service = self.application.services.get_service("AssetService") if not asset_service: pwarn("TimerPlugin: AssetService nicht verfuegbar, kein Notification-Sound") return audio_path: Path | None = asset_service.get_audio(sound_name) if not audio_path or not audio_path.exists(): pwarn(f"TimerPlugin: Notification-Sound nicht gefunden: {sound_name}") return try: raw = audio_path.read_bytes() # WAV-Header (44 Bytes) abschneiden -> rohe PCM-Daten self._notification_pcm = raw[44:] pdebug(f"TimerPlugin: Notification-Sound geladen ({len(self._notification_pcm)} bytes PCM)") except Exception as e: perror(f"TimerPlugin: Fehler beim Laden des Notification-Sounds: {e}") # ========================================================================= # Intent-Handler # ========================================================================= @intent("set_timer", description="Timer mit optionalem Namen setzen") @pattern("(setze|stelle|erstelle|starte) [einen|den] timer [auf|fuer] {duration}") @pattern("timer {duration}") @pattern("weck [mich] [in] {duration}") @example("Setze Timer auf 5 Minuten", "Timer 30 Sekunden") @example("Weck mich in einer Stunde", "Stell einen Timer auf 3 Minuten") async def handle_set_timer(self, data: IntentReceivedData, duration: int = 0, name: str = "") -> IntentResult: """Verarbeitet den set_timer Intent.""" # Dauer pruefen duration = data.get_slot("duration") if duration is None: return IntentResult.failure( "Keine Dauer angegeben", response_text="Ich habe keine Dauer verstanden. Wie lange soll der Timer laufen?", ) try: duration = int(duration) except (TypeError, ValueError): return IntentResult.failure( f"Ungueltige Dauer: {duration}", response_text="Ich konnte die Dauer nicht verstehen.", ) if duration <= 0: return IntentResult.failure( "Dauer muss positiv sein", response_text="Die Dauer muss groesser als null sein.", ) # Maximale Timer-Anzahl pruefen max_timers = self.get_config_value("max_timers", 10) if len(self._timers) >= max_timers: return IntentResult.failure( f"Maximale Timer-Anzahl erreicht ({max_timers})", response_text=f"Es laufen bereits {max_timers} Timer. Stoppe zuerst einen.", ) # Timer-Name bestimmen timer_name = data.get_slot("name", "Timer") if not timer_name: timer_name = "Timer" # Eindeutigen Namen sicherstellen base_name = timer_name counter = 2 while timer_name in self._timers: timer_name = f"{base_name} {counter}" counter += 1 # Timer erstellen entry = TimerEntry( name=timer_name, duration=duration, satellite_id=data.satellite_id, room_id=data.room_id, created_at=time.time(), ) entry.task = asyncio.create_task(self._run_timer(entry)) self._timers[timer_name] = entry # TTS-Bestaetigung response = f"{timer_name} auf {self._format_duration(duration)} gestellt." pinfo(f"TimerPlugin: {response} (satellite={data.satellite_id})") return IntentResult.success_with_response(response) @intent("stop_timer", description="Laufenden Timer stoppen") @pattern("(stoppe|stopp) [den] [timer] {name}") @pattern("timer [aus|abbrechen]") @example("Stoppe den Timer", "Timer aus", "Timer abbrechen") async def handle_stop_timer(self, data: IntentReceivedData, name: str = "") -> IntentResult: """Verarbeitet den stop_timer Intent.""" return await self._cancel_timer(data) @intent("remove_timer", description="Timer entfernen") @pattern("(entferne|loesche) [den] [timer] {name}") @pattern("timer (entfernen|loeschen)") @example("Entferne den Pizza Timer", "Loesche Timer") async def handle_remove_timer(self, data: IntentReceivedData, name: str = "") -> IntentResult: """Verarbeitet den remove_timer Intent (delegiert an stop).""" return await self._cancel_timer(data) @intent("list_timers", description="Alle aktiven Timer auflisten") @pattern("welche timer laufen") @pattern("timer status") @pattern("zeig [meine] timer") @pattern("wie viel zeit ist [noch]") @example("Welche Timer laufen?", "Timer Status") @example("Zeig mir meine Timer", "Wie viel Zeit ist noch?") async def handle_list_timers(self, data: IntentReceivedData) -> IntentResult: """Listet alle aktiven Timer auf.""" if not self._timers: return IntentResult.success_with_response( "Es laufen keine Timer." ) lines: list[str] = [] for entry in self._timers.values(): remaining = self._format_duration(entry.remaining) lines.append(f"{entry.name}: noch {remaining}") if len(lines) == 1: response = f"Ein Timer laeuft. {lines[0]}." else: timer_list = ", ".join(lines) response = f"Es laufen {len(lines)} Timer. {timer_list}." return IntentResult.success_with_response(response) # ========================================================================= # Timer-Logik # ========================================================================= async def _run_timer(self, entry: TimerEntry) -> None: """Fuehrt einen Timer aus und benachrichtigt bei Ablauf.""" try: await asyncio.sleep(entry.duration) except asyncio.CancelledError: pdebug(f"TimerPlugin: Timer '{entry.name}' abgebrochen") return pinfo(f"TimerPlugin: Timer '{entry.name}' abgelaufen") # Satellite holen satellites = getattr(self.application, "satellites", None) satellite = None if satellites: satellite = satellites.get(entry.satellite_id) if satellite and satellite.is_connected: # Notification-Sound abspielen if self._notification_pcm: try: await satellite.say(self._notification_pcm) except Exception as e: perror(f"TimerPlugin: Notification-Sound fehlgeschlagen: {e}") # TTS-Meldung try: await satellite.speak(f"Der {entry.name} ist abgelaufen.") except Exception as e: perror(f"TimerPlugin: TTS-Meldung fehlgeschlagen: {e}") else: pwarn(f"TimerPlugin: Satellite '{entry.satellite_id}' nicht erreichbar fuer Timer '{entry.name}'") # Timer aus dict entfernen self._timers.pop(entry.name, None) async def _cancel_timer(self, data: IntentReceivedData) -> IntentResult: """Stoppt und entfernt einen Timer.""" timer_name = data.get_slot("name", "") if not timer_name: # Kein Name angegeben: Falls nur ein Timer laeuft, diesen stoppen if len(self._timers) == 0: return IntentResult.success_with_response( "Es laeuft kein Timer." ) if len(self._timers) == 1: timer_name = next(iter(self._timers)) else: names = ", ".join(self._timers.keys()) return IntentResult.failure( "Mehrere Timer aktiv, kein Name angegeben", response_text=f"Es laufen mehrere Timer: {names}. Welchen soll ich stoppen?", ) entry = self._timers.pop(timer_name, None) if not entry: return IntentResult.failure( f"Timer '{timer_name}' nicht gefunden", response_text=f"Ich habe keinen Timer mit dem Namen {timer_name} gefunden.", ) # Task abbrechen if entry.task and not entry.task.done(): entry.task.cancel() response = f"{timer_name} gestoppt." pinfo(f"TimerPlugin: {response}") return IntentResult.success_with_response(response) # ========================================================================= # Hilfsmethoden # ========================================================================= @staticmethod def _format_duration(seconds: int) -> str: """Formatiert Sekunden als lesbaren Text.""" if seconds < 60: if seconds == 1: return "1 Sekunde" return f"{seconds} Sekunden" minutes = seconds // 60 remaining_secs = seconds % 60 if minutes >= 60: hours = minutes // 60 remaining_mins = minutes % 60 parts: list[str] = [] if hours == 1: parts.append("1 Stunde") else: parts.append(f"{hours} Stunden") if remaining_mins == 1: parts.append("1 Minute") elif remaining_mins > 0: parts.append(f"{remaining_mins} Minuten") return " und ".join(parts) if remaining_secs == 0: if minutes == 1: return "1 Minute" return f"{minutes} Minuten" if minutes == 1: min_text = "1 Minute" else: min_text = f"{minutes} Minuten" if remaining_secs == 1: sec_text = "1 Sekunde" else: sec_text = f"{remaining_secs} Sekunden" return f"{min_text} und {sec_text}"