"""Device storage using JSON files.""" import json import uuid from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional from wled_controller.utils import atomic_write_json, get_logger logger = get_logger(__name__) class Device: """Represents a WLED device configuration. A device holds connection state and output settings. Calibration, processing settings, and picture source assignments now live on ColorStripSource and WledOutputTarget respectively. """ def __init__( self, device_id: str, name: str, url: str, led_count: int, enabled: bool = True, device_type: str = "wled", baud_rate: Optional[int] = None, software_brightness: int = 255, auto_shutdown: bool = False, send_latency_ms: int = 0, rgbw: bool = False, zone_mode: str = "combined", tags: List[str] = None, # DMX (Art-Net / sACN) fields dmx_protocol: str = "artnet", dmx_start_universe: int = 0, dmx_start_channel: int = 1, # ESP-NOW fields espnow_peer_mac: str = "", espnow_channel: int = 1, # Philips Hue fields hue_username: str = "", hue_client_key: str = "", hue_entertainment_group_id: str = "", # SPI Direct fields spi_speed_hz: int = 800000, spi_led_type: str = "WS2812B", # Razer Chroma fields chroma_device_type: str = "chromalink", # SteelSeries GameSense fields gamesense_device_type: str = "keyboard", created_at: Optional[datetime] = None, updated_at: Optional[datetime] = None, ): self.id = device_id self.name = name self.url = url self.led_count = led_count self.enabled = enabled self.device_type = device_type self.baud_rate = baud_rate self.software_brightness = software_brightness self.auto_shutdown = auto_shutdown self.send_latency_ms = send_latency_ms self.rgbw = rgbw self.zone_mode = zone_mode self.tags = tags or [] self.dmx_protocol = dmx_protocol self.dmx_start_universe = dmx_start_universe self.dmx_start_channel = dmx_start_channel self.espnow_peer_mac = espnow_peer_mac self.espnow_channel = espnow_channel self.hue_username = hue_username self.hue_client_key = hue_client_key self.hue_entertainment_group_id = hue_entertainment_group_id self.spi_speed_hz = spi_speed_hz self.spi_led_type = spi_led_type self.chroma_device_type = chroma_device_type self.gamesense_device_type = gamesense_device_type self.created_at = created_at or datetime.now(timezone.utc) self.updated_at = updated_at or datetime.now(timezone.utc) def to_dict(self) -> dict: """Convert device to dictionary.""" d = { "id": self.id, "name": self.name, "url": self.url, "led_count": self.led_count, "enabled": self.enabled, "device_type": self.device_type, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), } if self.baud_rate is not None: d["baud_rate"] = self.baud_rate if self.software_brightness != 255: d["software_brightness"] = self.software_brightness if self.auto_shutdown: d["auto_shutdown"] = True if self.send_latency_ms: d["send_latency_ms"] = self.send_latency_ms if self.rgbw: d["rgbw"] = True if self.zone_mode != "combined": d["zone_mode"] = self.zone_mode if self.tags: d["tags"] = self.tags if self.dmx_protocol != "artnet": d["dmx_protocol"] = self.dmx_protocol if self.dmx_start_universe != 0: d["dmx_start_universe"] = self.dmx_start_universe if self.dmx_start_channel != 1: d["dmx_start_channel"] = self.dmx_start_channel if self.espnow_peer_mac: d["espnow_peer_mac"] = self.espnow_peer_mac if self.espnow_channel != 1: d["espnow_channel"] = self.espnow_channel if self.hue_username: d["hue_username"] = self.hue_username if self.hue_client_key: d["hue_client_key"] = self.hue_client_key if self.hue_entertainment_group_id: d["hue_entertainment_group_id"] = self.hue_entertainment_group_id if self.spi_speed_hz != 800000: d["spi_speed_hz"] = self.spi_speed_hz if self.spi_led_type != "WS2812B": d["spi_led_type"] = self.spi_led_type if self.chroma_device_type != "chromalink": d["chroma_device_type"] = self.chroma_device_type if self.gamesense_device_type != "keyboard": d["gamesense_device_type"] = self.gamesense_device_type return d @classmethod def from_dict(cls, data: dict) -> "Device": """Create device from dictionary.""" return cls( device_id=data["id"], name=data["name"], url=data["url"], led_count=data["led_count"], enabled=data.get("enabled", True), device_type=data.get("device_type", "wled"), baud_rate=data.get("baud_rate"), software_brightness=data.get("software_brightness", 255), auto_shutdown=data.get("auto_shutdown", False), send_latency_ms=data.get("send_latency_ms", 0), rgbw=data.get("rgbw", False), zone_mode=data.get("zone_mode", "combined"), tags=data.get("tags", []), dmx_protocol=data.get("dmx_protocol", "artnet"), dmx_start_universe=data.get("dmx_start_universe", 0), dmx_start_channel=data.get("dmx_start_channel", 1), espnow_peer_mac=data.get("espnow_peer_mac", ""), espnow_channel=data.get("espnow_channel", 1), hue_username=data.get("hue_username", ""), hue_client_key=data.get("hue_client_key", ""), hue_entertainment_group_id=data.get("hue_entertainment_group_id", ""), spi_speed_hz=data.get("spi_speed_hz", 800000), spi_led_type=data.get("spi_led_type", "WS2812B"), chroma_device_type=data.get("chroma_device_type", "chromalink"), gamesense_device_type=data.get("gamesense_device_type", "keyboard"), 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())), ) class DeviceStore: """Persistent storage for WLED devices.""" def __init__(self, storage_file: str | Path): 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("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 load_raw(self) -> dict: """Load raw JSON data from storage (for migration).""" if not self.storage_file.exists(): return {} try: with open(self.storage_file, "r") as f: return json.load(f) except Exception: return {} def save(self): """Save devices to storage file.""" try: data = { "devices": { device_id: device.to_dict() for device_id, device in self._devices.items() } } atomic_write_json(self.storage_file, data) 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, device_type: str = "wled", baud_rate: Optional[int] = None, auto_shutdown: bool = False, send_latency_ms: int = 0, rgbw: bool = False, zone_mode: str = "combined", tags: Optional[List[str]] = None, dmx_protocol: str = "artnet", dmx_start_universe: int = 0, dmx_start_channel: int = 1, espnow_peer_mac: str = "", espnow_channel: int = 1, hue_username: str = "", hue_client_key: str = "", hue_entertainment_group_id: str = "", spi_speed_hz: int = 800000, spi_led_type: str = "WS2812B", chroma_device_type: str = "chromalink", gamesense_device_type: str = "keyboard", ) -> Device: """Create a new device.""" device_id = f"device_{uuid.uuid4().hex[:8]}" # Mock devices use their device ID as the URL authority if device_type == "mock": url = f"mock://{device_id}" device = Device( device_id=device_id, name=name, url=url, led_count=led_count, device_type=device_type, baud_rate=baud_rate, auto_shutdown=auto_shutdown, send_latency_ms=send_latency_ms, rgbw=rgbw, zone_mode=zone_mode, tags=tags or [], dmx_protocol=dmx_protocol, dmx_start_universe=dmx_start_universe, dmx_start_channel=dmx_start_channel, espnow_peer_mac=espnow_peer_mac, espnow_channel=espnow_channel, hue_username=hue_username, hue_client_key=hue_client_key, hue_entertainment_group_id=hue_entertainment_group_id, spi_speed_hz=spi_speed_hz, spi_led_type=spi_led_type, chroma_device_type=chroma_device_type, gamesense_device_type=gamesense_device_type, ) 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.""" return self._devices.get(device_id) def get_all_devices(self) -> List[Device]: """Get 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, baud_rate: Optional[int] = None, auto_shutdown: Optional[bool] = None, send_latency_ms: Optional[int] = None, rgbw: Optional[bool] = None, zone_mode: Optional[str] = None, tags: Optional[List[str]] = None, dmx_protocol: Optional[str] = None, dmx_start_universe: Optional[int] = None, dmx_start_channel: Optional[int] = None, espnow_peer_mac: Optional[str] = None, espnow_channel: Optional[int] = None, hue_username: Optional[str] = None, hue_client_key: Optional[str] = None, hue_entertainment_group_id: Optional[str] = None, spi_speed_hz: Optional[int] = None, spi_led_type: Optional[str] = None, chroma_device_type: Optional[str] = None, gamesense_device_type: Optional[str] = None, ) -> Device: """Update device.""" device = self._devices.get(device_id) if not device: raise ValueError(f"Device {device_id} not found") 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 if enabled is not None: device.enabled = enabled if baud_rate is not None: device.baud_rate = baud_rate if auto_shutdown is not None: device.auto_shutdown = auto_shutdown if send_latency_ms is not None: device.send_latency_ms = send_latency_ms if rgbw is not None: device.rgbw = rgbw if zone_mode is not None: device.zone_mode = zone_mode if tags is not None: device.tags = tags if dmx_protocol is not None: device.dmx_protocol = dmx_protocol if dmx_start_universe is not None: device.dmx_start_universe = dmx_start_universe if dmx_start_channel is not None: device.dmx_start_channel = dmx_start_channel if espnow_peer_mac is not None: device.espnow_peer_mac = espnow_peer_mac if espnow_channel is not None: device.espnow_channel = espnow_channel if hue_username is not None: device.hue_username = hue_username if hue_client_key is not None: device.hue_client_key = hue_client_key if hue_entertainment_group_id is not None: device.hue_entertainment_group_id = hue_entertainment_group_id if spi_speed_hz is not None: device.spi_speed_hz = spi_speed_hz if spi_led_type is not None: device.spi_led_type = spi_led_type if chroma_device_type is not None: device.chroma_device_type = chroma_device_type if gamesense_device_type is not None: device.gamesense_device_type = gamesense_device_type device.updated_at = datetime.now(timezone.utc) self.save() logger.info(f"Updated device {device_id}") return device def delete_device(self, device_id: str): """Delete device.""" 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.""" return device_id in self._devices def count(self) -> int: """Get number of devices.""" return len(self._devices) def clear(self): """Clear all devices (for testing).""" self._devices.clear() self.save() logger.warning("Cleared all devices from storage")