Enhance CSS test preview with live capture, brightness display, and UX fixes

- Stream live JPEG frames from picture sources into the test preview rectangle
- Add composite layer brightness display via value source streaming
- Fix missing id on css-test-rect-screen element that prevented frame display
- Preload images before swapping to eliminate flicker on frame updates
- Increase preview resolution to 480x360 and add subtle outline
- Prevent auto-focus on name field in modals on touch devices (desktopFocus)
- Fix performance chart padding, color picker clipping, and subtitle offset
- Add calibration-style ticks and source name/LED count to rectangle preview

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-13 01:31:37 +03:00
parent 9b5686ac0a
commit 568a992a4e
12 changed files with 353 additions and 48 deletions

View File

@@ -1,8 +1,11 @@
"""Color strip source routes: CRUD, calibration test, preview, and API input push."""
import asyncio
import io as _io
import json as _json
import secrets
import time as _time
import uuid as _uuid
import numpy as np
from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect
@@ -706,7 +709,6 @@ async def test_color_strip_ws(
return
# 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 = f"__test_{_uuid.uuid4().hex[:8]}__"
@@ -733,6 +735,7 @@ async def test_color_strip_ws(
meta: dict = {
"type": "meta",
"source_type": source.source_type,
"source_name": source.name,
"led_count": stream.led_count,
}
if is_picture and stream.calibration:
@@ -752,9 +755,10 @@ async def test_color_strip_ws(
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}, ...]
layer_infos = [] # [{name, id, is_notification, has_brightness}, ...]
for layer in enabled_layers:
info = {"id": layer["source_id"], "name": layer.get("source_id", "?"), "is_notification": False}
info = {"id": layer["source_id"], "name": layer.get("source_id", "?"),
"is_notification": False, "has_brightness": bool(layer.get("brightness_source_id"))}
try:
layer_src = store.get_source(layer["source_id"])
info["name"] = layer_src.name
@@ -766,6 +770,13 @@ async def test_color_strip_ws(
meta["layer_infos"] = layer_infos
await websocket.send_text(_json.dumps(meta))
# For picture sources, grab the live stream for frame preview
_frame_live = None
if is_picture and hasattr(stream, '_live_stream'):
_frame_live = stream._live_stream
_last_aux_time = 0.0
_AUX_INTERVAL = 0.5 # send JPEG preview / brightness updates every 0.5s
# Stream binary RGB frames at ~20 Hz
while True:
# For composite sources, send per-layer data like target preview does
@@ -791,6 +802,50 @@ async def test_color_strip_ws(
colors = stream.get_latest_colors()
if colors is not None:
await websocket.send_bytes(colors.tobytes())
# Periodically send auxiliary data (frame preview, brightness)
now = _time.monotonic()
if now - _last_aux_time >= _AUX_INTERVAL:
_last_aux_time = now
# Send brightness values for composite layers
if is_composite and isinstance(stream, CompositeColorStripStream):
try:
bri_values = stream.get_layer_brightness()
if any(v is not None for v in bri_values):
bri_msg = {"type": "brightness", "values": [
round(v * 100) if v is not None else None for v in bri_values
]}
await websocket.send_text(_json.dumps(bri_msg))
except Exception:
pass
# Send JPEG frame preview for picture sources
if _frame_live:
try:
frame = _frame_live.get_latest_frame()
if frame is not None and frame.image is not None:
from PIL import Image as _PIL_Image
img = frame.image
# Ensure 3-channel RGB (some engines may produce BGRA)
if img.ndim == 3 and img.shape[2] == 4:
img = img[:, :, :3]
# Downscale for bandwidth
h, w = img.shape[:2]
scale = min(480 / w, 360 / h, 1.0)
if scale < 1.0:
new_w = max(1, int(w * scale))
new_h = max(1, int(h * scale))
pil = _PIL_Image.fromarray(img).resize((new_w, new_h), _PIL_Image.LANCZOS)
else:
pil = _PIL_Image.fromarray(img)
buf = _io.BytesIO()
pil.save(buf, format='JPEG', quality=70)
# Wire format: [0xFD] [jpeg_bytes]
await websocket.send_bytes(b'\xfd' + buf.getvalue())
except Exception as e:
logger.warning(f"JPEG frame preview error: {e}")
await asyncio.sleep(0.05)
except WebSocketDisconnect:
pass