# -*- coding: utf-8 -*- """ Sensor Conditions - Sensorbasierte Bedingungen. """ from enum import Enum from typing import Any from trixy_core.scheduler.condition.base import Condition from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError class ComparisonOperator(Enum): """Vergleichsoperatoren.""" EQUAL = "eq" NOT_EQUAL = "ne" GREATER = "gt" GREATER_EQUAL = "gte" LESS = "lt" LESS_EQUAL = "lte" BETWEEN = "between" IN = "in" NOT_IN = "not_in" class SensorThresholdCondition(Condition): """ Prüft ob Sensorwert Schwellenwert erreicht. Beispiel: - SensorThresholdCondition("temperature", ComparisonOperator.GREATER, 25) → Nur wenn Temperatur > 25 - SensorThresholdCondition("volume", ComparisonOperator.BETWEEN, (0.5, 0.8)) → Nur wenn Lautstärke zwischen 0.5 und 0.8 """ def __init__( self, sensor_key: str, operator: ComparisonOperator, value: Any, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: sensor_key: Schlüssel für Sensorwert im Kontext operator: Vergleichsoperator value: Vergleichswert (bei BETWEEN: Tuple (min, max)) name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name, invert) self._sensor_key = sensor_key self._operator = operator self._value = value @property def sensor_key(self) -> str: """Sensor-Schlüssel.""" return self._sensor_key @property def operator(self) -> ComparisonOperator: """Vergleichsoperator.""" return self._operator @property def value(self) -> Any: """Vergleichswert.""" return self._value def _evaluate(self, context: dict) -> bool: """Prüft ob Sensorwert Bedingung erfüllt.""" sensor_value = context.get(self._sensor_key) if sensor_value is None: return False try: if self._operator == ComparisonOperator.EQUAL: return sensor_value == self._value elif self._operator == ComparisonOperator.NOT_EQUAL: return sensor_value != self._value elif self._operator == ComparisonOperator.GREATER: return sensor_value > self._value elif self._operator == ComparisonOperator.GREATER_EQUAL: return sensor_value >= self._value elif self._operator == ComparisonOperator.LESS: return sensor_value < self._value elif self._operator == ComparisonOperator.LESS_EQUAL: return sensor_value <= self._value elif self._operator == ComparisonOperator.BETWEEN: min_val, max_val = self._value return min_val <= sensor_value <= max_val elif self._operator == ComparisonOperator.IN: return sensor_value in self._value elif self._operator == ComparisonOperator.NOT_IN: return sensor_value not in self._value except (TypeError, ValueError): return False return False def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "sensor_key": self._sensor_key, "operator": self._operator.value, "value": self._value, } @classmethod def from_dict(cls, data: dict) -> "SensorThresholdCondition": """Erstellt Condition aus Dictionary.""" return cls( sensor_key=data["sensor_key"], operator=ComparisonOperator(data["operator"]), value=data["value"], name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Sensor-Schwellwert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="sensor_key", label="Sensor-Schluessel", required=True, placeholder="temperature", help_text="Schluessel fuer den Sensorwert im Kontext", ), FormField( name="operator", label="Operator", field_type=FieldType.SELECT, default="gt", choices=[ ("eq", "Gleich (==)"), ("ne", "Ungleich (!=)"), ("gt", "Groesser (>)"), ("gte", "Groesser/Gleich (>=)"), ("lt", "Kleiner (<)"), ("lte", "Kleiner/Gleich (<=)"), ], ), FormField( name="threshold", label="Schwellwert", field_type=FieldType.NUMBER, required=True, is_float=True, ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] if not data.get("sensor_key", "").strip(): errors.append(FormValidationError("sensor_key", "Sensor-Schluessel ist erforderlich")) if data.get("threshold") is None: errors.append(FormValidationError("threshold", "Schwellwert ist erforderlich")) return errors class SensorChangeCondition(Condition): """ Prüft ob Sensorwert sich um bestimmten Betrag geändert hat. Beispiel: - SensorChangeCondition("temperature", min_change=5.0) → Nur wenn Temperatur sich um min. 5 Grad geändert hat """ def __init__( self, sensor_key: str, previous_key: str | None = None, min_change: float | None = None, max_change: float | None = None, percentage: bool = False, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: sensor_key: Schlüssel für aktuellen Sensorwert previous_key: Schlüssel für vorherigen Wert (default: {sensor_key}_previous) min_change: Mindeständerung max_change: Maximaländerung percentage: Änderung als Prozentsatz name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name, invert) self._sensor_key = sensor_key self._previous_key = previous_key or f"{sensor_key}_previous" self._min_change = min_change self._max_change = max_change self._percentage = percentage @property def sensor_key(self) -> str: """Sensor-Schlüssel.""" return self._sensor_key def _evaluate(self, context: dict) -> bool: """Prüft ob Sensorwert sich ausreichend geändert hat.""" current = context.get(self._sensor_key) previous = context.get(self._previous_key) if current is None or previous is None: return False try: current = float(current) previous = float(previous) except (TypeError, ValueError): return False # Änderung berechnen if self._percentage and previous != 0: change = abs((current - previous) / previous * 100) else: change = abs(current - previous) if self._min_change is not None and change < self._min_change: return False if self._max_change is not None and change > self._max_change: return False return True def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "sensor_key": self._sensor_key, "previous_key": self._previous_key, "min_change": self._min_change, "max_change": self._max_change, "percentage": self._percentage, } @classmethod def from_dict(cls, data: dict) -> "SensorChangeCondition": """Erstellt Condition aus Dictionary.""" return cls( sensor_key=data["sensor_key"], previous_key=data.get("previous_key"), min_change=data.get("min_change"), max_change=data.get("max_change"), percentage=data.get("percentage", False), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Sensor-Aenderung" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="sensor_key", label="Sensor-Schluessel", required=True, placeholder="temperature", help_text="Schluessel fuer den Sensorwert im Kontext", ), FormField( name="change_threshold", label="Aenderungs-Schwellwert", field_type=FieldType.NUMBER, is_float=True, help_text="Mindestaenderung fuer Ausloesung", ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] if not data.get("sensor_key", "").strip(): errors.append(FormValidationError("sensor_key", "Sensor-Schluessel ist erforderlich")) return errors