alexei.dolgolyov
d6cf45c873
- Add `static_color` capability to Adalight provider with `set_color()` method
- Add `static_color` field to Device model, DeviceState, and API schemas
- Add GET/PUT `/devices/{id}/color` API endpoints
- Change auto-shutdown behavior: restore device to idle state instead of
powering off (WLED uses snapshot/restore, Adalight sends static color
or black frame)
- Rename `_auto_shutdown_device_if_idle` to `_restore_device_idle_state`
- Add inline color picker on device cards for devices with static_color
- Add auto_shutdown toggle to device settings modal
- Update labels from "Auto Shutdown" to "Auto Restore" (en + ru)
- Remove backward-compat KC aliases from ProcessorManager
- Align card action buttons to bottom with flex column layout
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
302 lines
9.7 KiB
Python
302 lines
9.7 KiB
Python
"""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.core.capture.calibration import (
|
|
CalibrationConfig,
|
|
calibration_from_dict,
|
|
calibration_to_dict,
|
|
create_default_calibration,
|
|
)
|
|
from wled_controller.utils import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class Device:
|
|
"""Represents a WLED device configuration.
|
|
|
|
A device is a holder of connection state and calibration options.
|
|
Processing settings and picture source assignments live on PictureTargets.
|
|
"""
|
|
|
|
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,
|
|
calibration: Optional[CalibrationConfig] = 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.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."""
|
|
d = {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"url": self.url,
|
|
"led_count": self.led_count,
|
|
"enabled": self.enabled,
|
|
"device_type": self.device_type,
|
|
"calibration": calibration_to_dict(self.calibration),
|
|
"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: ignores legacy 'settings' and 'picture_source_id'
|
|
fields that have been migrated to PictureTarget.
|
|
"""
|
|
calibration_data = data.get("calibration")
|
|
calibration = (
|
|
calibration_from_dict(calibration_data)
|
|
if calibration_data
|
|
else create_default_calibration(data["led_count"])
|
|
)
|
|
|
|
static_color_raw = data.get("static_color")
|
|
static_color = tuple(static_color_raw) if static_color_raw else None
|
|
|
|
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),
|
|
static_color=static_color,
|
|
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):
|
|
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 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,
|
|
calibration: Optional[CalibrationConfig] = 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,
|
|
calibration=calibration,
|
|
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,
|
|
calibration: Optional[CalibrationConfig] = 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
|
|
device.calibration = create_default_calibration(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 calibration is not None:
|
|
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()
|
|
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")
|