Add per-layer LED preview for composite color strip sources

When a target uses a composite CSS source, the LED preview now shows
individual layer strips below the blended composite result. Backend
stores per-layer color snapshots and sends an extended binary wire
format; frontend renders separate canvases with hover labels.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 17:51:35 +03:00
parent f6f428515a
commit 2a73b92d4a
4 changed files with 140 additions and 17 deletions

View File

@@ -42,6 +42,7 @@ class CompositeColorStripStream(ColorStripStream):
self._running = False
self._thread: Optional[threading.Thread] = None
self._latest_colors: Optional[np.ndarray] = None
self._latest_layer_colors: Optional[List[np.ndarray]] = None
self._colors_lock = threading.Lock()
# layer_index -> (source_id, consumer_id, stream)
@@ -104,6 +105,11 @@ class CompositeColorStripStream(ColorStripStream):
with self._colors_lock:
return self._latest_colors
def get_layer_colors(self) -> Optional[List[np.ndarray]]:
"""Return per-layer color snapshots (after resize/brightness, before blending)."""
with self._colors_lock:
return self._latest_layer_colors
def configure(self, device_led_count: int) -> None:
if self._auto_size and device_led_count > 0 and device_led_count != self._led_count:
self._led_count = device_led_count
@@ -301,6 +307,7 @@ class CompositeColorStripStream(ColorStripStream):
result_buf = self._result_a if self._use_a else self._result_b
self._use_a = not self._use_a
has_result = False
layer_snapshots: List[np.ndarray] = []
with self._sub_lock:
sub_snapshot = dict(self._sub_streams)
@@ -327,6 +334,9 @@ class CompositeColorStripStream(ColorStripStream):
if bri < 1.0:
colors = (colors.astype(np.uint16) * int(bri * 256) >> 8).astype(np.uint8)
# Snapshot layer colors before blending (copy — may alias shared buf)
layer_snapshots.append(colors.copy())
opacity = layer.get("opacity", 1.0)
blend_mode = layer.get("blend_mode", _BLEND_NORMAL)
alpha = int(opacity * 256)
@@ -348,6 +358,7 @@ class CompositeColorStripStream(ColorStripStream):
if has_result:
with self._colors_lock:
self._latest_colors = result_buf
self._latest_layer_colors = layer_snapshots if len(layer_snapshots) > 1 else None
except Exception as e:
logger.error(f"CompositeColorStripStream processing error: {e}", exc_info=True)

View File

@@ -77,6 +77,7 @@ class WledTargetProcessor(TargetProcessor):
self._preview_clients: list = []
self._last_preview_colors: np.ndarray | None = None
self._last_preview_brightness: int = 255
self._last_preview_data: bytes | None = None # cached full binary frame
# ----- Properties -----
@@ -481,9 +482,8 @@ class WledTargetProcessor(TargetProcessor):
def add_led_preview_client(self, ws) -> None:
self._preview_clients.append(ws)
# Send last known frame immediately so late joiners see current state
if self._last_preview_colors is not None:
data = bytes([self._last_preview_brightness]) + self._last_preview_colors.tobytes()
asyncio.ensure_future(self._send_preview_to(ws, data))
if self._last_preview_data is not None:
asyncio.ensure_future(self._send_preview_to(ws, self._last_preview_data))
@staticmethod
async def _send_preview_to(ws, data: bytes) -> None:
@@ -499,13 +499,35 @@ class WledTargetProcessor(TargetProcessor):
async def _broadcast_led_preview(self, colors: np.ndarray, brightness: int = 255) -> None:
"""Broadcast LED colors as binary RGB bytes to preview WebSocket clients.
Wire format: [brightness_byte] [R G B R G B ...]
First byte is the effective brightness (0-255), rest is RGB pixel data.
Standard wire format: [brightness_byte] [R G B R G B ...]
Composite wire format: [brightness_byte] [0xFE] [layer_count] [led_count_hi] [led_count_lo]
[layer_0_rgb...] [layer_1_rgb...] ... [composite_rgb...]
"""
if not self._preview_clients:
return
data = bytes([brightness]) + colors.tobytes()
# Check if source is composite with multiple layers
from wled_controller.core.processing.composite_stream import CompositeColorStripStream
stream = self._css_stream
layer_colors = None
if isinstance(stream, CompositeColorStripStream):
layer_colors = stream.get_layer_colors()
if layer_colors and len(layer_colors) > 1:
led_count = len(colors)
header = bytes([brightness, 0xFE, len(layer_colors),
(led_count >> 8) & 0xFF, led_count & 0xFF])
parts = [header]
for lc in layer_colors:
if len(lc) != led_count:
lc = self._fit_to_device(lc, led_count)
parts.append(lc.tobytes())
parts.append(colors.tobytes())
data = b''.join(parts)
else:
data = bytes([brightness]) + colors.tobytes()
self._last_preview_data = data
async def _send_safe(ws):
try: