Add live LED strip preview via WebSocket on target cards

Stream real-time LED colors from running WLED targets to the browser via
binary WebSocket (RGB bytes, throttled to ~15 fps). Toggle button on
target card opens a compact canvas strip that renders each frame using
ImageData. Cached last frame is re-rendered after card reconciliation to
prevent flicker during auto-refresh.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-24 17:47:40 +03:00
parent a6253e8d96
commit 053a56eed3
8 changed files with 234 additions and 1 deletions

View File

@@ -541,6 +541,15 @@ class ProcessorManager:
proc = self._get_processor(target_id)
return proc.get_latest_colors()
def add_led_preview_client(self, target_id: str, ws) -> None:
proc = self._get_processor(target_id)
proc.add_led_preview_client(ws)
def remove_led_preview_client(self, target_id: str, ws) -> None:
proc = self._processors.get(target_id)
if proc:
proc.remove_led_preview_client(ws)
# ===== CALIBRATION TEST MODE (on device, driven by CSS calibration) =====
async def set_test_mode(

View File

@@ -224,6 +224,14 @@ class TargetProcessor(ABC):
"""Remove a WebSocket client."""
raise NotImplementedError(f"{type(self).__name__} does not support WebSockets")
def add_led_preview_client(self, ws) -> None:
"""Add a WebSocket client for live LED strip preview."""
raise NotImplementedError(f"{type(self).__name__} does not support LED preview")
def remove_led_preview_client(self, ws) -> None:
"""Remove a LED preview WebSocket client."""
raise NotImplementedError(f"{type(self).__name__} does not support LED preview")
def get_latest_colors(self) -> Dict[str, Tuple[int, int, int]]:
"""Get latest extracted colors (KC targets only)."""
return {}

View File

@@ -56,6 +56,9 @@ class WledTargetProcessor(TargetProcessor):
self._resolved_display_index: Optional[int] = None
# LED preview WebSocket clients
self._preview_clients: list = []
# ----- Properties -----
@property
@@ -403,6 +406,38 @@ class WledTargetProcessor(TargetProcessor):
def is_overlay_active(self) -> bool:
return self._overlay_active
# ----- LED Preview WebSocket -----
def supports_websocket(self) -> bool:
return True
def add_led_preview_client(self, ws) -> None:
self._preview_clients.append(ws)
def remove_led_preview_client(self, ws) -> None:
if ws in self._preview_clients:
self._preview_clients.remove(ws)
async def _broadcast_led_preview(self, colors: np.ndarray) -> None:
"""Broadcast LED colors as binary RGB bytes to preview WebSocket clients."""
if not self._preview_clients:
return
data = colors.astype(np.uint8).tobytes()
async def _send_safe(ws):
try:
await ws.send_bytes(data)
return True
except Exception:
return False
results = await asyncio.gather(*[_send_safe(ws) for ws in self._preview_clients])
disconnected = [ws for ws, ok in zip(self._preview_clients, results) if not ok]
for ws in disconnected:
self._preview_clients.remove(ws)
# ----- Private: processing loop -----
@staticmethod
@@ -426,6 +461,7 @@ class WledTargetProcessor(TargetProcessor):
fps_samples: collections.deque = collections.deque(maxlen=10)
send_timestamps: collections.deque = collections.deque()
last_send_time = 0.0
_last_preview_broadcast = 0.0
prev_frame_time_stamp = time.perf_counter()
loop = asyncio.get_running_loop()
_init_device_info = self._ctx.get_device_info(self._device_id)
@@ -540,6 +576,9 @@ class WledTargetProcessor(TargetProcessor):
last_send_time = now
send_timestamps.append(now)
self._metrics.frames_keepalive += 1
if self._preview_clients and (now - _last_preview_broadcast) >= 0.066:
await self._broadcast_led_preview(send_colors)
_last_preview_broadcast = now
self._metrics.frames_skipped += 1
while send_timestamps and send_timestamps[0] < now - 1.0:
send_timestamps.popleft()
@@ -570,6 +609,11 @@ class WledTargetProcessor(TargetProcessor):
last_send_time = now
send_timestamps.append(now)
# Broadcast to LED preview WebSocket clients (throttled to ~15 fps)
if self._preview_clients and (now - _last_preview_broadcast) >= 0.066:
await self._broadcast_led_preview(send_colors)
_last_preview_broadcast = now
self._metrics.timing_send_ms = send_ms
self._metrics.frames_processed += 1
self._metrics.last_update = datetime.utcnow()