# -*- coding: utf-8 -*- """ Interval Trigger - Zeitbasierter Trigger mit festen Intervallen. """ from datetime import datetime, timedelta from typing import Any from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError from trixy_core.scheduler.trigger.base import Trigger, TriggerContext, TriggerState class IntervalTrigger(Trigger): """ Trigger mit festem Zeitintervall. Beispiele: - IntervalTrigger(seconds=30) - Alle 30 Sekunden - IntervalTrigger(minutes=5) - Alle 5 Minuten - IntervalTrigger(hours=1) - Jede Stunde - IntervalTrigger(days=1, hours=12) - Alle 1,5 Tage """ def __init__( self, weeks: int = 0, days: int = 0, hours: int = 0, minutes: int = 0, seconds: int = 0, start_time: datetime | None = None, end_time: datetime | None = None, jitter_seconds: float = 0, name: str | None = None, ): """ Initialisiert den Interval-Trigger. Args: weeks: Wochen days: Tage hours: Stunden minutes: Minuten seconds: Sekunden start_time: Frühester Startzeitpunkt end_time: Spätester Endzeitpunkt jitter_seconds: Zufällige Abweichung (optional) name: Trigger-Name """ super().__init__(name) self._weeks = weeks self._days = days self._hours = hours self._minutes = minutes self._seconds = seconds self._start_time = start_time self._end_time = end_time self._jitter_seconds = jitter_seconds # Interval berechnen self._interval = timedelta( weeks=weeks, days=days, hours=hours, minutes=minutes, seconds=seconds, ) if self._interval.total_seconds() <= 0: raise ValueError("Interval muss größer als 0 sein") @property def interval(self) -> timedelta: """Intervall als timedelta.""" return self._interval @property def interval_seconds(self) -> float: """Intervall in Sekunden.""" return self._interval.total_seconds() @property def start_time(self) -> datetime | None: """Startzeitpunkt.""" return self._start_time @property def end_time(self) -> datetime | None: """Endzeitpunkt.""" return self._end_time def should_fire(self, context: TriggerContext) -> bool: """Prüft ob Trigger auslösen soll.""" if not self.is_active: return False now = context.current_time # Zeitfenster prüfen if self._start_time and now < self._start_time: return False if self._end_time and now > self._end_time: self._state = TriggerState.FINISHED return False # Erstes Mal if not context.last_fire_time: return True # Interval prüfen elapsed = now - context.last_fire_time return elapsed >= self._interval def get_next_fire_time( self, last_fire_time: datetime | None = None, ) -> datetime | None: """Berechnet nächsten Auslösezeitpunkt.""" if not self.is_active: return None now = datetime.now() # Basis für Berechnung if last_fire_time: next_time = last_fire_time + self._interval elif self._start_time and self._start_time > now: next_time = self._start_time else: next_time = now # Wenn nächster Zeitpunkt in Vergangenheit, auf jetzt setzen if next_time < now: # Berechne wie viele Intervalle vergangen sind elapsed = now - next_time intervals_passed = int(elapsed / self._interval) + 1 next_time = next_time + (self._interval * intervals_passed) # Jitter hinzufügen if self._jitter_seconds > 0: import random jitter = random.uniform(-self._jitter_seconds, self._jitter_seconds) next_time = next_time + timedelta(seconds=jitter) # Ende prüfen if self._end_time and next_time > self._end_time: return None return next_time def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "weeks": self._weeks, "days": self._days, "hours": self._hours, "minutes": self._minutes, "seconds": self._seconds, "start_time": ( self._start_time.isoformat() if self._start_time else None ), "end_time": ( self._end_time.isoformat() if self._end_time else None ), "jitter_seconds": self._jitter_seconds, } @classmethod def from_dict(cls, data: dict) -> "IntervalTrigger": """Erstellt Trigger aus Dictionary.""" start_time = None end_time = None if data.get("start_time"): start_time = datetime.fromisoformat(data["start_time"]) if data.get("end_time"): end_time = datetime.fromisoformat(data["end_time"]) return cls( weeks=data.get("weeks", 0), days=data.get("days", 0), hours=data.get("hours", 0), minutes=data.get("minutes", 0), seconds=data.get("seconds", 0), start_time=start_time, end_time=end_time, jitter_seconds=data.get("jitter_seconds", 0), name=data.get("name"), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Intervall" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="weeks", label="Wochen", field_type=FieldType.NUMBER, default=0, min_value=0, group="Intervall", ), FormField( name="days", label="Tage", field_type=FieldType.NUMBER, default=0, min_value=0, group="Intervall", ), FormField( name="hours", label="Stunden", field_type=FieldType.NUMBER, default=0, min_value=0, group="Intervall", ), FormField( name="minutes", label="Minuten", field_type=FieldType.NUMBER, default=0, min_value=0, group="Intervall", ), FormField( name="seconds", label="Sekunden", field_type=FieldType.NUMBER, default=0, min_value=0, group="Intervall", ), FormField( name="start_time", label="Startzeit", field_type=FieldType.DATETIME, placeholder="2024-01-01T00:00:00", group="Zeitfenster", ), FormField( name="end_time", label="Endzeit", field_type=FieldType.DATETIME, placeholder="2024-12-31T23:59:59", group="Zeitfenster", ), FormField( name="jitter_seconds", label="Jitter (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] = [] total = sum(data.get(k, 0) for k in ("weeks", "days", "hours", "minutes", "seconds")) if total <= 0: errors.append(FormValidationError("seconds", "Intervall muss groesser als 0 sein")) return errors def __repr__(self) -> str: return f"IntervalTrigger(interval={self._interval})"