remote/adaptive-lighting
1
0
Fork 0
mirror of https://github.com/basnijholt/adaptive-lighting.git synced 2026-09-12 06:44:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: track mixed targets during light turn-off

Browse source
This commit is contained in:
Bas Nijholt 2026-09-06 14:10:22 -07:00
commit 69df1fa0a4
3 changed files with 276 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

33
custom_components/adaptive_lighting/hass_utils.py
View file

@ -4,31 +4,30 @@ import logging
from collections.abc import Awaitable, Callable
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry, entity_registry
from homeassistant.helpers.target import async_extract_referenced_entity_ids
from homeassistant.util.read_only_dict import ReadOnlyDict
try:
from homeassistant.helpers.target import TargetSelection
except ImportError: # Compatibility with older Home Assistant releases
from homeassistant.helpers.target import TargetSelectorData as TargetSelection
from .adaptation_utils import ServiceData
_LOGGER = logging.getLogger(__name__)
def area_entities(hass: HomeAssistant, area_id: str):
"""Get all entities linked to an area."""
ent_reg = entity_registry.async_get(hass)
entity_ids = [
entry.entity_id
for entry in entity_registry.async_entries_for_area(ent_reg, area_id)
]
dev_reg = device_registry.async_get(hass)
entity_ids.extend(
[
entity.entity_id
for device in device_registry.async_entries_for_area(dev_reg, area_id)
for entity in entity_registry.async_entries_for_device(ent_reg, device.id)
if entity.area_id is None
],
def target_entities(
hass: HomeAssistant,
service_data: ServiceData,
) -> set[str]:
"""Resolve all directly and indirectly targeted entities without groups."""
selected = async_extract_referenced_entity_ids(
hass,
TargetSelection(service_data),
expand_group=False,
)
return entity_ids
return selected.referenced | selected.indirectly_referenced
def setup_service_call_interceptor(

100
custom_components/adaptive_lighting/switch.py
View file

@ -11,7 +11,6 @@ from copy import deepcopy
from datetime import timedelta
from typing import TYPE_CHECKING, Any
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
import ulid_transform
from homeassistant.components.light import (
@ -32,8 +31,11 @@ from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_DOMAIN,
ATTR_ENTITY_ID,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
ATTR_SERVICE,
ATTR_SERVICE_DATA,
ATTR_SUPPORTED_FEATURES,
@ -149,7 +151,7 @@ from .const import (
change_switch_settings_schema,
replace_none_str,
)
from .hass_utils import area_entities, setup_service_call_interceptor
from .hass_utils import setup_service_call_interceptor, target_entities
from .helpers import (
clamp,
color_difference_redmean,
@ -1970,13 +1972,6 @@ class AdaptiveLightingManager:
self._context_cnt += 1
return context
def _is_excluded_from_area(self, entity_id: str) -> bool:
"""Match Home Assistant's exclusions for indirect area targets."""
entry = entity_registry.async_get(self.hass).async_get(entity_id)
return entry is not None and (
entry.entity_category is not None or entry.hidden_by is not None
)
def _separate_entity_ids(
self,
entity_ids: list[str],
@ -2160,8 +2155,14 @@ class AdaptiveLightingManager:
entity_ids: list[str],
) -> dict[str, Any]:
"""Modify the service data to contain the entity IDs."""
service_data.pop(ATTR_ENTITY_ID, None)
service_data.pop(ATTR_AREA_ID, None)
for target_key in (
ATTR_ENTITY_ID,
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
):
service_data.pop(target_key, None)
service_data[ATTR_ENTITY_ID] = entity_ids
return service_data
@ -2633,31 +2634,11 @@ class AdaptiveLightingManager:
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])
if ATTR_AREA_ID in service_data:
entity_ids: list[str] = []
area_ids: list[str] = cv.ensure_list_csv(service_data[ATTR_AREA_ID])
for area_id in area_ids:
area_entity_ids = area_entities(self.hass, area_id)
eids = [
entity_id
for entity_id in area_entity_ids
if entity_id.startswith(LIGHT_DOMAIN)
and not self._is_excluded_from_area(entity_id)
]
entity_ids.extend(eids)
_LOGGER.debug(
"Found entity_ids '%s' for area_id '%s'",
entity_ids,
area_id,
)
return entity_ids
_LOGGER.debug(
"No entity_ids or area_ids found in service_data: %s",
service_data,
return sorted(
entity_id
for entity_id in target_entities(self.hass, service_data)
if entity_id.startswith(f"{LIGHT_DOMAIN}.")
)
return []
async def turn_on_off_event_listener(self, event: Event) -> None:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
@ -3052,7 +3033,7 @@ class AdaptiveLightingManager:
def _member_turn_on_explains_group_turn_on(
self,
entity_id: str,
on_to_off_event: Event[EventStateChangedData],
off_event: Event,
off_to_on_event: Event[EventStateChangedData],
) -> bool:
"""Check if a light group's 'off' → 'on' is caused by a member's 'light.turn_on'.
@ -3072,7 +3053,7 @@ class AdaptiveLightingManager:
member_turn_on = self.turn_on_event.get(member)
if (
member_turn_on is not None
and on_to_off_event.time_fired
and off_event.time_fired
< member_turn_on.time_fired
<= off_to_on_event.time_fired
):
@ -3087,6 +3068,49 @@ class AdaptiveLightingManager:
return True
return False
def _off_to_on_event_is_during_turn_off(
self,
entity_id: str,
off_to_on_event: Event[EventStateChangedData],
) -> bool:
"""Check if a reported turn-on belongs to a recent turn-off window."""
turn_off_event = self.turn_off_event.get(entity_id)
if (
turn_off_event is None
or off_to_on_event.context.id != turn_off_event.context.id
):
return False
turn_on_event = self.turn_on_event.get(entity_id)
if (
turn_on_event is not None
and turn_off_event.time_fired
< turn_on_event.time_fired
<= off_to_on_event.time_fired
):
return False
if self._member_turn_on_explains_group_turn_on(
entity_id,
turn_off_event,
off_to_on_event,
):
return False
transition = turn_off_event.data[ATTR_SERVICE_DATA].get(ATTR_TRANSITION)
delay = max(transition or 0, TURNING_OFF_DELAY)
elapsed = (dt_util.utcnow() - turn_off_event.time_fired).total_seconds()
if not 0 <= elapsed <= delay:
return False
_LOGGER.debug(
"just_turned_off: Fresh 'light.turn_off' for '%s' shares the"
" 'off' → 'on' context; ignoring the state during its %s second"
" transition window.",
entity_id,
delay,
)
return True
async def just_turned_off( # noqa: PLR0911, PLR0912
self,
entity_id: str,
@ -3105,6 +3129,8 @@ class AdaptiveLightingManager:
"""
off_to_on_event = self.off_to_on_event[entity_id]
on_to_off_event = self.on_to_off_event.get(entity_id)
if self._off_to_on_event_is_during_turn_off(entity_id, off_to_on_event):
return True
if on_to_off_event is None:
_LOGGER.debug(

197
tests/test_switch.py
View file

@ -108,6 +108,8 @@ from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_ENTITY_ID,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
ATTR_SUPPORTED_FEATURES,
CONF_LIGHTS,
CONF_NAME,
@ -4251,6 +4253,27 @@ def _turn_on_service_event(entity_ids: list[str], ts: float, context: Context) -
)
def _turn_off_service_event(
entity_ids: list[str],
ts: float,
context: Context,
transition: float,
) -> Event:
return Event(
EVENT_CALL_SERVICE,
{
"domain": LIGHT_DOMAIN,
"service": SERVICE_TURN_OFF,
"service_data": {
ATTR_ENTITY_ID: entity_ids,
ATTR_TRANSITION: transition,
},
},
time_fired_timestamp=ts,
context=context,
)
async def test_just_turned_off_group_context_reuse(hass, cleanup):
"""Group 'off' → 'on' with a reused 'turn_off' context must still adapt.
@ -4309,6 +4332,154 @@ async def test_just_turned_off_group_context_reuse(hass, cleanup):
assert await manager.just_turned_off(group)
def _register_mixed_target_lights(
hass,
device_registry,
floor_registry,
label_registry,
):
"""Assign the three test lights to mixed indirect HA targets."""
floor = floor_registry.async_create("Upstairs")
area_registry = ar.async_get(hass)
upstairs_area = area_registry.async_create(
"Upstairs room",
floor_id=floor.floor_id,
)
hall_area = area_registry.async_create("Hall")
config_entry = MockConfigEntry(domain="test")
config_entry.add_to_hass(hass)
device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={("test", "device-target")},
)
label = label_registry.async_create("Skipped light")
registry = entity_registry.async_get(hass)
registry.async_update_entity(ENTITY_LIGHT_1, area_id=upstairs_area.id)
registry.async_update_entity(ENTITY_LIGHT_2, area_id=hall_area.id)
registry.async_update_entity(
ENTITY_LIGHT_3,
device_id=device.id,
labels={label.label_id},
)
return {
ATTR_FLOOR_ID: floor.floor_id,
ATTR_AREA_ID: hall_area.id,
ATTR_DEVICE_ID: device.id,
ATTR_LABEL_ID: label.label_id,
}
async def test_mixed_turn_off_targets_do_not_readapt_off_device_light(
hass,
device_registry,
floor_registry,
label_registry,
cleanup,
):
"""A mixed-target turn-off must cover an already-off device light (#1069)."""
await setup_lights(hass)
targets = _register_mixed_target_lights(
hass,
device_registry,
floor_registry,
label_registry,
)
targets.pop(ATTR_LABEL_ID)
_, switch = await setup_switch(
hass,
{
CONF_LIGHTS: [ENTITY_LIGHT_1, ENTITY_LIGHT_2, ENTITY_LIGHT_3],
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INTERCEPT: True,
CONF_INITIAL_TRANSITION: 0,
},
)
assert hass.states.is_state(ENTITY_LIGHT_3, STATE_OFF)
turn_off_context = Context()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{
**targets,
ATTR_TRANSITION: 10,
},
blocking=True,
context=turn_off_context,
)
await hass.async_block_till_done()
calls = _track_adaptive_light_calls(hass)
off_state = hass.states.get(ENTITY_LIGHT_3)
assert off_state is not None
hass.states.async_set(
ENTITY_LIGHT_3,
STATE_ON,
off_state.attributes,
context=turn_off_context,
)
await hass.async_block_till_done()
assert not calls
async def test_intercept_replaces_all_mixed_target_selectors(
hass,
device_registry,
floor_registry,
label_registry,
cleanup,
):
"""A narrowed intercepted call must not retain indirect target selectors."""
lights = await setup_lights(hass)
targets = _register_mixed_target_lights(
hass,
device_registry,
floor_registry,
label_registry,
)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: [ENTITY_LIGHT_1, ENTITY_LIGHT_2, ENTITY_LIGHT_3],
},
blocking=True,
)
await setup_switch(
hass,
{
CONF_LIGHTS: [ENTITY_LIGHT_1, ENTITY_LIGHT_2],
CONF_INTERCEPT: True,
CONF_MULTI_LIGHT_INTERCEPT: True,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
with patch.object(
lights[2],
"async_turn_on",
wraps=lights[2].async_turn_on,
) as skipped_turn_on:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{**targets, ATTR_BRIGHTNESS: 200},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 128
assert hass.states.get(ENTITY_LIGHT_2).attributes[ATTR_BRIGHTNESS] == 128
skipped_turn_on.assert_awaited_once()
assert skipped_turn_on.call_args.kwargs[ATTR_BRIGHTNESS] == 200
async def test_just_turned_off_same_automation_context(hass, cleanup):
"""'turn_off' and 'turn_on' from one automation share a context.
@ -4325,6 +4496,12 @@ async def test_just_turned_off_same_automation_context(hass, cleanup):
now = dt_util.utcnow().timestamp()
automation_context = Context()
manager.turn_off_event[ENTITY_LIGHT_1] = _turn_off_service_event(
[ENTITY_LIGHT_1],
now - 2,
automation_context,
transition=10,
)
manager.on_to_off_event[ENTITY_LIGHT_1] = _state_changed_event(
ENTITY_LIGHT_1,
now - 2,
@ -4363,6 +4540,26 @@ async def test_just_turned_off_same_automation_context(hass, cleanup):
)
assert await manager.just_turned_off(ENTITY_LIGHT_1)
# A later physical turn-on has a fresh context and must not remain blocked by
# the old turn-off record after its transition window has elapsed.
manager.on_to_off_event[ENTITY_LIGHT_1] = _state_changed_event(
ENTITY_LIGHT_1,
now - 20,
automation_context,
)
manager.turn_off_event[ENTITY_LIGHT_1] = _turn_off_service_event(
[ENTITY_LIGHT_1],
now - 20,
automation_context,
transition=10,
)
manager.off_to_on_event[ENTITY_LIGHT_1] = _state_changed_event(
ENTITY_LIGHT_1,
now,
Context(),
)
assert not await manager.just_turned_off(ENTITY_LIGHT_1)
async def test_just_turned_off_group_context_reuse_end_to_end(hass, cleanup):
"""A tracked member turn-on explains a group's reused OFF context (#1378)."""