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

"""Automation storage using JSON files."""

import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

from wled_controller.storage.automation import Automation, Condition
from wled_controller.utils import atomic_write_json, get_logger

logger = get_logger(__name__)


class AutomationStore:
    """Persistent storage for automations."""

    def __init__(self, file_path: str):
        self.file_path = Path(file_path)
        self._automations: Dict[str, Automation] = {}
        self._load()

    def _load(self) -> None:
        if not self.file_path.exists():
            return

        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                data = json.load(f)

            automations_data = data.get("automations", {})
            loaded = 0
            for auto_id, auto_dict in automations_data.items():
                try:
                    automation = Automation.from_dict(auto_dict)
                    self._automations[auto_id] = automation
                    loaded += 1
                except Exception as e:
                    logger.error(f"Failed to load automation {auto_id}: {e}", exc_info=True)

            if loaded > 0:
                logger.info(f"Loaded {loaded} automations from storage")

        except Exception as e:
            logger.error(f"Failed to load automations from {self.file_path}: {e}")
            raise

        logger.info(f"Automation store initialized with {len(self._automations)} automations")

    def _save(self) -> None:
        try:
            data = {
                "version": "1.0.0",
                "automations": {
                    aid: a.to_dict() for aid, a in self._automations.items()
                },
            }
            atomic_write_json(self.file_path, data)
        except Exception as e:
            logger.error(f"Failed to save automations to {self.file_path}: {e}")
            raise

    def get_all_automations(self) -> List[Automation]:
        return list(self._automations.values())

    def get_automation(self, automation_id: str) -> Automation:
        if automation_id not in self._automations:
            raise ValueError(f"Automation not found: {automation_id}")
        return self._automations[automation_id]

    def create_automation(
        self,
        name: str,
        enabled: bool = True,
        condition_logic: str = "or",
        conditions: Optional[List[Condition]] = None,
        scene_preset_id: Optional[str] = None,
        deactivation_mode: str = "none",
        deactivation_scene_preset_id: Optional[str] = None,
    ) -> Automation:
        for a in self._automations.values():
            if a.name == name:
                raise ValueError(f"Automation with name '{name}' already exists")

        automation_id = f"auto_{uuid.uuid4().hex[:8]}"
        now = datetime.utcnow()

        automation = Automation(
            id=automation_id,
            name=name,
            enabled=enabled,
            condition_logic=condition_logic,
            conditions=conditions or [],
            scene_preset_id=scene_preset_id,
            deactivation_mode=deactivation_mode,
            deactivation_scene_preset_id=deactivation_scene_preset_id,
            created_at=now,
            updated_at=now,
        )

        self._automations[automation_id] = automation
        self._save()

        logger.info(f"Created automation: {name} ({automation_id})")
        return automation

    def update_automation(
        self,
        automation_id: str,
        name: Optional[str] = None,
        enabled: Optional[bool] = None,
        condition_logic: Optional[str] = None,
        conditions: Optional[List[Condition]] = None,
        scene_preset_id: str = "__unset__",
        deactivation_mode: Optional[str] = None,
        deactivation_scene_preset_id: str = "__unset__",
    ) -> Automation:
        if automation_id not in self._automations:
            raise ValueError(f"Automation not found: {automation_id}")

        automation = self._automations[automation_id]

        if name is not None:
            for aid, a in self._automations.items():
                if aid != automation_id and a.name == name:
                    raise ValueError(f"Automation with name '{name}' already exists")
            automation.name = name
        if enabled is not None:
            automation.enabled = enabled
        if condition_logic is not None:
            automation.condition_logic = condition_logic
        if conditions is not None:
            automation.conditions = conditions
        if scene_preset_id != "__unset__":
            automation.scene_preset_id = scene_preset_id
        if deactivation_mode is not None:
            automation.deactivation_mode = deactivation_mode
        if deactivation_scene_preset_id != "__unset__":
            automation.deactivation_scene_preset_id = deactivation_scene_preset_id

        automation.updated_at = datetime.utcnow()
        self._save()

        logger.info(f"Updated automation: {automation_id}")
        return automation

    def delete_automation(self, automation_id: str) -> None:
        if automation_id not in self._automations:
            raise ValueError(f"Automation not found: {automation_id}")

        del self._automations[automation_id]
        self._save()

        logger.info(f"Deleted automation: {automation_id}")

    def count(self) -> int:
        return len(self._automations)