wled-screen-controller-mixed/server/src/wled_controller/storage/automation.py

"""Automation and Condition data models."""

from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional


@dataclass
class Condition:
    """Base condition — polymorphic via condition_type discriminator."""

    condition_type: str

    def to_dict(self) -> dict:
        return {"condition_type": self.condition_type}

    @classmethod
    def from_dict(cls, data: dict) -> "Condition":
        """Factory: dispatch to the correct subclass."""
        ct = data.get("condition_type", "")
        if ct == "always":
            return AlwaysCondition.from_dict(data)
        if ct == "application":
            return ApplicationCondition.from_dict(data)
        if ct == "time_of_day":
            return TimeOfDayCondition.from_dict(data)
        if ct == "system_idle":
            return SystemIdleCondition.from_dict(data)
        if ct == "display_state":
            return DisplayStateCondition.from_dict(data)
        if ct == "mqtt":
            return MQTTCondition.from_dict(data)
        raise ValueError(f"Unknown condition type: {ct}")


@dataclass
class AlwaysCondition(Condition):
    """Always-true condition — automation activates unconditionally when enabled."""

    condition_type: str = "always"

    @classmethod
    def from_dict(cls, data: dict) -> "AlwaysCondition":
        return cls()


@dataclass
class ApplicationCondition(Condition):
    """Activate when specified applications are running or topmost."""

    condition_type: str = "application"
    apps: List[str] = field(default_factory=list)
    match_type: str = "running"  # "running" | "topmost"

    def to_dict(self) -> dict:
        d = super().to_dict()
        d["apps"] = list(self.apps)
        d["match_type"] = self.match_type
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "ApplicationCondition":
        return cls(
            apps=data.get("apps", []),
            match_type=data.get("match_type", "running"),
        )


@dataclass
class TimeOfDayCondition(Condition):
    """Activate during a specific time range (server local time).

    Supports overnight ranges: if start_time > end_time, the range wraps
    around midnight (e.g. 22:00 → 06:00).
    """

    condition_type: str = "time_of_day"
    start_time: str = "00:00"  # HH:MM
    end_time: str = "23:59"  # HH:MM

    def to_dict(self) -> dict:
        d = super().to_dict()
        d["start_time"] = self.start_time
        d["end_time"] = self.end_time
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "TimeOfDayCondition":
        return cls(
            start_time=data.get("start_time", "00:00"),
            end_time=data.get("end_time", "23:59"),
        )


@dataclass
class SystemIdleCondition(Condition):
    """Activate based on system idle time (keyboard/mouse inactivity)."""

    condition_type: str = "system_idle"
    idle_minutes: int = 5
    when_idle: bool = True  # True = active when idle; False = active when NOT idle

    def to_dict(self) -> dict:
        d = super().to_dict()
        d["idle_minutes"] = self.idle_minutes
        d["when_idle"] = self.when_idle
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "SystemIdleCondition":
        return cls(
            idle_minutes=data.get("idle_minutes", 5),
            when_idle=data.get("when_idle", True),
        )


@dataclass
class DisplayStateCondition(Condition):
    """Activate based on display/monitor power state."""

    condition_type: str = "display_state"
    state: str = "on"  # "on" | "off"

    def to_dict(self) -> dict:
        d = super().to_dict()
        d["state"] = self.state
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "DisplayStateCondition":
        return cls(
            state=data.get("state", "on"),
        )


@dataclass
class MQTTCondition(Condition):
    """Activate based on an MQTT topic value."""

    condition_type: str = "mqtt"
    topic: str = ""
    payload: str = ""
    match_mode: str = "exact"  # "exact" | "contains" | "regex"

    def to_dict(self) -> dict:
        d = super().to_dict()
        d["topic"] = self.topic
        d["payload"] = self.payload
        d["match_mode"] = self.match_mode
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "MQTTCondition":
        return cls(
            topic=data.get("topic", ""),
            payload=data.get("payload", ""),
            match_mode=data.get("match_mode", "exact"),
        )


@dataclass
class Automation:
    """Automation that activates a scene preset based on conditions."""

    id: str
    name: str
    enabled: bool
    condition_logic: str  # "or" | "and"
    conditions: List[Condition]
    scene_preset_id: Optional[str]  # scene to activate when conditions are met
    deactivation_mode: str  # "none" | "revert" | "fallback_scene"
    deactivation_scene_preset_id: Optional[str]  # scene for fallback_scene mode
    created_at: datetime
    updated_at: datetime

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "enabled": self.enabled,
            "condition_logic": self.condition_logic,
            "conditions": [c.to_dict() for c in self.conditions],
            "scene_preset_id": self.scene_preset_id,
            "deactivation_mode": self.deactivation_mode,
            "deactivation_scene_preset_id": self.deactivation_scene_preset_id,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
        }

    @classmethod
    def from_dict(cls, data: dict) -> "Automation":
        conditions = []
        for c_data in data.get("conditions", []):
            try:
                conditions.append(Condition.from_dict(c_data))
            except ValueError:
                pass  # skip unknown condition types on load

        return cls(
            id=data["id"],
            name=data["name"],
            enabled=data.get("enabled", True),
            condition_logic=data.get("condition_logic", "or"),
            conditions=conditions,
            scene_preset_id=data.get("scene_preset_id"),
            deactivation_mode=data.get("deactivation_mode", "none"),
            deactivation_scene_preset_id=data.get("deactivation_scene_preset_id"),
            created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
            updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
        )