feat: add expand_light_groups option (#1462)

* feat: add expand_light_groups option

Some light group entities act as a proxy that must receive a single combined
`light.turn_on` call to function correctly — virtual mixers like
<https://github.com/mion00/color-temperature-light-mixer> for instance,
that blend a warm and a cold white channel into one entity. In such setups the
individual member entities only expose `ColorMode.BRIGHTNESS`, so sending
separate per-member commands bypasses the mixing logic.

Setting `expand_light_groups: false` keeps the group entity in `self.lights`
instead of expanding it to its members. Adaptation commands go to the group,
and the interceptor no longer skips group entities for that switch.

Default is `true` — no behaviour change for existing configurations.

* tests: regression test for expand_light_groups=False

_switches_with_lights was expanding the incoming entity_id globally,
causing the switch to never be found when expand_light_groups=False

* Resolve group targets consistently across adaptation paths

* Discard delayed group events after target changes

* Stabilize delayed group target regression test

---------

Co-authored-by: Bas Nijholt <bas@nijho.lt>
This commit is contained in:
Leonhard Hesse 2026-09-06 21:13:35 +02:00 committed by GitHub
commit aa84eda871
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 547 additions and 56 deletions

View file

@ -48,6 +48,10 @@ This feature is available when `take_over_control` is enabled.
Additionally, enabling `detect_non_ha_changes` allows Adaptive Lighting to detect all state changes, including those made outside of Home Assistant, by comparing the light's state to its previously used settings.
The `adaptive_lighting.manual_control` event is fired when a light is marked as "manually controlled," allowing for integration with automations 🤖.
With `expand_light_groups: false`, manual control belongs to the group. A direct member change cannot pause adaptation for only that member; use group-level manual control or enable expansion for individual tracking.
Explicit member targets in Adaptive Lighting services stay individual targets and do not mark or command the whole group.
Changing expansion at runtime discards tracking and pending adaptation for targets no longer used by any profile.
The Adaptive Lighting switch exposes these read-only attributes for its lights:
- `manual_control`: lights with any attribute marked as manually controlled.
@ -166,6 +170,7 @@ The YAML and frontend configuration methods support all of the options listed be
| `intercept` | Intercept and adapt `light.turn_on` calls to enabling instantaneous color and brightness adaptation. 🏎️ Disable for lights that do not support `light.turn_on` with color and brightness. | `True` | `bool` |
| `multi_light_intercept` | Intercept and adapt `light.turn_on` calls that target multiple lights. ➗⚠️ This might result in splitting up a single `light.turn_on` call into multiple calls, e.g., when lights are in different switches. Requires `intercept` to be enabled. | `True` | `bool` |
| `include_config_in_attributes` | Show all options as attributes on the switch in Home Assistant when set to `true`. 📝 | `False` | `bool` |
| `expand_light_groups` | Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets. | `True` | `bool` |
<!-- OUTPUT:END -->

View file

@ -291,6 +291,14 @@ DOCS[CONF_MULTI_LIGHT_INTERCEPT] = (
"Requires `intercept` to be enabled."
)
CONF_EXPAND_LIGHT_GROUPS, DEFAULT_EXPAND_LIGHT_GROUPS = "expand_light_groups", True
DOCS[CONF_EXPAND_LIGHT_GROUPS] = (
"Expand light groups to their members (`true`, default). Set `false` to send "
"commands to the group and track manual control for the group. Explicit member "
"targets in services stay individual targets."
)
SLEEP_MODE_SWITCH = "sleep_mode_switch"
ADAPT_COLOR_SWITCH = "adapt_color_switch"
ADAPT_BRIGHTNESS_SWITCH = "adapt_brightness_switch"
@ -447,6 +455,7 @@ VALIDATION_TUPLES: list[tuple[str, Any, Any]] = [
(CONF_INTERCEPT, DEFAULT_INTERCEPT, bool),
(CONF_MULTI_LIGHT_INTERCEPT, DEFAULT_MULTI_LIGHT_INTERCEPT, bool),
(CONF_INCLUDE_CONFIG_IN_ATTRIBUTES, DEFAULT_INCLUDE_CONFIG_IN_ATTRIBUTES, bool),
(CONF_EXPAND_LIGHT_GROUPS, DEFAULT_EXPAND_LIGHT_GROUPS, bool),
]

View file

@ -139,6 +139,12 @@ change_switch_settings:
example: false
selector:
boolean: null
expand_light_groups:
description: Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets.
required: false
example: true
selector:
boolean: null
separate_turn_on_commands:
description: Use separate `light.turn_on` calls for color and brightness, needed for some light types. 🔀
required: false

View file

@ -78,7 +78,8 @@
"skip_redundant_commands": "skip_redundant_commands: Skip sending adaptation commands whose target state already equals the light's known state. Minimizes network traffic and improves the adaptation responsivity in some situations. 📉Disable if physical light states get out of sync with HA's recorded state.",
"intercept": "intercept: Intercept and adapt `light.turn_on` calls to enabling instantaneous color and brightness adaptation. 🏎️ Disable for lights that do not support `light.turn_on` with color and brightness.",
"multi_light_intercept": "multi_light_intercept: Intercept and adapt `light.turn_on` calls that target multiple lights. ➗⚠️ This might result in splitting up a single `light.turn_on` call into multiple calls, e.g., when lights are in different switches. Requires `intercept` to be enabled.",
"include_config_in_attributes": "include_config_in_attributes: Show all options as attributes on the switch in Home Assistant when set to `true`. 📝"
"include_config_in_attributes": "include_config_in_attributes: Show all options as attributes on the switch in Home Assistant when set to `true`. 📝",
"expand_light_groups": "expand_light_groups: Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets."
},
"data_description": {
"initial_transition": "Duration of the first transition when lights turn from `off` to `on` in seconds. ⏲️",
@ -211,6 +212,10 @@
"description": "Whether to prefer RGB color adjustment over light color temperature when possible. 🌈",
"name": "prefer_rgb_color"
},
"expand_light_groups": {
"description": "Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets.",
"name": "expand_light_groups"
},
"separate_turn_on_commands": {
"description": "Use separate `light.turn_on` calls for color and brightness, needed for some light types. 🔀",
"name": "separate_turn_on_commands"

View file

@ -98,6 +98,7 @@ from .const import (
CONF_BRIGHTNESS_MODE_TIME_DARK,
CONF_BRIGHTNESS_MODE_TIME_LIGHT,
CONF_DETECT_NON_HA_CHANGES,
CONF_EXPAND_LIGHT_GROUPS,
CONF_INCLUDE_CONFIG_IN_ATTRIBUTES,
CONF_INITIAL_TRANSITION,
CONF_INTERCEPT,
@ -252,14 +253,11 @@ def _switches_with_lights(
if not loaded_switches:
return []
all_check_lights = (
_expand_light_groups(hass, lights) if expand_light_groups else set(lights)
)
switches: AdaptiveSwitches = []
for switch in loaded_switches:
switch._expand_light_groups(hass=hass)
# Check if any of the lights are in the switch's lights
if set(switch.lights) & set(all_check_lights):
switch._expand_light_groups()
check_lights = switch._resolve_lights(lights) if expand_light_groups else lights
if set(switch.lights) & set(check_lights):
switches.append(switch)
return switches
@ -422,7 +420,7 @@ async def handle_apply_service(hass: HomeAssistant, service_call: ServiceCall) -
switches = _switches_from_service_call(hass, service_call)
lights = data[CONF_LIGHTS]
for switch in switches:
all_lights = switch.lights if not lights else _expand_light_groups(hass, lights)
all_lights = switch._resolve_lights(lights or None)
switch.manager.lights.update(all_lights)
for light in all_lights:
if data[CONF_TURN_ON_LIGHTS] or is_on(hass, light):
@ -457,7 +455,7 @@ async def handle_set_manual_control_service(
switches = _switches_from_service_call(hass, service_call)
lights = data[CONF_LIGHTS]
for switch in switches:
all_lights = switch.lights if not lights else _expand_light_groups(hass, lights)
all_lights = switch._resolve_lights(lights or None)
manual_attributes = manual_control_event_attribute_to_flags(
data[CONF_MANUAL_CONTROL],
)
@ -626,17 +624,22 @@ def _expand_light_groups(
hass: HomeAssistant,
lights: list[str],
) -> list[str]:
"""Resolve nested groups without changing another profile's tracked targets."""
all_lights: set[str] = set()
manager = hass.data[DOMAIN][ATTR_ADAPTIVE_LIGHTING_MANAGER]
for light in lights:
pending = list(lights)
visited: set[str] = set()
while pending:
light = pending.pop()
if light in visited:
continue
visited.add(light)
state = hass.states.get(light)
if state is None:
_LOGGER.debug("State of %s is None", light)
all_lights.add(light)
elif _is_light_group(state):
group = state.attributes["entity_id"]
manager.lights.discard(light)
all_lights.update(group)
pending.extend(group)
_LOGGER.debug("Expanded %s to %s", light, group)
else:
all_lights.add(light)
@ -889,7 +892,8 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._name = data[CONF_NAME]
self._interval: timedelta = data[CONF_INTERVAL]
self.lights: list[str] = data[CONF_LIGHTS]
self._configured_lights: list[str] = list(data[CONF_LIGHTS])
self.lights: list[str] = []
# backup data for use in change_switch_settings "configuration" CONF_USE_DEFAULTS
self._config_backup = deepcopy(data)
@ -990,6 +994,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._name,
)
self._multi_light_intercept = False
self._expand_light_groups_flag = data[CONF_EXPAND_LIGHT_GROUPS]
self._expand_light_groups() # updates manual control timers
observer = get_astral_observer(self.hass)
@ -1073,15 +1078,29 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
"""Remove the listeners upon removing the component."""
self._remove_listeners()
def _expand_light_groups(self, hass: HomeAssistant | None = None) -> None:
hass = hass or self.hass
all_lights = _expand_light_groups(hass, self.lights)
def _resolve_lights(self, lights: list[str] | None = None) -> list[str]:
"""Apply this profile's group policy, preserving explicit member targets."""
if lights is None:
lights = self._configured_lights
if self._expand_light_groups_flag:
return _expand_light_groups(self.hass, lights)
return sorted(set(lights))
def _expand_light_groups(self) -> None:
all_lights = self._resolve_lights()
removed = set(self.lights) - set(all_lights)
self.lights = all_lights
if removed:
# Other profiles may still own a retired member or group, even when off.
for entry in self.hass.data[DOMAIN].values():
if isinstance(entry, dict) and (switch := entry.get(SWITCH_DOMAIN)):
removed.difference_update(switch._resolve_lights())
self.manager.remove_lights(*removed)
self.manager.lights.update(all_lights)
self.manager.set_auto_reset_manual_control_times(
all_lights,
self._auto_reset_manual_control_time,
)
self.lights = list(all_lights)
async def _setup_listeners(self, _: Event[NoEventData] | None = None) -> None:
_LOGGER.debug("%s: Called '_setup_listeners'", self._name)
@ -1521,6 +1540,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return
if lights is None:
self._expand_light_groups()
lights = self.lights
on_lights = [light for light in lights if is_on(self.hass, light)]
@ -1668,6 +1688,10 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if self._adapt_delay > 0:
await asyncio.sleep(self._adapt_delay)
# Runtime settings may retire this profile's target while the event waits.
if entity_id not in self.lights:
return
await self._update_attrs_and_maybe_adapt_lights(
context=self.create_context("light_event", parent=event.context),
lights=[entity_id],
@ -1983,8 +2007,12 @@ class AdaptiveLightingManager:
if (
not switch.is_on
or not switch._intercept
# Never adapt on light groups, because HA will make a separate light.turn_on
or ((e := self.hass.states.get(entity_id)) and _is_light_group(e))
# Never adapt on light groups when expanding, because HA will make a separate light.turn_on
or (
switch._expand_light_groups_flag
and (e := self.hass.states.get(entity_id))
and _is_light_group(e)
)
# Prevent adaptation of TURN_ON calls when light is already on,
# and of TOGGLE calls when toggling off.
or self.hass.states.is_state(entity_id, STATE_ON)
@ -2381,7 +2409,11 @@ class AdaptiveLightingManager:
delay,
)
self.reset(light)
switches = _switches_with_lights(self.hass, [light])
switches = _switches_with_lights(
self.hass,
[light],
expand_light_groups=False,
)
for switch in switches:
if not switch.is_on:
continue
@ -2572,6 +2604,30 @@ class AdaptiveLightingManager:
if reset_manual_control:
self._schedule_manual_control_state_update(*lights)
def remove_lights(self, *lights: str) -> None:
"""Retire tracking and pending work for targets no profile owns anymore."""
self.reset(*lights)
for light in lights:
self.lights.discard(light)
self.clear_proactively_adapting(light)
if timer := self.transition_timers.pop(light, None):
timer.cancel()
if task := self.sleep_tasks.pop(light, None):
task.cancel()
for records in (
self.manual_control,
self.auto_reset_manual_control_times,
self.turn_on_event,
self.turn_off_event,
self.toggle_event,
self.on_to_off_event,
self.off_to_on_event,
self.turn_off_locks,
self.adaptation_tasks_brightness,
self.adaptation_tasks_color,
):
records.pop(light, None)
def _get_entity_list(self, service_data: ServiceData) -> list[str]:
if ATTR_ENTITY_ID in service_data:
return cv.ensure_list_csv(service_data[ATTR_ENTITY_ID])
@ -2806,7 +2862,11 @@ class AdaptiveLightingManager:
)
return
switches = _switches_with_lights(self.hass, [entity_id])
switches = _switches_with_lights(
self.hass,
[entity_id],
expand_light_groups=False,
)
for switch in switches:
if switch.is_on:
await switch._respond_to_off_to_on_event(

View file

@ -68,7 +68,8 @@
"skip_redundant_commands": "skip_redundant_commands: Überspringt das Senden von Anpassungsbefehlen, deren Zielzustand bereits mit dem bekannten Zustand der Leuchte übereinstimmt. Minimiert den Netzwerkverkehr und verbessert die Anpassungsreaktion in einigen Situationen. 📉 Deaktivieren, falls der physikalische Zustand der Lichter nicht mehr mit dem Zustand in HA übereinstimmt.",
"intercept": "intercept: Abfangen und Anpassen von `light.turn_on`-Aufrufen, um eine sofortige Anpassung von Farbe und Helligkeit zu ermöglichen. 🏎️ Deaktivieren für Leuchten, die `light.turn_on` mit Farbe und Helligkeit nicht unterstützen.",
"multi_light_intercept": "multi_light_intercept: Abfangen und Anpassen von `light.turn_on`-Aufrufen, die auf mehrere Lichter aufrufen. ➗⚠️ Dies kann dazu führen, dass ein einzelner `light.turn_on`-Aufruf in mehrere Aufrufe aufgeteilt wird, z.B. wenn Lichter in verschiedenen Schaltern sind. Erfordert, dass `intercept` aktiviert ist.",
"include_config_in_attributes": "include_config_in_attributes: Alle Optionen als Attribute auf dem Schalter im Home Assistant anzeigen, wenn auf `true` gesetzt. 📝"
"include_config_in_attributes": "include_config_in_attributes: Alle Optionen als Attribute auf dem Schalter im Home Assistant anzeigen, wenn auf `true` gesetzt. 📝",
"expand_light_groups": "expand_light_groups: Lichtgruppen auf einzelne Mitgliedsentitäten erweitern (`true`) oder die Gruppenentität direkt steuern (`false`)."
},
"data_description": {
"initial_transition": "Dauer des ersten Übergangs, wenn das Licht von `off` auf `on` schaltet, in Sekunden. ⏲️",
@ -88,7 +89,8 @@
"brightness_mode_time_light": "(Wird ignoriert, wenn `brightness_mode='default'`) Die Dauer in Sekunden, um die Helligkeit nach/vor Sonnenaufgang/Sonnenuntergang hoch/runter zu fahren. 📈📉.",
"autoreset_control_seconds": "Setzt die manuelle Steuerung nach einer bestimmten Anzahl von Sekunden automatisch zurück. Zum Deaktivieren auf 0 setzen. ⏲️",
"send_split_delay": "Verzögerung (ms) zwischen `separate_turn_on_commands` für Leuchten, die keine gleichzeitige Einstellung von Helligkeit und Farbe unterstützen. ⏲️",
"adapt_delay": "Wartezeit (Sekunden) zwischen dem Einschalten des Lichts und der Anwendung der adaptiven Beleuchtung. Könnte helfen, Flackern zu vermeiden. ⏲️"
"adapt_delay": "Wartezeit (Sekunden) zwischen dem Einschalten des Lichts und der Anwendung der adaptiven Beleuchtung. Könnte helfen, Flackern zu vermeiden. ⏲️",
"expand_light_groups": "Wenn auf `false` gesetzt, werden Anpassungsbefehle direkt an die Gruppenentität gesendet und nicht an die einzelnen Mitglieder."
}
}
}

View file

@ -79,7 +79,8 @@
"skip_redundant_commands": "skip_redundant_commands: Skip sending adaptation commands whose target state already equals the light's known state. Minimizes network traffic and improves the adaptation responsivity in some situations. 📉Disable if physical light states get out of sync with HA's recorded state.",
"intercept": "intercept: Intercept and adapt `light.turn_on` calls to enabling instantaneous color and brightness adaptation. 🏎️ Disable for lights that do not support `light.turn_on` with color and brightness.",
"multi_light_intercept": "multi_light_intercept: Intercept and adapt `light.turn_on` calls that target multiple lights. ➗⚠️ This might result in splitting up a single `light.turn_on` call into multiple calls, e.g., when lights are in different switches. Requires `intercept` to be enabled.",
"include_config_in_attributes": "include_config_in_attributes: Show all options as attributes on the switch in Home Assistant when set to `true`. 📝"
"include_config_in_attributes": "include_config_in_attributes: Show all options as attributes on the switch in Home Assistant when set to `true`. 📝",
"expand_light_groups": "expand_light_groups: Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets."
},
"data_description": {
"initial_transition": "Duration of the first transition when lights turn from `off` to `on` in seconds. ⏲️",
@ -212,6 +213,10 @@
"description": "Whether to prefer RGB color adjustment over light color temperature when possible. 🌈",
"name": "prefer_rgb_color"
},
"expand_light_groups": {
"description": "Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets.",
"name": "expand_light_groups"
},
"separate_turn_on_commands": {
"description": "Use separate `light.turn_on` calls for color and brightness, needed for some light types. 🔀",
"name": "separate_turn_on_commands"

View file

@ -20,6 +20,10 @@ This feature is available when `take_over_control` is enabled.
Additionally, enabling `detect_non_ha_changes` allows Adaptive Lighting to detect all state changes, including those made outside of Home Assistant, by comparing the light's state to its previously used settings.
The `adaptive_lighting.manual_control` event is fired when a light is marked as "manually controlled," allowing for integration with automations 🤖.
With `expand_light_groups: false`, manual control belongs to the group. A direct member change cannot pause adaptation for only that member; use group-level manual control or enable expansion for individual tracking.
Explicit member targets in Adaptive Lighting services stay individual targets and do not mark or command the whole group.
Changing expansion at runtime discards tracking and pending adaptation for targets no longer used by any profile.
The Adaptive Lighting switch exposes these read-only attributes for its lights:
- `manual_control`: lights with any attribute marked as manually controlled.

View file

@ -75,6 +75,7 @@ All configuration options are listed below with their default values. These opti
| `intercept` | Intercept and adapt `light.turn_on` calls to enabling instantaneous color and brightness adaptation. 🏎️ Disable for lights that do not support `light.turn_on` with color and brightness. | `True` | `bool` |
| `multi_light_intercept` | Intercept and adapt `light.turn_on` calls that target multiple lights. ➗⚠️ This might result in splitting up a single `light.turn_on` call into multiple calls, e.g., when lights are in different switches. Requires `intercept` to be enabled. | `True` | `bool` |
| `include_config_in_attributes` | Show all options as attributes on the switch in Home Assistant when set to `true`. 📝 | `False` | `bool` |
| `expand_light_groups` | Expand light groups to their members (`true`, default). Set `false` to send commands to the group and track manual control for the group. Explicit member targets in services stay individual targets. | `True` | `bool` |
<!-- OUTPUT:END -->

View file

@ -11,6 +11,7 @@ except ImportError:
from voluptuous_serialize import convert as to_field_list
from homeassistant.components.adaptive_lighting.const import (
BASIC_OPTIONS,
CONF_EXPAND_LIGHT_GROUPS,
CONF_INITIAL_TRANSITION,
CONF_MANUAL_CONTROL_ON_EXTERNAL_TURN_ON,
CONF_SUNRISE_TIME,
@ -106,6 +107,7 @@ async def test_options(hass):
# Build input with advanced options nested in "advanced" section
advanced_data = ADVANCED_DATA.copy()
advanced_data[CONF_INITIAL_TRANSITION] = 23
advanced_data[CONF_EXPAND_LIGHT_GROUPS] = False
advanced_data[CONF_SUNRISE_TIME] = NONE_STR
advanced_data[CONF_SUNSET_TIME] = NONE_STR
basic_data = {**BASIC_DATA, "min_brightness": 12}

View file

@ -37,6 +37,7 @@ from homeassistant.components.adaptive_lighting.const import (
CONF_BRIGHTNESS_MODE_TIME_DARK,
CONF_BRIGHTNESS_MODE_TIME_LIGHT,
CONF_DETECT_NON_HA_CHANGES,
CONF_EXPAND_LIGHT_GROUPS,
CONF_INITIAL_TRANSITION,
CONF_MANUAL_CONTROL,
CONF_MANUAL_CONTROL_ON_EXTERNAL_TURN_ON,
@ -2301,7 +2302,8 @@ def test_attributes_have_changed():
async def test_state_change_handlers(hass):
"""Test AdaptiveLightingManager's EVENT_STATE_CHANGED listener.
======================
===============
Sequence of events:
1. Transition from sleep mode to normal.
2. Create simulated transition events for that adapt.
@ -3824,6 +3826,407 @@ async def test_light_group(
assert len(events) == 3
def _track_adaptive_light_calls(hass, *, ours_only=True):
"""Capture commands emitted by Adaptive Lighting at the HA service boundary."""
calls = []
def track(event):
if (
event.data["domain"] == LIGHT_DOMAIN
and event.data["service"] == SERVICE_TURN_ON
and (not ours_only or is_our_context(event.context))
):
calls.append(event.data["service_data"])
hass.bus.async_listen(EVENT_CALL_SERVICE, track)
return calls
async def _setup_group_switch(hass, **settings):
return await setup_switch(
hass,
{
CONF_LIGHTS: ["light.light_group"],
CONF_INITIAL_TRANSITION: 0,
CONF_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
**settings,
},
)
@pytest.mark.parametrize("intercept", [False, True])
async def test_light_group_expand_disabled_off_to_on(hass, intercept, cleanup):
"""Group adaptation reaches members through the group on both turn-on paths."""
await setup_lights(hass, with_group=True)
_, switch = await _setup_group_switch(
hass,
expand_light_groups=False,
intercept=intercept,
)
calls = _track_adaptive_light_calls(hass, ours_only=False)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_group"},
blocking=True,
)
await hass.async_block_till_done()
await asyncio.gather(*switch.manager.adaptation_tasks)
# HA emits the original service event before the interceptor changes its data.
# The group forwards the injected values to its real member entities.
expected_target = (
["light.light_4", "light.light_5"] if intercept else "light.light_group"
)
assert any(
call[ATTR_ENTITY_ID] == expected_target and call.get(ATTR_BRIGHTNESS) == 128
for call in calls
), calls
for member in ["light.light_4", "light.light_5"]:
assert hass.states.get(member).attributes[ATTR_BRIGHTNESS] == 128
calls.clear()
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert any(call[ATTR_ENTITY_ID] == "light.light_group" for call in calls)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_group", ATTR_BRIGHTNESS: 77},
blocking=True,
)
await hass.async_block_till_done()
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert hass.states.get(switch.entity_id).attributes["manual_control"] == [
"light.light_group",
]
for member in ["light.light_4", "light.light_5"]:
assert hass.states.get(member).attributes[ATTR_BRIGHTNESS] == 77
@pytest.mark.parametrize("expand", [False, True])
@pytest.mark.parametrize("explicit_member", [False, True])
async def test_group_apply_respects_target_policy(
hass,
expand,
explicit_member,
cleanup,
):
"""Apply obeys profile expansion without widening an explicit member request."""
await setup_lights(hass, with_group=True)
_, switch = await _setup_group_switch(hass, expand_light_groups=expand)
calls = _track_adaptive_light_calls(hass)
target = "light.light_4" if explicit_member else "light.light_group"
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{
ATTR_ENTITY_ID: switch.entity_id,
CONF_LIGHTS: [target],
CONF_TURN_ON_LIGHTS: True,
},
blocking=True,
)
await hass.async_block_till_done()
expected = (
["light.light_4"]
if explicit_member
else (["light.light_4", "light.light_5"] if expand else ["light.light_group"])
)
# HA also emits forwarded member calls; AL's own commands use a scalar target.
assert (
sorted(
{
call[ATTR_ENTITY_ID]
for call in calls
if isinstance(call[ATTR_ENTITY_ID], str)
},
)
== expected
)
assert hass.states.get("light.light_4").attributes[ATTR_BRIGHTNESS] == 128
assert hass.states.get("light.light_5").state == (
STATE_OFF if explicit_member else STATE_ON
)
@pytest.mark.parametrize("expand", [False, True])
async def test_group_manual_control_service(hass, expand, cleanup):
"""Group lookup and manual service use the same target policy, including reset."""
await setup_lights(hass, with_group=True)
_, switch = await _setup_group_switch(hass, expand_light_groups=expand)
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
{CONF_LIGHTS: ["light.light_group"], CONF_MANUAL_CONTROL: True},
blocking=True,
)
await hass.async_block_till_done()
expected = ["light.light_4", "light.light_5"] if expand else ["light.light_group"]
assert hass.states.get(switch.entity_id).attributes["manual_control"] == expected
for target in expected:
assert switch.manager.manual_control[target] == LightControlAttributes.ALL
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
{CONF_LIGHTS: ["light.light_group"], CONF_MANUAL_CONTROL: False},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(switch.entity_id).attributes["manual_control"] == []
@pytest.mark.parametrize("shared_member", [False, True])
async def test_group_runtime_expansion_restores_targets(hass, shared_member, cleanup):
"""Changing expansion back and forth restores groups and retires member timers."""
await setup_lights(hass, with_group=True)
_, switch = await _setup_group_switch(hass, autoreset_control_seconds=60)
if shared_member:
_, other = await _setup_group_switch(
hass,
name="member",
lights=["light.light_4"],
autoreset_control_seconds=60,
)
await other.async_turn_off()
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{ATTR_ENTITY_ID: switch.entity_id, CONF_TURN_ON_LIGHTS: True},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
{
ATTR_ENTITY_ID: switch.entity_id,
CONF_LIGHTS: ["light.light_group"],
CONF_MANUAL_CONTROL: True,
},
blocking=True,
)
manager = switch.manager
old_timers = dict(manager.auto_reset_manual_control_timers)
assert len(old_timers) == 2
calls = _track_adaptive_light_calls(hass)
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{ATTR_ENTITY_ID: switch.entity_id, CONF_EXPAND_LIGHT_GROUPS: False},
blocking=True,
)
await hass.async_block_till_done()
assert switch.lights == ["light.light_group"]
retained = {"light.light_4"} if shared_member else set()
assert manager.lights == {"light.light_group"} | retained
assert set(manager.auto_reset_manual_control_timers) == retained
for light, timer in old_timers.items():
assert timer.is_running() == (light in retained)
assert (
manager.manual_control.get("light.light_5", LightControlAttributes.NONE)
== LightControlAttributes.NONE
)
assert any(call[ATTR_ENTITY_ID] == "light.light_group" for call in calls)
calls.clear()
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{ATTR_ENTITY_ID: switch.entity_id, CONF_EXPAND_LIGHT_GROUPS: True},
blocking=True,
)
await hass.async_block_till_done()
assert switch.lights == ["light.light_4", "light.light_5"]
assert manager.lights == {"light.light_4", "light.light_5"}
expected = (
["light.light_5"] if shared_member else ["light.light_4", "light.light_5"]
)
assert sorted({call[ATTR_ENTITY_ID] for call in calls}) == expected
if shared_member:
assert manager.manual_control["light.light_4"] == LightControlAttributes.ALL
@pytest.mark.parametrize("expand", [False, True])
@pytest.mark.parametrize("shared_target", [False, True])
async def test_group_runtime_change_retires_delayed_events(
hass,
expand,
shared_target,
cleanup,
):
"""Delayed reactive handlers must not command targets this profile retired."""
await setup_lights(hass, with_group=True)
_, switch = await _setup_group_switch(
hass,
expand_light_groups=expand,
adapt_delay=0.1234,
)
retired_targets = (
{"light.light_4", "light.light_5"} if expand else {"light.light_group"}
)
retained_target = "light.light_4" if expand else "light.light_group"
if shared_target:
_, other = await _setup_group_switch(
hass,
name="retained",
lights=[retained_target],
expand_light_groups=False,
)
await other.async_turn_off()
entered, release = asyncio.Event(), asyncio.Event()
original_sleep = asyncio.sleep
delayed_count = 0
async def controlled_sleep(delay, *args, **kwargs):
nonlocal delayed_count
if delay == 0.1234:
delayed_count += 1
if delayed_count == (2 if expand else 1):
entered.set()
await release.wait()
else:
await original_sleep(delay, *args, **kwargs)
calls = _track_adaptive_light_calls(hass)
with patch.object(asyncio, "sleep", controlled_sleep):
try:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_group"},
blocking=True,
)
await asyncio.wait_for(entered.wait(), timeout=2)
await hass.services.async_call(
DOMAIN,
SERVICE_CHANGE_SWITCH_SETTINGS,
{
ATTR_ENTITY_ID: switch.entity_id,
CONF_EXPAND_LIGHT_GROUPS: not expand,
},
blocking=True,
)
expected = (
["light.light_group"] if expand else ["light.light_4", "light.light_5"]
)
assert switch.lights == expected
assert switch.manager.lights == set(expected) | (
{retained_target} if shared_target else set()
)
calls.clear()
finally:
release.set()
await hass.async_block_till_done()
retired_calls = [
call
for call in calls
if isinstance(call[ATTR_ENTITY_ID], str)
and call[ATTR_ENTITY_ID] in retired_targets
]
assert (
not retired_calls
), f"Retired reactive handlers issued commands: {retired_calls}"
for member in ["light.light_4", "light.light_5"]:
assert hass.states.get(member).attributes[ATTR_BRIGHTNESS] == 128
@pytest.mark.parametrize("trigger", ["turn_on", "autoreset"])
async def test_group_mixed_profiles_preserve_tracking(hass, trigger, cleanup):
"""Expanding one profile must not remove another profile's group tracking."""
await setup_lights(hass, with_group=True)
_, proxy = await _setup_group_switch(
hass,
expand_light_groups=False,
autoreset_control_seconds=60,
detect_non_ha_changes=True,
)
_, expanded = await _setup_group_switch(
hass,
name="expanded",
min_brightness=70,
max_brightness=70,
detect_non_ha_changes=True,
)
calls = _track_adaptive_light_calls(hass)
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{
ATTR_ENTITY_ID: expanded.entity_id,
CONF_LIGHTS: ["light.light_group"],
CONF_TURN_ON_LIGHTS: True,
},
blocking=True,
)
await hass.async_block_till_done()
assert "light.light_group" in proxy.manager.lights
calls.clear()
if trigger == "turn_on":
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_group"},
blocking=True,
)
await hass.async_block_till_done()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "light.light_group"},
blocking=True,
)
else:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_MANUAL_CONTROL,
{
ATTR_ENTITY_ID: proxy.entity_id,
CONF_LIGHTS: ["light.light_group"],
CONF_MANUAL_CONTROL: True,
},
blocking=True,
)
timer = proxy.manager.auto_reset_manual_control_timers["light.light_group"]
timer.delay = 0
timer.start()
await timer.task
await hass.async_block_till_done()
# Only the proxy profile may command the group; the expanded profile uses members.
assert {
call[ATTR_BRIGHTNESS]
for call in calls
if call[ATTR_ENTITY_ID] == "light.light_group" and ATTR_BRIGHTNESS in call
} == {128}
async def test_nested_group_apply_targets_leaves(hass, cleanup):
"""Default expansion reaches nested leaves without sending commands to subgroups."""
await setup_lights(hass, with_group=True)
hass.states.async_set(
"light.outer",
STATE_OFF,
{ATTR_ENTITY_ID: ["light.light_group", "light.light_3"]},
)
_, switch = await _setup_group_switch(hass, lights=["light.outer"])
calls = _track_adaptive_light_calls(hass)
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{CONF_LIGHTS: ["light.outer"], CONF_TURN_ON_LIGHTS: True},
blocking=True,
)
await hass.async_block_till_done()
assert sorted({call[ATTR_ENTITY_ID] for call in calls}) == [
"light.light_3",
"light.light_4",
"light.light_5",
]
assert switch.lights == ["light.light_3", "light.light_4", "light.light_5"]
def _state_changed_event(entity_id: str, ts: float, context: Context) -> Event:
return Event(
EVENT_STATE_CHANGED,
@ -3960,25 +4363,21 @@ async def test_just_turned_off_same_automation_context(hass, cleanup):
async def test_just_turned_off_group_context_reuse_end_to_end(hass, cleanup):
"""Drive the issue #1378 scenario through the real event bus listeners.
Unlike `test_just_turned_off_group_context_reuse`, which calls
`just_turned_off` directly, this test fires the service and state-changed
events on the bus. Light groups are normally expanded out of
`manager.lights`, but they can remain tracked in real setups (e.g., when a
group is nested inside another configured group or is unavailable during
setup), which is the configuration under which issue #1378 was reported.
"""
"""A tracked member turn-on explains a group's reused OFF context (#1378)."""
await setup_lights(hass, with_group=True)
_, switch = await setup_switch(hass, {CONF_LIGHTS: ["light.light_group"]})
_, switch = await _setup_group_switch(
hass,
lights=["light.light_group", "light.light_4"],
expand_light_groups=False,
detect_non_ha_changes=True,
)
await hass.async_block_till_done()
manager = switch.manager
group = "light.light_group"
member = "light.light_4"
assert member in manager.lights
# Simulate a setup in which the group entity itself remains tracked.
manager.lights.add(group)
assert group in manager.lights
turn_off_context = Context()
# The group was turned off...
@ -4008,25 +4407,18 @@ async def test_just_turned_off_group_context_reuse_end_to_end(hass, cleanup):
assert member in manager.turn_on_event
# ...which turned the group back on, but HA reused the old turn_off context.
with patch.object(
AdaptiveSwitch,
"_respond_to_off_to_on_event",
AsyncMock(),
) as respond:
hass.bus.async_fire(
EVENT_STATE_CHANGED,
{
"entity_id": group,
"old_state": State(group, STATE_OFF),
"new_state": State(group, STATE_ON),
},
context=turn_off_context,
)
await hass.async_block_till_done()
calls = _track_adaptive_light_calls(hass)
state = hass.states.get(group)
hass.states.async_set(group, STATE_ON, state.attributes, context=turn_off_context)
await hass.async_block_till_done()
# Adaptation must not have been cancelled as a polling artifact.
respond.assert_called_once()
assert respond.call_args[0][0] == group
# The real group command must survive polling-artifact detection.
assert any(
call[ATTR_ENTITY_ID] == group and call.get(ATTR_BRIGHTNESS) == 128
for call in calls
)
for entity_id in ["light.light_4", "light.light_5"]:
assert hass.states.get(entity_id).attributes[ATTR_BRIGHTNESS] == 128
@pytest.mark.parametrize("brightness_mode", ["linear", "tanh"])