Trixy7/trixy_core/scheduler/condition/sensor.py

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