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

feat: expand color strip sources with gradient references and effect improvements

Browse Source 
Add gradient_id field to color strip sources for referencing reusable
gradient entities. Improve audio stream processing and effect stream
with new parameters.
This commit is contained in:
Admin
2026-03-24 13:58:33 +03:00
parent 6a881f8fdd
commit 227b82f522
8 changed files with 109 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
server/src/wled_controller/api/routes/color_strip_sources.py
View File

@@ -567,6 +567,13 @@ async def preview_color_strip_ws(
if not stream_cls:
raise ValueError(f"Unsupported preview source_type: {source.source_type}")
s = stream_cls(source)
# Inject gradient store for palette resolution
if hasattr(s, "set_gradient_store"):
try:
from wled_controller.api.dependencies import get_gradient_store
s.set_gradient_store(get_gradient_store())
except Exception:
pass
if hasattr(s, "configure"):
s.configure(led_count)
# Inject sync clock if requested

5
server/src/wled_controller/api/schemas/color_strip_sources.py
View File

@@ -70,6 +70,8 @@ class ColorStripSourceCreate(BaseModel):
scale: Optional[float] = Field(None, description="Spatial scale 0.5-5.0", ge=0.5, le=5.0)
mirror: Optional[bool] = Field(None, description="Mirror/bounce mode (meteor/comet)")
custom_palette: Optional[List[List[float]]] = Field(None, description="Custom palette stops [[pos,R,G,B],...]")
# gradient entity reference (effect, gradient, audio types)
gradient_id: Optional[str] = Field(None, description="Gradient entity ID (overrides palette/inline stops)")
# gradient-type easing
easing: Optional[str] = Field(None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic")
# composite-type fields
@@ -135,6 +137,8 @@ class ColorStripSourceUpdate(BaseModel):
scale: Optional[float] = Field(None, description="Spatial scale 0.5-5.0", ge=0.5, le=5.0)
mirror: Optional[bool] = Field(None, description="Mirror/bounce mode")
custom_palette: Optional[List[List[float]]] = Field(None, description="Custom palette stops [[pos,R,G,B],...]")
# gradient entity reference (effect, gradient, audio types)
gradient_id: Optional[str] = Field(None, description="Gradient entity ID (overrides palette/inline stops)")
# gradient-type easing
easing: Optional[str] = Field(None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic")
# composite-type fields
@@ -202,6 +206,7 @@ class ColorStripSourceResponse(BaseModel):
scale: Optional[float] = Field(None, description="Spatial scale")
mirror: Optional[bool] = Field(None, description="Mirror/bounce mode")
custom_palette: Optional[List[List[float]]] = Field(None, description="Custom palette stops")
gradient_id: Optional[str] = Field(None, description="Gradient entity ID")
# gradient-type easing
easing: Optional[str] = Field(None, description="Gradient interpolation easing")
# composite-type fields

20
server/src/wled_controller/core/processing/audio_stream.py
View File

@@ -58,14 +58,32 @@ class AudioColorStripStream(ColorStripStream):
self._prev_spectrum: Optional[np.ndarray] = None
self._prev_rms = 0.0
self._gradient_store = None # injected by stream manager
self._update_from_source(source)
def set_gradient_store(self, gradient_store) -> None:
"""Inject gradient store for palette resolution."""
self._gradient_store = gradient_store
self._resolve_palette_lut()
def _resolve_palette_lut(self) -> None:
"""Build palette LUT from gradient_id or legacy palette name."""
gradient_id = self._gradient_id
if gradient_id and self._gradient_store:
stops = self._gradient_store.resolve_stops(gradient_id)
if stops:
custom = [[s["position"], *s["color"]] for s in stops]
self._palette_lut = _build_palette_lut("custom", custom)
return
self._palette_lut = _build_palette_lut(self._palette_name)
def _update_from_source(self, source) -> None:
self._visualization_mode = getattr(source, "visualization_mode", "spectrum")
self._sensitivity = float(getattr(source, "sensitivity", 1.0))
self._smoothing = float(getattr(source, "smoothing", 0.3))
self._gradient_id = getattr(source, "gradient_id", None)
self._palette_name = getattr(source, "palette", "rainbow")
self._palette_lut = _build_palette_lut(self._palette_name)
self._resolve_palette_lut()
color = getattr(source, "color", None)
self._color = color if isinstance(color, list) and len(color) == 3 else [0, 255, 0]
color_peak = getattr(source, "color_peak", None)

18
server/src/wled_controller/core/processing/color_strip_stream.py
View File

@@ -851,10 +851,28 @@ class GradientColorStripStream(ColorStripStream):
self._fps = 30
self._frame_time = 1.0 / 30
self._clock = None # optional SyncClockRuntime
self._gradient_store = None # injected by stream manager
self._update_from_source(source)
def set_gradient_store(self, gradient_store) -> None:
"""Inject gradient store for resolving gradient_id to stops."""
self._gradient_store = gradient_store
# Re-resolve stops if gradient_id is set
gradient_id = getattr(self, "_gradient_id", None)
if gradient_id and self._gradient_store:
stops = self._gradient_store.resolve_stops(gradient_id)
if stops:
self._stops = stops
self._rebuild_colors()
def _update_from_source(self, source) -> None:
self._gradient_id = getattr(source, "gradient_id", None)
self._stops = list(source.stops) if source.stops else []
# Override inline stops with gradient entity if set
if self._gradient_id and self._gradient_store:
resolved = self._gradient_store.resolve_stops(self._gradient_id)
if resolved:
self._stops = resolved
_lc = getattr(source, "led_count", 0)
self._auto_size = not _lc
led_count = _lc if _lc and _lc > 0 else 1

7
server/src/wled_controller/core/processing/color_strip_stream_manager.py
View File

@@ -69,7 +69,7 @@ class ColorStripStreamManager:
keyed by ``{css_id}:{consumer_id}``.
"""
def __init__(self, color_strip_store, live_stream_manager, audio_capture_manager=None, audio_source_store=None, audio_template_store=None, sync_clock_manager=None, value_stream_manager=None, cspt_store=None):
def __init__(self, color_strip_store, live_stream_manager, audio_capture_manager=None, audio_source_store=None, audio_template_store=None, sync_clock_manager=None, value_stream_manager=None, cspt_store=None, gradient_store=None):
"""
Args:
color_strip_store: ColorStripStore for resolving source configs
@@ -79,6 +79,7 @@ class ColorStripStreamManager:
sync_clock_manager: SyncClockManager for acquiring clock runtimes
value_stream_manager: ValueStreamManager for per-layer brightness sources
cspt_store: ColorStripProcessingTemplateStore for per-layer filter chains
gradient_store: GradientStore for resolving gradient entity references
"""
self._color_strip_store = color_strip_store
self._live_stream_manager = live_stream_manager
@@ -88,6 +89,7 @@ class ColorStripStreamManager:
self._sync_clock_manager = sync_clock_manager
self._value_stream_manager = value_stream_manager
self._cspt_store = cspt_store
self._gradient_store = gradient_store
self._streams: Dict[str, _ColorStripEntry] = {}
def _inject_clock(self, css_stream, source) -> Optional[str]:
@@ -177,6 +179,9 @@ class ColorStripStreamManager:
f"Unsupported color strip source type '{source.source_type}' for {css_id}"
)
css_stream = stream_cls(source)
# Inject gradient store for palette resolution
if self._gradient_store and hasattr(css_stream, "set_gradient_store"):
css_stream.set_gradient_store(self._gradient_store)
# Inject sync clock runtime if source references a clock
acquired_clock_id = self._inject_clock(css_stream, source)
css_stream.start()

25
server/src/wled_controller/core/processing/effect_stream.py
View File

@@ -228,16 +228,37 @@ class EffectColorStripStream(ColorStripStream):
self._fw_last_launch = 0.0
# Sparkle rain state
self._sparkle_state: Optional[np.ndarray] = None # per-LED brightness 0..1
self._gradient_store = None # injected by stream manager
self._update_from_source(source)
def set_gradient_store(self, gradient_store) -> None:
"""Inject gradient store for palette resolution. Called by stream manager."""
self._gradient_store = gradient_store
# Re-resolve palette now that store is available
self._resolve_palette_lut()
def _resolve_palette_lut(self) -> None:
"""Build palette LUT from gradient_id or legacy palette name."""
gradient_id = self._gradient_id
if gradient_id and self._gradient_store:
stops = self._gradient_store.resolve_stops(gradient_id)
if stops:
# Convert gradient entity stops to palette LUT stops
custom = [[s["position"], *s["color"]] for s in stops]
self._palette_lut = _build_palette_lut("custom", custom)
return
# Fallback: legacy palette name or custom_palette
self._palette_lut = _build_palette_lut(self._palette_name, self._custom_palette)
def _update_from_source(self, source) -> None:
self._effect_type = getattr(source, "effect_type", "fire")
_lc = getattr(source, "led_count", 0)
self._auto_size = not _lc
self._led_count = _lc if _lc and _lc > 0 else 1
self._gradient_id = getattr(source, "gradient_id", None)
self._palette_name = getattr(source, "palette", None) or _EFFECT_DEFAULT_PALETTE.get(self._effect_type, "fire")
custom_palette = getattr(source, "custom_palette", None)
self._palette_lut = _build_palette_lut(self._palette_name, custom_palette)
self._custom_palette = getattr(source, "custom_palette", None)
self._resolve_palette_lut()
color = getattr(source, "color", None)
self._color = color if isinstance(color, list) and len(color) == 3 else [255, 80, 0]
self._intensity = float(getattr(source, "intensity", 1.0))

2
server/src/wled_controller/core/processing/processor_manager.py
View File

@@ -55,6 +55,7 @@ class ProcessorDependencies:
value_source_store: object = None
sync_clock_manager: object = None
cspt_store: object = None
gradient_store: object = None
@dataclass
@@ -129,6 +130,7 @@ class ProcessorManager(AutoRestartMixin, DeviceHealthMixin, DeviceTestModeMixin)
audio_template_store=deps.audio_template_store,
sync_clock_manager=deps.sync_clock_manager,
cspt_store=deps.cspt_store,
gradient_store=deps.gradient_store,
)
self._value_stream_manager = ValueStreamManager(
value_source_store=deps.value_source_store,

37
server/src/wled_controller/storage/color_strip_source.py
View File

@@ -155,6 +155,8 @@ class ColorStripSource:
created_at=created_at, updated_at=updated_at, description=description,
clock_id=clock_id, tags=tags, stops=stops,
animation=data.get("animation"),
easing=data.get("easing") or "linear",
gradient_id=data.get("gradient_id"),
)
if source_type == "color_cycle":
@@ -195,6 +197,7 @@ class ColorStripSource:
sensitivity=float(data.get("sensitivity") or 1.0),
smoothing=float(data.get("smoothing") or 0.3),
palette=data.get("palette") or "rainbow",
gradient_id=data.get("gradient_id"),
color=color,
color_peak=color_peak,
led_count=data.get("led_count") or 0,
@@ -212,10 +215,12 @@ class ColorStripSource:
created_at=created_at, updated_at=updated_at, description=description,
clock_id=clock_id, tags=tags, effect_type=data.get("effect_type") or "fire",
palette=data.get("palette") or "fire",
gradient_id=data.get("gradient_id"),
color=color,
intensity=float(data.get("intensity") or 1.0),
scale=float(data.get("scale") or 1.0),
mirror=bool(data.get("mirror", False)),
custom_palette=data.get("custom_palette"),
)
if source_type == "api_input":
@@ -494,19 +499,22 @@ class GradientColorStripSource(ColorStripSource):
])
animation: Optional[dict] = None # {"enabled": bool, "type": str, "speed": float} or None
easing: str = "linear" # linear | ease_in_out | step | cubic
gradient_id: Optional[str] = None # references a Gradient entity; overrides inline stops
def to_dict(self) -> dict:
d = super().to_dict()
d["stops"] = [dict(s) for s in self.stops]
d["animation"] = self.animation
d["easing"] = self.easing
d["gradient_id"] = self.gradient_id
return d
@classmethod
def create_from_kwargs(cls, *, id: str, name: str, source_type: str,
created_at: datetime, updated_at: datetime,
description=None, clock_id=None, tags=None,
stops=None, animation=None, easing=None, **_kwargs):
stops=None, animation=None, easing=None,
gradient_id=None, **_kwargs):
return cls(
id=id, name=name, source_type="gradient",
created_at=created_at, updated_at=updated_at,
@@ -517,6 +525,7 @@ class GradientColorStripSource(ColorStripSource):
],
animation=animation,
easing=easing if easing in ("linear", "ease_in_out", "step", "cubic") else "linear",
gradient_id=gradient_id,
)
def apply_update(self, **kwargs) -> None:
@@ -527,6 +536,8 @@ class GradientColorStripSource(ColorStripSource):
self.animation = kwargs["animation"]
if kwargs.get("easing") is not None:
self.easing = kwargs["easing"]
if "gradient_id" in kwargs:
self.gradient_id = kwargs["gradient_id"]
@dataclass
@@ -580,17 +591,19 @@ class EffectColorStripSource(ColorStripSource):
"""
effect_type: str = "fire" # fire | meteor | plasma | noise | aurora + new types
palette: str = "fire" # named color palette or "custom"
palette: str = "fire" # legacy palette name (kept for migration)
gradient_id: Optional[str] = None # references a Gradient entity (preferred over palette)
color: list = field(default_factory=lambda: [255, 80, 0]) # [R,G,B] for meteor/comet/bouncing_ball head
intensity: float = 1.0 # effect-specific intensity (0.1-2.0)
scale: float = 1.0 # spatial scale / zoom (0.5-5.0)
mirror: bool = False # bounce mode (meteor/comet)
custom_palette: Optional[list] = None # [[pos, R, G, B], ...] custom palette stops
custom_palette: Optional[list] = None # legacy [[pos, R, G, B], ...] custom palette stops
def to_dict(self) -> dict:
d = super().to_dict()
d["effect_type"] = self.effect_type
d["palette"] = self.palette
d["gradient_id"] = self.gradient_id
d["color"] = list(self.color)
d["intensity"] = self.intensity
d["scale"] = self.scale
@@ -602,8 +615,8 @@ class EffectColorStripSource(ColorStripSource):
def create_from_kwargs(cls, *, id: str, name: str, source_type: str,
created_at: datetime, updated_at: datetime,
description=None, clock_id=None, tags=None,
effect_type="fire", palette="fire", color=None,
intensity=1.0, scale=1.0, mirror=False,
effect_type="fire", palette="fire", gradient_id=None,
color=None, intensity=1.0, scale=1.0, mirror=False,
custom_palette=None, **_kwargs):
rgb = _validate_rgb(color, [255, 80, 0])
return cls(
@@ -611,6 +624,7 @@ class EffectColorStripSource(ColorStripSource):
created_at=created_at, updated_at=updated_at,
description=description, clock_id=clock_id, tags=tags or [],
effect_type=effect_type or "fire", palette=palette or "fire",
gradient_id=gradient_id,
color=rgb,
intensity=float(intensity) if intensity else 1.0,
scale=float(scale) if scale else 1.0,
@@ -623,6 +637,8 @@ class EffectColorStripSource(ColorStripSource):
self.effect_type = kwargs["effect_type"]
if kwargs.get("palette") is not None:
self.palette = kwargs["palette"]
if "gradient_id" in kwargs:
self.gradient_id = kwargs["gradient_id"]
color = kwargs.get("color")
if color is not None and isinstance(color, list) and len(color) == 3:
self.color = color
@@ -650,7 +666,8 @@ class AudioColorStripSource(ColorStripSource):
audio_source_id: str = "" # references a MonoAudioSource
sensitivity: float = 1.0 # gain multiplier (0.1-5.0)
smoothing: float = 0.3 # temporal smoothing (0.0-1.0)
palette: str = "rainbow" # named color palette
palette: str = "rainbow" # legacy palette name (kept for migration)
gradient_id: Optional[str] = None # references a Gradient entity (preferred)
color: list = field(default_factory=lambda: [0, 255, 0]) # base RGB for VU meter
color_peak: list = field(default_factory=lambda: [255, 0, 0]) # peak RGB for VU meter
led_count: int = 0 # 0 = use device LED count
@@ -663,6 +680,7 @@ class AudioColorStripSource(ColorStripSource):
d["sensitivity"] = self.sensitivity
d["smoothing"] = self.smoothing
d["palette"] = self.palette
d["gradient_id"] = self.gradient_id
d["color"] = list(self.color)
d["color_peak"] = list(self.color_peak)
d["led_count"] = self.led_count
@@ -675,8 +693,8 @@ class AudioColorStripSource(ColorStripSource):
description=None, clock_id=None, tags=None,
visualization_mode="spectrum", audio_source_id="",
sensitivity=1.0, smoothing=0.3, palette="rainbow",
color=None, color_peak=None, led_count=0,
mirror=False, **_kwargs):
gradient_id=None, color=None, color_peak=None,
led_count=0, mirror=False, **_kwargs):
rgb = _validate_rgb(color, [0, 255, 0])
peak = _validate_rgb(color_peak, [255, 0, 0])
return cls(
@@ -688,6 +706,7 @@ class AudioColorStripSource(ColorStripSource):
sensitivity=float(sensitivity) if sensitivity else 1.0,
smoothing=float(smoothing) if smoothing else 0.3,
palette=palette or "rainbow",
gradient_id=gradient_id,
color=rgb, color_peak=peak, led_count=led_count,
mirror=bool(mirror),
)
@@ -704,6 +723,8 @@ class AudioColorStripSource(ColorStripSource):
self.smoothing = float(kwargs["smoothing"])
if kwargs.get("palette") is not None:
self.palette = kwargs["palette"]
if "gradient_id" in kwargs:
self.gradient_id = kwargs["gradient_id"]
color = kwargs.get("color")
if color is not None and isinstance(color, list) and len(color) == 3:
self.color = color