Add LED axis ticks and calibration labels to color strip test preview

- Add horizontal axis with tick marks and LED index labels below strip
  and composite preview canvases (first/last labels edge-aligned)
- Show actual/calibration LED count label on picture-based composite
  layers (e.g. "25/934")
- Display warning icon in orange when LED counts don't match
- Send is_picture and calibration_led_count in composite layer_infos

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-14 22:47:22 +03:00
parent d1c8324c0f
commit 7e78323c9c
4 changed files with 171 additions and 182 deletions

View File

@@ -3,7 +3,6 @@
import asyncio
import io as _io
import json as _json
import secrets
import time as _time
import uuid as _uuid
@@ -43,22 +42,29 @@ from wled_controller.storage.picture_source import ProcessedPictureSource, Scree
from wled_controller.storage.picture_source_store import PictureSourceStore
from wled_controller.storage.output_target_store import OutputTargetStore
from wled_controller.utils import get_logger
from wled_controller.config import get_config
logger = get_logger(__name__)
router = APIRouter()
def _css_to_response(source, overlay_active: bool = False) -> ColorStripSourceResponse:
"""Convert a ColorStripSource to a ColorStripSourceResponse."""
calibration = None
if isinstance(source, (PictureColorStripSource, AdvancedPictureColorStripSource)) and source.calibration:
calibration = CalibrationSchema(**calibration_to_dict(source.calibration))
"""Convert a ColorStripSource to a ColorStripSourceResponse.
# Convert raw stop dicts to ColorStop schema objects for gradient sources
Uses the source's to_dict() for type-specific fields, then applies
schema conversions for calibration and gradient stops.
"""
from wled_controller.api.schemas.color_strip_sources import ColorStop as ColorStopSchema
raw_stops = getattr(source, "stops", None)
d = source.to_dict()
# Convert calibration dict → schema object
calibration = None
raw_cal = d.pop("calibration", None)
if raw_cal and isinstance(raw_cal, dict):
calibration = CalibrationSchema(**raw_cal)
# Convert stop dicts → schema objects
raw_stops = d.pop("stops", None)
stops = None
if raw_stops is not None:
try:
@@ -66,51 +72,20 @@ def _css_to_response(source, overlay_active: bool = False) -> ColorStripSourceRe
except Exception:
stops = None
# Remove serialized timestamp strings — use actual datetime objects
d.pop("created_at", None)
d.pop("updated_at", None)
# Filter to only keys accepted by the schema (to_dict may include extra
# fields like 'fps' that aren't in the response model)
valid_fields = ColorStripSourceResponse.model_fields
filtered = {k: v for k, v in d.items() if k in valid_fields}
return ColorStripSourceResponse(
id=source.id,
name=source.name,
source_type=source.source_type,
picture_source_id=getattr(source, "picture_source_id", None),
brightness=getattr(source, "brightness", None),
saturation=getattr(source, "saturation", None),
gamma=getattr(source, "gamma", None),
smoothing=getattr(source, "smoothing", None),
interpolation_mode=getattr(source, "interpolation_mode", None),
led_count=getattr(source, "led_count", 0),
**filtered,
calibration=calibration,
color=getattr(source, "color", None),
stops=stops,
colors=getattr(source, "colors", None),
effect_type=getattr(source, "effect_type", None),
palette=getattr(source, "palette", None),
intensity=getattr(source, "intensity", None),
scale=getattr(source, "scale", None),
mirror=getattr(source, "mirror", None),
description=source.description,
clock_id=source.clock_id,
frame_interpolation=getattr(source, "frame_interpolation", None),
animation=getattr(source, "animation", None),
layers=getattr(source, "layers", None),
zones=getattr(source, "zones", None),
visualization_mode=getattr(source, "visualization_mode", None),
audio_source_id=getattr(source, "audio_source_id", None),
sensitivity=getattr(source, "sensitivity", None),
color_peak=getattr(source, "color_peak", None),
fallback_color=getattr(source, "fallback_color", None),
timeout=getattr(source, "timeout", None),
notification_effect=getattr(source, "notification_effect", None),
duration_ms=getattr(source, "duration_ms", None),
default_color=getattr(source, "default_color", None),
app_colors=getattr(source, "app_colors", None),
app_filter_mode=getattr(source, "app_filter_mode", None),
app_filter_list=getattr(source, "app_filter_list", None),
os_listener=getattr(source, "os_listener", None),
speed=getattr(source, "speed", None),
use_real_time=getattr(source, "use_real_time", None),
latitude=getattr(source, "latitude", None),
num_candles=getattr(source, "num_candles", None),
overlay_active=overlay_active,
tags=getattr(source, 'tags', []),
created_at=source.created_at,
updated_at=source.updated_at,
)
@@ -145,6 +120,27 @@ async def list_color_strip_sources(
return ColorStripSourceListResponse(sources=responses, count=len(responses))
def _extract_css_kwargs(data) -> dict:
"""Extract store-compatible kwargs from a Pydantic CSS create/update schema.
Converts nested Pydantic models (calibration, stops, layers, zones,
animation) to plain dicts/lists that the store expects.
"""
kwargs = data.model_dump(exclude_unset=False, exclude={"calibration", "stops", "layers", "zones", "animation"})
# Remove fields that don't map to store kwargs
kwargs.pop("source_type", None)
if data.calibration is not None:
kwargs["calibration"] = calibration_from_dict(data.calibration.model_dump())
else:
kwargs["calibration"] = None
kwargs["stops"] = [s.model_dump() for s in data.stops] if data.stops is not None else None
kwargs["layers"] = [l.model_dump() for l in data.layers] if data.layers is not None else None
kwargs["zones"] = [z.model_dump() for z in data.zones] if data.zones is not None else None
kwargs["animation"] = data.animation.model_dump() if data.animation else None
return kwargs
@router.post("/api/v1/color-strip-sources", response_model=ColorStripSourceResponse, tags=["Color Strip Sources"], status_code=201)
async def create_color_strip_source(
data: ColorStripSourceCreate,
@@ -153,60 +149,8 @@ async def create_color_strip_source(
):
"""Create a new color strip source."""
try:
calibration = None
if data.calibration is not None:
calibration = calibration_from_dict(data.calibration.model_dump())
stops = [s.model_dump() for s in data.stops] if data.stops is not None else None
layers = [l.model_dump() for l in data.layers] if data.layers is not None else None
zones = [z.model_dump() for z in data.zones] if data.zones is not None else None
source = store.create_source(
name=data.name,
source_type=data.source_type,
picture_source_id=data.picture_source_id,
brightness=data.brightness,
saturation=data.saturation,
gamma=data.gamma,
smoothing=data.smoothing,
interpolation_mode=data.interpolation_mode,
led_count=data.led_count,
calibration=calibration,
color=data.color,
stops=stops,
description=data.description,
frame_interpolation=data.frame_interpolation,
animation=data.animation.model_dump() if data.animation else None,
colors=data.colors,
effect_type=data.effect_type,
palette=data.palette,
intensity=data.intensity,
scale=data.scale,
mirror=data.mirror,
layers=layers,
zones=zones,
visualization_mode=data.visualization_mode,
audio_source_id=data.audio_source_id,
sensitivity=data.sensitivity,
color_peak=data.color_peak,
fallback_color=data.fallback_color,
timeout=data.timeout,
clock_id=data.clock_id,
notification_effect=data.notification_effect,
duration_ms=data.duration_ms,
default_color=data.default_color,
app_colors=data.app_colors,
app_filter_mode=data.app_filter_mode,
app_filter_list=data.app_filter_list,
os_listener=data.os_listener,
speed=data.speed,
use_real_time=data.use_real_time,
latitude=data.latitude,
num_candles=data.num_candles,
tags=data.tags,
)
kwargs = _extract_css_kwargs(data)
source = store.create_source(source_type=data.source_type, **kwargs)
fire_entity_event("color_strip_source", "created", source.id)
return _css_to_response(source)
@@ -242,60 +186,8 @@ async def update_color_strip_source(
):
"""Update a color strip source and hot-reload any running streams."""
try:
calibration = None
if data.calibration is not None:
calibration = calibration_from_dict(data.calibration.model_dump())
stops = [s.model_dump() for s in data.stops] if data.stops is not None else None
layers = [l.model_dump() for l in data.layers] if data.layers is not None else None
zones = [z.model_dump() for z in data.zones] if data.zones is not None else None
source = store.update_source(
source_id=source_id,
name=data.name,
picture_source_id=data.picture_source_id,
brightness=data.brightness,
saturation=data.saturation,
gamma=data.gamma,
smoothing=data.smoothing,
interpolation_mode=data.interpolation_mode,
led_count=data.led_count,
calibration=calibration,
color=data.color,
stops=stops,
description=data.description,
frame_interpolation=data.frame_interpolation,
animation=data.animation.model_dump() if data.animation else None,
colors=data.colors,
effect_type=data.effect_type,
palette=data.palette,
intensity=data.intensity,
scale=data.scale,
mirror=data.mirror,
layers=layers,
zones=zones,
visualization_mode=data.visualization_mode,
audio_source_id=data.audio_source_id,
sensitivity=data.sensitivity,
color_peak=data.color_peak,
fallback_color=data.fallback_color,
timeout=data.timeout,
clock_id=data.clock_id,
notification_effect=data.notification_effect,
duration_ms=data.duration_ms,
default_color=data.default_color,
app_colors=data.app_colors,
app_filter_mode=data.app_filter_mode,
app_filter_list=data.app_filter_list,
os_listener=data.os_listener,
speed=data.speed,
use_real_time=data.use_real_time,
latitude=data.latitude,
num_candles=data.num_candles,
tags=data.tags,
)
kwargs = _extract_css_kwargs(data)
source = store.update_source(source_id=source_id, **kwargs)
# Hot-reload running stream (no restart needed for in-place param changes)
try:
@@ -583,7 +475,7 @@ async def os_notification_history(_auth: AuthRequired):
if listener is None:
return {"available": False, "history": []}
return {
"available": listener._available,
"available": listener.available,
"history": listener.recent_history,
}
@@ -599,16 +491,8 @@ async def css_api_input_ws(
Auth via ?token=<api_key>. Accepts JSON frames ({"colors": [[R,G,B], ...]})
or binary frames (raw RGBRGB... bytes, 3 bytes per LED).
"""
# Authenticate
authenticated = False
cfg = get_config()
if token and cfg.auth.api_keys:
for _label, api_key in cfg.auth.api_keys.items():
if secrets.compare_digest(token, api_key):
authenticated = True
break
if not authenticated:
from wled_controller.api.auth import verify_ws_token
if not verify_ws_token(token):
await websocket.close(code=4001, reason="Unauthorized")
return
@@ -689,15 +573,8 @@ async def test_color_strip_ws(
First message is JSON metadata (source_type, led_count, calibration segments).
Subsequent messages are binary RGB frames (``led_count * 3`` bytes).
"""
# Authenticate
authenticated = False
cfg = get_config()
if token and cfg.auth.api_keys:
for _label, api_key in cfg.auth.api_keys.items():
if secrets.compare_digest(token, api_key):
authenticated = True
break
if not authenticated:
from wled_controller.api.auth import verify_ws_token
if not verify_ws_token(token):
await websocket.close(code=4001, reason="Unauthorized")
return
@@ -760,7 +637,7 @@ 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, has_brightness}, ...]
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, "has_brightness": bool(layer.get("brightness_source_id"))}
@@ -768,6 +645,10 @@ async def test_color_strip_ws(
layer_src = store.get_source(layer["source_id"])
info["name"] = layer_src.name
info["is_notification"] = isinstance(layer_src, NotificationColorStripSource)
if isinstance(layer_src, (PictureColorStripSource, AdvancedPictureColorStripSource)):
info["is_picture"] = True
if hasattr(layer_src, "calibration") and layer_src.calibration:
info["calibration_led_count"] = layer_src.calibration.get_total_leds()
except (ValueError, KeyError):
pass
layer_infos.append(info)
@@ -777,8 +658,8 @@ async def test_color_strip_ws(
# 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
if is_picture and hasattr(stream, 'live_stream'):
_frame_live = stream.live_stream
_last_aux_time = 0.0
_AUX_INTERVAL = 0.08 # send JPEG preview / brightness updates ~12 FPS