Trixy7/trixy_core/scheduler/trigger/composite.py

# -*- coding: utf-8 -*-
"""
Composite Triggers - Kombiniert mehrere Trigger.
"""

from datetime import datetime
from typing import Any

from trixy_core.scheduler.trigger.base import Trigger, TriggerContext, TriggerState


class AndTrigger(Trigger):
    """
    Trigger der nur auslöst wenn ALLE Sub-Trigger auslösen.

    Beispiel:
        - AndTrigger([CronTrigger("0 9 * * *"), EventTrigger("user_present")])
          → Nur um 9 Uhr UND wenn User anwesend
    """

    def __init__(
        self,
        triggers: list[Trigger],
        name: str | None = None,
    ):
        """
        Initialisiert den AND-Trigger.

        Args:
            triggers: Liste von Sub-Triggern
            name: Trigger-Name
        """
        super().__init__(name)

        if not triggers:
            raise ValueError("AndTrigger benötigt mindestens einen Trigger")

        self._triggers = list(triggers)

    @property
    def triggers(self) -> list[Trigger]:
        """Sub-Trigger."""
        return list(self._triggers)

    def add_trigger(self, trigger: Trigger) -> None:
        """Fügt Trigger hinzu."""
        self._triggers.append(trigger)

    def remove_trigger(self, trigger: Trigger) -> bool:
        """Entfernt Trigger."""
        try:
            self._triggers.remove(trigger)
            return True
        except ValueError:
            return False

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

        if not self._triggers:
            return False

        return all(t.should_fire(context) for t in self._triggers)

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

        Für AND: Spätester Zeitpunkt aller Sub-Trigger.
        """
        if not self.is_active or not self._triggers:
            return None

        next_times = []
        for trigger in self._triggers:
            next_time = trigger.get_next_fire_time(last_fire_time)
            if next_time is None:
                # Wenn ein Trigger keinen Zeitpunkt hat (z.B. EventTrigger),
                # kann der AND-Trigger auch keinen planbaren Zeitpunkt haben
                return None
            next_times.append(next_time)

        # Spätester Zeitpunkt
        return max(next_times) if next_times else None

    def pause(self) -> None:
        """Pausiert alle Sub-Trigger."""
        super().pause()
        for trigger in self._triggers:
            trigger.pause()

    def resume(self) -> None:
        """Setzt alle Sub-Trigger fort."""
        super().resume()
        for trigger in self._triggers:
            trigger.resume()

    def reset(self) -> None:
        """Setzt alle Sub-Trigger zurück."""
        super().reset()
        for trigger in self._triggers:
            trigger.reset()

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "triggers": [t.to_dict() for t in self._triggers],
        }

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

    def __repr__(self) -> str:
        return f"AndTrigger(count={len(self._triggers)})"


class OrTrigger(Trigger):
    """
    Trigger der auslöst wenn MINDESTENS EIN Sub-Trigger auslöst.

    Beispiel:
        - OrTrigger([CronTrigger("0 9 * * *"), EventTrigger("urgent_task")])
          → Um 9 Uhr ODER bei dringendem Task
    """

    def __init__(
        self,
        triggers: list[Trigger],
        name: str | None = None,
    ):
        """
        Initialisiert den OR-Trigger.

        Args:
            triggers: Liste von Sub-Triggern
            name: Trigger-Name
        """
        super().__init__(name)

        if not triggers:
            raise ValueError("OrTrigger benötigt mindestens einen Trigger")

        self._triggers = list(triggers)
        self._last_fired_trigger: Trigger | None = None

    @property
    def triggers(self) -> list[Trigger]:
        """Sub-Trigger."""
        return list(self._triggers)

    @property
    def last_fired_trigger(self) -> Trigger | None:
        """Zuletzt ausgelöster Sub-Trigger."""
        return self._last_fired_trigger

    def add_trigger(self, trigger: Trigger) -> None:
        """Fügt Trigger hinzu."""
        self._triggers.append(trigger)

    def remove_trigger(self, trigger: Trigger) -> bool:
        """Entfernt Trigger."""
        try:
            self._triggers.remove(trigger)
            return True
        except ValueError:
            return False

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

        if not self._triggers:
            return False

        for trigger in self._triggers:
            if trigger.should_fire(context):
                self._last_fired_trigger = trigger
                return True

        return False

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

        Für OR: Frühester Zeitpunkt aller Sub-Trigger.
        """
        if not self.is_active or not self._triggers:
            return None

        next_times = []
        for trigger in self._triggers:
            next_time = trigger.get_next_fire_time(last_fire_time)
            if next_time:
                next_times.append(next_time)

        # Frühester Zeitpunkt
        return min(next_times) if next_times else None

    def pause(self) -> None:
        """Pausiert alle Sub-Trigger."""
        super().pause()
        for trigger in self._triggers:
            trigger.pause()

    def resume(self) -> None:
        """Setzt alle Sub-Trigger fort."""
        super().resume()
        for trigger in self._triggers:
            trigger.resume()

    def reset(self) -> None:
        """Setzt alle Sub-Trigger zurück."""
        super().reset()
        self._last_fired_trigger = None
        for trigger in self._triggers:
            trigger.reset()

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "triggers": [t.to_dict() for t in self._triggers],
        }

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

    def __repr__(self) -> str:
        return f"OrTrigger(count={len(self._triggers)})"