Add composite layer preview, configurable LED count, and notification fire button to CSS test modal

- Composite sources show per-layer strip canvases with composite result on top
- Server sends composite wire format with per-layer RGB data
- LED count is configurable via input field, persisted in localStorage
- Notification sources show a bell fire button on the strip preview
- Composite with notification layers shows per-layer fire buttons
- Fixed stale WS frame bug with generation counter and unique consumer IDs
- Modal width is now fixed at 700px to prevent layout jumps
- Target card composite layers now use same-height canvases

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-12 20:17:54 +03:00
parent f2162133a8
commit 97db63824e
9 changed files with 359 additions and 33 deletions

View File

@@ -34,7 +34,7 @@ from wled_controller.core.capture.calibration import (
)
from wled_controller.core.capture.screen_capture import get_available_displays
from wled_controller.core.processing.processor_manager import ProcessorManager
from wled_controller.storage.color_strip_source import AdvancedPictureColorStripSource, ApiInputColorStripSource, NotificationColorStripSource, PictureColorStripSource
from wled_controller.storage.color_strip_source import AdvancedPictureColorStripSource, ApiInputColorStripSource, CompositeColorStripSource, NotificationColorStripSource, PictureColorStripSource
from wled_controller.storage.color_strip_store import ColorStripStore
from wled_controller.storage.picture_source import ProcessedPictureSource, ScreenCapturePictureSource
from wled_controller.storage.picture_source_store import PictureSourceStore
@@ -705,10 +705,11 @@ async def test_color_strip_ws(
await websocket.close(code=4004, reason=str(e))
return
# Acquire stream
# Acquire stream unique consumer ID per WS to avoid release races
import uuid as _uuid
manager: ProcessorManager = get_processor_manager()
csm = manager.color_strip_stream_manager
consumer_id = "__test__"
consumer_id = f"__test_{_uuid.uuid4().hex[:8]}__"
try:
stream = csm.acquire(source_id, consumer_id)
except Exception as e:
@@ -724,8 +725,11 @@ async def test_color_strip_ws(
logger.info(f"CSS test WebSocket connected for {source_id}")
try:
from wled_controller.core.processing.composite_stream import CompositeColorStripStream
# Send metadata as first message
is_picture = isinstance(source, (PictureColorStripSource, AdvancedPictureColorStripSource))
is_composite = isinstance(source, CompositeColorStripSource)
meta: dict = {
"type": "meta",
"source_type": source.source_type,
@@ -745,13 +749,48 @@ async def test_color_strip_ws(
indices = [(idx + offset) % total for idx in indices]
edges.append({"edge": seg.edge, "indices": indices})
meta["edges"] = edges
if is_composite and hasattr(source, "layers"):
# Send layer info for composite preview
enabled_layers = [l for l in source.layers if l.get("enabled", True)]
layer_infos = [] # [{name, id, is_notification}, ...]
for layer in enabled_layers:
info = {"id": layer["source_id"], "name": layer.get("source_id", "?"), "is_notification": False}
try:
layer_src = store.get_source(layer["source_id"])
info["name"] = layer_src.name
info["is_notification"] = isinstance(layer_src, NotificationColorStripSource)
except (ValueError, KeyError):
pass
layer_infos.append(info)
meta["layers"] = [li["name"] for li in layer_infos]
meta["layer_infos"] = layer_infos
await websocket.send_text(_json.dumps(meta))
# Stream binary RGB frames at ~20 Hz
while True:
colors = stream.get_latest_colors()
if colors is not None:
await websocket.send_bytes(colors.tobytes())
# For composite sources, send per-layer data like target preview does
if is_composite and isinstance(stream, CompositeColorStripStream):
layer_colors = stream.get_layer_colors()
composite_colors = stream.get_latest_colors()
if composite_colors is not None and layer_colors and len(layer_colors) > 1:
led_count = composite_colors.shape[0]
rgb_size = led_count * 3
# Wire format: [0xFE] [layer_count] [led_count_hi] [led_count_lo] [layer0_rgb...] ... [composite_rgb]
header = bytes([0xFE, len(layer_colors), (led_count >> 8) & 0xFF, led_count & 0xFF])
parts = [header]
for lc in layer_colors:
if lc is not None and lc.shape[0] == led_count:
parts.append(lc.tobytes())
else:
parts.append(b'\x00' * rgb_size)
parts.append(composite_colors.tobytes())
await websocket.send_bytes(b''.join(parts))
elif composite_colors is not None:
await websocket.send_bytes(composite_colors.tobytes())
else:
colors = stream.get_latest_colors()
if colors is not None:
await websocket.send_bytes(colors.tobytes())
await asyncio.sleep(0.05)
except WebSocketDisconnect:
pass