Fix group member planning for low brightness

deploy/deploy.py

@@ -144,8 +144,7 @@ def _deploy(
 
     _ssh(host, "systemctl --user daemon-reload")
     if not no_start:
-        _ssh(host, "systemctl --user enable --now ha-deconz-bridge.service")
+        _ssh(host, "systemctl --user enable ha-deconz-bridge.service")
-        if config_only:
         _ssh(host, "systemctl --user restart ha-deconz-bridge.service")
         if restart_debug:
             _ssh(host, "systemctl --user restart ha-deconz-bridge-debug.service")

src/ha_deconz_bridge/debugger/web_debugger.py

@@ -13,14 +13,13 @@ from ha_deconz_bridge.app import load_config_module
 from ha_deconz_bridge.backends.deconz import DeconzBackend
 from ha_deconz_bridge.color import Color
 from ha_deconz_bridge.controller import PhysicalController
+from ha_deconz_bridge.group_planner import plan_group_command
 from ha_deconz_bridge.groups import (
     GroupMemberSpec,
     GroupSpec,
     derive_group_modes,
     fixed_brightness_scale,
     fixed_brightness_threshold,
-    fixed_color_mode,
-    fixed_color_payload,
     group_member_mode,
     normalize_group_specs,
 )
@@ -1239,16 +1238,15 @@ def _group_brightness_from_aggregate(
         return 0
     if requested_mode == "onoff":
         return 255 if _group_onoff_is_on(group, lights, controller_states, session, model_for_light) else 0
-    reference = sum(
+    plan = plan_group_command(
-        _group_member_reference_brightness(
+        group,
-            member,
+        lights,
-            lights[member.light_name],
+        group_target=report_color,
-            requested_mode,
+        group_mode=requested_mode,
-            report_color,
+        group_brightness=255,
-            model_for_light,
+        model_for_light=model_for_light,
-        )
-        for member in group.member_specs()
     )
+    reference = plan.reference_brightness
     if reference <= 0.0:
         return 0
     return int(round(max(0.0, min(1.0, actual_color.brightness / reference)) * 255.0))
@@ -1465,37 +1463,6 @@ def _group_debug_spec(
     )
 
 
-def _member_payload(
-    member: GroupMemberSpec,
-    member_spec: VirtualLightSpec,
-    group_mode: str,
-    request: dict[str, object],
-) -> dict[str, object]:
-    brightness = int(request.get("brightness", 255))
-    payload = fixed_color_payload(member, member_spec)
-    if payload is not None:
-        return {
-            "mode": fixed_color_mode(member, member_spec),
-            "brightness": int(member.fixed_brightness) if member.fixed_brightness is not None else brightness,
-            **payload,
-        }
-    if group_mode == "rgb" and "rgb" in member_spec.mode_set:
-        return {
-            "mode": "rgb",
-            "brightness": brightness,
-            "rgb": list(request.get("rgb", [255, 255, 255])),
-        }
-    if group_mode == "color_temp" and "color_temp" in member_spec.mode_set:
-        return {
-            "mode": "color_temp",
-            "brightness": brightness,
-            "color_temp": int(request.get("color_temp", _default_target(member_spec, "color_temp").color_temp)),
-        }
-    if member_spec.supported_color_modes == ("onoff",):
-        return {"mode": "onoff", "brightness": brightness}
-    return {"mode": "brightness", "brightness": brightness}
-
-
 def _build_target(spec: VirtualLightSpec, request: dict[str, object]) -> tuple[Color, str, int]:
     mode = str(request.get("mode") or spec.supported_color_modes[0])
     if mode not in spec.supported_color_modes:
@@ -1523,8 +1490,16 @@ def _solve_group_request(
     model_for_light,
 ) -> dict[str, object]:
     group_target, group_mode, group_brightness = _build_target(group_spec, request)
+    plan = plan_group_command(
+        group,
+        lights,
+        group_target=group_target,
+        group_mode=group_mode,
+        group_brightness=group_brightness,
+        model_for_light=model_for_light,
+    )
     achieved = Color.off()
-    reference_brightness = 0.0
+    reference_brightness = plan.reference_brightness
     chroma_error = 0.0
     brightness_error = 0.0
     channel_levels: dict[str, float] = {}
@@ -1533,8 +1508,16 @@ def _solve_group_request(
 
     for member in group.member_specs():
         member_spec = lights[member.light_name]
-        member_request = _member_payload(member, member_spec, group_mode, request)
+        member_plan = plan.members[member.light_name]
-        member_target, member_mode, member_brightness = _build_target(member_spec, member_request)
+        member_target = member_plan.target
+        member_mode = member_plan.mode
+        member_brightness = (
+            0
+            if not member_plan.active
+            else int(member_plan.fixed_brightness)
+            if member_plan.fixed_brightness is not None
+            else int(member_plan.brightness)
+        )
         member_result = solve_light(
             model_for_light(member.light_name),
             member_target,
@@ -1542,7 +1525,6 @@ def _solve_group_request(
             requested_mode=member_mode,
         )
         achieved += member_result.achieved_color
-        reference_brightness += member_result.reference_color.brightness
         chroma_error = max(chroma_error, float(member_result.chroma_error))
         brightness_error += float(member_result.brightness_error)
         channel_levels.update(member_result.channel_levels)
@@ -1582,7 +1564,6 @@ def _solve_group_request(
         },
     }
 
-
 def make_handler(
     lights: dict[str, VirtualLightSpec],
     initial_selection: list[str],

src/ha_deconz_bridge/group_light.py

@@ -4,6 +4,7 @@ from dataclasses import dataclass
 
 from ha_deconz_bridge.color import Color
 from ha_deconz_bridge.events import HaCommandEvent
+from ha_deconz_bridge.group_planner import GroupMemberPlan, plan_group_command
 from ha_deconz_bridge.groups import (
     GroupMemberSpec,
     GroupSpec,
@@ -69,14 +70,23 @@ class VirtualGroupLight:
 
     def handle_ha_command(self, event: HaCommandEvent) -> tuple[str, ...]:
         on, brightness, mode, rgb, color_temp = self._merge_command(event)
+        target = self._target_for_command(mode, rgb=rgb, color_temp=color_temp)
+        plan = plan_group_command(
+            self.group,
+            {name: member.spec for name, member in self.members.items()},
+            group_target=target,
+            group_mode=mode,
+            group_brightness=brightness if on else 0,
+            model_for_light=lambda name: self.members[name].light_model,
+        )
         for member_spec in self.member_specs:
             member = self.members[member_spec.light_name]
             member.handle_ha_command(
                 self._member_command(
                     member_spec,
                     member,
+                    plan=plan.members[member_spec.light_name],
                     on=on,
-                    brightness=brightness,
                     mode=mode,
                     rgb=rgb,
                     color_temp=color_temp,
@@ -150,20 +160,20 @@ class VirtualGroupLight:
         member_spec: GroupMemberSpec,
         member: VirtualLight,
         *,
+        plan: GroupMemberPlan,
         on: bool,
-        brightness: int,
         mode: str,
         rgb: tuple[int, int, int] | None,
         color_temp: int | None,
     ) -> HaCommandEvent:
-        if not on or brightness <= 0:
+        if not on or not plan.active or plan.brightness <= 0:
             return HaCommandEvent(member.spec.name, on=False)
 
         if member.spec.supported_color_modes == ("onoff",):
             return HaCommandEvent(member.spec.name, on=True)
 
-        fixed_brightness = member_spec.fixed_brightness
+        fixed_brightness = plan.fixed_brightness
-        member_brightness = int(fixed_brightness) if fixed_brightness is not None else brightness
+        member_brightness = int(fixed_brightness) if fixed_brightness is not None else plan.brightness
         payload = fixed_color_payload(member_spec, member.spec)
         if payload is not None:
             return HaCommandEvent(
@@ -172,11 +182,38 @@ class VirtualGroupLight:
                 brightness=member_brightness,
                 **payload,
             )
-        if mode == "rgb" and rgb is not None and "rgb" in member.spec.mode_set:
+        if plan.mode == "rgb" and "rgb" in member.spec.mode_set:
-            return HaCommandEvent(member.spec.name, on=True, brightness=brightness, rgb=rgb)
+            return HaCommandEvent(
-        if mode == "color_temp" and color_temp is not None and "color_temp" in member.spec.mode_set:
+                member.spec.name,
-            return HaCommandEvent(member.spec.name, on=True, brightness=brightness, color_temp=color_temp)
+                on=True,
-        return HaCommandEvent(member.spec.name, on=True, brightness=brightness)
+                brightness=plan.brightness,
+                rgb=tuple(int(round(component * 255.0)) for component in plan.target.rgb),  # type: ignore[arg-type]
+            )
+        if plan.mode == "color_temp" and "color_temp" in member.spec.mode_set:
+            return HaCommandEvent(
+                member.spec.name,
+                on=True,
+                brightness=plan.brightness,
+                color_temp=int(round(plan.target.color_temp)),
+            )
+        return HaCommandEvent(member.spec.name, on=True, brightness=plan.brightness)
+
+    def _target_for_command(
+        self,
+        mode: str,
+        *,
+        rgb: tuple[int, int, int] | None,
+        color_temp: int | None,
+    ) -> Color:
+        if mode == "color_temp":
+            return Color.from_color_temp(
+                color_temp or self._remembered.color_temp or self._default_color_temp() or 326,
+                brightness=1.0,
+                temp_range=self.spec.exposed_mireds or (153, 500),
+            )
+        if mode == "rgb":
+            return Color.from_rgb(rgb or self._remembered.rgb or (255, 255, 255), brightness=1.0)
+        return Color.from_rgb((255, 255, 255), brightness=1.0)
 
     def _recompute_state(
         self,
@@ -267,10 +304,15 @@ class VirtualGroupLight:
         mode: str,
         target: Color,
     ) -> float:
-        return sum(
+        plan = plan_group_command(
-            self._member_max_brightness(member_spec, member, mode, target)
+            self.group,
-            for member_spec, member, _state in members
+            {name: member.spec for name, member in self.members.items()},
+            group_target=target,
+            group_mode=mode,
+            group_brightness=255,
+            model_for_light=lambda name: self.members[name].light_model,
         )
+        return plan.reference_brightness
 
     def _member_max_brightness(
         self,

src/ha_deconz_bridge/group_planner.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable
+
+from ha_deconz_bridge.color import Color
+from ha_deconz_bridge.groups import GroupMemberSpec, GroupSpec, group_member_mode
+from ha_deconz_bridge.lights import ChannelSpec, ControllerModeSpec, ControllerSpec, VirtualLightSpec
+from ha_deconz_bridge.solver import CompiledLightModel, solve_light
+from ha_deconz_bridge.solver_model import compile_light_model
+
+
+@dataclass(slots=True, frozen=True)
+class GroupMemberPlan:
+    active: bool
+    brightness: int
+    mode: str
+    target: Color
+    fixed_brightness: int | None = None
+
+
+@dataclass(slots=True, frozen=True)
+class GroupSolvePlan:
+    members: dict[str, GroupMemberPlan]
+    achieved_color: Color
+    reference_brightness: float
+    chroma_error: float
+    brightness_error: float
+
+
+@dataclass(slots=True, frozen=True)
+class _MemberContribution:
+    member: GroupMemberSpec
+    spec: VirtualLightSpec
+    mode: str
+    target: Color
+    color: Color
+    switch_like: bool
+    fixed_brightness: int | None
+
+
+def plan_group_command(
+    group: GroupSpec,
+    lights: dict[str, VirtualLightSpec],
+    *,
+    group_target: Color,
+    group_mode: str,
+    group_brightness: int,
+    model_for_light: Callable[[str], CompiledLightModel],
+) -> GroupSolvePlan:
+    if group_brightness <= 0:
+        return GroupSolvePlan(
+            members={
+                member.light_name: GroupMemberPlan(
+                    active=False,
+                    brightness=0,
+                    mode=group_member_mode(member, lights[member.light_name], group_mode),  # type: ignore[arg-type]
+                    target=group_target,
+                )
+                for member in group.member_specs()
+            },
+            achieved_color=Color.off(),
+            reference_brightness=0.0,
+            chroma_error=0.0,
+            brightness_error=0.0,
+        )
+
+    contributions = tuple(
+        _member_contribution(member, lights[member.light_name], group_target, group_mode, model_for_light)
+        for member in group.member_specs()
+    )
+    if not contributions:
+        return GroupSolvePlan({}, Color.off(), 0.0, 0.0, 0.0)
+
+    plan_model = _compile_member_model(group.name, contributions)
+    requested_mode = group_mode if group_mode in {"rgb", "color_temp"} else "brightness"
+    result = solve_light(
+        plan_model,
+        group_target,
+        brightness=group_brightness,
+        requested_mode=requested_mode,
+    )
+
+    members: dict[str, GroupMemberPlan] = {}
+    for index, contribution in enumerate(contributions):
+        address = f"member_{index}.{'onoff' if contribution.switch_like else 'brightness'}.w"
+        level = result.channel_levels.get(address, 0.0)
+        active = level > 0.0
+        brightness = 255 if contribution.switch_like and active else int(round(max(0.0, min(1.0, level)) * 255.0))
+        members[contribution.member.light_name] = GroupMemberPlan(
+            active=active,
+            brightness=brightness,
+            mode=contribution.mode,
+            target=contribution.target,
+            fixed_brightness=contribution.fixed_brightness,
+        )
+
+    return GroupSolvePlan(
+        members=members,
+        achieved_color=result.achieved_color,
+        reference_brightness=result.reference_color.brightness,
+        chroma_error=float(result.chroma_error),
+        brightness_error=float(result.brightness_error),
+    )
+
+
+def _member_contribution(
+    member: GroupMemberSpec,
+    spec: VirtualLightSpec,
+    group_target: Color,
+    group_mode: str,
+    model_for_light: Callable[[str], CompiledLightModel],
+) -> _MemberContribution:
+    mode = group_member_mode(member, spec, group_mode)  # type: ignore[arg-type]
+    target = member.fixed_color.color if member.fixed_color is not None else group_target
+    fixed_brightness = int(member.fixed_brightness) if member.fixed_brightness is not None else None
+    brightness = fixed_brightness if fixed_brightness is not None else 255
+    model = model_for_light(spec.name)
+    result = solve_light(
+        model,
+        target,
+        brightness=brightness,
+        requested_mode=mode,
+    )
+    color = result.achieved_color
+    if (
+        group_mode == "color_temp"
+        and mode == "color_temp"
+        and color
+        and color.uv_distance(group_target) <= model.approximate_chroma_band
+    ):
+        # Group CT requests intentionally allow calibrated CT emitters to be
+        # approximate. Plan them on the requested CT chroma so a tiny exact
+        # emitter does not exclude much brighter near-CT emitters.
+        color = group_target.with_brightness(color.brightness)
+    return _MemberContribution(
+        member=member,
+        spec=spec,
+        mode=mode,
+        target=target,
+        color=color,
+        switch_like=member.fixed_brightness is not None or spec.supported_color_modes == ("onoff",),
+        fixed_brightness=fixed_brightness,
+    )
+
+
+def _compile_member_model(
+    group_name: str,
+    contributions: tuple[_MemberContribution, ...],
+) -> CompiledLightModel:
+    controllers: dict[str, ControllerSpec] = {}
+    for index, contribution in enumerate(contributions):
+        controller_id = f"member_{index}"
+        mode_name = "onoff" if contribution.switch_like else "brightness"
+        controllers[controller_id] = ControllerSpec(
+            zigbee_id=controller_id,
+            name=contribution.member.light_name,
+            modes={
+                mode_name: ControllerModeSpec(
+                    name=mode_name,  # type: ignore[arg-type]
+                    channels={
+                        "w": ChannelSpec(
+                            "w",
+                            contribution.color,
+                            _semantic_label(contribution.mode),
+                        )
+                    },
+                )
+            },
+        )
+    return compile_light_model(
+        VirtualLightSpec(
+            name=f"{group_name} member plan",
+            supported_color_modes=("rgb",),
+            controllers=controllers,
+        )
+    )
+
+
+def _semantic_label(mode: str) -> str | None:
+    if mode in {"rgb", "color_temp"}:
+        return mode
+    return None

tests/test_runtime.py

@@ -735,8 +735,8 @@ def test_example_group_readback_uses_aggregate_member_output(
     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["Floor Lamp Plug"].state.brightness == 0
-    assert app.lights["Desk Lamp Plug"].state.brightness == 255
+    assert app.lights["Desk Lamp Plug"].state.brightness == 0
 
 
 def test_example_native_ct_member_reports_requested_ct(settings) -> None:

tests/test_solver.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from lights_config import GROUPS, LIGHTS
 from ha_deconz_bridge.color import Color
-from ha_deconz_bridge.debugger.web_debugger import _combine_specs, _group_debug_spec
+from ha_deconz_bridge.debugger.web_debugger import _combine_specs, _group_debug_spec, _solve_group_request
 from ha_deconz_bridge.lights import ChannelSpec, ControllerModeSpec, ControllerSpec, VirtualLightSpec
 from ha_deconz_bridge.solver import solve_light
 from ha_deconz_bridge.solver_model import compile_light_model
@@ -686,3 +686,27 @@ def test_example_all_lights_ct_sweep_does_not_use_unused_onoff_reference() -> No
         previous = levels
 
     assert max_delta < 0.9
+
+
+def test_debugger_group_low_brightness_keeps_onoff_members_off() -> None:
+    group = GROUPS["Living Room"]
+    group_spec = _group_debug_spec(group, LIGHTS)
+    models = {
+        name: compile_light_model(spec)
+        for name, spec in LIGHTS.items()
+    }
+
+    result = _solve_group_request(
+        group,
+        LIGHTS,
+        group_spec,
+        {"mode": "color_temp", "brightness": 1, "color_temp": 500},
+        model_for_light=models.__getitem__,
+    )
+
+    assert result["commands"]["plug_a"]["on"] is False
+    assert result["commands"]["plug_b"]["on"] is False
+    assert result["channel_levels"].get("plug_a.onoff.w", 0.0) == 0.0
+    assert result["channel_levels"].get("plug_b.onoff.w", 0.0) == 0.0
+    assert result["channel_levels"].get("fixed_a.color_temp.ww", 0.0) == 0.0
+    assert result["channel_levels"].get("fixed_b.color_temp.ww", 0.0) == 0.0