Trixy7/plugins/timer/main.py

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