feat(ha-light): broadcast a single Color Value Source to all entities
HALightOutputTarget gains a `source_kind` field with two modes: - `css` (existing): per-mapping LED segments averaged from a ColorStripSource. - `color_vs` (new): one colour from a colour-returning ValueSource pushed to every mapped entity (mapping LED ranges are ignored in this mode). Backend wiring: - Schema/route: add `source_kind` + `color_value_source_id` to create/update/ response payloads, with VS existence + return_type=color validation. - Storage: persist new fields, with defensive `or ""` coalesce so legacy rows written via resolve_ref with None survive the str-typed response schema. - Processor: ha_light_target_processor reworked to drive both source kinds (incl. update_target_settings hot-swap of source mode). New unit tests in tests/core/test_ha_light_target_processor.py and extended store tests. Frontend: - ha-light editor modal: collapsed Color Strip + Color VS into one "Color Source" picker with grouped headers; mappings list shows a mode-aware hint when broadcasting a single colour. - EntityPalette: support non-selectable header rows (with keyboard / filter handling) for grouped source pickers. Bundled UI polish (icon inheritance + cleanup): - Custom card icons now flow into more surfaces: command palette, dashboard target cards, scene-preset target picker, calibration test-device picker, and the LED-target device picker. LED targets inherit their device's icon when none is set on the target itself. - Empty mod-card icon plates render as a dashed "+" placeholder when an icon-picker hook is wired, so the action stays discoverable. - Icon picker: distinct "HA light target" eyebrow label and supports HA-light cards (data-ha-target-id) for channel-colour resolution. - Update banner: "View release" now opens the in-app Update settings tab instead of an external link; uses the sparkles icon. - Color-strip delete: cleaner toast on 409 conflict.
This commit is contained in:
@@ -0,0 +1,252 @@
|
||||
"""Unit tests for HALightTargetProcessor — covers CSS and color_vs modes."""
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.processing.ha_light_target_processor import HALightTargetProcessor
|
||||
from ledgrab.core.processing.target_processor import TargetContext
|
||||
from ledgrab.storage.bindable import BindableFloat
|
||||
from ledgrab.storage.ha_light_output_target import HALightMapping
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test doubles
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class _RecordedCall:
|
||||
domain: str
|
||||
service: str
|
||||
service_data: dict
|
||||
target: dict
|
||||
|
||||
|
||||
class _FakeHARuntime:
|
||||
is_connected = True
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: List[_RecordedCall] = []
|
||||
self._states: Dict[str, Any] = {}
|
||||
|
||||
async def call_service(self, *, domain, service, service_data, target):
|
||||
self.calls.append(_RecordedCall(domain, service, service_data, target))
|
||||
|
||||
def get_state(self, _entity_id: str):
|
||||
return self._states.get(_entity_id)
|
||||
|
||||
|
||||
class _FakeHAManager:
|
||||
def __init__(self, runtime: _FakeHARuntime) -> None:
|
||||
self._runtime = runtime
|
||||
self.acquired_for: List[str] = []
|
||||
self.released_for: List[str] = []
|
||||
|
||||
async def acquire(self, source_id: str) -> _FakeHARuntime:
|
||||
self.acquired_for.append(source_id)
|
||||
return self._runtime
|
||||
|
||||
async def release(self, source_id: str) -> None:
|
||||
self.released_for.append(source_id)
|
||||
|
||||
|
||||
class _FakeColorStream:
|
||||
"""A ValueStream that returns a constant RGB triple."""
|
||||
|
||||
def __init__(self, color: Tuple[int, int, int]) -> None:
|
||||
self._color = color
|
||||
self.get_color_calls = 0
|
||||
|
||||
def get_value(self) -> float:
|
||||
r, g, b = self._color
|
||||
return (0.299 * r + 0.587 * g + 0.114 * b) / 255.0
|
||||
|
||||
def get_color(self) -> Tuple[int, int, int]:
|
||||
self.get_color_calls += 1
|
||||
return self._color
|
||||
|
||||
|
||||
class _FakeCSSStream:
|
||||
"""A CSS stream that returns a fixed numpy LED frame."""
|
||||
|
||||
def __init__(self, frame: np.ndarray) -> None:
|
||||
self._frame = frame
|
||||
|
||||
def get_latest_colors(self) -> np.ndarray:
|
||||
return self._frame
|
||||
|
||||
|
||||
class _FakeVSManager:
|
||||
def __init__(self, stream) -> None:
|
||||
self._stream = stream
|
||||
self.acquired: List[str] = []
|
||||
self.released: List[str] = []
|
||||
|
||||
def acquire(self, vs_id: str):
|
||||
self.acquired.append(vs_id)
|
||||
return self._stream
|
||||
|
||||
def release(self, vs_id: str) -> None:
|
||||
self.released.append(vs_id)
|
||||
|
||||
|
||||
class _FakeCSSManager:
|
||||
def __init__(self, stream) -> None:
|
||||
self._stream = stream
|
||||
self.acquired: List[Tuple[str, str]] = []
|
||||
self.released: List[Tuple[str, str]] = []
|
||||
|
||||
def acquire(self, css_id: str, target_id: str):
|
||||
self.acquired.append((css_id, target_id))
|
||||
return self._stream
|
||||
|
||||
def release(self, css_id: str, target_id: str) -> None:
|
||||
self.released.append((css_id, target_id))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_ctx(
|
||||
*,
|
||||
ha_manager: Optional[_FakeHAManager] = None,
|
||||
css_manager: Optional[_FakeCSSManager] = None,
|
||||
vs_manager: Optional[_FakeVSManager] = None,
|
||||
) -> TargetContext:
|
||||
return TargetContext(
|
||||
live_stream_manager=None, # type: ignore[arg-type]
|
||||
overlay_manager=None, # type: ignore[arg-type]
|
||||
color_strip_stream_manager=css_manager,
|
||||
value_stream_manager=vs_manager,
|
||||
ha_manager=ha_manager,
|
||||
)
|
||||
|
||||
|
||||
def _mapping(entity_id: str, *, scale: float = 1.0) -> HALightMapping:
|
||||
return HALightMapping(
|
||||
entity_id=entity_id,
|
||||
led_start=0,
|
||||
led_end=-1,
|
||||
brightness_scale=BindableFloat(scale),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_color_vs_mode_broadcasts_same_color_to_all_mappings():
|
||||
runtime = _FakeHARuntime()
|
||||
ha_mgr = _FakeHAManager(runtime)
|
||||
color_stream = _FakeColorStream((128, 64, 200))
|
||||
vs_mgr = _FakeVSManager(color_stream)
|
||||
|
||||
proc = HALightTargetProcessor(
|
||||
target_id="t_color_vs",
|
||||
ha_source_id="ha_1",
|
||||
source_kind="color_vs",
|
||||
color_value_source_id="vs_purple",
|
||||
light_mappings=[_mapping("light.a"), _mapping("light.b"), _mapping("light.c")],
|
||||
color_tolerance=0,
|
||||
ctx=_make_ctx(ha_manager=ha_mgr, vs_manager=vs_mgr),
|
||||
)
|
||||
|
||||
await proc.start()
|
||||
# Wait until each entity has been called at least once.
|
||||
for _ in range(40):
|
||||
await asyncio.sleep(0.05)
|
||||
entities = {c.target["entity_id"] for c in runtime.calls}
|
||||
if {"light.a", "light.b", "light.c"} <= entities:
|
||||
break
|
||||
await proc.stop()
|
||||
|
||||
assert vs_mgr.acquired == ["vs_purple"], "color VS must be acquired in color_vs mode"
|
||||
assert vs_mgr.released == ["vs_purple"], "color VS must be released on stop"
|
||||
|
||||
# Each entity received the same RGB triple
|
||||
by_entity: Dict[str, set] = {}
|
||||
for c in runtime.calls:
|
||||
eid = c.target["entity_id"]
|
||||
by_entity.setdefault(eid, set()).add(tuple(c.service_data["rgb_color"]))
|
||||
assert {"light.a", "light.b", "light.c"} <= set(by_entity.keys())
|
||||
for colors in by_entity.values():
|
||||
assert colors == {(128, 64, 200)}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_css_mode_still_uses_per_segment_average():
|
||||
runtime = _FakeHARuntime()
|
||||
ha_mgr = _FakeHAManager(runtime)
|
||||
# 6-LED frame; light.a covers 0..2 (red), light.b covers 3..5 (blue)
|
||||
frame = np.array(
|
||||
[
|
||||
[255, 0, 0],
|
||||
[255, 0, 0],
|
||||
[255, 0, 0],
|
||||
[0, 0, 255],
|
||||
[0, 0, 255],
|
||||
[0, 0, 255],
|
||||
],
|
||||
dtype=np.int32,
|
||||
)
|
||||
css_mgr = _FakeCSSManager(_FakeCSSStream(frame))
|
||||
proc = HALightTargetProcessor(
|
||||
target_id="t_css",
|
||||
ha_source_id="ha_1",
|
||||
source_kind="css",
|
||||
color_strip_source_id="css_1",
|
||||
light_mappings=[
|
||||
HALightMapping(
|
||||
entity_id="light.a",
|
||||
led_start=0,
|
||||
led_end=3,
|
||||
brightness_scale=BindableFloat(1.0),
|
||||
),
|
||||
HALightMapping(
|
||||
entity_id="light.b",
|
||||
led_start=3,
|
||||
led_end=6,
|
||||
brightness_scale=BindableFloat(1.0),
|
||||
),
|
||||
],
|
||||
color_tolerance=0,
|
||||
ctx=_make_ctx(ha_manager=ha_mgr, css_manager=css_mgr),
|
||||
)
|
||||
|
||||
await proc.start()
|
||||
for _ in range(40):
|
||||
await asyncio.sleep(0.05)
|
||||
if {c.target["entity_id"] for c in runtime.calls} >= {"light.a", "light.b"}:
|
||||
break
|
||||
await proc.stop()
|
||||
|
||||
by_entity = {c.target["entity_id"]: tuple(c.service_data["rgb_color"]) for c in runtime.calls}
|
||||
assert by_entity["light.a"] == (255, 0, 0)
|
||||
assert by_entity["light.b"] == (0, 0, 255)
|
||||
assert css_mgr.released == [("css_1", "t_css")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_state_reports_source_kind():
|
||||
runtime = _FakeHARuntime()
|
||||
ha_mgr = _FakeHAManager(runtime)
|
||||
proc = HALightTargetProcessor(
|
||||
target_id="t_state",
|
||||
ha_source_id="ha_1",
|
||||
source_kind="color_vs",
|
||||
color_value_source_id="vs_x",
|
||||
light_mappings=[_mapping("light.a")],
|
||||
ctx=_make_ctx(ha_manager=ha_mgr, vs_manager=_FakeVSManager(_FakeColorStream((0, 0, 0)))),
|
||||
)
|
||||
state = proc.get_state()
|
||||
assert state["source_kind"] == "color_vs"
|
||||
assert state["color_value_source_id"] == "vs_x"
|
||||
assert state["css_id"] == ""
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import pytest
|
||||
|
||||
from ledgrab.storage.ha_light_output_target import HALightMapping, HALightOutputTarget
|
||||
from ledgrab.storage.output_target import OutputTarget
|
||||
from ledgrab.storage.output_target_store import OutputTargetStore
|
||||
from ledgrab.storage.wled_output_target import WledOutputTarget
|
||||
@@ -188,3 +189,98 @@ class TestOutputTargetPersistence:
|
||||
assert isinstance(loaded, WledOutputTarget)
|
||||
assert loaded.tags == ["tv"]
|
||||
db.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HA-light dual-mode source: CSS and color value source
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestHALightSourceModes:
|
||||
def test_legacy_payload_defaults_to_css_kind(self):
|
||||
"""Records persisted before source_kind existed must load as CSS mode."""
|
||||
data = {
|
||||
"id": "pt_legacy",
|
||||
"name": "Legacy HA light",
|
||||
"target_type": "ha_light",
|
||||
"ha_source_id": "ha_1",
|
||||
"color_strip_source_id": "css_1",
|
||||
"light_mappings": [],
|
||||
"created_at": "2025-01-01T00:00:00+00:00",
|
||||
"updated_at": "2025-01-01T00:00:00+00:00",
|
||||
}
|
||||
target = OutputTarget.from_dict(data)
|
||||
assert isinstance(target, HALightOutputTarget)
|
||||
assert target.source_kind == "css"
|
||||
assert target.color_strip_source_id == "css_1"
|
||||
assert target.color_value_source_id == ""
|
||||
|
||||
def test_invalid_source_kind_falls_back_to_css(self):
|
||||
data = {
|
||||
"id": "pt_bad_kind",
|
||||
"name": "Bad kind",
|
||||
"target_type": "ha_light",
|
||||
"ha_source_id": "ha_1",
|
||||
"source_kind": "garbage",
|
||||
"color_strip_source_id": "css_1",
|
||||
"created_at": "2025-01-01T00:00:00+00:00",
|
||||
"updated_at": "2025-01-01T00:00:00+00:00",
|
||||
}
|
||||
target = OutputTarget.from_dict(data)
|
||||
assert target.source_kind == "css"
|
||||
|
||||
def test_create_ha_light_color_vs(self, store):
|
||||
target = store.create_target(
|
||||
"Mood lamp",
|
||||
"ha_light",
|
||||
ha_source_id="ha_1",
|
||||
source_kind="color_vs",
|
||||
color_value_source_id="vs_static_red",
|
||||
ha_light_mappings=[HALightMapping(entity_id="light.lamp")],
|
||||
)
|
||||
assert isinstance(target, HALightOutputTarget)
|
||||
assert target.source_kind == "color_vs"
|
||||
assert target.color_value_source_id == "vs_static_red"
|
||||
# Round-trip via dict
|
||||
assert target.to_dict()["source_kind"] == "color_vs"
|
||||
|
||||
def test_round_trip_color_vs_persists(self, tmp_path):
|
||||
from ledgrab.storage.database import Database
|
||||
|
||||
db_path = str(tmp_path / "ha_color_vs.db")
|
||||
db = Database(db_path)
|
||||
s1 = OutputTargetStore(db)
|
||||
t = s1.create_target(
|
||||
"Sunrise",
|
||||
"ha_light",
|
||||
ha_source_id="ha_1",
|
||||
source_kind="color_vs",
|
||||
color_value_source_id="vs_anim_1",
|
||||
ha_light_mappings=[HALightMapping(entity_id="light.bed")],
|
||||
)
|
||||
s2 = OutputTargetStore(db)
|
||||
loaded = s2.get_target(t.id)
|
||||
assert isinstance(loaded, HALightOutputTarget)
|
||||
assert loaded.source_kind == "color_vs"
|
||||
assert loaded.color_value_source_id == "vs_anim_1"
|
||||
db.close()
|
||||
|
||||
def test_update_switches_source_kind(self, store):
|
||||
target = store.create_target(
|
||||
"Switcher",
|
||||
"ha_light",
|
||||
ha_source_id="ha_1",
|
||||
color_strip_source_id="css_1",
|
||||
ha_light_mappings=[HALightMapping(entity_id="light.x")],
|
||||
)
|
||||
assert target.source_kind == "css"
|
||||
updated = store.update_target(
|
||||
target.id,
|
||||
source_kind="color_vs",
|
||||
color_value_source_id="vs_color_1",
|
||||
)
|
||||
assert isinstance(updated, HALightOutputTarget)
|
||||
assert updated.source_kind == "color_vs"
|
||||
assert updated.color_value_source_id == "vs_color_1"
|
||||
# CSS id is preserved (non-destructive switch)
|
||||
assert updated.color_strip_source_id == "css_1"
|
||||
|
||||
Reference in New Issue
Block a user