Trixy7/trixy_core/scheduler/trigger/event.py

# -*- coding: utf-8 -*-
"""
Event Trigger - Reagiert auf Events des EventManagers.
"""

from datetime import datetime
from typing import Any, Callable

from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError
from trixy_core.scheduler.trigger.base import Trigger, TriggerContext, TriggerState


class EventTrigger(Trigger):
    """
    Trigger der auf Events des EventManagers reagiert.

    Beispiele:
        - EventTrigger("wakeword_received") - Bei Wakeword
        - EventTrigger("satellite_connected", filter_fn=lambda d: d.get("room") == "kitchen")
        - EventTrigger(["motion_detected", "door_opened"]) - Bei mehreren Events
    """

    def __init__(
        self,
        event_names: str | list[str],
        filter_fn: Callable[[dict], bool] | None = None,
        debounce_seconds: float = 0,
        name: str | None = None,
    ):
        """
        Initialisiert den Event-Trigger.

        Args:
            event_names: Event-Name(n) auf die reagiert wird
            filter_fn: Optionale Filter-Funktion für Event-Daten
            debounce_seconds: Mindestzeit zwischen Auslösungen
            name: Trigger-Name
        """
        super().__init__(name)

        if isinstance(event_names, str):
            self._event_names = [event_names]
        else:
            self._event_names = list(event_names)

        self._filter_fn = filter_fn
        self._debounce_seconds = debounce_seconds
        self._pending_event: dict | None = None

    @property
    def event_names(self) -> list[str]:
        """Event-Namen auf die reagiert wird."""
        return list(self._event_names)

    @property
    def debounce_seconds(self) -> float:
        """Debounce-Zeit in Sekunden."""
        return self._debounce_seconds

    def on_event(self, event_name: str, event_data: dict | None = None) -> bool:
        """
        Wird aufgerufen wenn ein Event eintrifft.

        Args:
            event_name: Name des Events
            event_data: Event-Daten

        Returns:
            True wenn Trigger aktiviert wurde
        """
        if not self.is_active:
            return False

        # Event-Name prüfen
        if event_name not in self._event_names:
            return False

        # Filter prüfen
        if self._filter_fn and event_data:
            try:
                if not self._filter_fn(event_data):
                    return False
            except Exception:
                return False

        # Debounce prüfen
        if self._debounce_seconds > 0 and self._last_fire_time:
            elapsed = (datetime.now() - self._last_fire_time).total_seconds()
            if elapsed < self._debounce_seconds:
                return False

        # Event speichern für should_fire
        self._pending_event = {
            "event_name": event_name,
            "event_data": event_data or {},
            "timestamp": datetime.now(),
        }

        return True

    def should_fire(self, context: TriggerContext) -> bool:
        """Prüft ob Trigger auslösen soll."""
        if not self.is_active:
            return False

        # Event-basiert: Prüfe ob Event im Kontext
        if context.event_name:
            # Event-Name prüfen
            if context.event_name not in self._event_names:
                return False

            # Filter prüfen
            if self._filter_fn and context.event_data:
                try:
                    if not self._filter_fn(context.event_data):
                        return False
                except Exception:
                    return False

            # Debounce prüfen
            if self._debounce_seconds > 0 and context.last_fire_time:
                elapsed = (context.current_time - context.last_fire_time).total_seconds()
                if elapsed < self._debounce_seconds:
                    return False

            return True

        # Alternativ: Pending Event prüfen
        if self._pending_event:
            self._pending_event = None  # Einmalig
            return True

        return False

    def get_next_fire_time(
        self,
        last_fire_time: datetime | None = None,
    ) -> datetime | None:
        """
        Berechnet nächsten Auslösezeitpunkt.

        Event-Trigger haben keinen planbaren Zeitpunkt.
        """
        # Event-basierte Trigger haben keinen festen Zeitpunkt
        return None

    def clear_pending(self) -> None:
        """Löscht pending Event."""
        self._pending_event = None

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "event_names": self._event_names,
            "debounce_seconds": self._debounce_seconds,
            # filter_fn kann nicht serialisiert werden
        }

    @classmethod
    def from_dict(cls, data: dict) -> "EventTrigger":
        """Erstellt Trigger aus Dictionary."""
        return cls(
            event_names=data["event_names"],
            debounce_seconds=data.get("debounce_seconds", 0),
            name=data.get("name"),
            # filter_fn muss separat gesetzt werden
        )

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "Event"

    @classmethod
    def form_fields(cls) -> list[FormField]:
        """Gibt die Formular-Felder fuer die TUI zurueck."""
        return [
            FormField(
                name="event_names", label="Event-Namen",
                field_type=FieldType.TAGS, required=True,
                placeholder="wakeword_received, satellite_connected",
                help_text="Komma-separierte Event-Namen",
            ),
            FormField(
                name="debounce_seconds", label="Debounce (Sek.)",
                field_type=FieldType.NUMBER,
                default=0, min_value=0, is_float=True,
            ),
        ]

    @classmethod
    def validate_form(cls, data: dict) -> list[FormValidationError]:
        """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck."""
        errors: list[FormValidationError] = []
        names = data.get("event_names", "")
        if isinstance(names, str):
            names = [n.strip() for n in names.split(",") if n.strip()]
        if not names:
            errors.append(FormValidationError("event_names", "Mindestens ein Event-Name ist erforderlich"))
        return errors

    def __repr__(self) -> str:
        return f"EventTrigger(events={self._event_names})"