"""Device 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.core.calibration import ( CalibrationConfig, calibration_from_dict, calibration_to_dict, create_default_calibration, ) from wled_controller.core.processor_manager import DEFAULT_STATE_CHECK_INTERVAL, ProcessingSettings from wled_controller.utils import get_logger logger = get_logger(__name__) class Device: """Represents a WLED device configuration.""" def __init__( self, device_id: str, name: str, url: str, led_count: int, enabled: bool = True, settings: Optional[ProcessingSettings] = None, calibration: Optional[CalibrationConfig] = None, created_at: Optional[datetime] = None, updated_at: Optional[datetime] = None, ): """Initialize device. Args: device_id: Unique device identifier name: Device name url: WLED device URL led_count: Number of LEDs enabled: Whether device is enabled settings: Processing settings calibration: Calibration configuration created_at: Creation timestamp updated_at: Last update timestamp """ self.id = device_id self.name = name self.url = url self.led_count = led_count self.enabled = enabled self.settings = settings or ProcessingSettings() self.calibration = calibration or create_default_calibration(led_count) self.created_at = created_at or datetime.utcnow() self.updated_at = updated_at or datetime.utcnow() def to_dict(self) -> dict: """Convert device to dictionary. Returns: Dictionary representation """ return { "id": self.id, "name": self.name, "url": self.url, "led_count": self.led_count, "enabled": self.enabled, "settings": { "display_index": self.settings.display_index, "fps": self.settings.fps, "border_width": self.settings.border_width, "brightness": self.settings.brightness, "gamma": self.settings.gamma, "saturation": self.settings.saturation, "smoothing": self.settings.smoothing, "interpolation_mode": self.settings.interpolation_mode, "state_check_interval": self.settings.state_check_interval, }, "calibration": calibration_to_dict(self.calibration), "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), } @classmethod def from_dict(cls, data: dict) -> "Device": """Create device from dictionary. Args: data: Dictionary with device data Returns: Device instance """ settings_data = data.get("settings", {}) settings = ProcessingSettings( display_index=settings_data.get("display_index", 0), fps=settings_data.get("fps", 30), border_width=settings_data.get("border_width", 10), brightness=settings_data.get("brightness", 1.0), gamma=settings_data.get("gamma", 2.2), saturation=settings_data.get("saturation", 1.0), smoothing=settings_data.get("smoothing", 0.3), interpolation_mode=settings_data.get("interpolation_mode", "average"), state_check_interval=settings_data.get( "state_check_interval", settings_data.get("health_check_interval", DEFAULT_STATE_CHECK_INTERVAL), ), ) calibration_data = data.get("calibration") calibration = ( calibration_from_dict(calibration_data) if calibration_data else create_default_calibration(data["led_count"]) ) return cls( device_id=data["id"], name=data["name"], url=data["url"], led_count=data["led_count"], enabled=data.get("enabled", True), settings=settings, calibration=calibration, created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())), updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())), ) class DeviceStore: """Persistent storage for WLED devices.""" def __init__(self, storage_file: str | Path): """Initialize device store. Args: storage_file: Path to JSON storage file """ self.storage_file = Path(storage_file) self._devices: Dict[str, Device] = {} # Ensure directory exists self.storage_file.parent.mkdir(parents=True, exist_ok=True) # Load existing devices self.load() logger.info(f"Device store initialized with {len(self._devices)} devices") def load(self): """Load devices from storage file.""" if not self.storage_file.exists(): logger.info(f"Storage file does not exist, starting with empty store") return try: with open(self.storage_file, "r") as f: data = json.load(f) devices_data = data.get("devices", {}) self._devices = { device_id: Device.from_dict(device_data) for device_id, device_data in devices_data.items() } logger.info(f"Loaded {len(self._devices)} devices from storage") except json.JSONDecodeError as e: logger.error(f"Failed to parse storage file: {e}") raise except Exception as e: logger.error(f"Failed to load devices: {e}") raise def save(self): """Save devices to storage file.""" try: data = { "devices": { device_id: device.to_dict() for device_id, device in self._devices.items() } } # Write to temporary file first temp_file = self.storage_file.with_suffix(".tmp") with open(temp_file, "w") as f: json.dump(data, f, indent=2) # Atomic rename temp_file.replace(self.storage_file) logger.debug(f"Saved {len(self._devices)} devices to storage") except Exception as e: logger.error(f"Failed to save devices: {e}") raise def create_device( self, name: str, url: str, led_count: int, settings: Optional[ProcessingSettings] = None, calibration: Optional[CalibrationConfig] = None, ) -> Device: """Create a new device. Args: name: Device name url: WLED device URL led_count: Number of LEDs settings: Processing settings calibration: Calibration configuration Returns: Created device Raises: ValueError: If validation fails """ # Generate unique ID device_id = f"device_{uuid.uuid4().hex[:8]}" # Create device device = Device( device_id=device_id, name=name, url=url, led_count=led_count, settings=settings, calibration=calibration, ) # Store self._devices[device_id] = device self.save() logger.info(f"Created device {device_id}: {name}") return device def get_device(self, device_id: str) -> Optional[Device]: """Get device by ID. Args: device_id: Device identifier Returns: Device or None if not found """ return self._devices.get(device_id) def get_all_devices(self) -> List[Device]: """Get all devices. Returns: List of all devices """ return list(self._devices.values()) def update_device( self, device_id: str, name: Optional[str] = None, url: Optional[str] = None, led_count: Optional[int] = None, enabled: Optional[bool] = None, settings: Optional[ProcessingSettings] = None, calibration: Optional[CalibrationConfig] = None, ) -> Device: """Update device. Args: device_id: Device identifier name: New name (optional) url: New URL (optional) led_count: New LED count (optional) enabled: New enabled state (optional) settings: New settings (optional) calibration: New calibration (optional) Returns: Updated device Raises: ValueError: If device not found or validation fails """ device = self._devices.get(device_id) if not device: raise ValueError(f"Device {device_id} not found") # Update fields if name is not None: device.name = name if url is not None: device.url = url if led_count is not None: device.led_count = led_count # Reset calibration if LED count changed device.calibration = create_default_calibration(led_count) if enabled is not None: device.enabled = enabled if settings is not None: device.settings = settings if calibration is not None: # Validate LED count matches if calibration.get_total_leds() != device.led_count: raise ValueError( f"Calibration LED count ({calibration.get_total_leds()}) " f"does not match device LED count ({device.led_count})" ) device.calibration = calibration device.updated_at = datetime.utcnow() # Save self.save() logger.info(f"Updated device {device_id}") return device def delete_device(self, device_id: str): """Delete device. Args: device_id: Device identifier Raises: ValueError: If device not found """ if device_id not in self._devices: raise ValueError(f"Device {device_id} not found") del self._devices[device_id] self.save() logger.info(f"Deleted device {device_id}") def device_exists(self, device_id: str) -> bool: """Check if device exists. Args: device_id: Device identifier Returns: True if device exists """ return device_id in self._devices def count(self) -> int: """Get number of devices. Returns: Device count """ return len(self._devices) def clear(self): """Clear all devices (for testing).""" self._devices.clear() self.save() logger.warning("Cleared all devices from storage")