from __future__ import annotations from dataclasses import dataclass, field from ha_deconz_bridge.app import HaDeconzBridgeApp from ha_deconz_bridge.app import load_config_module from ha_deconz_bridge.automation import ActionSpec, AutomationRule, RemoteSpec from ha_deconz_bridge.backends.fake import FakeControllerBackend from ha_deconz_bridge.color import Color from ha_deconz_bridge.controller import PhysicalController from ha_deconz_bridge.events import DeconzEvent, DeconzSensorEvent, HaCommandEvent from ha_deconz_bridge.groups import GroupMemberSpec, GroupSpec from ha_deconz_bridge.light import VirtualLight from ha_deconz_bridge.lights import ChannelSpec, ControllerModeSpec, ControllerSpec, VirtualLightSpec from ha_deconz_bridge.settings import Settings from ha_deconz_bridge.solver_model import compile_light_model from ha_deconz_bridge.state import ControllerCommand, ControllerState from ha_deconz_bridge.translate import controller_state_to_actual_color from ha_deconz_bridge.warmup import warmup_requests @dataclass class RecordingBridge: discoveries: list[str] = field(default_factory=list) states: list[dict[str, object]] = field(default_factory=list) def publish_discovery(self, spec) -> None: self.discoveries.append(spec.name) def publish_state(self, spec, state) -> None: self.states.append( { "name": spec.name, "on": state.on, "mode": state.ha_mode, "brightness": state.brightness, "color_temp": state.color.color_temp, "rgb": tuple(round(value, 3) for value in state.color.rgb), } ) def _resource_for_controller(controller) -> dict[str, object]: if "rgb" in controller.modes: return {"state": {"on": False, "reachable": True, "bri": 0, "xy": [0.3127, 0.3290], "colormode": "xy"}} if "color_temp" in controller.modes: return { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 153, "ctmax": 500, } if "brightness" in controller.modes: return {"state": {"on": False, "reachable": True, "bri": 0}} return {"state": {"on": False, "reachable": True}} def _build_light(spec, settings: Settings): resources = { controller_id: _resource_for_controller(controller) for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) bridge = RecordingBridge() controllers = { controller_id: PhysicalController( controller, backend=backend, settings=settings, fallback_mireds=spec.exposed_mireds, ) for controller_id, controller in spec.controllers.items() } light = VirtualLight( spec, controllers=controllers, light_model=compile_light_model(spec), ha_bridge=bridge, ) return light, bridge, backend def test_runtime_handles_rgb_command_roundtrip(rgb_light_spec, settings) -> None: light, bridge, _backend = _build_light(rgb_light_spec, settings) state = light.handle_ha_command( HaCommandEvent("rgb_light", on=True, brightness=128, rgb=(0, 0, 255)) ) assert state.on is True assert state.ha_mode == "rgb" assert abs(state.brightness - 128) <= 2 assert state.color.rgb[2] > 0.9 assert bridge.states def test_runtime_handles_color_temp_command_roundtrip(ct_light_spec, settings) -> None: light, _bridge, _backend = _build_light(ct_light_spec, settings) state = light.handle_ha_command( HaCommandEvent("ct_light", on=True, brightness=96, color_temp=450) ) assert state.on is True assert state.ha_mode == "color_temp" assert abs(state.brightness - 96) <= 2 assert state.color.brightness > 0.3 assert state.color.color_temp >= 450 def test_runtime_handles_off_command(mixed_light_spec, settings) -> None: light, _bridge, _backend = _build_light(mixed_light_spec, settings) light.handle_ha_command( HaCommandEvent("mixed_light", on=True, brightness=180, color_temp=250) ) off_state = light.handle_ha_command(HaCommandEvent("mixed_light", on=False)) assert off_state.on is False assert off_state.brightness == 0 def test_runtime_external_change_is_published(mixed_light_spec, settings) -> None: light, bridge, backend = _build_light(mixed_light_spec, settings) backend.push_state( "2", {"on": True, "reachable": True, "bri": 120, "ct": 450, "colormode": "ct"}, ) state = light.handle_controller_event("2", light.controllers["2"].backend.read_controller_state("2")) assert state.on is True assert bridge.states[-1]["on"] is True def test_runtime_ignores_stale_deconz_event_after_successful_command(ct_light_spec, settings) -> None: light, _bridge, _backend = _build_light(ct_light_spec, settings) commanded = light.handle_ha_command( HaCommandEvent("ct_light", on=True, brightness=255, color_temp=500) ) accepted_state = light.controllers["2"].state stale_state = ControllerState( on=True, reachable=True, brightness=255, mode="color_temp", color_temp=182, ) after_stale = light.handle_controller_event("2", stale_state) assert commanded.on is True assert after_stale.on is True assert light.controllers["2"].state == accepted_state def test_deconz_command_payload_disables_transition() -> None: from ha_deconz_bridge.backends.deconz import DeconzBackend payload = DeconzBackend._command_payload( ControllerCommand(on=True, brightness=255, mode="color_temp", color_temp=500) ) assert payload["transitiontime"] == 0 assert payload["bri"] == 254 assert "colormode" not in payload def test_deconz_off_payload_only_turns_off_with_no_transition() -> None: from ha_deconz_bridge.backends.deconz import DeconzBackend payload = DeconzBackend._command_payload( ControllerCommand(on=False, brightness=0, mode="color_temp", color_temp=500) ) assert payload == {"on": False, "transitiontime": 0} def test_runtime_deconz_event_updates_all_overlapping_virtual_lights( ct_light_spec, settings, monkeypatch, ) -> None: group_spec = VirtualLightSpec( name="ct_group", supported_color_modes=("color_temp",), controllers=ct_light_spec.controllers, ) resources = { "2": { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 153, "ctmax": 500, } } backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, { ct_light_spec.name: ct_light_spec, group_spec.name: group_spec, }, ) assert app.lights[ct_light_spec.name].controllers["2"] is app.controllers["2"] assert app.lights[group_spec.name].controllers["2"] is app.controllers["2"] assert len(app.controller_lights["2"]) == 2 app.handle_deconz_event( DeconzEvent( "2", {"on": True, "reachable": True, "bri": 120, "ct": 450, "colormode": "ct"}, ) ) published_names = {state["name"] for state in bridge.states[-2:]} assert published_names == {ct_light_spec.name, group_spec.name} assert all(state["on"] is True for state in bridge.states[-2:]) def test_runtime_ha_command_updates_all_overlapping_virtual_lights( ct_light_spec, settings, monkeypatch, ) -> None: group_spec = VirtualLightSpec( name="ct_group", supported_color_modes=("color_temp",), controllers=ct_light_spec.controllers, ) resources = { "2": { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 153, "ctmax": 500, } } backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, { ct_light_spec.name: ct_light_spec, group_spec.name: group_spec, }, ) app.handle_ha_command( HaCommandEvent(group_spec.name, on=True, brightness=120, color_temp=450) ) published_names = {state["name"] for state in bridge.states[-2:]} assert published_names == {ct_light_spec.name, group_spec.name} assert all(state["on"] is True for state in bridge.states[-2:]) def test_load_config_module_by_path_supports_repo_config_imports() -> None: module = load_config_module("config.example/lights_config.py") assert "Living Room Ceiling" in module.LIGHTS assert "Living Room" in module.GROUPS def test_warmup_requests_cover_supported_modes(rgb_light_spec, ct_light_spec) -> None: onoff_spec = VirtualLightSpec( name="onoff", supported_color_modes=("onoff",), controllers={ "onoff": ControllerSpec( zigbee_id="onoff", name="On/off", modes={ "onoff": ControllerModeSpec( name="onoff", channels={ "w": ChannelSpec( "w", Color.from_rgb((1.0, 1.0, 1.0), brightness=1.0), ) }, ) }, ) }, ) rgb_modes = {mode for _target, mode, _brightness in warmup_requests(rgb_light_spec)} ct_modes = {mode for _target, mode, _brightness in warmup_requests(ct_light_spec)} onoff_modes = {mode for _target, mode, _brightness in warmup_requests(onoff_spec)} assert rgb_modes == {"rgb"} assert ct_modes == {"color_temp"} assert onoff_modes == {"onoff"} def test_runtime_warmup_solvers_uses_compiled_light_models( ct_light_spec, settings, monkeypatch, ) -> None: resources = { "2": { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 153, "ctmax": 500, } } calls = [] backend = FakeControllerBackend(resources) monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: RecordingBridge()) monkeypatch.setattr( "ha_deconz_bridge.app.solve_light", lambda model, target, *, brightness, requested_mode: calls.append( (model.spec.name, brightness, requested_mode) ), ) app = HaDeconzBridgeApp(settings, {ct_light_spec.name: ct_light_spec}) app.warmup_solvers() assert calls assert {mode for _name, _brightness, mode in calls} == {"color_temp"} def test_runtime_sensor_event_runs_remote_automation( ct_light_spec, settings, monkeypatch, ) -> None: resources = { "2": { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 153, "ctmax": 500, } } backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, {ct_light_spec.name: ct_light_spec}, remotes={"4": RemoteSpec("4", "room", "Room remote")}, automations=( AutomationRule( remote_id="room", button="top", gesture="single", actions=( ActionSpec( "set_scene", target=ct_light_spec.name, brightness=96, color_temp=450, ), ), ), ), ) app.handle_sensor_event( DeconzSensorEvent("4", {"buttonevent": 1002, "eventduration": 1}) ) assert bridge.states[-1]["name"] == ct_light_spec.name assert bridge.states[-1]["on"] is True assert abs(bridge.states[-1]["brightness"] - 96) <= 2 def test_runtime_action_target_expressions_use_group_members( ct_light_spec, red_only_spec, settings, monkeypatch, ) -> None: specs = { ct_light_spec.name: ct_light_spec, red_only_spec.name: red_only_spec, } groups = { "All test": (ct_light_spec.name, red_only_spec.name), "Only red": (red_only_spec.name,), } resources = { controller_id: _resource_for_controller(controller) for spec in specs.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: RecordingBridge()) app = HaDeconzBridgeApp(settings, specs, groups=groups) assert app._action_target_names("All test") == ("All test",) assert app._action_target_names("All test-Only red") == (ct_light_spec.name,) assert app._action_target_names("-Only red") == (ct_light_spec.name,) assert app._action_target_names("All") == (ct_light_spec.name, red_only_spec.name) def test_runtime_group_member_fixed_ct_restricts_member_command( ct_light_spec, rgb_light_spec, settings, monkeypatch, ) -> None: specs = { ct_light_spec.name: ct_light_spec, rgb_light_spec.name: rgb_light_spec, } resources = { controller_id: _resource_for_controller(controller) for spec in specs.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, specs, groups={ "fixed": GroupSpec( name="fixed", members=(GroupMemberSpec(ct_light_spec.name, fixed_color=500),), balance="relative_member", ) }, ) assert app.group_lights["fixed"].spec.supported_color_modes == ("brightness",) app.handle_ha_command(HaCommandEvent("fixed", on=True, brightness=128)) controller_state = backend.read_controller_state("2") assert controller_state.on is True assert controller_state.brightness == 128 assert abs(controller_state.color_temp - 500) <= 8 assert bridge.states[-1]["name"] == "fixed" def test_runtime_group_relative_member_balance_sends_same_brightness( ct_light_spec, settings, monkeypatch, ) -> None: other_ct = VirtualLightSpec( name="other_ct", supported_color_modes=("color_temp",), controllers={"9": ct_light_spec.controllers["2"].__class__( zigbee_id="9", name="Other CT", modes=ct_light_spec.controllers["2"].modes, mireds=ct_light_spec.controllers["2"].mireds, )}, ) specs = {ct_light_spec.name: ct_light_spec, other_ct.name: other_ct} resources = { controller_id: _resource_for_controller(controller) for spec in specs.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: RecordingBridge()) app = HaDeconzBridgeApp( settings, specs, groups={ "balanced": GroupSpec( name="balanced", members=( GroupMemberSpec(ct_light_spec.name, fixed_color=500), GroupMemberSpec(other_ct.name, fixed_color=500), ), balance="relative_member", ) }, ) app.handle_ha_command(HaCommandEvent("balanced", on=True, brightness=100)) assert backend.read_controller_state("2").brightness == 100 assert backend.read_controller_state("9").brightness == 100 def test_runtime_group_fixed_brightness_sends_fixed_output( ct_light_spec, settings, monkeypatch, ) -> None: resources = {"2": _resource_for_controller(ct_light_spec.controllers["2"])} backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, {ct_light_spec.name: ct_light_spec}, groups={ "fixed": GroupSpec( name="fixed", members=(GroupMemberSpec(ct_light_spec.name, fixed_color=500, fixed_brightness=0.5),), ) }, ) assert app.group_lights["fixed"].spec.supported_color_modes == ("onoff",) app.handle_ha_command(HaCommandEvent("fixed", on=True)) controller_state = backend.read_controller_state("2") assert controller_state.on is True assert controller_state.brightness == 128 assert abs(controller_state.color_temp - 500) <= 8 def test_runtime_group_fixed_brightness_readback_threshold( ct_light_spec, settings, monkeypatch, ) -> None: resources = {"2": _resource_for_controller(ct_light_spec.controllers["2"])} backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, {ct_light_spec.name: ct_light_spec}, groups={ "fixed": GroupSpec( name="fixed", members=(GroupMemberSpec(ct_light_spec.name, fixed_color=500, fixed_brightness=128),), ) }, ) app.group_lights["fixed"].clear_accepted() app.handle_deconz_event( DeconzEvent("2", {"on": True, "reachable": True, "bri": 100, "ct": 500, "colormode": "ct"}) ) assert app.group_lights["fixed"].state.on is False app.handle_deconz_event( DeconzEvent("2", {"on": True, "reachable": True, "bri": 116, "ct": 500, "colormode": "ct"}) ) assert app.group_lights["fixed"].state.on is True assert app.group_lights["fixed"].state.brightness == 255 def test_runtime_group_keeps_accepted_low_brightness_during_quantized_readback( ct_light_spec, settings, monkeypatch, ) -> None: resources = {"2": _resource_for_controller(ct_light_spec.controllers["2"])} backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, {ct_light_spec.name: ct_light_spec}, groups={ "fixed": GroupSpec( name="fixed", members=(GroupMemberSpec(ct_light_spec.name, fixed_color=500),), balance="relative_member", ) }, ) app.handle_ha_command(HaCommandEvent("fixed", on=True, brightness=1)) app.handle_deconz_event( DeconzEvent( "2", {"on": True, "reachable": True, "bri": 3, "ct": 500, "colormode": "ct"}, ) ) assert app.group_lights["fixed"].state.brightness == 1 assert bridge.states[-1]["name"] == "fixed" assert bridge.states[-1]["brightness"] == 1 def test_runtime_direct_member_command_clears_group_accepted_brightness( ct_light_spec, settings, monkeypatch, ) -> None: resources = {"2": _resource_for_controller(ct_light_spec.controllers["2"])} backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp( settings, {ct_light_spec.name: ct_light_spec}, groups={ "fixed": GroupSpec( name="fixed", members=(GroupMemberSpec(ct_light_spec.name, fixed_color=500),), balance="relative_member", ) }, ) app.handle_ha_command(HaCommandEvent("fixed", on=True, brightness=1)) app.handle_ha_command(HaCommandEvent(ct_light_spec.name, on=True, brightness=12, color_temp=450)) assert abs(app.group_lights["fixed"].state.brightness - 12) <= 2 def test_runtime_relative_group_readback_uses_aggregate_output(settings, monkeypatch) -> None: def fixed_spec(name: str, controller_id: str) -> VirtualLightSpec: return VirtualLightSpec( name=name, supported_color_modes=("brightness",), controllers={ controller_id: ControllerSpec( zigbee_id=controller_id, name=name, modes={ "color_temp": ControllerModeSpec( name="color_temp", channels={ "ww": ChannelSpec( "ww", Color.from_color_temp(182, brightness=255.0, temp_range=(153, 500)), "color_temp", ) }, ) }, mireds=(153, 500), ) }, ) specs = { "a": fixed_spec("a", "a"), "b": fixed_spec("b", "b"), "c": fixed_spec("c", "c"), } resources = { controller_id: _resource_for_controller(controller) for spec in specs.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: RecordingBridge()) app = HaDeconzBridgeApp( settings, specs, groups={ "plants": GroupSpec( name="plants", members=( GroupMemberSpec("a", fixed_color=182), GroupMemberSpec("b", fixed_color=182), GroupMemberSpec("c", fixed_color=182), ), balance="relative_member", ) }, ) app.group_lights["plants"].clear_accepted() app.handle_deconz_event( DeconzEvent("a", {"on": True, "reachable": True, "bri": 1, "ct": 182, "colormode": "ct"}) ) app.handle_deconz_event( DeconzEvent("b", {"on": True, "reachable": True, "bri": 1, "ct": 182, "colormode": "ct"}) ) app.handle_deconz_event( DeconzEvent("c", {"on": True, "reachable": True, "bri": 8, "ct": 182, "colormode": "ct"}) ) assert app.group_lights["plants"].state.brightness == 3 def test_example_room_scene_reports_requested_brightness_and_ct(settings, monkeypatch) -> None: from actions import scene from groups import GROUPS from lights import LIGHTS resources = { controller_id: _resource_for_controller(controller) for spec in LIGHTS.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) bridge = RecordingBridge() monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: bridge) app = HaDeconzBridgeApp(settings, LIGHTS, groups=GROUPS) for kelvin, brightness, mireds in ( (2200, 51, 455), (2500, 102, 400), (3000, 153, 333), (3500, 204, 286), (4000, 255, 250), (5500, 255, 182), ): app._run_action(scene("Kitchen", brightness=brightness, kelvin=kelvin)) state = app.group_lights["Kitchen"].state assert state.brightness == brightness assert state.color.color_temp == mireds def test_example_group_readback_uses_aggregate_member_output( settings, monkeypatch, ) -> None: from actions import scene from groups import GROUPS from lights import LIGHTS resources = { controller_id: _resource_for_controller(controller) for spec in LIGHTS.values() for controller_id, controller in spec.controllers.items() } backend = FakeControllerBackend(resources) monkeypatch.setattr("ha_deconz_bridge.app.DeconzBackend", lambda _settings: backend) monkeypatch.setattr("ha_deconz_bridge.app.HomeAssistantBridge", lambda _settings: RecordingBridge()) app = HaDeconzBridgeApp(settings, LIGHTS, groups=GROUPS) app._run_action(scene("Living Room", brightness=51, kelvin=2200)) app.group_lights["Living Room"].clear_accepted() state = app.group_lights["Living Room"].handle_member_update() assert 45 <= state.brightness <= 60 assert app.lights["Floor Lamp Plug"].state.brightness == 255 assert app.lights["Desk Lamp Plug"].state.brightness == 255 def test_example_native_ct_member_reports_requested_ct(settings) -> None: from lights import LIGHTS light, _bridge, _backend = _build_light(LIGHTS["Bedroom Left Lamp"], settings) state = light.handle_ha_command(HaCommandEvent("Bedroom Left Lamp", on=True, brightness=153, color_temp=333)) assert state.brightness == 153 assert state.color.color_temp == 333 def test_runtime_reports_brightness_for_fixed_white_ct_backed_light(settings) -> None: spec = VirtualLightSpec( name="plant_light", supported_color_modes=("brightness",), controllers={ "6": ControllerSpec( zigbee_id="6", name="Plant", modes={ "color_temp": ControllerModeSpec( name="color_temp", channels={ "ww": ChannelSpec( "ww", Color.from_color_temp(182, brightness=280.0, temp_range=(153, 500)), "color_temp", ) }, ) }, ) }, ) light, _bridge, _backend = _build_light(spec, settings) state = light.handle_ha_command( HaCommandEvent("plant_light", on=True, brightness=200) ) assert state.on is True assert state.ha_mode == "brightness" assert abs(state.brightness - 200) <= 2 def test_runtime_reports_rgb_brightness_consistently_with_ct_boost(settings) -> None: spec = VirtualLightSpec( name="boosted_rgb", supported_color_modes=("rgb", "color_temp"), controllers={ "rgb": ControllerSpec( zigbee_id="rgb", name="RGB", modes={ "rgb": ControllerModeSpec( name="rgb", channels={ "r": ChannelSpec("r", Color.from_xy((0.68, 0.31), brightness=180.0), "rgb"), "g": ChannelSpec("g", Color.from_xy((0.18, 0.70), brightness=360.0), "rgb"), "b": ChannelSpec("b", Color.from_xy((0.14, 0.05), brightness=110.0), "rgb"), }, ) }, ), "ct": ControllerSpec( zigbee_id="ct", name="CT", modes={ "color_temp": ControllerModeSpec( name="color_temp", channels={ "ww": ChannelSpec( "ww", Color.from_color_temp(500, brightness=340.0, temp_range=(158, 500)), "color_temp", ), "cw": ChannelSpec( "cw", Color.from_color_temp(158, brightness=360.0, temp_range=(158, 500)), "color_temp", ), }, ) }, ), }, ) resources = { "rgb": {"state": {"on": False, "reachable": True, "bri": 0, "xy": [0.3127, 0.3290], "colormode": "xy"}}, "ct": { "state": {"on": False, "reachable": True, "bri": 0, "ct": 326, "colormode": "ct"}, "ctmin": 158, "ctmax": 500, }, } backend = FakeControllerBackend(resources) bridge = RecordingBridge() controllers = { controller_id: PhysicalController( controller, backend=backend, settings=settings, fallback_mireds=spec.exposed_mireds, ) for controller_id, controller in spec.controllers.items() } light = VirtualLight( spec, controllers=controllers, light_model=compile_light_model(spec), ha_bridge=bridge, ) state = light.handle_ha_command( HaCommandEvent("boosted_rgb", on=True, brightness=200, rgb=(128, 255, 128)) ) assert state.on is True assert state.ha_mode == "rgb" assert abs(state.brightness - 200) <= 2 white_state = light.handle_ha_command( HaCommandEvent("boosted_rgb", rgb=(255, 255, 255)) ) assert white_state.on is True assert white_state.ha_mode == "rgb" assert abs(white_state.brightness - 200) <= 2 ct_state = light.handle_ha_command( HaCommandEvent("boosted_rgb", color_temp=450) ) assert ct_state.on is True assert ct_state.ha_mode == "color_temp" assert abs(ct_state.brightness - 200) <= 4 def test_runtime_keeps_accepted_state_when_readback_is_within_tolerance(ct_light_spec, settings) -> None: settings.ct_tolerance = 1 light, _bridge, backend = _build_light(ct_light_spec, settings) state = light.handle_ha_command( HaCommandEvent("ct_light", on=True, brightness=181, color_temp=311) ) accepted_color_temp = state.color.color_temp accepted_command_ct = light.controllers["2"].accepted_state.command.color_temp assert state.brightness == 181 # Simulate deCONZ reporting the closest integer state instead of the exact # physical command we sent. This should not erase the accepted logical state. backend.push_state( "2", {"on": True, "reachable": True, "bri": 181, "ct": accepted_command_ct, "colormode": "ct"}, ) state = light.handle_controller_event("2", light.controllers["2"].backend.read_controller_state("2")) assert state.brightness == 181 assert state.color.color_temp == accepted_color_temp def test_physical_ct_readback_uses_calibrated_channel_mix() -> None: controller = ControllerSpec( zigbee_id="ct", name="Off locus CT", modes={ "color_temp": ControllerModeSpec( name="color_temp", channels={ "ww": ChannelSpec("ww", Color.from_xy((0.62, 0.32), brightness=2.0)), "cw": ChannelSpec("cw", Color.from_xy((0.20, 0.66), brightness=1.0)), }, ) }, mireds=(153, 500), ) state = ControllerState(on=True, reachable=True, brightness=255, mode="color_temp", color_temp=326) actual, family = controller_state_to_actual_color(controller, state, mireds_range=(153, 500)) ideal = Color.from_color_temp(326, brightness=actual.brightness, temp_range=(153, 500)) assert family == "color_temp" assert actual.uv_distance(ideal) > 0.01 def test_accepted_ct_state_preserves_requested_ct_for_close_readback(settings) -> None: spec = VirtualLightSpec( name="off_locus", supported_color_modes=("color_temp",), controllers={ "ct": ControllerSpec( zigbee_id="ct", name="Off locus CT", modes={ "color_temp": ControllerModeSpec( name="color_temp", channels={ "ww": ChannelSpec("ww", Color.from_xy((0.62, 0.32), brightness=2.0)), "cw": ChannelSpec("cw", Color.from_xy((0.20, 0.66), brightness=1.0)), }, ) }, mireds=(153, 500), ) }, ) light, _bridge, backend = _build_light(spec, settings) state = light.handle_ha_command(HaCommandEvent("off_locus", on=True, brightness=128, color_temp=326)) assert state.color.color_temp == 326 accepted_command_ct = light.controllers["ct"].accepted_state.command.color_temp backend.push_state( "ct", {"on": True, "reachable": True, "bri": 128, "ct": accepted_command_ct, "colormode": "ct"}, ) state = light.handle_controller_event("ct", backend.read_controller_state("ct")) assert state.brightness == 128 assert state.color.color_temp == 326 def test_runtime_drops_accepted_state_when_external_change_is_not_close(ct_light_spec, settings) -> None: settings.deconz_retry_timeout = -1.0 light, _bridge, backend = _build_light(ct_light_spec, settings) light.handle_ha_command( HaCommandEvent("ct_light", on=True, brightness=181, color_temp=311) ) backend.push_state( "2", {"on": True, "reachable": True, "bri": 120, "ct": 450, "colormode": "ct"}, ) state = light.handle_controller_event("2", light.controllers["2"].backend.read_controller_state("2")) assert state.color.color_temp >= 440 def test_controller_can_trust_write_without_blocking_on_stale_readback(ct_light_spec, settings) -> None: class StaleReadbackBackend(FakeControllerBackend): def __init__(self, resources): super().__init__(resources) self.reads = 0 self.writes: list[ControllerCommand] = [] def read_controller_state(self, controller_id: str) -> ControllerState: self.reads += 1 return super().read_controller_state(controller_id) def write_controller_command(self, controller_id: str, command: ControllerCommand) -> None: self.writes.append(command) backend = StaleReadbackBackend({"2": _resource_for_controller(ct_light_spec.controllers["2"])}) controller = PhysicalController( ct_light_spec.controllers["2"], backend=backend, settings=settings, fallback_mireds=(153, 500), ) state = controller.apply( ControllerCommand(on=True, brightness=255, mode="color_temp", color_temp=500), verify=False, ) assert state.on is True assert state.brightness == 255 assert state.color_temp == 500 assert len(backend.writes) == 1 assert backend.reads == 1 def test_off_readback_with_retained_mode_is_similar(settings) -> None: command = ControllerCommand(on=False) readback = ControllerState(on=False, brightness=254, mode="color_temp", color_temp=500) assert readback.similar_to( command, brightness_tolerance=settings.brightness_tolerance, ct_tolerance=settings.ct_tolerance, xy_distance_tolerance=settings.xy_distance_tolerance, ) def test_controller_accepts_clamped_ct_min_endpoint(ct_light_spec, settings) -> None: controller_spec = ct_light_spec.controllers["2"] backend = FakeControllerBackend( { "2": { "state": { "on": False, "reachable": True, "bri": 0, "ct": 158, "colormode": "ct", } } } ) controller = PhysicalController( controller_spec, backend=backend, settings=settings, fallback_mireds=(153, 500), ) assert controller._state_matches_command( ControllerState(on=True, brightness=56, mode="color_temp", color_temp=158), ControllerCommand(on=True, brightness=56, mode="color_temp", color_temp=153), )