alexei.dolgolyov/wled-screen-controller-mixed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduce Picture Targets to separate processing from devices

Browse Source 
Add PictureTarget entity that bridges PictureSource to output device,
separating processing settings from device connection/calibration state.
This enables future target types (Art-Net, E1.31) and cleanly decouples
"what to stream" from "where to stream."

- Add PictureTarget/WledPictureTarget dataclasses and storage
- Split ProcessorManager into DeviceState (health) + TargetState (processing)
- Add /api/v1/picture-targets endpoints (CRUD, start/stop, settings, metrics)
- Simplify device API (remove processing/settings/metrics endpoints)
- Auto-migrate existing device settings to picture targets on first startup
- Add Targets tab to WebUI with target cards and editor modal
- Add en/ru locale keys for targets UI

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Admin
2026-02-12 15:27:41 +03:00
parent c3828e10fa
commit 55814a3c30
20 changed files with 1976 additions and 1489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
server/src/wled_controller/storage/device_store.py
View File

@@ -12,14 +12,17 @@ from wled_controller.core.calibration import (
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."""
"""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,
@@ -28,62 +31,28 @@ class Device:
url: str,
led_count: int,
enabled: bool = True,
settings: Optional[ProcessingSettings] = None,
calibration: Optional[CalibrationConfig] = None,
picture_source_id: str = "",
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
picture_source_id: ID of assigned picture source
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.picture_source_id = picture_source_id
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
"""
"""Convert device to dictionary."""
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),
"picture_source_id": self.picture_source_id,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
}
@@ -92,25 +61,9 @@ class Device:
def from_dict(cls, data: dict) -> "Device":
"""Create device from dictionary.
Args:
data: Dictionary with device data
Returns:
Device instance
Backward-compatible: ignores legacy 'settings' and 'picture_source_id'
fields that have been migrated to PictureTarget.
"""
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", DEFAULT_STATE_CHECK_INTERVAL),
)
calibration_data = data.get("calibration")
calibration = (
calibration_from_dict(calibration_data)
@@ -118,17 +71,13 @@ class Device:
else create_default_calibration(data["led_count"])
)
picture_source_id = data.get("picture_source_id", "")
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,
picture_source_id=picture_source_id,
created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
)
@@ -138,11 +87,6 @@ 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] = {}
@@ -179,6 +123,16 @@ class DeviceStore:
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:
@@ -189,12 +143,10 @@ class DeviceStore:
}
}
# 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")
@@ -208,41 +160,19 @@ class DeviceStore:
name: str,
url: str,
led_count: int,
settings: Optional[ProcessingSettings] = None,
calibration: Optional[CalibrationConfig] = None,
picture_source_id: str = "",
) -> Device:
"""Create a new device.
Args:
name: Device name
url: WLED device URL
led_count: Number of LEDs
settings: Processing settings
calibration: Calibration configuration
picture_source_id: ID of assigned picture source
Returns:
Created device
Raises:
ValueError: If validation fails
"""
# Generate unique ID
"""Create a new device."""
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,
picture_source_id=picture_source_id,
)
# Store
self._devices[device_id] = device
self.save()
@@ -250,22 +180,11 @@ class DeviceStore:
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
"""
"""Get device by ID."""
return self._devices.get(device_id)
def get_all_devices(self) -> List[Device]:
"""Get all devices.
Returns:
List of all devices
"""
"""Get all devices."""
return list(self._devices.values())
def update_device(
@@ -275,73 +194,38 @@ class DeviceStore:
url: Optional[str] = None,
led_count: Optional[int] = None,
enabled: Optional[bool] = None,
settings: Optional[ProcessingSettings] = None,
calibration: Optional[CalibrationConfig] = None,
picture_source_id: Optional[str] = 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)
picture_source_id: New picture source ID (optional)
Returns:
Updated device
Raises:
ValueError: If device not found or validation fails
"""
"""Update device."""
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
if picture_source_id is not None:
device.picture_source_id = picture_source_id
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
"""
"""Delete device."""
if device_id not in self._devices:
raise ValueError(f"Device {device_id} not found")
@@ -351,22 +235,11 @@ class DeviceStore:
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
"""
"""Check if device exists."""
return device_id in self._devices
def count(self) -> int:
"""Get number of devices.
Returns:
Device count
"""
"""Get number of devices."""
return len(self._devices)
def clear(self):

13
server/src/wled_controller/storage/picture_source_store.py
View File

@@ -301,20 +301,17 @@ class PictureSourceStore:
logger.info(f"Deleted picture source: {stream_id}")
def is_referenced_by_device(self, stream_id: str, device_store) -> bool:
"""Check if this stream is referenced by any device.
def is_referenced_by_target(self, stream_id: str, target_store) -> bool:
"""Check if this stream is referenced by any picture target.
Args:
stream_id: Stream ID to check
device_store: DeviceStore instance
target_store: PictureTargetStore instance
Returns:
True if any device references this stream
True if any target references this stream
"""
for device in device_store.get_all_devices():
if getattr(device, "picture_source_id", None) == stream_id:
return True
return False
return target_store.is_referenced_by_source(stream_id)
def resolve_stream_chain(self, stream_id: str) -> dict:
"""Resolve a stream chain to get the terminal stream and collected postprocessing templates.

96
server/src/wled_controller/storage/picture_target.py Normal file
View File

@@ -0,0 +1,96 @@
"""Picture target data models."""
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from wled_controller.core.processor_manager import ProcessingSettings
@dataclass
class PictureTarget:
"""Base class for picture targets."""
id: str
name: str
target_type: str # "wled" (future: "artnet", "e131", ...)
created_at: datetime
updated_at: datetime
description: Optional[str] = None
def to_dict(self) -> dict:
"""Convert to dictionary."""
return {
"id": self.id,
"name": self.name,
"target_type": self.target_type,
"description": self.description,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
}
@classmethod
def from_dict(cls, data: dict) -> "PictureTarget":
"""Create from dictionary, dispatching to the correct subclass."""
target_type = data.get("target_type", "wled")
if target_type == "wled":
return WledPictureTarget.from_dict(data)
raise ValueError(f"Unknown target type: {target_type}")
@dataclass
class WledPictureTarget(PictureTarget):
"""WLED picture target — streams a picture source to a WLED device."""
device_id: str = ""
picture_source_id: str = ""
settings: ProcessingSettings = field(default_factory=ProcessingSettings)
def to_dict(self) -> dict:
"""Convert to dictionary."""
d = super().to_dict()
d["device_id"] = self.device_id
d["picture_source_id"] = self.picture_source_id
d["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,
}
return d
@classmethod
def from_dict(cls, data: dict) -> "WledPictureTarget":
"""Create from dictionary."""
from wled_controller.core.processor_manager import DEFAULT_STATE_CHECK_INTERVAL
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", DEFAULT_STATE_CHECK_INTERVAL),
)
return cls(
id=data["id"],
name=data["name"],
target_type=data.get("target_type", "wled"),
device_id=data.get("device_id", ""),
picture_source_id=data.get("picture_source_id", ""),
settings=settings,
description=data.get("description"),
created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
)

220
server/src/wled_controller/storage/picture_target_store.py Normal file
View File

@@ -0,0 +1,220 @@
"""Picture target 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.processor_manager import ProcessingSettings
from wled_controller.storage.picture_target import PictureTarget, WledPictureTarget
from wled_controller.utils import get_logger
logger = get_logger(__name__)
class PictureTargetStore:
"""Persistent storage for picture targets."""
def __init__(self, file_path: str):
"""Initialize picture target store.
Args:
file_path: Path to targets JSON file
"""
self.file_path = Path(file_path)
self._targets: Dict[str, PictureTarget] = {}
self._load()
def _load(self) -> None:
"""Load targets from file."""
if not self.file_path.exists():
return
try:
with open(self.file_path, "r", encoding="utf-8") as f:
data = json.load(f)
targets_data = data.get("picture_targets", {})
loaded = 0
for target_id, target_dict in targets_data.items():
try:
target = PictureTarget.from_dict(target_dict)
self._targets[target_id] = target
loaded += 1
except Exception as e:
logger.error(f"Failed to load picture target {target_id}: {e}", exc_info=True)
if loaded > 0:
logger.info(f"Loaded {loaded} picture targets from storage")
except Exception as e:
logger.error(f"Failed to load picture targets from {self.file_path}: {e}")
raise
logger.info(f"Picture target store initialized with {len(self._targets)} targets")
def _save(self) -> None:
"""Save all targets to file."""
try:
self.file_path.parent.mkdir(parents=True, exist_ok=True)
targets_dict = {
target_id: target.to_dict()
for target_id, target in self._targets.items()
}
data = {
"version": "1.0.0",
"picture_targets": targets_dict,
}
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 picture targets to {self.file_path}: {e}")
raise
def get_all_targets(self) -> List[PictureTarget]:
"""Get all picture targets."""
return list(self._targets.values())
def get_target(self, target_id: str) -> PictureTarget:
"""Get target by ID.
Raises:
ValueError: If target not found
"""
if target_id not in self._targets:
raise ValueError(f"Picture target not found: {target_id}")
return self._targets[target_id]
def create_target(
self,
name: str,
target_type: str,
device_id: str = "",
picture_source_id: str = "",
settings: Optional[ProcessingSettings] = None,
description: Optional[str] = None,
) -> PictureTarget:
"""Create a new picture target.
Args:
name: Target name
target_type: Target type ("wled")
device_id: WLED device ID (for wled targets)
picture_source_id: Picture source ID
settings: Processing settings
description: Optional description
Raises:
ValueError: If validation fails
"""
if target_type not in ("wled",):
raise ValueError(f"Invalid target type: {target_type}")
# Check for duplicate name
for target in self._targets.values():
if target.name == name:
raise ValueError(f"Picture target with name '{name}' already exists")
target_id = f"pt_{uuid.uuid4().hex[:8]}"
now = datetime.utcnow()
if target_type == "wled":
target: PictureTarget = WledPictureTarget(
id=target_id,
name=name,
target_type="wled",
device_id=device_id,
picture_source_id=picture_source_id,
settings=settings or ProcessingSettings(),
description=description,
created_at=now,
updated_at=now,
)
else:
raise ValueError(f"Unknown target type: {target_type}")
self._targets[target_id] = target
self._save()
logger.info(f"Created picture target: {name} ({target_id}, type={target_type})")
return target
def update_target(
self,
target_id: str,
name: Optional[str] = None,
device_id: Optional[str] = None,
picture_source_id: Optional[str] = None,
settings: Optional[ProcessingSettings] = None,
description: Optional[str] = None,
) -> PictureTarget:
"""Update a picture target.
Raises:
ValueError: If target not found or validation fails
"""
if target_id not in self._targets:
raise ValueError(f"Picture target not found: {target_id}")
target = self._targets[target_id]
if name is not None:
# Check for duplicate name (exclude self)
for other in self._targets.values():
if other.id != target_id and other.name == name:
raise ValueError(f"Picture target with name '{name}' already exists")
target.name = name
if description is not None:
target.description = description
if isinstance(target, WledPictureTarget):
if device_id is not None:
target.device_id = device_id
if picture_source_id is not None:
target.picture_source_id = picture_source_id
if settings is not None:
target.settings = settings
target.updated_at = datetime.utcnow()
self._save()
logger.info(f"Updated picture target: {target_id}")
return target
def delete_target(self, target_id: str) -> None:
"""Delete a picture target.
Raises:
ValueError: If target not found
"""
if target_id not in self._targets:
raise ValueError(f"Picture target not found: {target_id}")
del self._targets[target_id]
self._save()
logger.info(f"Deleted picture target: {target_id}")
def get_targets_for_device(self, device_id: str) -> List[PictureTarget]:
"""Get all targets that reference a specific device."""
return [
t for t in self._targets.values()
if isinstance(t, WledPictureTarget) and t.device_id == device_id
]
def is_referenced_by_source(self, source_id: str) -> bool:
"""Check if any target references a picture source."""
for target in self._targets.values():
if isinstance(target, WledPictureTarget) and target.picture_source_id == source_id:
return True
return False
def count(self) -> int:
"""Get number of targets."""
return len(self._targets)