Add multi-segment LED targets, replace single color strip source + skip fields
Each target now has a segments list where each segment maps a color strip source to a pixel range (start/end) on the device with optional reverse. This enables composing multiple visualizations on a single LED strip. Old targets auto-migrate from the single source format on load. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -7,7 +7,7 @@ from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from wled_controller.storage.picture_target import PictureTarget
|
||||
from wled_controller.storage.wled_picture_target import WledPictureTarget
|
||||
from wled_controller.storage.wled_picture_target import TargetSegment, WledPictureTarget
|
||||
from wled_controller.storage.key_colors_picture_target import (
|
||||
KeyColorsSettings,
|
||||
KeyColorsPictureTarget,
|
||||
@@ -101,38 +101,21 @@ class PictureTargetStore:
|
||||
name: str,
|
||||
target_type: str,
|
||||
device_id: str = "",
|
||||
color_strip_source_id: str = "",
|
||||
segments: Optional[List[dict]] = None,
|
||||
fps: int = 30,
|
||||
keepalive_interval: float = 1.0,
|
||||
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL,
|
||||
led_skip_start: int = 0,
|
||||
led_skip_end: int = 0,
|
||||
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 ("led", "wled", "key_colors")
|
||||
device_id: WLED device ID (for led targets)
|
||||
color_strip_source_id: Color strip source ID (for led targets)
|
||||
keepalive_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
|
||||
|
||||
Raises:
|
||||
ValueError: If validation fails
|
||||
"""
|
||||
if target_type not in ("led", "wled", "key_colors"):
|
||||
if target_type not in ("led", "key_colors"):
|
||||
raise ValueError(f"Invalid target type: {target_type}")
|
||||
# Normalize legacy "wled" to "led"
|
||||
if target_type == "wled":
|
||||
target_type = "led"
|
||||
|
||||
# Check for duplicate name
|
||||
for target in self._targets.values():
|
||||
@@ -143,17 +126,17 @@ class PictureTargetStore:
|
||||
now = datetime.utcnow()
|
||||
|
||||
if target_type == "led":
|
||||
seg_list = [TargetSegment.from_dict(s) for s in segments] if segments else []
|
||||
|
||||
target: PictureTarget = WledPictureTarget(
|
||||
id=target_id,
|
||||
name=name,
|
||||
target_type="led",
|
||||
device_id=device_id,
|
||||
color_strip_source_id=color_strip_source_id,
|
||||
segments=seg_list,
|
||||
fps=fps,
|
||||
keepalive_interval=keepalive_interval,
|
||||
state_check_interval=state_check_interval,
|
||||
led_skip_start=led_skip_start,
|
||||
led_skip_end=led_skip_end,
|
||||
description=description,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
@@ -183,17 +166,12 @@ class PictureTargetStore:
|
||||
target_id: str,
|
||||
name: Optional[str] = None,
|
||||
device_id: Optional[str] = None,
|
||||
color_strip_source_id: Optional[str] = None,
|
||||
segments: Optional[List[dict]] = None,
|
||||
fps: Optional[int] = None,
|
||||
keepalive_interval: Optional[float] = None,
|
||||
state_check_interval: Optional[int] = None,
|
||||
led_skip_start: Optional[int] = None,
|
||||
led_skip_end: 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.
|
||||
|
||||
@@ -214,12 +192,10 @@ class PictureTargetStore:
|
||||
target.update_fields(
|
||||
name=name,
|
||||
device_id=device_id,
|
||||
color_strip_source_id=color_strip_source_id,
|
||||
segments=segments,
|
||||
fps=fps,
|
||||
keepalive_interval=keepalive_interval,
|
||||
state_check_interval=state_check_interval,
|
||||
led_skip_start=led_skip_start,
|
||||
led_skip_end=led_skip_end,
|
||||
key_colors_settings=key_colors_settings,
|
||||
description=description,
|
||||
)
|
||||
@@ -262,7 +238,8 @@ class PictureTargetStore:
|
||||
"""Return names of LED targets that reference a color strip source."""
|
||||
return [
|
||||
target.name for target in self._targets.values()
|
||||
if isinstance(target, WledPictureTarget) and target.color_strip_source_id == css_id
|
||||
if isinstance(target, WledPictureTarget)
|
||||
and any(seg.color_strip_source_id == css_id for seg in target.segments)
|
||||
]
|
||||
|
||||
def count(self) -> int:
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""LED picture target — sends a color strip source to an LED device."""
|
||||
"""LED picture target — sends color strip sources to an LED device."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from wled_controller.storage.picture_target import PictureTarget
|
||||
|
||||
@@ -9,20 +10,50 @@ DEFAULT_STATE_CHECK_INTERVAL = 30 # seconds
|
||||
|
||||
|
||||
@dataclass
|
||||
class WledPictureTarget(PictureTarget):
|
||||
"""LED picture target — pairs an LED device with a ColorStripSource.
|
||||
class TargetSegment:
|
||||
"""Maps a color strip source to a pixel range on the LED device.
|
||||
|
||||
The ColorStripSource produces LED colors (calibration, color correction,
|
||||
smoothing). The target controls device-specific settings including send FPS.
|
||||
``start`` is inclusive, ``end`` is exclusive. When a target has a single
|
||||
segment with ``end == 0`` the range auto-fits to the full device LED count.
|
||||
"""
|
||||
|
||||
color_strip_source_id: str = ""
|
||||
start: int = 0
|
||||
end: int = 0
|
||||
reverse: bool = False
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"color_strip_source_id": self.color_strip_source_id,
|
||||
"start": self.start,
|
||||
"end": self.end,
|
||||
"reverse": self.reverse,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict) -> "TargetSegment":
|
||||
return TargetSegment(
|
||||
color_strip_source_id=d.get("color_strip_source_id", ""),
|
||||
start=d.get("start", 0),
|
||||
end=d.get("end", 0),
|
||||
reverse=d.get("reverse", False),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WledPictureTarget(PictureTarget):
|
||||
"""LED picture target — pairs an LED device with one or more ColorStripSources.
|
||||
|
||||
Each segment maps a ColorStripSource to a pixel range on the device.
|
||||
Gaps between segments stay black. A single segment with ``end == 0``
|
||||
auto-fits to the full device LED count.
|
||||
"""
|
||||
|
||||
device_id: str = ""
|
||||
color_strip_source_id: str = ""
|
||||
segments: List[TargetSegment] = field(default_factory=list)
|
||||
fps: int = 30 # target send FPS (1-90)
|
||||
keepalive_interval: float = 1.0 # seconds between keepalive sends when screen is static
|
||||
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL
|
||||
led_skip_start: int = 0 # first N LEDs forced to black
|
||||
led_skip_end: int = 0 # last M LEDs forced to black
|
||||
|
||||
def register_with_manager(self, manager) -> None:
|
||||
"""Register this WLED target with the processor manager."""
|
||||
@@ -30,80 +61,90 @@ class WledPictureTarget(PictureTarget):
|
||||
manager.add_target(
|
||||
target_id=self.id,
|
||||
device_id=self.device_id,
|
||||
color_strip_source_id=self.color_strip_source_id,
|
||||
segments=[s.to_dict() for s in self.segments],
|
||||
fps=self.fps,
|
||||
keepalive_interval=self.keepalive_interval,
|
||||
state_check_interval=self.state_check_interval,
|
||||
led_skip_start=self.led_skip_start,
|
||||
led_skip_end=self.led_skip_end,
|
||||
)
|
||||
|
||||
def sync_with_manager(self, manager, *, settings_changed: bool, source_changed: bool, device_changed: bool) -> None:
|
||||
def sync_with_manager(self, manager, *, settings_changed: bool,
|
||||
segments_changed: bool = False,
|
||||
device_changed: bool = False) -> None:
|
||||
"""Push changed fields to the processor manager."""
|
||||
if settings_changed:
|
||||
manager.update_target_settings(self.id, {
|
||||
"fps": self.fps,
|
||||
"keepalive_interval": self.keepalive_interval,
|
||||
"state_check_interval": self.state_check_interval,
|
||||
"led_skip_start": self.led_skip_start,
|
||||
"led_skip_end": self.led_skip_end,
|
||||
})
|
||||
if source_changed:
|
||||
manager.update_target_color_strip_source(self.id, self.color_strip_source_id)
|
||||
if segments_changed:
|
||||
manager.update_target_segments(self.id, [s.to_dict() for s in self.segments])
|
||||
if device_changed:
|
||||
manager.update_target_device(self.id, self.device_id)
|
||||
|
||||
def update_fields(self, *, name=None, device_id=None, color_strip_source_id=None,
|
||||
def update_fields(self, *, name=None, device_id=None, segments=None,
|
||||
fps=None, keepalive_interval=None, state_check_interval=None,
|
||||
led_skip_start=None, led_skip_end=None,
|
||||
description=None, **_kwargs) -> 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 color_strip_source_id is not None:
|
||||
self.color_strip_source_id = color_strip_source_id
|
||||
if segments is not None:
|
||||
self.segments = [
|
||||
TargetSegment.from_dict(s) if isinstance(s, dict) else s
|
||||
for s in segments
|
||||
]
|
||||
if fps is not None:
|
||||
self.fps = fps
|
||||
if keepalive_interval is not None:
|
||||
self.keepalive_interval = keepalive_interval
|
||||
if state_check_interval is not None:
|
||||
self.state_check_interval = state_check_interval
|
||||
if led_skip_start is not None:
|
||||
self.led_skip_start = led_skip_start
|
||||
if led_skip_end is not None:
|
||||
self.led_skip_end = led_skip_end
|
||||
|
||||
@property
|
||||
def has_picture_source(self) -> bool:
|
||||
return bool(self.color_strip_source_id)
|
||||
return any(s.color_strip_source_id for s in self.segments)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Convert to dictionary."""
|
||||
d = super().to_dict()
|
||||
d["device_id"] = self.device_id
|
||||
d["color_strip_source_id"] = self.color_strip_source_id
|
||||
d["segments"] = [s.to_dict() for s in self.segments]
|
||||
d["fps"] = self.fps
|
||||
d["keepalive_interval"] = self.keepalive_interval
|
||||
d["state_check_interval"] = self.state_check_interval
|
||||
d["led_skip_start"] = self.led_skip_start
|
||||
d["led_skip_end"] = self.led_skip_end
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "WledPictureTarget":
|
||||
"""Create from dictionary."""
|
||||
"""Create from dictionary with backward compatibility."""
|
||||
# Migrate old single-source format to segments
|
||||
if "segments" in data:
|
||||
segments = [TargetSegment.from_dict(s) for s in data["segments"]]
|
||||
elif "color_strip_source_id" in data:
|
||||
css_id = data.get("color_strip_source_id", "")
|
||||
skip_start = data.get("led_skip_start", 0)
|
||||
skip_end = data.get("led_skip_end", 0)
|
||||
if css_id:
|
||||
segments = [TargetSegment(
|
||||
color_strip_source_id=css_id,
|
||||
start=skip_start,
|
||||
end=0, # auto-fit; skip_end handled by processor
|
||||
)]
|
||||
else:
|
||||
segments = []
|
||||
else:
|
||||
segments = []
|
||||
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
target_type="led",
|
||||
device_id=data.get("device_id", ""),
|
||||
color_strip_source_id=data.get("color_strip_source_id", ""),
|
||||
segments=segments,
|
||||
fps=data.get("fps", 30),
|
||||
keepalive_interval=data.get("keepalive_interval", data.get("standby_interval", 1.0)),
|
||||
state_check_interval=data.get("state_check_interval", DEFAULT_STATE_CHECK_INTERVAL),
|
||||
led_skip_start=data.get("led_skip_start", 0),
|
||||
led_skip_end=data.get("led_skip_end", 0),
|
||||
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())),
|
||||
|
||||
Reference in New Issue
Block a user