ledgrab/server/src/wled_controller/storage/device_store.py

"""Device storage using JSON files."""

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

from wled_controller.utils import 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 WledPictureTarget 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,
        static_color: Optional[Tuple[int, int, int]] = None,
        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.static_color = static_color
        self.created_at = created_at or datetime.utcnow()
        self.updated_at = updated_at or datetime.utcnow()
        # Preserved from old JSON for migration — not written back
        self._legacy_calibration = None

    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.static_color is not None:
            d["static_color"] = list(self.static_color)
        return d

    @classmethod
    def from_dict(cls, data: dict) -> "Device":
        """Create device from dictionary.

        Backward-compatible: reads legacy 'calibration' field and stores it
        in _legacy_calibration for migration use only.
        """
        static_color_raw = data.get("static_color")
        static_color = tuple(static_color_raw) if static_color_raw else None

        device = 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),
            static_color=static_color,
            created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
            updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
        )

        # Preserve old calibration for migration (never written back by to_dict)
        calibration_data = data.get("calibration")
        if calibration_data:
            try:
                from wled_controller.core.capture.calibration import calibration_from_dict
                device._legacy_calibration = calibration_from_dict(calibration_data)
            except Exception:
                pass

        return device


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()
                }
            }

            temp_file = self.storage_file.with_suffix(".tmp")
            with open(temp_file, "w") as f:
                json.dump(data, f, indent=2)

            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,
        device_type: str = "wled",
        baud_rate: Optional[int] = None,
        auto_shutdown: bool = False,
    ) -> Device:
        """Create a new device."""
        device_id = f"device_{uuid.uuid4().hex[:8]}"

        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,
        )

        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,
    ) -> 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

        device.updated_at = datetime.utcnow()
        self.save()

        logger.info(f"Updated device {device_id}")
        return device

    def set_static_color(
        self, device_id: str, color: Optional[Tuple[int, int, int]]
    ) -> "Device":
        """Set or clear the static idle color for a device."""
        device = self._devices.get(device_id)
        if not device:
            raise ValueError(f"Device {device_id} not found")
        device.static_color = color
        device.updated_at = datetime.utcnow()
        self.save()
        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")