Add API Input color strip source type with REST and WebSocket push

New source_type "api_input" allows external clients to push raw LED
color arrays ([R,G,B] per LED) via REST POST or WebSocket. Includes
configurable fallback color and timeout for automatic revert when no
data is received. Stream auto-sizes LED count from the target device.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-24 17:07:47 +03:00
parent 1e4a7a067f
commit 67a15776b2
10 changed files with 512 additions and 10 deletions

View File

@@ -10,6 +10,7 @@ Current types:
GradientColorStripSource — linear gradient across all LEDs from user-defined color stops
ColorCycleColorStripSource — smoothly cycles through a user-defined list of colors
AudioColorStripSource — audio-reactive visualization (spectrum, beat pulse, VU meter)
ApiInputColorStripSource — receives raw LED colors from external clients via REST/WebSocket
"""
from dataclasses import dataclass, field
@@ -78,6 +79,8 @@ class ColorStripSource:
"audio_source_id": None,
"sensitivity": None,
"color_peak": None,
"fallback_color": None,
"timeout": None,
}
@staticmethod
@@ -200,6 +203,20 @@ class ColorStripSource:
mirror=bool(data.get("mirror", False)),
)
if source_type == "api_input":
raw_fallback = data.get("fallback_color")
fallback_color = (
raw_fallback if isinstance(raw_fallback, list) and len(raw_fallback) == 3
else [0, 0, 0]
)
return ApiInputColorStripSource(
id=sid, name=name, source_type="api_input",
created_at=created_at, updated_at=updated_at, description=description,
led_count=data.get("led_count") or 0,
fallback_color=fallback_color,
timeout=float(data.get("timeout") or 5.0),
)
# Default: "picture" type
return PictureColorStripSource(
id=sid, name=name, source_type=source_type,
@@ -433,3 +450,25 @@ class MappedColorStripSource(ColorStripSource):
d["zones"] = [dict(z) for z in self.zones]
d["led_count"] = self.led_count
return d
@dataclass
class ApiInputColorStripSource(ColorStripSource):
"""Color strip source that receives raw LED color arrays from external clients.
External clients push [R,G,B] arrays via REST POST or WebSocket. The stream
buffers the latest frame and serves it to targets. When no data has been
received within `timeout` seconds, LEDs revert to `fallback_color`.
LED count auto-sizes from the connected device when led_count == 0.
"""
led_count: int = 0 # 0 = auto-size from device
fallback_color: list = field(default_factory=lambda: [0, 0, 0]) # [R, G, B]
timeout: float = 5.0 # seconds before reverting to fallback
def to_dict(self) -> dict:
d = super().to_dict()
d["led_count"] = self.led_count
d["fallback_color"] = list(self.fallback_color)
d["timeout"] = self.timeout
return d

View File

@@ -8,6 +8,7 @@ from typing import Dict, List, Optional
from wled_controller.core.capture.calibration import CalibrationConfig, calibration_to_dict
from wled_controller.storage.color_strip_source import (
ApiInputColorStripSource,
AudioColorStripSource,
ColorCycleColorStripSource,
ColorStripSource,
@@ -125,6 +126,8 @@ class ColorStripStore:
audio_source_id: str = "",
sensitivity: float = 1.0,
color_peak: Optional[list] = None,
fallback_color: Optional[list] = None,
timeout: Optional[float] = None,
) -> ColorStripSource:
"""Create a new color strip source.
@@ -245,6 +248,19 @@ class ColorStripStore:
zones=zones if isinstance(zones, list) else [],
led_count=led_count,
)
elif source_type == "api_input":
fb = fallback_color if isinstance(fallback_color, list) and len(fallback_color) == 3 else [0, 0, 0]
source = ApiInputColorStripSource(
id=source_id,
name=name,
source_type="api_input",
created_at=now,
updated_at=now,
description=description,
led_count=led_count,
fallback_color=fb,
timeout=float(timeout) if timeout is not None else 5.0,
)
else:
if calibration is None:
calibration = CalibrationConfig(layout="clockwise", start_position="bottom_left")
@@ -305,6 +321,8 @@ class ColorStripStore:
audio_source_id: Optional[str] = None,
sensitivity: Optional[float] = None,
color_peak: Optional[list] = None,
fallback_color: Optional[list] = None,
timeout: Optional[float] = None,
) -> ColorStripSource:
"""Update an existing color strip source.
@@ -414,6 +432,13 @@ class ColorStripStore:
source.zones = zones
if led_count is not None:
source.led_count = led_count
elif isinstance(source, ApiInputColorStripSource):
if fallback_color is not None and isinstance(fallback_color, list) and len(fallback_color) == 3:
source.fallback_color = fallback_color
if timeout is not None:
source.timeout = float(timeout)
if led_count is not None:
source.led_count = led_count
source.updated_at = datetime.utcnow()
self._save()