Resolve group targets consistently across adaptation paths

This commit is contained in:
Bas Nijholt 2026-09-06 09:22:10 -07:00
commit 4cda1c544a
11 changed files with 410 additions and 96 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.
@ -165,6 +169,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

@ -283,11 +283,12 @@ DOCS[CONF_MULTI_LIGHT_INTERCEPT] = (
CONF_EXPAND_LIGHT_GROUPS, DEFAULT_EXPAND_LIGHT_GROUPS = "expand_light_groups", True
DOCS[CONF_EXPAND_LIGHT_GROUPS] = (
"Expand light groups to their individual member entities (`true`, default). "
"Set to `false` to send adaptation commands to the group entity directly "
"instead of its members."
"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"

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,7 @@
"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`. 📝",
"expand_light_groups": "expand_light_groups: Expand light groups to their individual member entities (`true`, default). Set to `false` to send adaptation commands to the group entity directly instead of its members."
"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 +211,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

@ -252,15 +252,10 @@ 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)
# A switch with expand_light_groups=False stores the group entity in
# switch.lights (not its children), so don't expand the incoming lights for it.
check_lights = all_check_lights if switch._expand_light_groups_flag else 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
@ -424,7 +419,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):
@ -459,7 +454,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],
)
@ -628,17 +623,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)
@ -891,7 +891,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)
@ -1071,18 +1072,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
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:
all_lights = _expand_light_groups(hass, self.lights)
else:
all_lights = list(self.lights) # keep group entities as-is
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)
@ -1522,6 +1534,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)]
@ -2376,7 +2389,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
@ -2567,6 +2584,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])
@ -2803,7 +2844,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,7 @@
"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`. 📝",
"expand_light_groups": "expand_light_groups: Expand light groups to their individual member entities (`true`, default). Set to `false` to send adaptation commands to the group entity directly instead of its members."
"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 +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

@ -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

@ -74,6 +74,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_SUNRISE_TIME,
CONF_SUNSET_TIME,
@ -104,6 +105,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

@ -3675,55 +3675,46 @@ async def test_light_group(
assert len(events) == 3
async def test_light_group_expand_disabled_off_to_on(hass):
"""Regression test: off→on reactive adaptation works when expand_light_groups=False.
def _track_adaptive_light_calls(hass, *, ours_only=True):
"""Capture commands emitted by Adaptive Lighting at the HA service boundary."""
calls = []
When expand_light_groups=False the switch stores the group entity in switch.lights.
_switches_with_lights must not expand the incoming state-change entity_id either,
otherwise the intersection is empty and the switch is never found, silently skipping
reactive adaptation.
"""
await setup_lights(hass, with_group=True)
entity_ids = ["light.light_group"]
_, switch = await setup_switch(
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: entity_ids,
CONF_EXPAND_LIGHT_GROUPS: False,
CONF_LIGHTS: ["light.light_group"],
CONF_INITIAL_TRANSITION: 0,
CONF_TRANSITION: 0,
CONF_INTERCEPT: False, # disable interceptor so reactive path is exercised
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
**settings,
},
)
await hass.async_block_till_done()
assert switch.is_on
# With expand disabled, switch.lights should contain the group entity, not children
assert "light.light_group" in switch.lights
assert "light.light_4" not in switch.lights
assert "light.light_5" not in switch.lights
# Turn the group off first
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.light_group"},
blocking=True,
@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,
)
await hass.async_block_till_done()
# Track adaptation calls triggered by the off→on event
adapted = []
original = switch._respond_to_off_to_on_event
async def track_adapt(entity_id, event):
adapted.append(entity_id)
return await original(entity_id, event)
switch._respond_to_off_to_on_event = track_adapt
# Turn the group back on from OFF — this fires a state_changed event for
# "light.light_group"; the reactive path must find the switch via this entity.
calls = _track_adaptive_light_calls(hass, ours_only=False)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
@ -3732,11 +3723,271 @@ async def test_light_group_expand_disabled_off_to_on(hass):
)
await hass.async_block_till_done()
await asyncio.gather(*switch.manager.adaptation_tasks)
assert "light.light_group" in adapted, (
"Switch was not found via _switches_with_lights for the group entity "
"when expand_light_groups=False — reactive off→on adaptation was skipped."
# 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("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:
@ -3875,25 +4126,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...
@ -3923,25 +4170,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"])