"""Profile 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.profile import Condition, Profile from wled_controller.utils import get_logger logger = get_logger(__name__) class ProfileStore: """Persistent storage for automation profiles.""" def __init__(self, file_path: str): self.file_path = Path(file_path) self._profiles: Dict[str, Profile] = {} 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) profiles_data = data.get("profiles", {}) loaded = 0 for profile_id, profile_dict in profiles_data.items(): try: profile = Profile.from_dict(profile_dict) self._profiles[profile_id] = profile loaded += 1 except Exception as e: logger.error(f"Failed to load profile {profile_id}: {e}", exc_info=True) if loaded > 0: logger.info(f"Loaded {loaded} profiles from storage") except Exception as e: logger.error(f"Failed to load profiles from {self.file_path}: {e}") raise logger.info(f"Profile store initialized with {len(self._profiles)} profiles") def _save(self) -> None: try: self.file_path.parent.mkdir(parents=True, exist_ok=True) data = { "version": "1.0.0", "profiles": { pid: p.to_dict() for pid, p in self._profiles.items() }, } with open(self.file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Failed to save profiles to {self.file_path}: {e}") raise def get_all_profiles(self) -> List[Profile]: return list(self._profiles.values()) def get_profile(self, profile_id: str) -> Profile: if profile_id not in self._profiles: raise ValueError(f"Profile not found: {profile_id}") return self._profiles[profile_id] def create_profile( self, name: str, enabled: bool = True, condition_logic: str = "or", conditions: Optional[List[Condition]] = None, target_ids: Optional[List[str]] = None, ) -> Profile: profile_id = f"prof_{uuid.uuid4().hex[:8]}" now = datetime.utcnow() profile = Profile( id=profile_id, name=name, enabled=enabled, condition_logic=condition_logic, conditions=conditions or [], target_ids=target_ids or [], created_at=now, updated_at=now, ) self._profiles[profile_id] = profile self._save() logger.info(f"Created profile: {name} ({profile_id})") return profile def update_profile( self, profile_id: str, name: Optional[str] = None, enabled: Optional[bool] = None, condition_logic: Optional[str] = None, conditions: Optional[List[Condition]] = None, target_ids: Optional[List[str]] = None, ) -> Profile: if profile_id not in self._profiles: raise ValueError(f"Profile not found: {profile_id}") profile = self._profiles[profile_id] if name is not None: profile.name = name if enabled is not None: profile.enabled = enabled if condition_logic is not None: profile.condition_logic = condition_logic if conditions is not None: profile.conditions = conditions if target_ids is not None: profile.target_ids = target_ids profile.updated_at = datetime.utcnow() self._save() logger.info(f"Updated profile: {profile_id}") return profile def delete_profile(self, profile_id: str) -> None: if profile_id not in self._profiles: raise ValueError(f"Profile not found: {profile_id}") del self._profiles[profile_id] self._save() logger.info(f"Deleted profile: {profile_id}") def count(self) -> int: return len(self._profiles)