"""Automation and Condition data models.""" from dataclasses import dataclass, field from datetime import datetime, timezone 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) if ct == "webhook": return WebhookCondition.from_dict(data) if ct == "startup": return StartupCondition.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 WebhookCondition(Condition): """Activate via an HTTP webhook call with a secret token.""" condition_type: str = "webhook" token: str = "" # auto-generated 128-bit hex secret def to_dict(self) -> dict: d = super().to_dict() d["token"] = self.token return d @classmethod def from_dict(cls, data: dict) -> "WebhookCondition": return cls(token=data.get("token", "")) @dataclass class StartupCondition(Condition): """Activate when the server starts — stays active while enabled.""" condition_type: str = "startup" @classmethod def from_dict(cls, data: dict) -> "StartupCondition": return cls() @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 tags: List[str] = field(default_factory=list) 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, "tags": self.tags, "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"), tags=data.get("tags", []), created_at=datetime.fromisoformat(data.get("created_at", datetime.now(timezone.utc).isoformat())), updated_at=datetime.fromisoformat(data.get("updated_at", datetime.now(timezone.utc).isoformat())), )