refactor: key colors targets → CSS source type, HA target improvements
Lint & Test / test (push) Successful in 1m26s
Lint & Test / test (push) Successful in 1m26s
Key Colors refactor: - New `key_colors` CSS source type with inline rectangles - KeyColorsColorStripStream: extracts N colors from screen regions - CSS editor: EntitySelect for picture source, IconSelect for color mode - Configure Regions button on card opens pattern canvas editor - Live WS preview at 5 FPS with rectangle overlay + color swatches - Removed KC target type, pattern template entity, and related API routes - Removed KC/pattern template sections from Targets tab HA light target improvements: - Update rate, transition, mappings, brightness VS now editable via PUT - Card crosslinks for HA source, CSS source, brightness VS - HA connection status icon, text metrics (Hz, uptime) - Brightness value source selector in editor
This commit is contained in:
@@ -9,10 +9,8 @@ from .routes.devices import router as devices_router
|
||||
from .routes.templates import router as templates_router
|
||||
from .routes.postprocessing import router as postprocessing_router
|
||||
from .routes.picture_sources import router as picture_sources_router
|
||||
from .routes.pattern_templates import router as pattern_templates_router
|
||||
from .routes.output_targets import router as output_targets_router
|
||||
from .routes.output_targets_control import router as output_targets_control_router
|
||||
from .routes.output_targets_keycolors import router as output_targets_keycolors_router
|
||||
from .routes.color_strip_sources import router as color_strip_sources_router
|
||||
from .routes.audio import router as audio_router
|
||||
from .routes.audio_sources import router as audio_sources_router
|
||||
@@ -36,7 +34,6 @@ router.include_router(system_settings_router)
|
||||
router.include_router(devices_router)
|
||||
router.include_router(templates_router)
|
||||
router.include_router(postprocessing_router)
|
||||
router.include_router(pattern_templates_router)
|
||||
router.include_router(picture_sources_router)
|
||||
router.include_router(color_strip_sources_router)
|
||||
router.include_router(audio_router)
|
||||
@@ -45,7 +42,6 @@ router.include_router(audio_templates_router)
|
||||
router.include_router(value_sources_router)
|
||||
router.include_router(output_targets_router)
|
||||
router.include_router(output_targets_control_router)
|
||||
router.include_router(output_targets_keycolors_router)
|
||||
router.include_router(automations_router)
|
||||
router.include_router(scene_presets_router)
|
||||
router.include_router(webhooks_router)
|
||||
|
||||
@@ -11,7 +11,6 @@ from wled_controller.storage.database import Database
|
||||
from wled_controller.storage import DeviceStore
|
||||
from wled_controller.storage.template_store import TemplateStore
|
||||
from wled_controller.storage.postprocessing_template_store import PostprocessingTemplateStore
|
||||
from wled_controller.storage.pattern_template_store import PatternTemplateStore
|
||||
from wled_controller.storage.picture_source_store import PictureSourceStore
|
||||
from wled_controller.storage.output_target_store import OutputTargetStore
|
||||
from wled_controller.storage.color_strip_store import ColorStripStore
|
||||
@@ -64,10 +63,6 @@ def get_pp_template_store() -> PostprocessingTemplateStore:
|
||||
return _get("pp_template_store", "Postprocessing template store")
|
||||
|
||||
|
||||
def get_pattern_template_store() -> PatternTemplateStore:
|
||||
return _get("pattern_template_store", "Pattern template store")
|
||||
|
||||
|
||||
def get_picture_source_store() -> PictureSourceStore:
|
||||
return _get("picture_source_store", "Picture source store")
|
||||
|
||||
@@ -188,7 +183,6 @@ def init_dependencies(
|
||||
processor_manager: ProcessorManager,
|
||||
database: Database | None = None,
|
||||
pp_template_store: PostprocessingTemplateStore | None = None,
|
||||
pattern_template_store: PatternTemplateStore | None = None,
|
||||
picture_source_store: PictureSourceStore | None = None,
|
||||
output_target_store: OutputTargetStore | None = None,
|
||||
color_strip_store: ColorStripStore | None = None,
|
||||
@@ -218,7 +212,6 @@ def init_dependencies(
|
||||
"template_store": template_store,
|
||||
"processor_manager": processor_manager,
|
||||
"pp_template_store": pp_template_store,
|
||||
"pattern_template_store": pattern_template_store,
|
||||
"picture_source_store": picture_source_store,
|
||||
"output_target_store": output_target_store,
|
||||
"color_strip_store": color_strip_store,
|
||||
|
||||
@@ -12,9 +12,12 @@ from wled_controller.api.auth import AuthRequired
|
||||
from wled_controller.api.dependencies import (
|
||||
fire_entity_event,
|
||||
get_color_strip_store,
|
||||
get_device_store,
|
||||
get_picture_source_store,
|
||||
get_output_target_store,
|
||||
get_pp_template_store,
|
||||
get_processor_manager,
|
||||
get_template_store,
|
||||
)
|
||||
from wled_controller.api.schemas.color_strip_sources import (
|
||||
ColorPushRequest,
|
||||
@@ -34,13 +37,25 @@ 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, CompositeColorStripSource, NotificationColorStripSource, PictureColorStripSource
|
||||
from wled_controller.storage.color_strip_source import (
|
||||
AdvancedPictureColorStripSource,
|
||||
ApiInputColorStripSource,
|
||||
CompositeColorStripSource,
|
||||
NotificationColorStripSource,
|
||||
PictureColorStripSource,
|
||||
)
|
||||
from wled_controller.storage import DeviceStore
|
||||
from wled_controller.storage.color_strip_store import ColorStripStore
|
||||
from wled_controller.storage.picture_source import ProcessedPictureSource, ScreenCapturePictureSource
|
||||
from wled_controller.storage.template_store import TemplateStore
|
||||
from wled_controller.storage.picture_source import (
|
||||
ProcessedPictureSource,
|
||||
ScreenCapturePictureSource,
|
||||
)
|
||||
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.storage.base_store import EntityNotFoundError
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
@@ -90,14 +105,18 @@ def _css_to_response(source, overlay_active: bool = False) -> ColorStripSourceRe
|
||||
)
|
||||
|
||||
|
||||
def _resolve_display_index(picture_source_id: str, picture_source_store: PictureSourceStore, depth: int = 0) -> int:
|
||||
def _resolve_display_index(
|
||||
picture_source_id: str, picture_source_store: PictureSourceStore, depth: int = 0
|
||||
) -> int:
|
||||
"""Resolve display index from a picture source, following processed source chains."""
|
||||
if not picture_source_id or depth > 5:
|
||||
return 0
|
||||
try:
|
||||
ps = picture_source_store.get_stream(picture_source_id)
|
||||
except Exception as e:
|
||||
logger.debug("Failed to resolve display index for picture source %s: %s", picture_source_id, e)
|
||||
logger.debug(
|
||||
"Failed to resolve display index for picture source %s: %s", picture_source_id, e
|
||||
)
|
||||
return 0
|
||||
if isinstance(ps, ScreenCapturePictureSource):
|
||||
return ps.display_index
|
||||
@@ -108,7 +127,12 @@ def _resolve_display_index(picture_source_id: str, picture_source_store: Picture
|
||||
|
||||
# ===== CRUD ENDPOINTS =====
|
||||
|
||||
@router.get("/api/v1/color-strip-sources", response_model=ColorStripSourceListResponse, tags=["Color Strip Sources"])
|
||||
|
||||
@router.get(
|
||||
"/api/v1/color-strip-sources",
|
||||
response_model=ColorStripSourceListResponse,
|
||||
tags=["Color Strip Sources"],
|
||||
)
|
||||
async def list_color_strip_sources(
|
||||
_auth: AuthRequired,
|
||||
store: ColorStripStore = Depends(get_color_strip_store),
|
||||
@@ -126,7 +150,9 @@ def _extract_css_kwargs(data) -> dict:
|
||||
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"})
|
||||
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)
|
||||
|
||||
@@ -135,13 +161,20 @@ def _extract_css_kwargs(data) -> dict:
|
||||
else:
|
||||
kwargs["calibration"] = None
|
||||
kwargs["stops"] = [s.model_dump() for s in data.stops] if data.stops is not None else None
|
||||
kwargs["layers"] = [layer.model_dump() for layer in data.layers] if data.layers is not None else None
|
||||
kwargs["layers"] = (
|
||||
[layer.model_dump() for layer 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)
|
||||
@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,
|
||||
_auth: AuthRequired,
|
||||
@@ -157,7 +190,6 @@ async def create_color_strip_source(
|
||||
except EntityNotFoundError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
@@ -165,7 +197,11 @@ async def create_color_strip_source(
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.get("/api/v1/color-strip-sources/{source_id}", response_model=ColorStripSourceResponse, tags=["Color Strip Sources"])
|
||||
@router.get(
|
||||
"/api/v1/color-strip-sources/{source_id}",
|
||||
response_model=ColorStripSourceResponse,
|
||||
tags=["Color Strip Sources"],
|
||||
)
|
||||
async def get_color_strip_source(
|
||||
source_id: str,
|
||||
_auth: AuthRequired,
|
||||
@@ -180,7 +216,11 @@ async def get_color_strip_source(
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
|
||||
@router.put("/api/v1/color-strip-sources/{source_id}", response_model=ColorStripSourceResponse, tags=["Color Strip Sources"])
|
||||
@router.put(
|
||||
"/api/v1/color-strip-sources/{source_id}",
|
||||
response_model=ColorStripSourceResponse,
|
||||
tags=["Color Strip Sources"],
|
||||
)
|
||||
async def update_color_strip_source(
|
||||
source_id: str,
|
||||
data: ColorStripSourceUpdate,
|
||||
@@ -209,7 +249,9 @@ async def update_color_strip_source(
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.delete("/api/v1/color-strip-sources/{source_id}", status_code=204, tags=["Color Strip Sources"])
|
||||
@router.delete(
|
||||
"/api/v1/color-strip-sources/{source_id}", status_code=204, tags=["Color Strip Sources"]
|
||||
)
|
||||
async def delete_color_strip_source(
|
||||
source_id: str,
|
||||
_auth: AuthRequired,
|
||||
@@ -224,7 +266,7 @@ async def delete_color_strip_source(
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Color strip source is referenced by target(s): {names}. "
|
||||
"Delete or reassign the target(s) first.",
|
||||
"Delete or reassign the target(s) first.",
|
||||
)
|
||||
composite_names = store.get_composites_referencing(source_id)
|
||||
if composite_names:
|
||||
@@ -232,7 +274,7 @@ async def delete_color_strip_source(
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Color strip source is used as a layer in composite source(s): {names}. "
|
||||
"Remove it from the composite(s) first.",
|
||||
"Remove it from the composite(s) first.",
|
||||
)
|
||||
mapped_names = store.get_mapped_referencing(source_id)
|
||||
if mapped_names:
|
||||
@@ -240,7 +282,7 @@ async def delete_color_strip_source(
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Color strip source is used as a zone in mapped source(s): {names}. "
|
||||
"Remove it from the mapped source(s) first.",
|
||||
"Remove it from the mapped source(s) first.",
|
||||
)
|
||||
processed_names = store.get_processed_referencing(source_id)
|
||||
if processed_names:
|
||||
@@ -248,7 +290,7 @@ async def delete_color_strip_source(
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Color strip source is used as input in processed source(s): {names}. "
|
||||
"Delete or reassign the processed source(s) first.",
|
||||
"Delete or reassign the processed source(s) first.",
|
||||
)
|
||||
store.delete_source(source_id)
|
||||
fire_entity_event("color_strip_source", "deleted", source_id)
|
||||
@@ -261,8 +303,360 @@ async def delete_color_strip_source(
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
# ===== KEY COLORS TEST =====
|
||||
|
||||
|
||||
@router.post(
|
||||
"/api/v1/color-strip-sources/{source_id}/key-colors/test",
|
||||
tags=["Color Strip Sources"],
|
||||
)
|
||||
async def test_key_colors_source(
|
||||
source_id: str,
|
||||
_auth: AuthRequired,
|
||||
store: ColorStripStore = Depends(get_color_strip_store),
|
||||
source_store: PictureSourceStore = Depends(get_picture_source_store),
|
||||
template_store: TemplateStore = Depends(get_template_store),
|
||||
processor_manager: ProcessorManager = Depends(get_processor_manager),
|
||||
device_store: DeviceStore = Depends(get_device_store),
|
||||
pp_template_store=Depends(get_pp_template_store),
|
||||
):
|
||||
"""Test a key_colors source: capture a frame, extract colors from each rectangle."""
|
||||
from wled_controller.storage.color_strip_source import KeyColorsColorStripSource
|
||||
from wled_controller.core.capture.screen_capture import (
|
||||
calculate_average_color,
|
||||
calculate_dominant_color,
|
||||
calculate_median_color,
|
||||
)
|
||||
from wled_controller.core.capture_engines import EngineRegistry
|
||||
from wled_controller.core.filters import FilterRegistry, ImagePool
|
||||
from wled_controller.storage.picture_source import (
|
||||
ScreenCapturePictureSource,
|
||||
StaticImagePictureSource,
|
||||
)
|
||||
from wled_controller.utils.image_codec import encode_jpeg_data_uri
|
||||
|
||||
stream = None
|
||||
try:
|
||||
source = store.get_source(source_id)
|
||||
if not isinstance(source, KeyColorsColorStripSource):
|
||||
raise HTTPException(status_code=400, detail="Source is not a key_colors type")
|
||||
|
||||
if not source.rectangles:
|
||||
raise HTTPException(status_code=400, detail="No screen regions configured")
|
||||
if not source.picture_source_id:
|
||||
raise HTTPException(status_code=400, detail="No picture source configured")
|
||||
|
||||
# Resolve picture source and capture a frame
|
||||
chain = source_store.resolve_stream_chain(source.picture_source_id)
|
||||
raw_stream = chain["raw_stream"]
|
||||
|
||||
from wled_controller.utils.image_codec import load_image_file
|
||||
|
||||
if isinstance(raw_stream, StaticImagePictureSource):
|
||||
from wled_controller.api.dependencies import get_asset_store as _get_asset_store
|
||||
|
||||
asset_store = _get_asset_store()
|
||||
image_path = (
|
||||
asset_store.get_file_path(raw_stream.image_asset_id)
|
||||
if raw_stream.image_asset_id
|
||||
else None
|
||||
)
|
||||
if not image_path:
|
||||
raise HTTPException(status_code=400, detail="Image asset not found")
|
||||
image = load_image_file(image_path)
|
||||
elif isinstance(raw_stream, ScreenCapturePictureSource):
|
||||
capture_template = template_store.get_template(raw_stream.capture_template_id)
|
||||
display_index = raw_stream.display_index
|
||||
|
||||
if capture_template.engine_type not in EngineRegistry.get_available_engines():
|
||||
raise HTTPException(
|
||||
status_code=400, detail=f"Engine '{capture_template.engine_type}' not available"
|
||||
)
|
||||
|
||||
locked = processor_manager.get_display_lock_info(display_index)
|
||||
if locked:
|
||||
try:
|
||||
device_name = device_store.get_device(locked).name
|
||||
except Exception:
|
||||
device_name = locked
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Display {display_index} is captured by '{device_name}'. Stop it first.",
|
||||
)
|
||||
|
||||
stream = EngineRegistry.create_stream(
|
||||
capture_template.engine_type, display_index, capture_template.engine_config
|
||||
)
|
||||
stream.initialize()
|
||||
sc = stream.capture_frame()
|
||||
if sc is None:
|
||||
raise RuntimeError("No frame captured")
|
||||
image = sc.image
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="Unsupported picture source type")
|
||||
|
||||
# Apply postprocessing filters
|
||||
pp_ids = chain.get("postprocessing_template_ids", [])
|
||||
if pp_ids and pp_template_store:
|
||||
pool = ImagePool()
|
||||
for pp_id in pp_ids:
|
||||
try:
|
||||
pp = pp_template_store.get_template(pp_id)
|
||||
for fi in pp_template_store.resolve_filter_instances(pp.filters):
|
||||
f = FilterRegistry.create_instance(fi.filter_id, fi.options)
|
||||
result = f.process_image(image, pool)
|
||||
if result is not None:
|
||||
image = result
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Extract colors from each rectangle
|
||||
h, w = image.shape[:2]
|
||||
calc_fns = {
|
||||
"average": calculate_average_color,
|
||||
"median": calculate_median_color,
|
||||
"dominant": calculate_dominant_color,
|
||||
}
|
||||
calc_fn = calc_fns.get(source.interpolation_mode, calculate_average_color)
|
||||
|
||||
result_rects = []
|
||||
for rect in source.rectangles:
|
||||
px_x = max(0, int(rect.x * w))
|
||||
px_y = max(0, int(rect.y * h))
|
||||
px_w = max(1, int(rect.width * w))
|
||||
px_h = max(1, int(rect.height * h))
|
||||
px_x, px_y = min(px_x, w - 1), min(px_y, h - 1)
|
||||
px_w, px_h = min(px_w, w - px_x), min(px_h, h - px_y)
|
||||
sub_img = image[px_y : px_y + px_h, px_x : px_x + px_w]
|
||||
r, g, b = calc_fn(sub_img)
|
||||
result_rects.append(
|
||||
{
|
||||
"name": rect.name,
|
||||
"x": rect.x,
|
||||
"y": rect.y,
|
||||
"width": rect.width,
|
||||
"height": rect.height,
|
||||
"color": {
|
||||
"r": int(r),
|
||||
"g": int(g),
|
||||
"b": int(b),
|
||||
"hex": f"#{int(r):02x}{int(g):02x}{int(b):02x}",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
image_data_uri = encode_jpeg_data_uri(image, quality=90)
|
||||
|
||||
return {
|
||||
"image": image_data_uri,
|
||||
"rectangles": result_rects,
|
||||
"interpolation_mode": source.interpolation_mode,
|
||||
}
|
||||
except HTTPException:
|
||||
raise
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error("Key colors test failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
finally:
|
||||
if stream:
|
||||
try:
|
||||
stream.stop()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@router.websocket("/api/v1/color-strip-sources/{source_id}/key-colors/test/ws")
|
||||
async def test_key_colors_ws(
|
||||
websocket: WebSocket,
|
||||
source_id: str,
|
||||
token: str = Query(""),
|
||||
fps: int = Query(3),
|
||||
preview_width: int = Query(480),
|
||||
):
|
||||
"""WebSocket for real-time key_colors test preview with frame + rectangle overlay."""
|
||||
import json as ws_json
|
||||
import time as ws_time
|
||||
from wled_controller.api.auth import verify_ws_token
|
||||
from wled_controller.storage.color_strip_source import KeyColorsColorStripSource
|
||||
from wled_controller.core.capture.screen_capture import (
|
||||
calculate_average_color,
|
||||
calculate_dominant_color,
|
||||
calculate_median_color,
|
||||
)
|
||||
from wled_controller.core.filters import FilterRegistry, ImagePool
|
||||
from wled_controller.storage.picture_source import ScreenCapturePictureSource
|
||||
from wled_controller.utils.image_codec import encode_jpeg_data_uri, resize_down
|
||||
|
||||
if not verify_ws_token(token):
|
||||
await websocket.close(code=4001, reason="Unauthorized")
|
||||
return
|
||||
|
||||
store = get_color_strip_store()
|
||||
source_store = get_picture_source_store()
|
||||
manager = get_processor_manager()
|
||||
device_store = get_device_store()
|
||||
pp_store = get_pp_template_store()
|
||||
|
||||
try:
|
||||
source = store.get_source(source_id)
|
||||
except ValueError as e:
|
||||
await websocket.close(code=4004, reason=str(e))
|
||||
return
|
||||
|
||||
if not isinstance(source, KeyColorsColorStripSource):
|
||||
await websocket.close(code=4003, reason="Not a key_colors source")
|
||||
return
|
||||
if not source.rectangles:
|
||||
await websocket.close(code=4003, reason="No regions configured")
|
||||
return
|
||||
if not source.picture_source_id:
|
||||
await websocket.close(code=4003, reason="No picture source configured")
|
||||
return
|
||||
|
||||
try:
|
||||
chain = source_store.resolve_stream_chain(source.picture_source_id)
|
||||
except ValueError as e:
|
||||
await websocket.close(code=4003, reason=str(e))
|
||||
return
|
||||
|
||||
raw_stream = chain["raw_stream"]
|
||||
if isinstance(raw_stream, ScreenCapturePictureSource):
|
||||
locked = manager.get_display_lock_info(raw_stream.display_index)
|
||||
if locked:
|
||||
try:
|
||||
name = device_store.get_device(locked).name
|
||||
except Exception:
|
||||
name = locked
|
||||
await websocket.close(code=4003, reason=f"Display captured by '{name}'")
|
||||
return
|
||||
|
||||
fps = max(1, min(30, fps))
|
||||
preview_width = max(120, min(1920, preview_width))
|
||||
frame_interval = 1.0 / fps
|
||||
|
||||
calc_fns = {
|
||||
"average": calculate_average_color,
|
||||
"median": calculate_median_color,
|
||||
"dominant": calculate_dominant_color,
|
||||
}
|
||||
calc_fn = calc_fns.get(source.interpolation_mode, calculate_average_color)
|
||||
|
||||
await websocket.accept()
|
||||
logger.info(f"KC CSS test WS connected for {source_id} (fps={fps})")
|
||||
|
||||
live_stream_mgr = manager._live_stream_manager
|
||||
live_stream = None
|
||||
|
||||
try:
|
||||
live_stream = await asyncio.to_thread(live_stream_mgr.acquire, source.picture_source_id)
|
||||
prev_frame_ref = None
|
||||
|
||||
while True:
|
||||
loop_start = ws_time.monotonic()
|
||||
try:
|
||||
capture = await asyncio.to_thread(live_stream.get_latest_frame)
|
||||
if capture is None or capture.image is None:
|
||||
await asyncio.sleep(frame_interval)
|
||||
continue
|
||||
if capture is prev_frame_ref:
|
||||
await asyncio.sleep(frame_interval * 0.5)
|
||||
continue
|
||||
prev_frame_ref = capture
|
||||
|
||||
cur_image = capture.image
|
||||
if not isinstance(cur_image, np.ndarray):
|
||||
await asyncio.sleep(frame_interval)
|
||||
continue
|
||||
|
||||
# Apply postprocessing
|
||||
pp_ids = chain.get("postprocessing_template_ids", [])
|
||||
if pp_ids and pp_store:
|
||||
pool = ImagePool()
|
||||
for pp_id in pp_ids:
|
||||
try:
|
||||
pp = pp_store.get_template(pp_id)
|
||||
for fi in pp_store.resolve_filter_instances(pp.filters):
|
||||
f = FilterRegistry.create_instance(fi.filter_id, fi.options)
|
||||
result = f.process_image(cur_image, pool)
|
||||
if result is not None:
|
||||
cur_image = result
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Re-read source for hot-update support
|
||||
try:
|
||||
source = store.get_source(source_id)
|
||||
calc_fn = calc_fns.get(source.interpolation_mode, calculate_average_color)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
h, w = cur_image.shape[:2]
|
||||
result_rects = []
|
||||
for rect in source.rectangles:
|
||||
px_x = max(0, int(rect.x * w))
|
||||
px_y = max(0, int(rect.y * h))
|
||||
px_w = max(1, int(rect.width * w))
|
||||
px_h = max(1, int(rect.height * h))
|
||||
px_x, px_y = min(px_x, w - 1), min(px_y, h - 1)
|
||||
px_w, px_h = min(px_w, w - px_x), min(px_h, h - px_y)
|
||||
sub = cur_image[px_y : px_y + px_h, px_x : px_x + px_w]
|
||||
r, g, b = calc_fn(sub)
|
||||
result_rects.append(
|
||||
{
|
||||
"name": rect.name,
|
||||
"x": rect.x,
|
||||
"y": rect.y,
|
||||
"width": rect.width,
|
||||
"height": rect.height,
|
||||
"color": {
|
||||
"r": int(r),
|
||||
"g": int(g),
|
||||
"b": int(b),
|
||||
"hex": f"#{int(r):02x}{int(g):02x}{int(b):02x}",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
frame_to_encode = resize_down(cur_image, preview_width)
|
||||
frame_uri = encode_jpeg_data_uri(frame_to_encode, quality=85)
|
||||
|
||||
await websocket.send_text(
|
||||
ws_json.dumps(
|
||||
{
|
||||
"type": "frame",
|
||||
"image": frame_uri,
|
||||
"rectangles": result_rects,
|
||||
"interpolation_mode": source.interpolation_mode,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
except (WebSocketDisconnect, Exception) as inner_e:
|
||||
if isinstance(inner_e, WebSocketDisconnect):
|
||||
raise
|
||||
logger.warning(f"KC CSS test WS frame error: {inner_e}")
|
||||
|
||||
elapsed = ws_time.monotonic() - loop_start
|
||||
if frame_interval - elapsed > 0:
|
||||
await asyncio.sleep(frame_interval - elapsed)
|
||||
|
||||
except WebSocketDisconnect:
|
||||
logger.info(f"KC CSS test WS disconnected for {source_id}")
|
||||
except Exception as e:
|
||||
logger.error(f"KC CSS test WS error: {e}", exc_info=True)
|
||||
finally:
|
||||
if live_stream is not None:
|
||||
try:
|
||||
await asyncio.to_thread(live_stream_mgr.release, source.picture_source_id)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ===== CALIBRATION TEST =====
|
||||
|
||||
|
||||
@router.put(
|
||||
"/api/v1/color-strip-sources/{source_id}/calibration/test",
|
||||
response_model=CalibrationTestModeResponse,
|
||||
@@ -291,7 +685,7 @@ async def test_css_calibration(
|
||||
if edge_name not in valid_edges:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Invalid edge '{edge_name}'. Must be one of: {', '.join(sorted(valid_edges))}"
|
||||
detail=f"Invalid edge '{edge_name}'. Must be one of: {', '.join(sorted(valid_edges))}",
|
||||
)
|
||||
if len(color) != 3 or not all(0 <= c <= 255 for c in color):
|
||||
raise HTTPException(
|
||||
@@ -304,7 +698,9 @@ async def test_css_calibration(
|
||||
if body.edges:
|
||||
try:
|
||||
source = store.get_source(source_id)
|
||||
if not isinstance(source, (PictureColorStripSource, AdvancedPictureColorStripSource)):
|
||||
if not isinstance(
|
||||
source, (PictureColorStripSource, AdvancedPictureColorStripSource)
|
||||
):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Calibration test is only available for picture color strip sources",
|
||||
@@ -339,6 +735,7 @@ async def test_css_calibration(
|
||||
|
||||
# ===== OVERLAY VISUALIZATION =====
|
||||
|
||||
|
||||
@router.post("/api/v1/color-strip-sources/{source_id}/overlay/start", tags=["Color Strip Sources"])
|
||||
async def start_css_overlay(
|
||||
source_id: str,
|
||||
@@ -351,9 +748,13 @@ async def start_css_overlay(
|
||||
try:
|
||||
source = store.get_source(source_id)
|
||||
if not isinstance(source, (PictureColorStripSource, AdvancedPictureColorStripSource)):
|
||||
raise HTTPException(status_code=400, detail="Overlay is only supported for picture color strip sources")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Overlay is only supported for picture color strip sources"
|
||||
)
|
||||
if not source.calibration:
|
||||
raise HTTPException(status_code=400, detail="Color strip source has no calibration configured")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Color strip source has no calibration configured"
|
||||
)
|
||||
|
||||
ps_id = getattr(source, "picture_source_id", "") or ""
|
||||
display_index = _resolve_display_index(ps_id, picture_source_store)
|
||||
@@ -404,6 +805,7 @@ async def get_css_overlay_status(
|
||||
|
||||
# ===== API INPUT: COLOR PUSH =====
|
||||
|
||||
|
||||
@router.post("/api/v1/color-strip-sources/{source_id}/colors", tags=["Color Strip Sources"])
|
||||
async def push_colors(
|
||||
source_id: str,
|
||||
@@ -442,7 +844,9 @@ async def push_colors(
|
||||
# Legacy flat colors path
|
||||
colors_array = np.array(body.colors, dtype=np.uint8)
|
||||
if colors_array.ndim != 2 or colors_array.shape[1] != 3:
|
||||
raise HTTPException(status_code=400, detail="Colors must be an array of [R,G,B] triplets")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Colors must be an array of [R,G,B] triplets"
|
||||
)
|
||||
for stream in streams:
|
||||
if hasattr(stream, "push_colors"):
|
||||
stream.push_colors(colors_array)
|
||||
@@ -495,7 +899,10 @@ async def notify_source(
|
||||
@router.get("/api/v1/color-strip-sources/os-notifications/history", tags=["Color Strip Sources"])
|
||||
async def os_notification_history(_auth: AuthRequired):
|
||||
"""Return recent OS notification capture history (newest first)."""
|
||||
from wled_controller.core.processing.os_notification_listener import get_os_notification_listener
|
||||
from wled_controller.core.processing.os_notification_listener import (
|
||||
get_os_notification_listener,
|
||||
)
|
||||
|
||||
listener = get_os_notification_listener()
|
||||
if listener is None:
|
||||
return {"available": False, "history": []}
|
||||
@@ -507,7 +914,15 @@ async def os_notification_history(_auth: AuthRequired):
|
||||
|
||||
# ── Transient Preview WebSocket ────────────────────────────────────────
|
||||
|
||||
_PREVIEW_ALLOWED_TYPES = {"static", "gradient", "color_cycle", "effect", "daylight", "candlelight", "notification"}
|
||||
_PREVIEW_ALLOWED_TYPES = {
|
||||
"static",
|
||||
"gradient",
|
||||
"color_cycle",
|
||||
"effect",
|
||||
"daylight",
|
||||
"candlelight",
|
||||
"notification",
|
||||
}
|
||||
|
||||
|
||||
@router.websocket("/api/v1/color-strip-sources/preview/ws")
|
||||
@@ -529,6 +944,7 @@ async def preview_color_strip_ws(
|
||||
changed the old stream is replaced; otherwise ``update_source()`` is used.
|
||||
"""
|
||||
from wled_controller.api.auth import verify_ws_token
|
||||
|
||||
if not verify_ws_token(token):
|
||||
await websocket.close(code=4001, reason="Unauthorized")
|
||||
return
|
||||
@@ -557,6 +973,7 @@ async def preview_color_strip_ws(
|
||||
def _build_source(config: dict):
|
||||
"""Build a ColorStripSource from a raw config dict, injecting synthetic id/name."""
|
||||
from wled_controller.storage.color_strip_source import ColorStripSource
|
||||
|
||||
config.setdefault("id", "__preview__")
|
||||
config.setdefault("name", "__preview__")
|
||||
return ColorStripSource.from_dict(config)
|
||||
@@ -564,6 +981,7 @@ async def preview_color_strip_ws(
|
||||
def _create_stream(source):
|
||||
"""Instantiate and start the appropriate stream class for *source*."""
|
||||
from wled_controller.core.processing.color_strip_stream_manager import _SIMPLE_STREAM_MAP
|
||||
|
||||
stream_cls = _SIMPLE_STREAM_MAP.get(source.source_type)
|
||||
if not stream_cls:
|
||||
raise ValueError(f"Unsupported preview source_type: {source.source_type}")
|
||||
@@ -572,6 +990,7 @@ async def preview_color_strip_ws(
|
||||
if hasattr(s, "set_gradient_store"):
|
||||
try:
|
||||
from wled_controller.api.dependencies import get_gradient_store
|
||||
|
||||
s.set_gradient_store(get_gradient_store())
|
||||
except Exception:
|
||||
pass
|
||||
@@ -626,7 +1045,14 @@ async def preview_color_strip_ws(
|
||||
config = _json.loads(initial_text)
|
||||
source_type = config.get("source_type")
|
||||
if source_type not in _PREVIEW_ALLOWED_TYPES:
|
||||
await websocket.send_text(_json.dumps({"type": "error", "detail": f"source_type must be one of {sorted(_PREVIEW_ALLOWED_TYPES)}"}))
|
||||
await websocket.send_text(
|
||||
_json.dumps(
|
||||
{
|
||||
"type": "error",
|
||||
"detail": f"source_type must be one of {sorted(_PREVIEW_ALLOWED_TYPES)}",
|
||||
}
|
||||
)
|
||||
)
|
||||
await websocket.close(code=4003, reason="Invalid source_type")
|
||||
return
|
||||
source = _build_source(config)
|
||||
@@ -639,7 +1065,9 @@ async def preview_color_strip_ws(
|
||||
return
|
||||
|
||||
await _send_meta(current_source_type)
|
||||
logger.info(f"Preview WS connected: source_type={current_source_type}, led_count={led_count}, fps={fps}")
|
||||
logger.info(
|
||||
f"Preview WS connected: source_type={current_source_type}, led_count={led_count}, fps={fps}"
|
||||
)
|
||||
|
||||
# Frame loop ─────────────────────────────────────────────────────────
|
||||
|
||||
@@ -659,7 +1087,10 @@ async def preview_color_strip_ws(
|
||||
|
||||
# Handle "fire" command for notification streams
|
||||
if new_config.get("action") == "fire":
|
||||
from wled_controller.core.processing.notification_stream import NotificationColorStripStream
|
||||
from wled_controller.core.processing.notification_stream import (
|
||||
NotificationColorStripStream,
|
||||
)
|
||||
|
||||
if isinstance(stream, NotificationColorStripStream):
|
||||
stream.fire(
|
||||
app_name=new_config.get("app", ""),
|
||||
@@ -669,7 +1100,14 @@ async def preview_color_strip_ws(
|
||||
|
||||
new_type = new_config.get("source_type")
|
||||
if new_type not in _PREVIEW_ALLOWED_TYPES:
|
||||
await websocket.send_text(_json.dumps({"type": "error", "detail": f"source_type must be one of {sorted(_PREVIEW_ALLOWED_TYPES)}"}))
|
||||
await websocket.send_text(
|
||||
_json.dumps(
|
||||
{
|
||||
"type": "error",
|
||||
"detail": f"source_type must be one of {sorted(_PREVIEW_ALLOWED_TYPES)}",
|
||||
}
|
||||
)
|
||||
)
|
||||
continue
|
||||
new_source = _build_source(new_config)
|
||||
if new_type != current_source_type:
|
||||
@@ -692,7 +1130,7 @@ async def preview_color_strip_ws(
|
||||
await websocket.send_bytes(colors.tobytes())
|
||||
else:
|
||||
# Stream hasn't produced a frame yet — send black
|
||||
await websocket.send_bytes(b'\x00' * led_count * 3)
|
||||
await websocket.send_bytes(b"\x00" * led_count * 3)
|
||||
|
||||
except WebSocketDisconnect:
|
||||
pass
|
||||
@@ -716,6 +1154,7 @@ async def css_api_input_ws(
|
||||
or binary frames (raw RGBRGB... bytes, 3 bytes per LED).
|
||||
"""
|
||||
from wled_controller.api.auth import verify_ws_token
|
||||
|
||||
if not verify_ws_token(token):
|
||||
await websocket.close(code=4001, reason="Unauthorized")
|
||||
return
|
||||
@@ -746,6 +1185,7 @@ async def css_api_input_ws(
|
||||
if "text" in message:
|
||||
# JSON frame: {"colors": [[R,G,B], ...]} or {"segments": [...]}
|
||||
import json
|
||||
|
||||
try:
|
||||
data = json.loads(message["text"])
|
||||
except (json.JSONDecodeError, ValueError) as e:
|
||||
@@ -756,6 +1196,7 @@ async def css_api_input_ws(
|
||||
# Segment-based path — validate and push
|
||||
try:
|
||||
from wled_controller.api.schemas.color_strip_sources import SegmentPayload
|
||||
|
||||
seg_dicts = [SegmentPayload(**s).model_dump() for s in data["segments"]]
|
||||
except Exception as e:
|
||||
await websocket.send_json({"error": f"Invalid segment: {e}"})
|
||||
@@ -777,7 +1218,9 @@ async def css_api_input_ws(
|
||||
await websocket.send_json({"error": str(e)})
|
||||
continue
|
||||
else:
|
||||
await websocket.send_json({"error": "JSON frame must contain 'colors' or 'segments'"})
|
||||
await websocket.send_json(
|
||||
{"error": "JSON frame must contain 'colors' or 'segments'"}
|
||||
)
|
||||
continue
|
||||
|
||||
elif "bytes" in message:
|
||||
@@ -807,6 +1250,7 @@ async def css_api_input_ws(
|
||||
|
||||
# ── Test / Preview WebSocket ──────────────────────────────────────────
|
||||
|
||||
|
||||
@router.websocket("/api/v1/color-strip-sources/{source_id}/test/ws")
|
||||
async def test_color_strip_ws(
|
||||
websocket: WebSocket,
|
||||
@@ -821,6 +1265,7 @@ async def test_color_strip_ws(
|
||||
Subsequent messages are binary RGB frames (``led_count * 3`` bytes).
|
||||
"""
|
||||
from wled_controller.api.auth import verify_ws_token
|
||||
|
||||
if not verify_ws_token(token):
|
||||
await websocket.close(code=4001, reason="Unauthorized")
|
||||
return
|
||||
@@ -868,6 +1313,7 @@ async def test_color_strip_ws(
|
||||
from wled_controller.core.processing.composite_stream import CompositeColorStripStream
|
||||
|
||||
from wled_controller.core.processing.api_input_stream import ApiInputColorStripStream
|
||||
|
||||
is_api_input = isinstance(stream, ApiInputColorStripStream)
|
||||
_last_push_gen = 0 # track api_input push generation to skip unchanged frames
|
||||
|
||||
@@ -900,13 +1346,19 @@ async def test_color_strip_ws(
|
||||
enabled_layers = [layer for layer in source.layers if layer.get("enabled", True)]
|
||||
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"))}
|
||||
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
|
||||
info["is_notification"] = isinstance(layer_src, NotificationColorStripSource)
|
||||
if isinstance(layer_src, (PictureColorStripSource, AdvancedPictureColorStripSource)):
|
||||
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()
|
||||
@@ -927,7 +1379,7 @@ 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'):
|
||||
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
|
||||
@@ -943,15 +1395,17 @@ async def test_color_strip_ws(
|
||||
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])
|
||||
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(b"\x00" * rgb_size)
|
||||
parts.append(composite_colors.tobytes())
|
||||
await websocket.send_bytes(b''.join(parts))
|
||||
await websocket.send_bytes(b"".join(parts))
|
||||
elif composite_colors is not None:
|
||||
await websocket.send_bytes(composite_colors.tobytes())
|
||||
else:
|
||||
@@ -978,9 +1432,12 @@ async def test_color_strip_ws(
|
||||
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
|
||||
]}
|
||||
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
|
||||
@@ -992,6 +1449,7 @@ async def test_color_strip_ws(
|
||||
if frame is not None and frame.image is not None:
|
||||
from wled_controller.utils.image_codec import encode_jpeg
|
||||
import cv2 as _cv2
|
||||
|
||||
img = frame.image
|
||||
# Ensure 3-channel RGB (some engines may produce BGRA)
|
||||
if img.ndim == 3 and img.shape[2] == 4:
|
||||
@@ -1000,19 +1458,25 @@ async def test_color_strip_ws(
|
||||
# Send frame dimensions once so client can compute border overlay
|
||||
if not _frame_dims_sent:
|
||||
_frame_dims_sent = True
|
||||
await websocket.send_text(_json.dumps({
|
||||
"type": "frame_dims",
|
||||
"width": w,
|
||||
"height": h,
|
||||
}))
|
||||
await websocket.send_text(
|
||||
_json.dumps(
|
||||
{
|
||||
"type": "frame_dims",
|
||||
"width": w,
|
||||
"height": h,
|
||||
}
|
||||
)
|
||||
)
|
||||
# Downscale for bandwidth
|
||||
scale = min(960 / w, 540 / h, 1.0)
|
||||
if scale < 1.0:
|
||||
new_w = max(1, int(w * scale))
|
||||
new_h = max(1, int(h * scale))
|
||||
img = _cv2.resize(img, (new_w, new_h), interpolation=_cv2.INTER_AREA)
|
||||
img = _cv2.resize(
|
||||
img, (new_w, new_h), interpolation=_cv2.INTER_AREA
|
||||
)
|
||||
# Wire format: [0xFD] [jpeg_bytes]
|
||||
await websocket.send_bytes(b'\xfd' + encode_jpeg(img, quality=70))
|
||||
await websocket.send_bytes(b"\xfd" + encode_jpeg(img, quality=70))
|
||||
except Exception as e:
|
||||
logger.warning(f"JPEG frame preview error: {e}")
|
||||
|
||||
|
||||
@@ -12,7 +12,6 @@ from wled_controller.api.dependencies import (
|
||||
get_processor_manager,
|
||||
)
|
||||
from wled_controller.api.schemas.output_targets import (
|
||||
KeyColorsSettingsSchema,
|
||||
OutputTargetCreate,
|
||||
OutputTargetListResponse,
|
||||
OutputTargetResponse,
|
||||
@@ -21,10 +20,6 @@ from wled_controller.api.schemas.output_targets import (
|
||||
from wled_controller.core.processing.processor_manager import ProcessorManager
|
||||
from wled_controller.storage import DeviceStore
|
||||
from wled_controller.storage.wled_output_target import WledOutputTarget
|
||||
from wled_controller.storage.key_colors_output_target import (
|
||||
KeyColorsSettings,
|
||||
KeyColorsOutputTarget,
|
||||
)
|
||||
from wled_controller.storage.ha_light_output_target import (
|
||||
HALightMapping,
|
||||
HALightOutputTarget,
|
||||
@@ -39,30 +34,6 @@ logger = get_logger(__name__)
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _kc_settings_to_schema(settings: KeyColorsSettings) -> KeyColorsSettingsSchema:
|
||||
"""Convert core KeyColorsSettings to schema."""
|
||||
return KeyColorsSettingsSchema(
|
||||
fps=settings.fps,
|
||||
interpolation_mode=settings.interpolation_mode,
|
||||
smoothing=settings.smoothing,
|
||||
pattern_template_id=settings.pattern_template_id,
|
||||
brightness=settings.brightness,
|
||||
brightness_value_source_id=settings.brightness_value_source_id,
|
||||
)
|
||||
|
||||
|
||||
def _kc_schema_to_settings(schema: KeyColorsSettingsSchema) -> KeyColorsSettings:
|
||||
"""Convert schema KeyColorsSettings to core."""
|
||||
return KeyColorsSettings(
|
||||
fps=schema.fps,
|
||||
interpolation_mode=schema.interpolation_mode,
|
||||
smoothing=schema.smoothing,
|
||||
pattern_template_id=schema.pattern_template_id,
|
||||
brightness=schema.brightness,
|
||||
brightness_value_source_id=schema.brightness_value_source_id,
|
||||
)
|
||||
|
||||
|
||||
def _target_to_response(target) -> OutputTargetResponse:
|
||||
"""Convert an OutputTarget to OutputTargetResponse."""
|
||||
if isinstance(target, WledOutputTarget):
|
||||
@@ -84,18 +55,6 @@ def _target_to_response(target) -> OutputTargetResponse:
|
||||
created_at=target.created_at,
|
||||
updated_at=target.updated_at,
|
||||
)
|
||||
elif isinstance(target, KeyColorsOutputTarget):
|
||||
return OutputTargetResponse(
|
||||
id=target.id,
|
||||
name=target.name,
|
||||
target_type=target.target_type,
|
||||
picture_source_id=target.picture_source_id,
|
||||
key_colors_settings=_kc_settings_to_schema(target.settings),
|
||||
description=target.description,
|
||||
tags=target.tags,
|
||||
created_at=target.created_at,
|
||||
updated_at=target.updated_at,
|
||||
)
|
||||
elif isinstance(target, HALightOutputTarget):
|
||||
return OutputTargetResponse(
|
||||
id=target.id,
|
||||
@@ -155,9 +114,6 @@ async def create_target(
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=422, detail=f"Device {data.device_id} not found")
|
||||
|
||||
kc_settings = (
|
||||
_kc_schema_to_settings(data.key_colors_settings) if data.key_colors_settings else None
|
||||
)
|
||||
ha_mappings = (
|
||||
[
|
||||
HALightMapping(
|
||||
@@ -185,8 +141,6 @@ async def create_target(
|
||||
min_brightness_threshold=data.min_brightness_threshold,
|
||||
adaptive_fps=data.adaptive_fps,
|
||||
protocol=data.protocol,
|
||||
picture_source_id=data.picture_source_id,
|
||||
key_colors_settings=kc_settings,
|
||||
description=data.description,
|
||||
tags=data.tags,
|
||||
ha_source_id=data.ha_source_id,
|
||||
@@ -282,31 +236,18 @@ async def update_target(
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=422, detail=f"Device {data.device_id} not found")
|
||||
|
||||
# Build KC settings with partial-update support: only apply fields that were
|
||||
# explicitly provided in the request body, merging with the existing settings.
|
||||
kc_settings = None
|
||||
if data.key_colors_settings is not None:
|
||||
incoming = data.key_colors_settings.model_dump(exclude_unset=True)
|
||||
try:
|
||||
existing_target = target_store.get_target(target_id)
|
||||
except ValueError:
|
||||
existing_target = None
|
||||
|
||||
if isinstance(existing_target, KeyColorsOutputTarget):
|
||||
ex = existing_target.settings
|
||||
merged = KeyColorsSettingsSchema(
|
||||
fps=incoming.get("fps", ex.fps),
|
||||
interpolation_mode=incoming.get("interpolation_mode", ex.interpolation_mode),
|
||||
smoothing=incoming.get("smoothing", ex.smoothing),
|
||||
pattern_template_id=incoming.get("pattern_template_id", ex.pattern_template_id),
|
||||
brightness=incoming.get("brightness", ex.brightness),
|
||||
brightness_value_source_id=incoming.get(
|
||||
"brightness_value_source_id", ex.brightness_value_source_id
|
||||
),
|
||||
# Build HA light mappings if provided
|
||||
ha_mappings = None
|
||||
if data.ha_light_mappings is not None:
|
||||
ha_mappings = [
|
||||
HALightMapping(
|
||||
entity_id=m.entity_id,
|
||||
led_start=m.led_start,
|
||||
led_end=m.led_end,
|
||||
brightness_scale=m.brightness_scale,
|
||||
)
|
||||
kc_settings = _kc_schema_to_settings(merged)
|
||||
else:
|
||||
kc_settings = _kc_schema_to_settings(data.key_colors_settings)
|
||||
for m in data.ha_light_mappings
|
||||
]
|
||||
|
||||
# Update in store
|
||||
target = target_store.update_target(
|
||||
@@ -321,18 +262,15 @@ async def update_target(
|
||||
min_brightness_threshold=data.min_brightness_threshold,
|
||||
adaptive_fps=data.adaptive_fps,
|
||||
protocol=data.protocol,
|
||||
key_colors_settings=kc_settings,
|
||||
description=data.description,
|
||||
tags=data.tags,
|
||||
ha_source_id=data.ha_source_id,
|
||||
ha_light_mappings=ha_mappings,
|
||||
update_rate=data.update_rate,
|
||||
transition=data.transition,
|
||||
color_tolerance=data.color_tolerance,
|
||||
)
|
||||
|
||||
# Detect KC brightness VS change (inside key_colors_settings)
|
||||
kc_brightness_vs_changed = False
|
||||
if data.key_colors_settings is not None:
|
||||
kc_incoming = data.key_colors_settings.model_dump(exclude_unset=True)
|
||||
if "brightness_value_source_id" in kc_incoming:
|
||||
kc_brightness_vs_changed = True
|
||||
|
||||
# Sync processor manager (run in thread — css release/acquire can block)
|
||||
try:
|
||||
await asyncio.to_thread(
|
||||
@@ -344,12 +282,13 @@ async def update_target(
|
||||
or data.state_check_interval is not None
|
||||
or data.min_brightness_threshold is not None
|
||||
or data.adaptive_fps is not None
|
||||
or data.key_colors_settings is not None
|
||||
or data.update_rate is not None
|
||||
or data.transition is not None
|
||||
or data.color_tolerance is not None
|
||||
or data.ha_light_mappings is not None
|
||||
),
|
||||
css_changed=data.color_strip_source_id is not None,
|
||||
brightness_vs_changed=(
|
||||
data.brightness_value_source_id is not None or kc_brightness_vs_changed
|
||||
),
|
||||
brightness_vs_changed=data.brightness_value_source_id is not None,
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.debug("Processor config update skipped for target %s: %s", target_id, e)
|
||||
|
||||
@@ -26,7 +26,6 @@ from wled_controller.api.dependencies import (
|
||||
get_ha_manager,
|
||||
get_ha_store,
|
||||
get_output_target_store,
|
||||
get_pattern_template_store,
|
||||
get_picture_source_store,
|
||||
get_pp_template_store,
|
||||
get_processor_manager,
|
||||
@@ -155,7 +154,6 @@ async def list_all_tags(_: AuthRequired):
|
||||
get_template_store,
|
||||
get_audio_template_store,
|
||||
get_pp_template_store,
|
||||
get_pattern_template_store,
|
||||
get_asset_store,
|
||||
]
|
||||
for getter in store_getters:
|
||||
|
||||
@@ -61,12 +61,6 @@ from .postprocessing import (
|
||||
PostprocessingTemplateUpdate,
|
||||
PPTemplateTestRequest,
|
||||
)
|
||||
from .pattern_templates import (
|
||||
PatternTemplateCreate,
|
||||
PatternTemplateListResponse,
|
||||
PatternTemplateResponse,
|
||||
PatternTemplateUpdate,
|
||||
)
|
||||
from .picture_sources import (
|
||||
ImageValidateRequest,
|
||||
ImageValidateResponse,
|
||||
|
||||
@@ -11,8 +11,12 @@ from wled_controller.api.schemas.devices import Calibration
|
||||
class AppSoundOverride(BaseModel):
|
||||
"""Per-application sound override for notification sources."""
|
||||
|
||||
sound_asset_id: Optional[str] = Field(None, description="Asset ID for the sound (None = mute this app)")
|
||||
volume: Optional[float] = Field(None, ge=0.0, le=1.0, description="Volume override (None = use global)")
|
||||
sound_asset_id: Optional[str] = Field(
|
||||
None, description="Asset ID for the sound (None = mute this app)"
|
||||
)
|
||||
volume: Optional[float] = Field(
|
||||
None, ge=0.0, le=1.0, description="Volume override (None = use global)"
|
||||
)
|
||||
|
||||
|
||||
class AnimationConfig(BaseModel):
|
||||
@@ -26,7 +30,9 @@ class AnimationConfig(BaseModel):
|
||||
class ColorStop(BaseModel):
|
||||
"""A single color stop in a gradient."""
|
||||
|
||||
position: float = Field(description="Relative position along the strip (0.0–1.0)", ge=0.0, le=1.0)
|
||||
position: float = Field(
|
||||
description="Relative position along the strip (0.0–1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
color: List[int] = Field(description="Primary RGB color [R, G, B] (0–255 each)")
|
||||
color_right: Optional[List[int]] = Field(
|
||||
None,
|
||||
@@ -38,13 +44,21 @@ class CompositeLayer(BaseModel):
|
||||
"""A single layer in a composite color strip source."""
|
||||
|
||||
source_id: str = Field(description="ID of the layer's color strip source")
|
||||
blend_mode: str = Field(default="normal", description="Blend mode: normal|add|multiply|screen|override")
|
||||
blend_mode: str = Field(
|
||||
default="normal", description="Blend mode: normal|add|multiply|screen|override"
|
||||
)
|
||||
opacity: float = Field(default=1.0, ge=0.0, le=1.0, description="Layer opacity 0.0-1.0")
|
||||
enabled: bool = Field(default=True, description="Whether this layer is active")
|
||||
brightness_source_id: Optional[str] = Field(None, description="Optional value source ID for dynamic brightness")
|
||||
processing_template_id: Optional[str] = Field(None, description="Optional color strip processing template ID")
|
||||
brightness_source_id: Optional[str] = Field(
|
||||
None, description="Optional value source ID for dynamic brightness"
|
||||
)
|
||||
processing_template_id: Optional[str] = Field(
|
||||
None, description="Optional color strip processing template ID"
|
||||
)
|
||||
start: int = Field(default=0, ge=0, description="First LED index for range (0 = full strip)")
|
||||
end: int = Field(default=0, ge=0, description="Last LED index exclusive for range (0 = full strip)")
|
||||
end: int = Field(
|
||||
default=0, ge=0, description="Last LED index exclusive for range (0 = full strip)"
|
||||
)
|
||||
reverse: bool = Field(default=False, description="Reverse layer output within its range")
|
||||
|
||||
|
||||
@@ -61,74 +75,179 @@ class ColorStripSourceCreate(BaseModel):
|
||||
"""Request to create a color strip source."""
|
||||
|
||||
name: str = Field(description="Source name", min_length=1, max_length=100)
|
||||
source_type: Literal["picture", "picture_advanced", "static", "gradient", "color_cycle", "effect", "composite", "mapped", "audio", "api_input", "notification", "daylight", "candlelight", "processed", "weather"] = Field(default="picture", description="Source type")
|
||||
source_type: Literal[
|
||||
"picture",
|
||||
"picture_advanced",
|
||||
"static",
|
||||
"gradient",
|
||||
"color_cycle",
|
||||
"effect",
|
||||
"composite",
|
||||
"mapped",
|
||||
"audio",
|
||||
"api_input",
|
||||
"notification",
|
||||
"daylight",
|
||||
"candlelight",
|
||||
"processed",
|
||||
"weather",
|
||||
"key_colors",
|
||||
] = Field(default="picture", description="Source type")
|
||||
# picture-type fields
|
||||
picture_source_id: str = Field(default="", description="Picture source ID (for picture type)")
|
||||
smoothing: float = Field(default=0.3, description="Temporal smoothing (0.0=none, 1.0=full)", ge=0.0, le=1.0)
|
||||
interpolation_mode: str = Field(default="average", description="LED color interpolation mode (average, median, dominant)")
|
||||
calibration: Optional[Calibration] = Field(None, description="LED calibration (position and count per edge)")
|
||||
smoothing: float = Field(
|
||||
default=0.3, description="Temporal smoothing (0.0=none, 1.0=full)", ge=0.0, le=1.0
|
||||
)
|
||||
interpolation_mode: str = Field(
|
||||
default="average", description="LED color interpolation mode (average, median, dominant)"
|
||||
)
|
||||
calibration: Optional[Calibration] = Field(
|
||||
None, description="LED calibration (position and count per edge)"
|
||||
)
|
||||
# static-type fields
|
||||
color: Optional[List[int]] = Field(None, description="Static RGB color [R, G, B] (0-255 each, for static type)")
|
||||
color: Optional[List[int]] = Field(
|
||||
None, description="Static RGB color [R, G, B] (0-255 each, for static type)"
|
||||
)
|
||||
# gradient-type fields
|
||||
stops: Optional[List[ColorStop]] = Field(None, description="Color stops for gradient type")
|
||||
# color_cycle-type fields
|
||||
colors: Optional[List[List[int]]] = Field(None, description="List of [R,G,B] colors to cycle (color_cycle type)")
|
||||
colors: Optional[List[List[int]]] = Field(
|
||||
None, description="List of [R,G,B] colors to cycle (color_cycle type)"
|
||||
)
|
||||
# effect-type fields
|
||||
effect_type: Optional[str] = Field(None, description="Effect algorithm: fire|meteor|plasma|noise|aurora|rain|comet|bouncing_ball|fireworks|sparkle_rain|lava_lamp|wave_interference")
|
||||
palette: Optional[str] = Field(None, description="Named palette (fire/ocean/lava/forest/rainbow/aurora/sunset/ice) or 'custom'")
|
||||
effect_type: Optional[str] = Field(
|
||||
None,
|
||||
description="Effect algorithm: fire|meteor|plasma|noise|aurora|rain|comet|bouncing_ball|fireworks|sparkle_rain|lava_lamp|wave_interference",
|
||||
)
|
||||
palette: Optional[str] = Field(
|
||||
None,
|
||||
description="Named palette (fire/ocean/lava/forest/rainbow/aurora/sunset/ice) or 'custom'",
|
||||
)
|
||||
intensity: Optional[float] = Field(None, description="Effect intensity 0.1-2.0", ge=0.1, le=2.0)
|
||||
scale: Optional[float] = Field(None, description="Spatial scale 0.5-5.0", ge=0.5, le=5.0)
|
||||
mirror: Optional[bool] = Field(None, description="Mirror/bounce mode (meteor/comet)")
|
||||
custom_palette: Optional[List[List[float]]] = Field(None, description="Custom palette stops [[pos,R,G,B],...]")
|
||||
custom_palette: Optional[List[List[float]]] = Field(
|
||||
None, description="Custom palette stops [[pos,R,G,B],...]"
|
||||
)
|
||||
# gradient entity reference (effect, gradient, audio types)
|
||||
gradient_id: Optional[str] = Field(None, description="Gradient entity ID (overrides palette/inline stops)")
|
||||
gradient_id: Optional[str] = Field(
|
||||
None, description="Gradient entity ID (overrides palette/inline stops)"
|
||||
)
|
||||
# gradient-type easing
|
||||
easing: Optional[str] = Field(None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic")
|
||||
easing: Optional[str] = Field(
|
||||
None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic"
|
||||
)
|
||||
# composite-type fields
|
||||
layers: Optional[List[CompositeLayer]] = Field(None, description="Layers for composite type")
|
||||
# mapped-type fields
|
||||
zones: Optional[List[MappedZone]] = Field(None, description="Zones for mapped type")
|
||||
# audio-type fields
|
||||
visualization_mode: Optional[str] = Field(None, description="Audio visualization: spectrum|beat_pulse|vu_meter")
|
||||
audio_source_id: Optional[str] = Field(None, description="Mono audio source ID (for audio type)")
|
||||
sensitivity: Optional[float] = Field(None, description="Audio sensitivity/gain 0.1-5.0", ge=0.1, le=5.0)
|
||||
color_peak: Optional[List[int]] = Field(None, description="Peak/high RGB color for VU meter [R,G,B]")
|
||||
visualization_mode: Optional[str] = Field(
|
||||
None, description="Audio visualization: spectrum|beat_pulse|vu_meter"
|
||||
)
|
||||
audio_source_id: Optional[str] = Field(
|
||||
None, description="Mono audio source ID (for audio type)"
|
||||
)
|
||||
sensitivity: Optional[float] = Field(
|
||||
None, description="Audio sensitivity/gain 0.1-5.0", ge=0.1, le=5.0
|
||||
)
|
||||
color_peak: Optional[List[int]] = Field(
|
||||
None, description="Peak/high RGB color for VU meter [R,G,B]"
|
||||
)
|
||||
# shared
|
||||
led_count: int = Field(default=0, description="Total LED count (0 = auto from calibration / device)", ge=0)
|
||||
led_count: int = Field(
|
||||
default=0, description="Total LED count (0 = auto from calibration / device)", ge=0
|
||||
)
|
||||
description: Optional[str] = Field(None, description="Optional description", max_length=500)
|
||||
animation: Optional[AnimationConfig] = Field(None, description="Procedural animation config (static/gradient only)")
|
||||
animation: Optional[AnimationConfig] = Field(
|
||||
None, description="Procedural animation config (static/gradient only)"
|
||||
)
|
||||
# api_input-type fields
|
||||
fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] when no data received (api_input type)")
|
||||
timeout: Optional[float] = Field(None, description="Seconds before reverting to fallback (api_input type)", ge=0.0, le=300.0)
|
||||
interpolation: Optional[str] = Field(None, description="LED count interpolation mode: none|linear|nearest (api_input type)")
|
||||
fallback_color: Optional[List[int]] = Field(
|
||||
None, description="Fallback RGB color [R,G,B] when no data received (api_input type)"
|
||||
)
|
||||
timeout: Optional[float] = Field(
|
||||
None, description="Seconds before reverting to fallback (api_input type)", ge=0.0, le=300.0
|
||||
)
|
||||
interpolation: Optional[str] = Field(
|
||||
None, description="LED count interpolation mode: none|linear|nearest (api_input type)"
|
||||
)
|
||||
# notification-type fields
|
||||
notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
|
||||
duration_ms: Optional[int] = Field(None, ge=100, le=10000, description="Effect duration in milliseconds")
|
||||
default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB) for notifications")
|
||||
app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color (#RRGGBB)")
|
||||
app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
|
||||
notification_effect: Optional[str] = Field(
|
||||
None, description="Notification effect: flash|pulse|sweep"
|
||||
)
|
||||
duration_ms: Optional[int] = Field(
|
||||
None, ge=100, le=10000, description="Effect duration in milliseconds"
|
||||
)
|
||||
default_color: Optional[str] = Field(
|
||||
None, description="Default hex color (#RRGGBB) for notifications"
|
||||
)
|
||||
app_colors: Optional[Dict[str, str]] = Field(
|
||||
None, description="Map of app name to hex color (#RRGGBB)"
|
||||
)
|
||||
app_filter_mode: Optional[str] = Field(
|
||||
None, description="App filter mode: off|whitelist|blacklist"
|
||||
)
|
||||
app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
|
||||
os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
|
||||
sound_asset_id: Optional[str] = Field(None, description="Global notification sound asset ID")
|
||||
sound_volume: Optional[float] = Field(None, ge=0.0, le=1.0, description="Global notification sound volume")
|
||||
app_sounds: Optional[Dict[str, AppSoundOverride]] = Field(None, description="Per-app sound overrides")
|
||||
sound_volume: Optional[float] = Field(
|
||||
None, ge=0.0, le=1.0, description="Global notification sound volume"
|
||||
)
|
||||
app_sounds: Optional[Dict[str, AppSoundOverride]] = Field(
|
||||
None, description="Per-app sound overrides"
|
||||
)
|
||||
# daylight-type fields
|
||||
speed: Optional[float] = Field(None, description="Cycle/flicker speed multiplier", ge=0.1, le=10.0)
|
||||
use_real_time: Optional[bool] = Field(None, description="Use wall-clock time for daylight cycle")
|
||||
latitude: Optional[float] = Field(None, description="Latitude for daylight timing (-90 to 90)", ge=-90.0, le=90.0)
|
||||
longitude: Optional[float] = Field(None, description="Longitude for daylight timing (-180 to 180)", ge=-180.0, le=180.0)
|
||||
speed: Optional[float] = Field(
|
||||
None, description="Cycle/flicker speed multiplier", ge=0.1, le=10.0
|
||||
)
|
||||
use_real_time: Optional[bool] = Field(
|
||||
None, description="Use wall-clock time for daylight cycle"
|
||||
)
|
||||
latitude: Optional[float] = Field(
|
||||
None, description="Latitude for daylight timing (-90 to 90)", ge=-90.0, le=90.0
|
||||
)
|
||||
longitude: Optional[float] = Field(
|
||||
None, description="Longitude for daylight timing (-180 to 180)", ge=-180.0, le=180.0
|
||||
)
|
||||
# candlelight-type fields
|
||||
num_candles: Optional[int] = Field(None, description="Number of independent candle sources (1-20)", ge=1, le=20)
|
||||
wind_strength: Optional[float] = Field(None, description="Wind simulation strength (0.0-2.0)", ge=0.0, le=2.0)
|
||||
candle_type: Optional[str] = Field(None, description="Candle type preset: default|taper|votive|bonfire")
|
||||
num_candles: Optional[int] = Field(
|
||||
None, description="Number of independent candle sources (1-20)", ge=1, le=20
|
||||
)
|
||||
wind_strength: Optional[float] = Field(
|
||||
None, description="Wind simulation strength (0.0-2.0)", ge=0.0, le=2.0
|
||||
)
|
||||
candle_type: Optional[str] = Field(
|
||||
None, description="Candle type preset: default|taper|votive|bonfire"
|
||||
)
|
||||
# processed-type fields
|
||||
input_source_id: Optional[str] = Field(None, description="Input color strip source ID (for processed type)")
|
||||
processing_template_id: Optional[str] = Field(None, description="Color strip processing template ID (for processed type)")
|
||||
input_source_id: Optional[str] = Field(
|
||||
None, description="Input color strip source ID (for processed type)"
|
||||
)
|
||||
processing_template_id: Optional[str] = Field(
|
||||
None, description="Color strip processing template ID (for processed type)"
|
||||
)
|
||||
# weather-type fields
|
||||
weather_source_id: Optional[str] = Field(None, description="Weather source entity ID (for weather type)")
|
||||
temperature_influence: Optional[float] = Field(None, description="Temperature color shift strength (0.0-1.0)", ge=0.0, le=1.0)
|
||||
weather_source_id: Optional[str] = Field(
|
||||
None, description="Weather source entity ID (for weather type)"
|
||||
)
|
||||
temperature_influence: Optional[float] = Field(
|
||||
None, description="Temperature color shift strength (0.0-1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
# key_colors-type fields
|
||||
rectangles: Optional[List[dict]] = Field(
|
||||
None, description="Named screen regions [{name,x,y,width,height}] for key_colors type"
|
||||
)
|
||||
brightness: Optional[float] = Field(
|
||||
None, description="Static brightness (0.0-1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
brightness_value_source_id: Optional[str] = Field(
|
||||
None, description="Dynamic brightness value source ID"
|
||||
)
|
||||
# sync clock
|
||||
clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
|
||||
clock_id: Optional[str] = Field(
|
||||
None, description="Optional sync clock ID for synchronized animation"
|
||||
)
|
||||
tags: List[str] = Field(default_factory=list, description="User-defined tags")
|
||||
|
||||
|
||||
@@ -138,71 +257,147 @@ class ColorStripSourceUpdate(BaseModel):
|
||||
name: Optional[str] = Field(None, description="Source name", min_length=1, max_length=100)
|
||||
# picture-type fields
|
||||
picture_source_id: Optional[str] = Field(None, description="Picture source ID")
|
||||
smoothing: Optional[float] = Field(None, description="Temporal smoothing (0.0-1.0)", ge=0.0, le=1.0)
|
||||
interpolation_mode: Optional[str] = Field(None, description="Interpolation mode (average, median, dominant)")
|
||||
smoothing: Optional[float] = Field(
|
||||
None, description="Temporal smoothing (0.0-1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
interpolation_mode: Optional[str] = Field(
|
||||
None, description="Interpolation mode (average, median, dominant)"
|
||||
)
|
||||
calibration: Optional[Calibration] = Field(None, description="LED calibration")
|
||||
# static-type fields
|
||||
color: Optional[List[int]] = Field(None, description="Static RGB color [R, G, B] (0-255 each, for static type)")
|
||||
color: Optional[List[int]] = Field(
|
||||
None, description="Static RGB color [R, G, B] (0-255 each, for static type)"
|
||||
)
|
||||
# gradient-type fields
|
||||
stops: Optional[List[ColorStop]] = Field(None, description="Color stops for gradient type")
|
||||
# color_cycle-type fields
|
||||
colors: Optional[List[List[int]]] = Field(None, description="List of [R,G,B] colors to cycle (color_cycle type)")
|
||||
colors: Optional[List[List[int]]] = Field(
|
||||
None, description="List of [R,G,B] colors to cycle (color_cycle type)"
|
||||
)
|
||||
# effect-type fields
|
||||
effect_type: Optional[str] = Field(None, description="Effect algorithm")
|
||||
palette: Optional[str] = Field(None, description="Named palette")
|
||||
intensity: Optional[float] = Field(None, description="Effect intensity 0.1-2.0", ge=0.1, le=2.0)
|
||||
scale: Optional[float] = Field(None, description="Spatial scale 0.5-5.0", ge=0.5, le=5.0)
|
||||
mirror: Optional[bool] = Field(None, description="Mirror/bounce mode")
|
||||
custom_palette: Optional[List[List[float]]] = Field(None, description="Custom palette stops [[pos,R,G,B],...]")
|
||||
custom_palette: Optional[List[List[float]]] = Field(
|
||||
None, description="Custom palette stops [[pos,R,G,B],...]"
|
||||
)
|
||||
# gradient entity reference (effect, gradient, audio types)
|
||||
gradient_id: Optional[str] = Field(None, description="Gradient entity ID (overrides palette/inline stops)")
|
||||
gradient_id: Optional[str] = Field(
|
||||
None, description="Gradient entity ID (overrides palette/inline stops)"
|
||||
)
|
||||
# gradient-type easing
|
||||
easing: Optional[str] = Field(None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic")
|
||||
easing: Optional[str] = Field(
|
||||
None, description="Gradient interpolation easing: linear|ease_in_out|step|cubic"
|
||||
)
|
||||
# composite-type fields
|
||||
layers: Optional[List[CompositeLayer]] = Field(None, description="Layers for composite type")
|
||||
# mapped-type fields
|
||||
zones: Optional[List[MappedZone]] = Field(None, description="Zones for mapped type")
|
||||
# audio-type fields
|
||||
visualization_mode: Optional[str] = Field(None, description="Audio visualization: spectrum|beat_pulse|vu_meter")
|
||||
audio_source_id: Optional[str] = Field(None, description="Mono audio source ID (for audio type)")
|
||||
sensitivity: Optional[float] = Field(None, description="Audio sensitivity/gain 0.1-5.0", ge=0.1, le=5.0)
|
||||
color_peak: Optional[List[int]] = Field(None, description="Peak/high RGB color for VU meter [R,G,B]")
|
||||
visualization_mode: Optional[str] = Field(
|
||||
None, description="Audio visualization: spectrum|beat_pulse|vu_meter"
|
||||
)
|
||||
audio_source_id: Optional[str] = Field(
|
||||
None, description="Mono audio source ID (for audio type)"
|
||||
)
|
||||
sensitivity: Optional[float] = Field(
|
||||
None, description="Audio sensitivity/gain 0.1-5.0", ge=0.1, le=5.0
|
||||
)
|
||||
color_peak: Optional[List[int]] = Field(
|
||||
None, description="Peak/high RGB color for VU meter [R,G,B]"
|
||||
)
|
||||
# shared
|
||||
led_count: Optional[int] = Field(None, description="Total LED count (0 = auto from calibration / device)", ge=0)
|
||||
led_count: Optional[int] = Field(
|
||||
None, description="Total LED count (0 = auto from calibration / device)", ge=0
|
||||
)
|
||||
description: Optional[str] = Field(None, description="Optional description", max_length=500)
|
||||
animation: Optional[AnimationConfig] = Field(None, description="Procedural animation config (static/gradient only)")
|
||||
animation: Optional[AnimationConfig] = Field(
|
||||
None, description="Procedural animation config (static/gradient only)"
|
||||
)
|
||||
# api_input-type fields
|
||||
fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] (api_input type)")
|
||||
timeout: Optional[float] = Field(None, description="Timeout before fallback (api_input type)", ge=0.0, le=300.0)
|
||||
interpolation: Optional[str] = Field(None, description="LED count interpolation mode: none|linear|nearest (api_input type)")
|
||||
fallback_color: Optional[List[int]] = Field(
|
||||
None, description="Fallback RGB color [R,G,B] (api_input type)"
|
||||
)
|
||||
timeout: Optional[float] = Field(
|
||||
None, description="Timeout before fallback (api_input type)", ge=0.0, le=300.0
|
||||
)
|
||||
interpolation: Optional[str] = Field(
|
||||
None, description="LED count interpolation mode: none|linear|nearest (api_input type)"
|
||||
)
|
||||
# notification-type fields
|
||||
notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
|
||||
duration_ms: Optional[int] = Field(None, ge=100, le=10000, description="Effect duration in milliseconds")
|
||||
notification_effect: Optional[str] = Field(
|
||||
None, description="Notification effect: flash|pulse|sweep"
|
||||
)
|
||||
duration_ms: Optional[int] = Field(
|
||||
None, ge=100, le=10000, description="Effect duration in milliseconds"
|
||||
)
|
||||
default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB)")
|
||||
app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color")
|
||||
app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
|
||||
app_filter_mode: Optional[str] = Field(
|
||||
None, description="App filter mode: off|whitelist|blacklist"
|
||||
)
|
||||
app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
|
||||
os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
|
||||
sound_asset_id: Optional[str] = Field(None, description="Global notification sound asset ID")
|
||||
sound_volume: Optional[float] = Field(None, ge=0.0, le=1.0, description="Global notification sound volume")
|
||||
app_sounds: Optional[Dict[str, AppSoundOverride]] = Field(None, description="Per-app sound overrides")
|
||||
sound_volume: Optional[float] = Field(
|
||||
None, ge=0.0, le=1.0, description="Global notification sound volume"
|
||||
)
|
||||
app_sounds: Optional[Dict[str, AppSoundOverride]] = Field(
|
||||
None, description="Per-app sound overrides"
|
||||
)
|
||||
# daylight-type fields
|
||||
speed: Optional[float] = Field(None, description="Cycle/flicker speed multiplier", ge=0.1, le=10.0)
|
||||
use_real_time: Optional[bool] = Field(None, description="Use wall-clock time for daylight cycle")
|
||||
latitude: Optional[float] = Field(None, description="Latitude for daylight timing (-90 to 90)", ge=-90.0, le=90.0)
|
||||
longitude: Optional[float] = Field(None, description="Longitude for daylight timing (-180 to 180)", ge=-180.0, le=180.0)
|
||||
speed: Optional[float] = Field(
|
||||
None, description="Cycle/flicker speed multiplier", ge=0.1, le=10.0
|
||||
)
|
||||
use_real_time: Optional[bool] = Field(
|
||||
None, description="Use wall-clock time for daylight cycle"
|
||||
)
|
||||
latitude: Optional[float] = Field(
|
||||
None, description="Latitude for daylight timing (-90 to 90)", ge=-90.0, le=90.0
|
||||
)
|
||||
longitude: Optional[float] = Field(
|
||||
None, description="Longitude for daylight timing (-180 to 180)", ge=-180.0, le=180.0
|
||||
)
|
||||
# candlelight-type fields
|
||||
num_candles: Optional[int] = Field(None, description="Number of independent candle sources (1-20)", ge=1, le=20)
|
||||
wind_strength: Optional[float] = Field(None, description="Wind simulation strength (0.0-2.0)", ge=0.0, le=2.0)
|
||||
candle_type: Optional[str] = Field(None, description="Candle type preset: default|taper|votive|bonfire")
|
||||
num_candles: Optional[int] = Field(
|
||||
None, description="Number of independent candle sources (1-20)", ge=1, le=20
|
||||
)
|
||||
wind_strength: Optional[float] = Field(
|
||||
None, description="Wind simulation strength (0.0-2.0)", ge=0.0, le=2.0
|
||||
)
|
||||
candle_type: Optional[str] = Field(
|
||||
None, description="Candle type preset: default|taper|votive|bonfire"
|
||||
)
|
||||
# processed-type fields
|
||||
input_source_id: Optional[str] = Field(None, description="Input color strip source ID (for processed type)")
|
||||
processing_template_id: Optional[str] = Field(None, description="Color strip processing template ID (for processed type)")
|
||||
input_source_id: Optional[str] = Field(
|
||||
None, description="Input color strip source ID (for processed type)"
|
||||
)
|
||||
processing_template_id: Optional[str] = Field(
|
||||
None, description="Color strip processing template ID (for processed type)"
|
||||
)
|
||||
# weather-type fields
|
||||
weather_source_id: Optional[str] = Field(None, description="Weather source entity ID (for weather type)")
|
||||
temperature_influence: Optional[float] = Field(None, description="Temperature color shift strength (0.0-1.0)", ge=0.0, le=1.0)
|
||||
weather_source_id: Optional[str] = Field(
|
||||
None, description="Weather source entity ID (for weather type)"
|
||||
)
|
||||
temperature_influence: Optional[float] = Field(
|
||||
None, description="Temperature color shift strength (0.0-1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
# key_colors-type fields
|
||||
rectangles: Optional[List[dict]] = Field(
|
||||
None, description="Named screen regions for key_colors type"
|
||||
)
|
||||
brightness: Optional[float] = Field(
|
||||
None, description="Static brightness (0.0-1.0)", ge=0.0, le=1.0
|
||||
)
|
||||
brightness_value_source_id: Optional[str] = Field(
|
||||
None, description="Dynamic brightness value source ID"
|
||||
)
|
||||
# sync clock
|
||||
clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
|
||||
clock_id: Optional[str] = Field(
|
||||
None, description="Optional sync clock ID for synchronized animation"
|
||||
)
|
||||
tags: Optional[List[str]] = None
|
||||
|
||||
|
||||
@@ -222,7 +417,9 @@ class ColorStripSourceResponse(BaseModel):
|
||||
# gradient-type fields
|
||||
stops: Optional[List[ColorStop]] = Field(None, description="Color stops for gradient type")
|
||||
# color_cycle-type fields
|
||||
colors: Optional[List[List[int]]] = Field(None, description="List of [R,G,B] colors to cycle (color_cycle type)")
|
||||
colors: Optional[List[List[int]]] = Field(
|
||||
None, description="List of [R,G,B] colors to cycle (color_cycle type)"
|
||||
)
|
||||
# effect-type fields
|
||||
effect_type: Optional[str] = Field(None, description="Effect algorithm")
|
||||
palette: Optional[str] = Field(None, description="Named palette")
|
||||
@@ -245,17 +442,27 @@ class ColorStripSourceResponse(BaseModel):
|
||||
# shared
|
||||
led_count: int = Field(0, description="Total LED count (0 = auto from calibration / device)")
|
||||
description: Optional[str] = Field(None, description="Description")
|
||||
animation: Optional[AnimationConfig] = Field(None, description="Procedural animation config (static/gradient only)")
|
||||
animation: Optional[AnimationConfig] = Field(
|
||||
None, description="Procedural animation config (static/gradient only)"
|
||||
)
|
||||
# api_input-type fields
|
||||
fallback_color: Optional[List[int]] = Field(None, description="Fallback RGB color [R,G,B] (api_input type)")
|
||||
fallback_color: Optional[List[int]] = Field(
|
||||
None, description="Fallback RGB color [R,G,B] (api_input type)"
|
||||
)
|
||||
timeout: Optional[float] = Field(None, description="Timeout before fallback (api_input type)")
|
||||
interpolation: Optional[str] = Field(None, description="LED count interpolation mode: none|linear|nearest (api_input type)")
|
||||
interpolation: Optional[str] = Field(
|
||||
None, description="LED count interpolation mode: none|linear|nearest (api_input type)"
|
||||
)
|
||||
# notification-type fields
|
||||
notification_effect: Optional[str] = Field(None, description="Notification effect: flash|pulse|sweep")
|
||||
notification_effect: Optional[str] = Field(
|
||||
None, description="Notification effect: flash|pulse|sweep"
|
||||
)
|
||||
duration_ms: Optional[int] = Field(None, description="Effect duration in milliseconds")
|
||||
default_color: Optional[str] = Field(None, description="Default hex color (#RRGGBB)")
|
||||
app_colors: Optional[Dict[str, str]] = Field(None, description="Map of app name to hex color")
|
||||
app_filter_mode: Optional[str] = Field(None, description="App filter mode: off|whitelist|blacklist")
|
||||
app_filter_mode: Optional[str] = Field(
|
||||
None, description="App filter mode: off|whitelist|blacklist"
|
||||
)
|
||||
app_filter_list: Optional[List[str]] = Field(None, description="App names for filter")
|
||||
os_listener: Optional[bool] = Field(None, description="Whether to listen for OS notifications")
|
||||
sound_asset_id: Optional[str] = Field(None, description="Global notification sound asset ID")
|
||||
@@ -263,7 +470,9 @@ class ColorStripSourceResponse(BaseModel):
|
||||
app_sounds: Optional[Dict[str, dict]] = Field(None, description="Per-app sound overrides")
|
||||
# daylight-type fields
|
||||
speed: Optional[float] = Field(None, description="Cycle/flicker speed multiplier")
|
||||
use_real_time: Optional[bool] = Field(None, description="Use wall-clock time for daylight cycle")
|
||||
use_real_time: Optional[bool] = Field(
|
||||
None, description="Use wall-clock time for daylight cycle"
|
||||
)
|
||||
latitude: Optional[float] = Field(None, description="Latitude for daylight timing")
|
||||
longitude: Optional[float] = Field(None, description="Longitude for daylight timing")
|
||||
# candlelight-type fields
|
||||
@@ -272,14 +481,30 @@ class ColorStripSourceResponse(BaseModel):
|
||||
candle_type: Optional[str] = Field(None, description="Candle type preset")
|
||||
# processed-type fields
|
||||
input_source_id: Optional[str] = Field(None, description="Input color strip source ID")
|
||||
processing_template_id: Optional[str] = Field(None, description="Color strip processing template ID")
|
||||
processing_template_id: Optional[str] = Field(
|
||||
None, description="Color strip processing template ID"
|
||||
)
|
||||
# weather-type fields
|
||||
weather_source_id: Optional[str] = Field(None, description="Weather source entity ID")
|
||||
temperature_influence: Optional[float] = Field(None, description="Temperature color shift strength")
|
||||
temperature_influence: Optional[float] = Field(
|
||||
None, description="Temperature color shift strength"
|
||||
)
|
||||
# key_colors-type fields
|
||||
rectangles: Optional[List[dict]] = Field(
|
||||
None, description="Named screen regions for key_colors type"
|
||||
)
|
||||
brightness: Optional[float] = Field(None, description="Static brightness (0.0-1.0)")
|
||||
brightness_value_source_id: Optional[str] = Field(
|
||||
None, description="Dynamic brightness value source ID"
|
||||
)
|
||||
# sync clock
|
||||
clock_id: Optional[str] = Field(None, description="Optional sync clock ID for synchronized animation")
|
||||
clock_id: Optional[str] = Field(
|
||||
None, description="Optional sync clock ID for synchronized animation"
|
||||
)
|
||||
tags: List[str] = Field(default_factory=list, description="User-defined tags")
|
||||
overlay_active: bool = Field(False, description="Whether the screen overlay is currently active")
|
||||
overlay_active: bool = Field(
|
||||
False, description="Whether the screen overlay is currently active"
|
||||
)
|
||||
created_at: datetime = Field(description="Creation timestamp")
|
||||
updated_at: datetime = Field(description="Last update timestamp")
|
||||
|
||||
@@ -298,7 +523,9 @@ class SegmentPayload(BaseModel):
|
||||
length: int = Field(ge=1, description="Number of LEDs in segment")
|
||||
mode: Literal["solid", "per_pixel", "gradient"] = Field(description="Fill mode")
|
||||
color: Optional[List[int]] = Field(None, description="RGB for solid mode [R,G,B]")
|
||||
colors: Optional[List[List[int]]] = Field(None, description="Colors for per_pixel/gradient [[R,G,B],...]")
|
||||
colors: Optional[List[List[int]]] = Field(
|
||||
None, description="Colors for per_pixel/gradient [[R,G,B],...]"
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_mode_fields(self) -> "SegmentPayload":
|
||||
@@ -329,8 +556,12 @@ class ColorPushRequest(BaseModel):
|
||||
At least one must be provided.
|
||||
"""
|
||||
|
||||
colors: Optional[List[List[int]]] = Field(None, description="LED color array [[R,G,B], ...] (0-255 each)")
|
||||
segments: Optional[List[SegmentPayload]] = Field(None, description="Segment-based color updates")
|
||||
colors: Optional[List[List[int]]] = Field(
|
||||
None, description="LED color array [[R,G,B], ...] (0-255 each)"
|
||||
)
|
||||
segments: Optional[List[SegmentPayload]] = Field(
|
||||
None, description="Segment-based color updates"
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _require_colors_or_segments(self) -> "ColorPushRequest":
|
||||
|
||||
Reference in New Issue
Block a user