CSS: add StaticColorStripSource type with auto-sized LED count
Introduces a new 'static' source type that fills all device LEDs with a single constant RGB color — no screen capture or processing required. - StaticColorStripSource storage model (color + led_count=0 auto-size) - StaticColorStripStream: no background thread, configure() sizes to device LED count at processor start; hot-updates preserve runtime size - ColorStripStreamManager dispatches static sources (no LiveStream needed) - WledTargetProcessor calls stream.configure(device_led_count) on start - API schemas/routes: source_type Literal["picture","static"]; color field; overlay/calibration-test endpoints return 400 for static - Frontend: type selector modal, color picker, type-aware card rendering (🎨 icon + color swatch), LED count field hidden for static type - Locale keys: color_strip.type, color_strip.static_color (en + ru) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -331,3 +331,72 @@ class PictureColorStripStream(ColorStripStream):
|
||||
remaining = frame_time - elapsed
|
||||
if remaining > 0:
|
||||
time.sleep(remaining)
|
||||
|
||||
|
||||
class StaticColorStripStream(ColorStripStream):
|
||||
"""Color strip stream that returns a constant single-color array.
|
||||
|
||||
No background thread needed — every call to get_latest_colors() returns
|
||||
the same pre-built numpy array. Parameters can be hot-updated via
|
||||
update_source().
|
||||
"""
|
||||
|
||||
def __init__(self, source):
|
||||
"""
|
||||
Args:
|
||||
source: StaticColorStripSource config
|
||||
"""
|
||||
self._update_from_source(source)
|
||||
|
||||
def _update_from_source(self, source) -> None:
|
||||
color = source.color if isinstance(source.color, list) and len(source.color) == 3 else [255, 255, 255]
|
||||
self._source_color = color # stored separately so configure() can rebuild
|
||||
self._auto_size = not source.led_count # True when led_count == 0
|
||||
led_count = source.led_count if source.led_count and source.led_count > 0 else 1
|
||||
self._led_count = led_count
|
||||
self._rebuild_colors()
|
||||
|
||||
def _rebuild_colors(self) -> None:
|
||||
self._colors = np.tile(
|
||||
np.array(self._source_color, dtype=np.uint8),
|
||||
(self._led_count, 1),
|
||||
)
|
||||
|
||||
def configure(self, device_led_count: int) -> None:
|
||||
"""Set LED count from the target device (called by WledTargetProcessor on start).
|
||||
|
||||
Only takes effect when led_count was 0 (auto-size). Silently ignored
|
||||
when an explicit led_count was configured on the source.
|
||||
"""
|
||||
if self._auto_size and device_led_count > 0 and device_led_count != self._led_count:
|
||||
self._led_count = device_led_count
|
||||
self._rebuild_colors()
|
||||
logger.debug(f"StaticColorStripStream auto-sized to {device_led_count} LEDs")
|
||||
|
||||
@property
|
||||
def target_fps(self) -> int:
|
||||
return 30 # static output; any reasonable value is fine
|
||||
|
||||
@property
|
||||
def led_count(self) -> int:
|
||||
return self._led_count
|
||||
|
||||
def start(self) -> None:
|
||||
logger.info(f"StaticColorStripStream started (leds={self._led_count})")
|
||||
|
||||
def stop(self) -> None:
|
||||
logger.info("StaticColorStripStream stopped")
|
||||
|
||||
def get_latest_colors(self) -> Optional[np.ndarray]:
|
||||
return self._colors
|
||||
|
||||
def update_source(self, source) -> None:
|
||||
from wled_controller.storage.color_strip_source import StaticColorStripSource
|
||||
if isinstance(source, StaticColorStripSource):
|
||||
prev_led_count = self._led_count if self._auto_size else None
|
||||
self._update_from_source(source)
|
||||
# If we were auto-sized, preserve the runtime LED count across updates
|
||||
if prev_led_count and self._auto_size:
|
||||
self._led_count = prev_led_count
|
||||
self._rebuild_colors()
|
||||
logger.info("StaticColorStripStream params updated in-place")
|
||||
|
||||
@@ -15,6 +15,7 @@ from typing import Dict, Optional
|
||||
from wled_controller.core.processing.color_strip_stream import (
|
||||
ColorStripStream,
|
||||
PictureColorStripStream,
|
||||
StaticColorStripStream,
|
||||
)
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
@@ -78,10 +79,21 @@ class ColorStripStreamManager:
|
||||
)
|
||||
return entry.stream
|
||||
|
||||
from wled_controller.storage.color_strip_source import PictureColorStripSource
|
||||
from wled_controller.storage.color_strip_source import PictureColorStripSource, StaticColorStripSource
|
||||
|
||||
source = self._color_strip_store.get_source(css_id)
|
||||
|
||||
if isinstance(source, StaticColorStripSource):
|
||||
css_stream = StaticColorStripStream(source)
|
||||
css_stream.start()
|
||||
self._streams[css_id] = _ColorStripEntry(
|
||||
stream=css_stream,
|
||||
ref_count=1,
|
||||
picture_source_id="", # no live stream to manage
|
||||
)
|
||||
logger.info(f"Created static color strip stream for source {css_id}")
|
||||
return css_stream
|
||||
|
||||
if not isinstance(source, PictureColorStripSource):
|
||||
raise ValueError(
|
||||
f"Unsupported color strip source type '{source.source_type}' for {css_id}"
|
||||
@@ -110,7 +122,7 @@ class ColorStripStreamManager:
|
||||
picture_source_id=source.picture_source_id,
|
||||
)
|
||||
|
||||
logger.info(f"Created color strip stream for source {css_id}")
|
||||
logger.info(f"Created picture color strip stream for source {css_id}")
|
||||
return css_stream
|
||||
|
||||
def release(self, css_id: str) -> None:
|
||||
@@ -140,8 +152,9 @@ class ColorStripStreamManager:
|
||||
del self._streams[css_id]
|
||||
logger.info(f"Removed color strip stream for source {css_id}")
|
||||
|
||||
# Release the underlying live stream
|
||||
self._live_stream_manager.release(picture_source_id)
|
||||
# Release the underlying live stream (not needed for static sources)
|
||||
if picture_source_id:
|
||||
self._live_stream_manager.release(picture_source_id)
|
||||
|
||||
def update_source(self, css_id: str, new_source) -> None:
|
||||
"""Hot-update processing params on a running stream.
|
||||
|
||||
@@ -114,6 +114,12 @@ class WledTargetProcessor(TargetProcessor):
|
||||
self._color_strip_stream = stream
|
||||
self._resolved_display_index = stream.display_index
|
||||
self._resolved_target_fps = stream.target_fps
|
||||
|
||||
# For auto-sized static streams (led_count == 0), size to device LED count
|
||||
from wled_controller.core.processing.color_strip_stream import StaticColorStripStream
|
||||
if isinstance(stream, StaticColorStripStream) and device_info.led_count > 0:
|
||||
stream.configure(device_info.led_count)
|
||||
|
||||
logger.info(
|
||||
f"Acquired color strip stream for target {self._target_id} "
|
||||
f"(css={self._color_strip_source_id}, display={self._resolved_display_index}, "
|
||||
|
||||
Reference in New Issue
Block a user