Fix autoreset timer renewal for unchanged physical light states (#1561)

* fix: preserve manual-control timeout across polls

* fix: update manual baseline after adaptive writes

* fix: seed manual baseline from tracked changes
This commit is contained in:
Bas Nijholt 2026-09-06 14:25:24 +02:00 committed by GitHub
commit bbe5f3837d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 472 additions and 11 deletions

View file

@ -1437,6 +1437,10 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
data.context.id,
)
light = service_data[ATTR_ENTITY_ID]
self.manager.invalidate_manual_control_state(
light,
get_light_control_attributes(service_data),
)
self.manager.last_service_data[light] = {
**self.manager.last_service_data.get(light, {}),
**service_data,
@ -1624,10 +1628,18 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
and not is_our_context(event.context)
):
service_data = self.manager.turn_on_event[entity_id].data[ATTR_SERVICE_DATA]
manual_attributes = get_light_control_attributes(service_data)
if self.manager._mark_manual_control_if_non_bare_turn_on(
entity_id,
service_data,
):
new_state = event.data["new_state"]
assert new_state is not None
self.manager.update_manual_control_state(
entity_id,
new_state,
manual_attributes,
)
_LOGGER.debug(
"Marked attributes from service_data as manually controlled for '%s' "
"with context.id='%s'. Continuing to adapt remaining attributes. "
@ -1802,6 +1814,15 @@ class AdaptiveLightingManager:
self.our_last_state_on_change: dict[str, list[State]] = {}
# Track last 'service_data' to 'light.turn_on' resulting from this integration
self.last_service_data: dict[str, dict[str, Any]] = {}
# Track reported states that established manual control of each axis
self.last_manual_control_state: dict[
str,
dict[LightControlAttributes, dict[str, Any]],
] = {}
self.pending_manual_control_state: dict[
str,
dict[LightControlAttributes, str],
] = {}
# Track ongoing split adaptations to be able to cancel them
self.adaptation_tasks_brightness: dict[str, asyncio.Task[None]] = {}
self.adaptation_tasks_color: dict[str, asyncio.Task[None]] = {}
@ -2382,6 +2403,68 @@ class AdaptiveLightingManager:
new = current | attributes
self.set_manual_control_attributes(light, new)
def invalidate_manual_control_state(
self,
light: str,
attributes: LightControlAttributes,
) -> None:
"""Stop comparing adapted attributes with an older physical state."""
states = self.last_manual_control_state.get(light)
pending = self.pending_manual_control_state.get(light)
for attribute in LightControlAttributes:
if attribute in attributes and states is not None:
states.pop(attribute, None)
if attribute in attributes and pending is not None:
pending.pop(attribute, None)
if states == {}:
self.last_manual_control_state.pop(light)
if pending == {}:
self.pending_manual_control_state.pop(light)
def update_manual_control_state(
self,
light: str,
state: State,
attributes: LightControlAttributes,
) -> None:
"""Record the reported state that established manual control of each axis."""
states = self.last_manual_control_state.setdefault(light, {})
for attribute in LightControlAttributes:
if attribute in attributes:
states[attribute] = dict(state.attributes)
def mark_manual_control_state_pending(
self,
light: str,
attributes: LightControlAttributes,
context_id: str,
) -> None:
"""Wait for the reported state produced by a tracked service call."""
pending = self.pending_manual_control_state.setdefault(light, {})
for attribute in LightControlAttributes:
if attribute in attributes:
pending[attribute] = context_id
def consume_pending_manual_control_state(
self,
light: str,
state: State,
context_id: str | None = None,
) -> None:
"""Record a tracked service's reported state once it is available."""
pending = self.pending_manual_control_state.get(light)
if pending is None:
return
attributes = LightControlAttributes.NONE
for attribute, pending_context_id in tuple(pending.items()):
if context_id is None or context_id == pending_context_id:
attributes |= attribute
pending.pop(attribute)
if not pending:
self.pending_manual_control_state.pop(light)
if attributes:
self.update_manual_control_state(light, state, attributes)
def get_adaption_control_attributes(
self,
switch: AdaptiveSwitch,
@ -2460,6 +2543,8 @@ class AdaptiveLightingManager:
light,
)
self.manual_control[light] = LightControlAttributes.NONE
self.last_manual_control_state.pop(light, None)
self.pending_manual_control_state.pop(light, None)
if timer := self.auto_reset_manual_control_timers.pop(light, None):
timer.cancel()
self.our_last_state_on_change.pop(light, None)
@ -2615,7 +2700,6 @@ class AdaptiveLightingManager:
if old_state is not None and old_state.state == STATE_OFF
else None
)
if new_on:
_LOGGER.debug(
"Detected a '%s' 'state_changed' event: '%s' with context.id='%s'",
@ -2660,6 +2744,11 @@ class AdaptiveLightingManager:
self.start_transition_timer(entity_id)
elif last_state is not None:
self.our_last_state_on_change[entity_id].append(new_on)
self.consume_pending_manual_control_state(
entity_id,
new_on,
new_on.context.id,
)
if old_on and new_off:
# Tracks 'on' → 'off' state changes
@ -2737,6 +2826,11 @@ class AdaptiveLightingManager:
# Light was already on and 'light.turn_on' was not called by
# the adaptive_lighting integration.
self.mark_manual_control_state_pending(
light,
turn_on_attributes,
turn_on_event.context.id,
)
self.add_manual_control_attributes(light, turn_on_attributes)
switch.fire_manual_control_event(light, turn_on_event.context)
_LOGGER.debug(
@ -2804,14 +2898,29 @@ class AdaptiveLightingManager:
await async_update_entity(self.hass, light)
refreshed_state = self.hass.states.get(light)
assert refreshed_state is not None
self.consume_pending_manual_control_state(light, refreshed_state)
changed_attributes = _attributes_have_changed(
old_attributes=last_service_data,
new_attributes=refreshed_state.attributes,
light=light,
context=context,
)
manual_control = self.get_manual_control_attributes(light)
manual_control_states = self.last_manual_control_state.get(light, {})
changed_attributes = LightControlAttributes.NONE
for attribute in LightControlAttributes:
old_attributes = (
manual_control_states.get(attribute, last_service_data)
if attribute in manual_control
else last_service_data
)
changed_attributes |= attribute & _attributes_have_changed(
old_attributes=dict(old_attributes),
new_attributes=refreshed_state.attributes,
light=light,
context=context,
)
if changed_attributes:
self.update_manual_control_state(
light,
refreshed_state,
changed_attributes,
)
_LOGGER.debug(
"%s: State attributes %s of '%s' changed (%s) wrt 'last_service_data' (%s) (context.id=%s)",
switch._name,

View file

@ -27,6 +27,8 @@ from homeassistant.components.adaptive_lighting.color_and_brightness import (
from homeassistant.components.adaptive_lighting.const import (
ADAPT_BRIGHTNESS_SWITCH,
ADAPT_COLOR_SWITCH,
ATTR_ADAPT_BRIGHTNESS,
ATTR_ADAPT_COLOR,
ATTR_ADAPTIVE_LIGHTING_MANAGER,
CONF_ADAPT_ONLY_ON_BARE_TURN_ON,
CONF_ADAPT_UNTIL_SLEEP,
@ -117,7 +119,10 @@ from homeassistant.helpers import area_registry as ar
from homeassistant.helpers import entity_registry
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.setup import async_setup_component
from homeassistant.util.color import color_temperature_mired_to_kelvin
from homeassistant.util.color import (
color_temperature_kelvin_to_mired,
color_temperature_mired_to_kelvin,
)
from tests.common import MockConfigEntry
from tests.common import mock_area_registry as mock_ha_area_registry
@ -722,7 +727,7 @@ async def test_manual_control(
_LOGGER.debug("End of change_manual_control")
def increased_brightness():
return (light._attr_brightness + 100) % 255
return max(1, (light._attr_brightness + 100) % 255)
def increased_color_temp():
return max(
@ -1122,6 +1127,353 @@ async def test_interval_adaptation_preserves_manual_control_timeout(
)
@pytest.mark.parametrize("intercept", [False, True])
@pytest.mark.parametrize("mode", list(TakeOverControlMode))
@pytest.mark.parametrize(
"manual_attribute",
[LightControlAttributes.BRIGHTNESS, LightControlAttributes.COLOR],
)
async def test_tracked_change_seeds_non_ha_baseline(
hass,
freezer,
cleanup,
intercept,
mode,
manual_attribute,
):
"""A tracked service change must not be detected again by the next poll."""
switch, (light, *_) = await setup_lights_and_switch(
hass,
{
CONF_AUTORESET_CONTROL: 7200,
CONF_TAKE_OVER_CONTROL_MODE: mode,
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INTERCEPT: intercept,
},
)
await switch._update_attrs_and_maybe_adapt_lights(
context=switch.create_context("test"),
force=True,
transition=0,
)
await hass.async_block_till_done()
if manual_attribute == LightControlAttributes.BRIGHTNESS:
adaptive_value = light.brightness
attribute = ATTR_BRIGHTNESS
difference = 120
else:
adaptive_value = light.color_temp_kelvin
attribute = ATTR_COLOR_TEMP_KELVIN
difference = 500
assert adaptive_value is not None
manual_value = (
adaptive_value - difference
if adaptive_value >= difference
else adaptive_value + difference
)
events = []
hass.bus.async_listen(f"{DOMAIN}.manual_control", events.append)
service_context = Context()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: light.entity_id, attribute: manual_value},
blocking=True,
context=service_context,
)
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== manual_attribute
)
assert len(events) == 1
freezer.tick(90)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7110
)
assert len(events) == 1
if manual_attribute == LightControlAttributes.BRIGHTNESS:
set_light_brightness(light, adaptive_value)
else:
light._attr_color_temp_kelvin = adaptive_value
if hasattr(light, "_temperature"):
light._temperature = color_temperature_kelvin_to_mired(adaptive_value)
light.async_set_context(service_context)
light.async_write_ha_state()
await hass.async_block_till_done()
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7200
)
assert len(events) == 2
@pytest.mark.parametrize("intercept", [False, True])
@pytest.mark.parametrize("mode", list(TakeOverControlMode))
@pytest.mark.parametrize(
("first_manual_attribute", "second_manual_attribute"),
[
(LightControlAttributes.BRIGHTNESS, LightControlAttributes.COLOR),
(LightControlAttributes.COLOR, LightControlAttributes.BRIGHTNESS),
],
)
async def test_unchanged_non_ha_change_preserves_manual_control_timeout(
hass,
freezer,
cleanup,
intercept,
mode,
first_manual_attribute,
second_manual_attribute,
):
"""An unchanged physical state must not renew its manual-control timeout."""
switch, lights = await setup_lights_and_switch(
hass,
{
CONF_AUTORESET_CONTROL: 7200,
CONF_TAKE_OVER_CONTROL_MODE: mode,
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INTERCEPT: intercept,
},
)
light = lights[0]
lights_by_entity = {item.entity_id: item for item in lights}
await switch._update_attrs_and_maybe_adapt_lights(
context=switch.create_context("test"),
force=True,
transition=0,
)
await hass.async_block_till_done()
adaptive_brightness = light.brightness
assert adaptive_brightness is not None
adaptive_color_temp = light.color_temp_kelvin
assert adaptive_color_temp is not None
manual_values = {
LightControlAttributes.BRIGHTNESS: (
adaptive_brightness - 120
if adaptive_brightness >= 120
else adaptive_brightness + 120
),
LightControlAttributes.COLOR: (
adaptive_color_temp - 500
if adaptive_color_temp >= 2500
else adaptive_color_temp + 500
),
}
def set_physical_state(attribute):
if attribute == LightControlAttributes.BRIGHTNESS:
set_light_brightness(light, manual_values[attribute])
else:
color_temp_kelvin = manual_values[attribute]
light._attr_color_temp_kelvin = color_temp_kelvin
if hasattr(light, "_temperature"):
light._temperature = color_temperature_kelvin_to_mired(
color_temp_kelvin,
)
async def flush_physical_state(hass, entity_id):
lights_by_entity[entity_id].async_write_ha_state()
with patch(
"homeassistant.components.adaptive_lighting.switch.async_update_entity",
new=AsyncMock(side_effect=flush_physical_state),
):
set_physical_state(first_manual_attribute)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== first_manual_attribute
)
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7200
)
freezer.tick(90)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== first_manual_attribute
)
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7110
)
freezer.tick(90)
set_physical_state(second_manual_attribute)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== LightControlAttributes.ALL
)
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7200
)
@pytest.mark.parametrize("intercept", [False, True])
@pytest.mark.parametrize("mode", list(TakeOverControlMode))
@pytest.mark.parametrize(
"manual_attribute",
[LightControlAttributes.BRIGHTNESS, LightControlAttributes.COLOR],
)
async def test_apply_updates_non_ha_change_baseline(
hass,
freezer,
cleanup,
intercept,
mode,
manual_attribute,
):
"""An adaptive apply must become the baseline for later physical changes."""
switch, lights = await setup_lights_and_switch(
hass,
{
CONF_AUTORESET_CONTROL: 7200,
CONF_TAKE_OVER_CONTROL_MODE: mode,
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INTERCEPT: intercept,
},
)
light = lights[0]
lights_by_entity = {item.entity_id: item for item in lights}
await switch._update_attrs_and_maybe_adapt_lights(
context=switch.create_context("test"),
force=True,
transition=0,
)
await hass.async_block_till_done()
adaptive_value = (
light.brightness
if manual_attribute == LightControlAttributes.BRIGHTNESS
else light.color_temp_kelvin
)
assert adaptive_value is not None
difference = 120 if manual_attribute == LightControlAttributes.BRIGHTNESS else 500
manual_value = (
adaptive_value - difference
if adaptive_value >= difference
else adaptive_value + difference
)
def set_physical_state(value=manual_value):
if manual_attribute == LightControlAttributes.BRIGHTNESS:
set_light_brightness(light, value)
else:
light._attr_color_temp_kelvin = value
if hasattr(light, "_temperature"):
light._temperature = color_temperature_kelvin_to_mired(value)
async def flush_physical_state(hass, entity_id):
lights_by_entity[entity_id].async_write_ha_state()
with patch(
"homeassistant.components.adaptive_lighting.switch.async_update_entity",
new=AsyncMock(side_effect=flush_physical_state),
):
set_physical_state()
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== manual_attribute
)
direction = 1 if manual_value < adaptive_value else -1
small_change = (
15 if manual_attribute == LightControlAttributes.BRIGHTNESS else 60
)
freezer.tick(90)
set_physical_state(manual_value + direction * small_change)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7110
)
adapt_brightness = manual_attribute == LightControlAttributes.COLOR
adapt_color = manual_attribute == LightControlAttributes.BRIGHTNESS
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{
ATTR_ENTITY_ID: switch.entity_id,
CONF_LIGHTS: [light.entity_id],
ATTR_ADAPT_BRIGHTNESS: adapt_brightness,
ATTR_ADAPT_COLOR: adapt_color,
},
blocking=True,
)
await hass.async_block_till_done()
assert (
switch.manager.get_manual_control_attributes(light.entity_id)
== manual_attribute
)
freezer.tick(90)
pre_apply_value = manual_value + direction * small_change * 2
set_physical_state(pre_apply_value)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7200
)
await hass.services.async_call(
DOMAIN,
SERVICE_APPLY,
{
ATTR_ENTITY_ID: switch.entity_id,
CONF_LIGHTS: [light.entity_id],
ATTR_ADAPT_BRIGHTNESS: not adapt_brightness,
ATTR_ADAPT_COLOR: not adapt_color,
},
blocking=True,
)
await hass.async_block_till_done()
applied_value = (
light.brightness
if manual_attribute == LightControlAttributes.BRIGHTNESS
else light.color_temp_kelvin
)
assert applied_value != pre_apply_value
freezer.tick(90)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7110
)
set_physical_state(pre_apply_value)
await switch._async_update_at_interval_action()
await hass.async_block_till_done()
assert (
switch.extra_state_attributes["autoreset_time_remaining"][light.entity_id]
== 7200
)
@pytest.mark.parametrize("mode", list(TakeOverControlMode))
@pytest.mark.parametrize("service_data", [{}, {ATTR_BRIGHTNESS: 20}])
async def test_mixed_turn_on_restarts_manual_control_timeout(
@ -2808,7 +3160,7 @@ async def test_two_switches_for_single_light(hass):
_LOGGER.debug("Turn light %s, to %s", state, kwargs)
def increased_brightness():
return (light1._attr_brightness + 100) % 255
return max(1, (light1._attr_brightness + 100) % 255)
def increased_color_temp():
return max(