Introduce ColorStripSource as first-class entity
Extracts color processing and calibration out of WledPictureTarget into a new PictureColorStripSource entity, enabling multiple LED targets to share one capture/processing pipeline. New entities & processing: - storage/color_strip_source.py: ColorStripSource + PictureColorStripSource models - storage/color_strip_store.py: JSON-backed CRUD store (prefix css_) - core/processing/color_strip_stream.py: ColorStripStream ABC + PictureColorStripStream (runs border-extract → map → smooth → brightness/sat/gamma in background thread) - core/processing/color_strip_stream_manager.py: ref-counted shared stream manager Modified storage/processing: - WledPictureTarget simplified to device_id + color_strip_source_id + standby_interval + state_check_interval - Device model: calibration field removed - WledTargetProcessor: acquires ColorStripStream from manager instead of running its own pipeline - ProcessorManager: wires ColorStripStreamManager into TargetContext API layer: - New routes: GET/POST/PUT/DELETE /api/v1/color-strip-sources, PUT calibration/test - Removed calibration endpoints from /devices - Updated /picture-targets CRUD for new target structure Frontend: - New color-strips.js module with CSS editor modal and card rendering - Calibration modal extended with CSS mode (css-id hidden field + device picker) - targets.js: Color Strip Sources section added to LED tab; target editor/card updated - app.js: imports and window globals for CSS + showCSSCalibration - en.json / ru.json: color_strip.* and targets.section.color_strips keys added Data migration runs at startup: existing WledPictureTargets are converted to reference a new PictureColorStripSource created from their old settings. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
130
server/src/wled_controller/storage/color_strip_source.py
Normal file
130
server/src/wled_controller/storage/color_strip_source.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Color strip source data model with inheritance-based source types.
|
||||
|
||||
A ColorStripSource produces a stream of LED color arrays (np.ndarray shape (N, 3))
|
||||
from some input, encapsulating everything needed to drive a physical LED strip:
|
||||
calibration, color correction, smoothing, and FPS.
|
||||
|
||||
Current types:
|
||||
PictureColorStripSource — derives LED colors from a PictureSource (screen capture)
|
||||
|
||||
Future types (not yet implemented):
|
||||
StaticColorStripSource — constant solid colors
|
||||
GradientColorStripSource — animated gradient
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from wled_controller.core.capture.calibration import (
|
||||
CalibrationConfig,
|
||||
calibration_from_dict,
|
||||
calibration_to_dict,
|
||||
create_default_calibration,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ColorStripSource:
|
||||
"""Base class for color strip source configurations."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
source_type: str # "picture" | future types
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
description: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convert source to dictionary. Subclasses extend this."""
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"source_type": self.source_type,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"updated_at": self.updated_at.isoformat(),
|
||||
"description": self.description,
|
||||
# Subclass fields default to None for forward compat
|
||||
"picture_source_id": None,
|
||||
"fps": None,
|
||||
"brightness": None,
|
||||
"saturation": None,
|
||||
"gamma": None,
|
||||
"smoothing": None,
|
||||
"interpolation_mode": None,
|
||||
"calibration": None,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def from_dict(data: dict) -> "ColorStripSource":
|
||||
"""Factory: dispatch to the correct subclass based on source_type."""
|
||||
source_type: str = data.get("source_type", "picture") or "picture"
|
||||
sid: str = data["id"]
|
||||
name: str = data["name"]
|
||||
description: str | None = data.get("description")
|
||||
|
||||
raw_created = data.get("created_at")
|
||||
created_at: datetime = (
|
||||
datetime.fromisoformat(raw_created)
|
||||
if isinstance(raw_created, str)
|
||||
else raw_created if isinstance(raw_created, datetime)
|
||||
else datetime.utcnow()
|
||||
)
|
||||
raw_updated = data.get("updated_at")
|
||||
updated_at: datetime = (
|
||||
datetime.fromisoformat(raw_updated)
|
||||
if isinstance(raw_updated, str)
|
||||
else raw_updated if isinstance(raw_updated, datetime)
|
||||
else datetime.utcnow()
|
||||
)
|
||||
|
||||
calibration_data = data.get("calibration")
|
||||
calibration = (
|
||||
calibration_from_dict(calibration_data)
|
||||
if calibration_data
|
||||
else create_default_calibration(0)
|
||||
)
|
||||
|
||||
# Only "picture" type for now; extend with elif branches for future types
|
||||
return PictureColorStripSource(
|
||||
id=sid, name=name, source_type=source_type,
|
||||
created_at=created_at, updated_at=updated_at, description=description,
|
||||
picture_source_id=data.get("picture_source_id") or "",
|
||||
fps=data.get("fps") or 30,
|
||||
brightness=data["brightness"] if data.get("brightness") is not None else 1.0,
|
||||
saturation=data["saturation"] if data.get("saturation") is not None else 1.0,
|
||||
gamma=data["gamma"] if data.get("gamma") is not None else 1.0,
|
||||
smoothing=data["smoothing"] if data.get("smoothing") is not None else 0.3,
|
||||
interpolation_mode=data.get("interpolation_mode") or "average",
|
||||
calibration=calibration,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PictureColorStripSource(ColorStripSource):
|
||||
"""Color strip source driven by a PictureSource (screen capture / static image).
|
||||
|
||||
Contains everything required to produce LED color arrays from a picture stream:
|
||||
calibration (LED positions), color correction, smoothing, FPS target.
|
||||
"""
|
||||
|
||||
picture_source_id: str = ""
|
||||
fps: int = 30
|
||||
brightness: float = 1.0 # color correction multiplier (0.0–2.0; 1.0 = unchanged)
|
||||
saturation: float = 1.0 # 1.0 = unchanged, 0.0 = grayscale, 2.0 = double saturation
|
||||
gamma: float = 1.0 # 1.0 = no correction; <1 = brighter, >1 = darker mids
|
||||
smoothing: float = 0.3 # temporal smoothing (0.0 = none, 1.0 = full)
|
||||
interpolation_mode: str = "average" # "average" | "median" | "dominant"
|
||||
calibration: CalibrationConfig = field(default_factory=lambda: create_default_calibration(0))
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["picture_source_id"] = self.picture_source_id
|
||||
d["fps"] = self.fps
|
||||
d["brightness"] = self.brightness
|
||||
d["saturation"] = self.saturation
|
||||
d["gamma"] = self.gamma
|
||||
d["smoothing"] = self.smoothing
|
||||
d["interpolation_mode"] = self.interpolation_mode
|
||||
d["calibration"] = calibration_to_dict(self.calibration)
|
||||
return d
|
||||
224
server/src/wled_controller/storage/color_strip_store.py
Normal file
224
server/src/wled_controller/storage/color_strip_store.py
Normal file
@@ -0,0 +1,224 @@
|
||||
"""Color strip source 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.capture.calibration import calibration_to_dict
|
||||
from wled_controller.storage.color_strip_source import (
|
||||
ColorStripSource,
|
||||
PictureColorStripSource,
|
||||
)
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class ColorStripStore:
|
||||
"""Persistent storage for color strip sources."""
|
||||
|
||||
def __init__(self, file_path: str):
|
||||
self.file_path = Path(file_path)
|
||||
self._sources: Dict[str, ColorStripSource] = {}
|
||||
self._load()
|
||||
|
||||
def _load(self) -> None:
|
||||
if not self.file_path.exists():
|
||||
logger.info("Color strip store file not found — starting empty")
|
||||
return
|
||||
|
||||
try:
|
||||
with open(self.file_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
sources_data = data.get("color_strip_sources", {})
|
||||
loaded = 0
|
||||
for source_id, source_dict in sources_data.items():
|
||||
try:
|
||||
source = ColorStripSource.from_dict(source_dict)
|
||||
self._sources[source_id] = source
|
||||
loaded += 1
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load color strip source {source_id}: {e}", exc_info=True)
|
||||
|
||||
if loaded > 0:
|
||||
logger.info(f"Loaded {loaded} color strip sources from storage")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load color strip sources from {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
logger.info(f"Color strip store initialized with {len(self._sources)} sources")
|
||||
|
||||
def _save(self) -> None:
|
||||
try:
|
||||
self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
sources_dict = {
|
||||
sid: source.to_dict()
|
||||
for sid, source in self._sources.items()
|
||||
}
|
||||
|
||||
data = {
|
||||
"version": "1.0.0",
|
||||
"color_strip_sources": sources_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 color strip sources to {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
def get_all_sources(self) -> List[ColorStripSource]:
|
||||
return list(self._sources.values())
|
||||
|
||||
def get_source(self, source_id: str) -> ColorStripSource:
|
||||
"""Get a color strip source by ID.
|
||||
|
||||
Raises:
|
||||
ValueError: If source not found
|
||||
"""
|
||||
if source_id not in self._sources:
|
||||
raise ValueError(f"Color strip source not found: {source_id}")
|
||||
return self._sources[source_id]
|
||||
|
||||
def create_source(
|
||||
self,
|
||||
name: str,
|
||||
source_type: str = "picture",
|
||||
picture_source_id: str = "",
|
||||
fps: int = 30,
|
||||
brightness: float = 1.0,
|
||||
saturation: float = 1.0,
|
||||
gamma: float = 1.0,
|
||||
smoothing: float = 0.3,
|
||||
interpolation_mode: str = "average",
|
||||
calibration=None,
|
||||
description: Optional[str] = None,
|
||||
) -> ColorStripSource:
|
||||
"""Create a new color strip source.
|
||||
|
||||
Raises:
|
||||
ValueError: If validation fails
|
||||
"""
|
||||
from wled_controller.core.capture.calibration import create_default_calibration
|
||||
|
||||
if not name or not name.strip():
|
||||
raise ValueError("Name is required")
|
||||
|
||||
for source in self._sources.values():
|
||||
if source.name == name:
|
||||
raise ValueError(f"Color strip source with name '{name}' already exists")
|
||||
|
||||
if calibration is None:
|
||||
calibration = create_default_calibration(0)
|
||||
|
||||
source_id = f"css_{uuid.uuid4().hex[:8]}"
|
||||
now = datetime.utcnow()
|
||||
|
||||
source = PictureColorStripSource(
|
||||
id=source_id,
|
||||
name=name,
|
||||
source_type=source_type,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
description=description,
|
||||
picture_source_id=picture_source_id,
|
||||
fps=fps,
|
||||
brightness=brightness,
|
||||
saturation=saturation,
|
||||
gamma=gamma,
|
||||
smoothing=smoothing,
|
||||
interpolation_mode=interpolation_mode,
|
||||
calibration=calibration,
|
||||
)
|
||||
|
||||
self._sources[source_id] = source
|
||||
self._save()
|
||||
|
||||
logger.info(f"Created color strip source: {name} ({source_id}, type={source_type})")
|
||||
return source
|
||||
|
||||
def update_source(
|
||||
self,
|
||||
source_id: str,
|
||||
name: Optional[str] = None,
|
||||
picture_source_id: Optional[str] = None,
|
||||
fps: Optional[int] = None,
|
||||
brightness: Optional[float] = None,
|
||||
saturation: Optional[float] = None,
|
||||
gamma: Optional[float] = None,
|
||||
smoothing: Optional[float] = None,
|
||||
interpolation_mode: Optional[str] = None,
|
||||
calibration=None,
|
||||
description: Optional[str] = None,
|
||||
) -> ColorStripSource:
|
||||
"""Update an existing color strip source.
|
||||
|
||||
Raises:
|
||||
ValueError: If source not found
|
||||
"""
|
||||
if source_id not in self._sources:
|
||||
raise ValueError(f"Color strip source not found: {source_id}")
|
||||
|
||||
source = self._sources[source_id]
|
||||
|
||||
if name is not None:
|
||||
for other in self._sources.values():
|
||||
if other.id != source_id and other.name == name:
|
||||
raise ValueError(f"Color strip source with name '{name}' already exists")
|
||||
source.name = name
|
||||
|
||||
if description is not None:
|
||||
source.description = description
|
||||
|
||||
if isinstance(source, PictureColorStripSource):
|
||||
if picture_source_id is not None:
|
||||
source.picture_source_id = picture_source_id
|
||||
if fps is not None:
|
||||
source.fps = fps
|
||||
if brightness is not None:
|
||||
source.brightness = brightness
|
||||
if saturation is not None:
|
||||
source.saturation = saturation
|
||||
if gamma is not None:
|
||||
source.gamma = gamma
|
||||
if smoothing is not None:
|
||||
source.smoothing = smoothing
|
||||
if interpolation_mode is not None:
|
||||
source.interpolation_mode = interpolation_mode
|
||||
if calibration is not None:
|
||||
source.calibration = calibration
|
||||
|
||||
source.updated_at = datetime.utcnow()
|
||||
self._save()
|
||||
|
||||
logger.info(f"Updated color strip source: {source_id}")
|
||||
return source
|
||||
|
||||
def delete_source(self, source_id: str) -> None:
|
||||
"""Delete a color strip source.
|
||||
|
||||
Raises:
|
||||
ValueError: If source not found
|
||||
"""
|
||||
if source_id not in self._sources:
|
||||
raise ValueError(f"Color strip source not found: {source_id}")
|
||||
|
||||
del self._sources[source_id]
|
||||
self._save()
|
||||
|
||||
logger.info(f"Deleted color strip source: {source_id}")
|
||||
|
||||
def is_referenced_by_target(self, source_id: str, target_store) -> bool:
|
||||
"""Check if this source is referenced by any picture target."""
|
||||
from wled_controller.storage.wled_picture_target import WledPictureTarget
|
||||
|
||||
for target in target_store.get_all_targets():
|
||||
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == source_id:
|
||||
return True
|
||||
return False
|
||||
@@ -6,12 +6,6 @@ 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__)
|
||||
@@ -20,8 +14,9 @@ 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.
|
||||
A device holds connection state and output settings.
|
||||
Calibration, processing settings, and picture source assignments
|
||||
now live on ColorStripSource and WledPictureTarget respectively.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -36,7 +31,6 @@ class Device:
|
||||
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,
|
||||
):
|
||||
@@ -50,9 +44,10 @@ class Device:
|
||||
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()
|
||||
# Preserved from old JSON for migration — not written back
|
||||
self._legacy_calibration = None
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convert device to dictionary."""
|
||||
@@ -63,7 +58,6 @@ class Device:
|
||||
"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(),
|
||||
}
|
||||
@@ -81,20 +75,13 @@ class Device:
|
||||
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.
|
||||
Backward-compatible: reads legacy 'calibration' field and stores it
|
||||
in _legacy_calibration for migration use only.
|
||||
"""
|
||||
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 = cls(
|
||||
device_id=data["id"],
|
||||
name=data["name"],
|
||||
url=data["url"],
|
||||
@@ -105,11 +92,21 @@ class Device:
|
||||
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())),
|
||||
)
|
||||
|
||||
# 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."""
|
||||
@@ -129,7 +126,7 @@ class DeviceStore:
|
||||
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")
|
||||
logger.info("Storage file does not exist, starting with empty store")
|
||||
return
|
||||
|
||||
try:
|
||||
@@ -190,7 +187,6 @@ class DeviceStore:
|
||||
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."""
|
||||
@@ -203,7 +199,6 @@ class DeviceStore:
|
||||
led_count=led_count,
|
||||
device_type=device_type,
|
||||
baud_rate=baud_rate,
|
||||
calibration=calibration,
|
||||
auto_shutdown=auto_shutdown,
|
||||
)
|
||||
|
||||
@@ -229,7 +224,6 @@ class DeviceStore:
|
||||
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."""
|
||||
@@ -249,13 +243,6 @@ class DeviceStore:
|
||||
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()
|
||||
|
||||
@@ -6,7 +6,6 @@ from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from wled_controller.core.processing.processing_settings import ProcessingSettings
|
||||
from wled_controller.storage.picture_target import PictureTarget
|
||||
from wled_controller.storage.wled_picture_target import WledPictureTarget
|
||||
from wled_controller.storage.key_colors_picture_target import (
|
||||
@@ -17,6 +16,8 @@ from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DEFAULT_STATE_CHECK_INTERVAL = 30 # seconds
|
||||
|
||||
|
||||
class PictureTargetStore:
|
||||
"""Persistent storage for picture targets."""
|
||||
@@ -100,19 +101,24 @@ class PictureTargetStore:
|
||||
name: str,
|
||||
target_type: str,
|
||||
device_id: str = "",
|
||||
picture_source_id: str = "",
|
||||
settings: Optional[ProcessingSettings] = None,
|
||||
color_strip_source_id: str = "",
|
||||
standby_interval: float = 1.0,
|
||||
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL,
|
||||
key_colors_settings: Optional[KeyColorsSettings] = None,
|
||||
description: Optional[str] = None,
|
||||
# Legacy params — accepted but ignored for backward compat
|
||||
picture_source_id: str = "",
|
||||
settings=None,
|
||||
) -> PictureTarget:
|
||||
"""Create a new picture target.
|
||||
|
||||
Args:
|
||||
name: Target name
|
||||
target_type: Target type ("wled", "key_colors")
|
||||
device_id: WLED device ID (for wled targets)
|
||||
picture_source_id: Picture source ID
|
||||
settings: Processing settings (for wled targets)
|
||||
target_type: Target type ("led", "wled", "key_colors")
|
||||
device_id: WLED device ID (for led targets)
|
||||
color_strip_source_id: Color strip source ID (for led targets)
|
||||
standby_interval: Keepalive interval in seconds (for led targets)
|
||||
state_check_interval: State check interval in seconds (for led targets)
|
||||
key_colors_settings: Key colors settings (for key_colors targets)
|
||||
description: Optional description
|
||||
|
||||
@@ -139,8 +145,9 @@ class PictureTargetStore:
|
||||
name=name,
|
||||
target_type="led",
|
||||
device_id=device_id,
|
||||
picture_source_id=picture_source_id,
|
||||
settings=settings or ProcessingSettings(),
|
||||
color_strip_source_id=color_strip_source_id,
|
||||
standby_interval=standby_interval,
|
||||
state_check_interval=state_check_interval,
|
||||
description=description,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
@@ -170,10 +177,14 @@ class PictureTargetStore:
|
||||
target_id: str,
|
||||
name: Optional[str] = None,
|
||||
device_id: Optional[str] = None,
|
||||
picture_source_id: Optional[str] = None,
|
||||
settings: Optional[ProcessingSettings] = None,
|
||||
color_strip_source_id: Optional[str] = None,
|
||||
standby_interval: Optional[float] = None,
|
||||
state_check_interval: Optional[int] = None,
|
||||
key_colors_settings: Optional[KeyColorsSettings] = None,
|
||||
description: Optional[str] = None,
|
||||
# Legacy params — accepted but ignored
|
||||
picture_source_id: Optional[str] = None,
|
||||
settings=None,
|
||||
) -> PictureTarget:
|
||||
"""Update a picture target.
|
||||
|
||||
@@ -194,8 +205,9 @@ class PictureTargetStore:
|
||||
target.update_fields(
|
||||
name=name,
|
||||
device_id=device_id,
|
||||
picture_source_id=picture_source_id,
|
||||
settings=settings,
|
||||
color_strip_source_id=color_strip_source_id,
|
||||
standby_interval=standby_interval,
|
||||
state_check_interval=state_check_interval,
|
||||
key_colors_settings=key_colors_settings,
|
||||
description=description,
|
||||
)
|
||||
@@ -228,9 +240,16 @@ class PictureTargetStore:
|
||||
]
|
||||
|
||||
def is_referenced_by_source(self, source_id: str) -> bool:
|
||||
"""Check if any target references a picture source."""
|
||||
"""Check if any KC target directly references a picture source."""
|
||||
for target in self._targets.values():
|
||||
if target.has_picture_source and target.picture_source_id == source_id:
|
||||
if isinstance(target, KeyColorsPictureTarget) and target.picture_source_id == source_id:
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_referenced_by_color_strip_source(self, css_id: str) -> bool:
|
||||
"""Check if any WLED target references a color strip source."""
|
||||
for target in self._targets.values():
|
||||
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == css_id:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
@@ -1,20 +1,31 @@
|
||||
"""LED picture target — streams a picture source to an LED device."""
|
||||
"""LED picture target — sends a color strip source to an LED device."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from wled_controller.core.processing.processing_settings import ProcessingSettings
|
||||
from wled_controller.storage.picture_target import PictureTarget
|
||||
|
||||
DEFAULT_STATE_CHECK_INTERVAL = 30 # seconds
|
||||
|
||||
|
||||
@dataclass
|
||||
class WledPictureTarget(PictureTarget):
|
||||
"""LED picture target — streams a picture source to an LED device."""
|
||||
"""LED picture target — pairs an LED device with a ColorStripSource.
|
||||
|
||||
The ColorStripSource encapsulates everything needed to produce LED colors
|
||||
(calibration, color correction, smoothing, fps). The LED target itself only
|
||||
holds device-specific timing/keepalive settings.
|
||||
"""
|
||||
|
||||
device_id: str = ""
|
||||
picture_source_id: str = ""
|
||||
settings: ProcessingSettings = field(default_factory=ProcessingSettings)
|
||||
color_strip_source_id: str = ""
|
||||
standby_interval: float = 1.0 # seconds between keepalive sends when screen is static
|
||||
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL
|
||||
|
||||
# Legacy fields — populated from old JSON data during migration; not written back
|
||||
_legacy_picture_source_id: str = field(default="", repr=False, compare=False)
|
||||
_legacy_settings: Optional[dict] = field(default=None, repr=False, compare=False)
|
||||
|
||||
def register_with_manager(self, manager) -> None:
|
||||
"""Register this WLED target with the processor manager."""
|
||||
@@ -22,74 +33,72 @@ class WledPictureTarget(PictureTarget):
|
||||
manager.add_target(
|
||||
target_id=self.id,
|
||||
device_id=self.device_id,
|
||||
settings=self.settings,
|
||||
picture_source_id=self.picture_source_id,
|
||||
color_strip_source_id=self.color_strip_source_id,
|
||||
standby_interval=self.standby_interval,
|
||||
state_check_interval=self.state_check_interval,
|
||||
)
|
||||
|
||||
def sync_with_manager(self, manager, *, settings_changed: bool, source_changed: bool, device_changed: bool) -> None:
|
||||
"""Push changed fields to the processor manager."""
|
||||
if settings_changed:
|
||||
manager.update_target_settings(self.id, self.settings)
|
||||
manager.update_target_settings(self.id, {
|
||||
"standby_interval": self.standby_interval,
|
||||
"state_check_interval": self.state_check_interval,
|
||||
})
|
||||
if source_changed:
|
||||
manager.update_target_source(self.id, self.picture_source_id)
|
||||
manager.update_target_color_strip_source(self.id, self.color_strip_source_id)
|
||||
if device_changed:
|
||||
manager.update_target_device(self.id, self.device_id)
|
||||
|
||||
def update_fields(self, *, name=None, device_id=None, picture_source_id=None,
|
||||
settings=None, key_colors_settings=None, description=None) -> None:
|
||||
def update_fields(self, *, name=None, device_id=None, color_strip_source_id=None,
|
||||
standby_interval=None, state_check_interval=None,
|
||||
# Legacy params accepted but ignored to keep base class compat:
|
||||
picture_source_id=None, settings=None,
|
||||
key_colors_settings=None, description=None) -> None:
|
||||
"""Apply mutable field updates for WLED targets."""
|
||||
super().update_fields(name=name, description=description)
|
||||
if device_id is not None:
|
||||
self.device_id = device_id
|
||||
if picture_source_id is not None:
|
||||
self.picture_source_id = picture_source_id
|
||||
if settings is not None:
|
||||
self.settings = settings
|
||||
if color_strip_source_id is not None:
|
||||
self.color_strip_source_id = color_strip_source_id
|
||||
if standby_interval is not None:
|
||||
self.standby_interval = standby_interval
|
||||
if state_check_interval is not None:
|
||||
self.state_check_interval = state_check_interval
|
||||
|
||||
@property
|
||||
def has_picture_source(self) -> bool:
|
||||
return True
|
||||
return bool(self.color_strip_source_id)
|
||||
|
||||
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,
|
||||
"brightness": self.settings.brightness,
|
||||
"smoothing": self.settings.smoothing,
|
||||
"interpolation_mode": self.settings.interpolation_mode,
|
||||
"standby_interval": self.settings.standby_interval,
|
||||
"state_check_interval": self.settings.state_check_interval,
|
||||
}
|
||||
d["color_strip_source_id"] = self.color_strip_source_id
|
||||
d["standby_interval"] = self.standby_interval
|
||||
d["state_check_interval"] = self.state_check_interval
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "WledPictureTarget":
|
||||
"""Create from dictionary."""
|
||||
from wled_controller.core.processing.processing_settings 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),
|
||||
brightness=settings_data.get("brightness", 1.0),
|
||||
smoothing=settings_data.get("smoothing", 0.3),
|
||||
interpolation_mode=settings_data.get("interpolation_mode", "average"),
|
||||
standby_interval=settings_data.get("standby_interval", 1.0),
|
||||
state_check_interval=settings_data.get("state_check_interval", DEFAULT_STATE_CHECK_INTERVAL),
|
||||
)
|
||||
|
||||
return cls(
|
||||
"""Create from dictionary. Reads legacy picture_source_id/settings for migration."""
|
||||
obj = cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
target_type="led",
|
||||
device_id=data.get("device_id", ""),
|
||||
picture_source_id=data.get("picture_source_id", ""),
|
||||
settings=settings,
|
||||
color_strip_source_id=data.get("color_strip_source_id", ""),
|
||||
standby_interval=data.get("standby_interval", 1.0),
|
||||
state_check_interval=data.get("state_check_interval", DEFAULT_STATE_CHECK_INTERVAL),
|
||||
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())),
|
||||
)
|
||||
|
||||
# Preserve legacy fields for migration — never written back by to_dict()
|
||||
obj._legacy_picture_source_id = data.get("picture_source_id", "")
|
||||
settings_data = data.get("settings", {})
|
||||
if settings_data:
|
||||
obj._legacy_settings = settings_data
|
||||
|
||||
return obj
|
||||
|
||||
Reference in New Issue
Block a user