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

Handle lights recovering from unavailable

Browse source
This commit is contained in:
Bas Nijholt 2026-09-06 14:13:29 -07:00
commit da75a16d36
3 changed files with 359 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -768,7 +768,7 @@ If lights stop adapting after you turn them on with a physical switch or a Zigbe
To adapt these turn-ons while still detecting later manual changes, enable `detect_non_ha_changes` and leave `manual_control_on_external_turn_on` disabled. This requires the light integration to report its state reliably. If you want Adaptive Lighting to keep adapting regardless of manual changes, disable `take_over_control` along with the options that require it: `detect_non_ha_changes`, `adapt_only_on_bare_turn_on`, and `manual_control_on_external_turn_on`.
This explains the physical-switch case in [#1056](https://github.com/basnijholt/adaptive-lighting/issues/1056), but not every report in that thread. If the light is not listed in `manual_control`, include diagnostics and debug logs from the failed turn-on when reporting it. Lights returning from `unavailable` after a power cut are a separate case from an `off` to `on` state change.
This explains the physical-switch case in [#1056](https://github.com/basnijholt/adaptive-lighting/issues/1056), but not every report in that thread. If the light is not listed in `manual_control`, include diagnostics and debug logs from the failed turn-on when reporting it. When a light returns directly from `unavailable` to `on`, Adaptive Lighting uses the same turn-on policy and, if adaptation is allowed, applies `adapt_delay` and `initial_transition` even with `only_once`. Existing `manual_control` remains set because availability alone cannot distinguish a power cycle from a temporary connection loss.
#### :bulb: Lights Not Responding or Turning On by Themselves

43
custom_components/adaptive_lighting/switch.py
View file

@ -47,6 +47,7 @@ from homeassistant.const import (
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import (
CALLBACK_TYPE,
@ -1564,9 +1565,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
# being turned off in 'interval' update, see #726
not self._detect_non_ha_changes
and is_our_context(context, "interval")
and (turn_on := self.manager.turn_on_event.get(light))
and (turn_off := self.manager.turn_off_event.get(light))
and turn_off.time_fired > turn_on.time_fired
and self.manager.last_service_call_was_turn_off(light)
):
_LOGGER.debug(
"%s: Light '%s' was turned just turned off, context.id='%s'",
@ -1688,8 +1687,22 @@ 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:
# Runtime settings may disable the profile or retire its target while waiting.
if not self.is_on or entity_id not in self.lights:
return
old_state = event.data["old_state"]
if (
old_state is not None
and old_state.state == STATE_UNAVAILABLE
and not self._detect_non_ha_changes
and self.manager.last_service_call_was_turn_off(entity_id)
):
_LOGGER.debug(
"%s: Skipping recovery of '%s' after a light.turn_off call",
self._name,
entity_id,
)
return
await self._update_attrs_and_maybe_adapt_lights(
@ -2777,6 +2790,7 @@ class AdaptiveLightingManager:
if old_state is not None and old_state.state == STATE_OFF
else None
)
old_unavailable = old_state is not None and old_state.state == STATE_UNAVAILABLE
if new_on:
_LOGGER.debug(
"Detected a '%s' 'state_changed' event: '%s' with context.id='%s'",
@ -2836,11 +2850,14 @@ class AdaptiveLightingManager:
entity_id,
event.context.id,
)
elif old_off and new_on:
# Tracks 'off' → 'on' state changes
elif (old_off or old_unavailable) and new_on:
# Treat a recovered light like one that was turned on, without
# treating the preceding loss of availability as a turn-off.
self.off_to_on_event[entity_id] = event
old_state_name = STATE_OFF if old_off else STATE_UNAVAILABLE
_LOGGER.debug(
"Detected an 'off' → 'on' event for '%s' with context.id='%s'",
"Detected an '%s' → 'on' event for '%s' with context.id='%s'",
old_state_name,
entity_id,
event.context.id,
)
@ -2858,7 +2875,7 @@ class AdaptiveLightingManager:
self.reset(entity_id, reset_manual_control=False)
lock = self.turn_off_locks.setdefault(entity_id, asyncio.Lock())
async with lock:
if await self.just_turned_off(entity_id):
if old_off and await self.just_turned_off(entity_id):
# Stop if a rapid 'off' → 'on' → 'off' happens.
_LOGGER.debug(
"Cancelling adjusting lights for %s",
@ -2878,6 +2895,14 @@ class AdaptiveLightingManager:
event,
)
def last_service_call_was_turn_off(self, entity_id: str) -> bool:
"""Return whether the most recent tracked service call turned a light off."""
turn_on = self.turn_on_event.get(entity_id)
turn_off = self.turn_off_event.get(entity_id)
return turn_off is not None and (
turn_on is None or turn_off.time_fired > turn_on.time_fired
)
async def update_manually_controlled_from_event(
self,
switch: AdaptiveSwitch,

324
tests/test_switch.py
View file

@ -30,6 +30,7 @@ from homeassistant.components.adaptive_lighting.const import (
ATTR_ADAPT_BRIGHTNESS,
ATTR_ADAPT_COLOR,
ATTR_ADAPTIVE_LIGHTING_MANAGER,
CONF_ADAPT_DELAY,
CONF_ADAPT_ONLY_ON_BARE_TURN_ON,
CONF_ADAPT_UNTIL_SLEEP,
CONF_AUTORESET_CONTROL,
@ -46,6 +47,7 @@ from homeassistant.components.adaptive_lighting.const import (
CONF_MIN_BRIGHTNESS,
CONF_MIN_COLOR_TEMP,
CONF_MULTI_LIGHT_INTERCEPT,
CONF_ONLY_ONCE,
CONF_PREFER_RGB_COLOR,
CONF_RESET_MANUAL_CONTROL_ON_SLEEP_MODE_CHANGE,
CONF_SEND_SPLIT_DELAY,
@ -117,6 +119,8 @@ from homeassistant.const import (
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
EntityCategory,
)
from homeassistant.core import Context, CoreState, Event, HomeAssistant, State
@ -4976,6 +4980,326 @@ async def test_manual_control_on_external_turn_on_external_state_change(
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
async def _recover_light_from_unavailable(hass, brightness=200):
"""Publish the state sequence emitted when a powered-off bulb reconnects."""
attributes = dict(hass.states.get(ENTITY_LIGHT_1).attributes)
attributes[ATTR_BRIGHTNESS] = brightness
hass.states.async_set(
ENTITY_LIGHT_1,
STATE_UNAVAILABLE,
attributes,
context=Context(id="became_unavailable"),
)
await hass.async_block_till_done()
hass.states.async_set(
ENTITY_LIGHT_1,
STATE_ON,
attributes,
context=Context(id="recovered_on"),
)
await hass.async_block_till_done()
@pytest.mark.parametrize(
("only_once", "take_over_control", "detect_non_ha_changes"),
[(False, True, True), (True, True, True), (False, False, False)],
)
async def test_unavailable_light_recovery_adapts_immediately(
hass,
only_once,
take_over_control,
detect_non_ha_changes,
):
"""A trusted unavailable-to-on recovery is an initial adaptation (#307)."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL: take_over_control,
CONF_DETECT_NON_HA_CHANGES: detect_non_ha_changes,
CONF_ONLY_ONCE: only_once,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 128
assert (
switch.manager.get_manual_control_attributes(ENTITY_LIGHT_1)
== LightControlAttributes.NONE
)
@pytest.mark.parametrize(
("detect_non_ha_changes", "external_turn_on_is_manual"),
[(False, False), (True, True)],
)
async def test_unavailable_light_recovery_respects_external_turn_on_policy(
hass,
detect_non_ha_changes,
external_turn_on_is_manual,
):
"""Recovery follows the same opt-in policy as an unmatched turn-on."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_DETECT_NON_HA_CHANGES: detect_non_ha_changes,
CONF_MANUAL_CONTROL_ON_EXTERNAL_TURN_ON: external_turn_on_is_manual,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
assert (
switch.manager.get_manual_control_attributes(ENTITY_LIGHT_1)
== LightControlAttributes.ALL
)
async def test_unavailable_light_recovery_preserves_manual_control(hass):
"""A reconnect does not discard existing manual-control intent."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
switch.manager.set_manual_control_attributes(ENTITY_LIGHT_1)
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
assert (
switch.manager.get_manual_control_attributes(ENTITY_LIGHT_1)
== LightControlAttributes.ALL
)
async def test_unavailable_light_recovery_adapts_non_manual_attributes(hass):
"""PAUSE_CHANGED keeps manual brightness while adapting color."""
switch, lights = await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL_MODE: TakeOverControlMode.PAUSE_CHANGED.value,
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
switch.manager.set_manual_control_attributes(
ENTITY_LIGHT_1,
LightControlAttributes.BRIGHTNESS,
)
switch.manager.last_service_data.pop(ENTITY_LIGHT_1, None)
set_light_brightness(lights[0], 200)
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
assert (
switch.manager.get_manual_control_attributes(ENTITY_LIGHT_1)
== LightControlAttributes.BRIGHTNESS
)
service_data = switch.manager.last_service_data[ENTITY_LIGHT_1]
assert ATTR_BRIGHTNESS not in service_data
assert ATTR_COLOR_TEMP_KELVIN in service_data
async def test_unavailable_light_recovery_ignores_disabled_profile(hass):
"""A disabled profile does not act when one of its lights reconnects."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_DETECT_NON_HA_CHANGES: True,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
await switch.async_turn_off()
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
async def test_unknown_to_on_does_not_trigger_recovery(hass):
"""An unknown-to-on update remains outside unavailable recovery."""
await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL: False,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
attributes = dict(hass.states.get(ENTITY_LIGHT_1).attributes)
attributes[ATTR_BRIGHTNESS] = 200
hass.states.async_set(ENTITY_LIGHT_1, STATE_UNKNOWN, attributes)
await hass.async_block_till_done()
hass.states.async_set(ENTITY_LIGHT_1, STATE_ON, attributes)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
async def test_unavailable_light_is_skipped_by_interval(hass):
"""Periodic updates do not send commands to unavailable lights."""
switch, _ = await setup_lights_and_switch(hass)
attributes = dict(hass.states.get(ENTITY_LIGHT_1).attributes)
hass.states.async_set(ENTITY_LIGHT_1, STATE_UNAVAILABLE, attributes)
await hass.async_block_till_done()
switch.manager.last_service_data.pop(ENTITY_LIGHT_1, None)
await switch._update_attrs_and_maybe_adapt_lights(
context=switch.create_context("interval"),
lights=[ENTITY_LIGHT_1],
)
assert ENTITY_LIGHT_1 not in switch.manager.last_service_data
@pytest.mark.parametrize(
("turn_on_timestamp", "should_adapt"),
[(None, False), (1.0, False), (3.0, True)],
)
async def test_interval_respects_latest_turn_off_service(
hass,
turn_on_timestamp,
should_adapt,
):
"""Interval adaptation follows the latest tracked on/off service intent."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL: False,
CONF_DETECT_NON_HA_CHANGES: False,
CONF_INITIAL_TRANSITION: 0,
},
)
manager = switch.manager
manager.last_service_data.pop(ENTITY_LIGHT_1, None)
manager.turn_off_event[ENTITY_LIGHT_1] = Event(
EVENT_CALL_SERVICE,
{},
time_fired_timestamp=2.0,
)
manager.turn_on_event.pop(ENTITY_LIGHT_1, None)
if turn_on_timestamp is not None:
manager.turn_on_event[ENTITY_LIGHT_1] = Event(
EVENT_CALL_SERVICE,
{},
time_fired_timestamp=turn_on_timestamp,
)
await switch._update_attrs_and_maybe_adapt_lights(
context=switch.create_context("interval"),
lights=[ENTITY_LIGHT_1],
)
assert (ENTITY_LIGHT_1 in manager.last_service_data) is should_adapt
async def test_unavailable_light_recovery_preserves_recent_turn_off(hass):
"""A false on report does not override a newer explicit turn-off."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL: False,
CONF_DETECT_NON_HA_CHANGES: False,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
switch.manager.turn_on_event.pop(ENTITY_LIGHT_1, None)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_LIGHT_1},
blocking=True,
)
await hass.async_block_till_done()
assert ENTITY_LIGHT_1 not in switch.manager.turn_on_event
await _recover_light_from_unavailable(hass)
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
@pytest.mark.parametrize("cancel_reason", ["light_turn_off", "profile_disabled"])
async def test_unavailable_light_recovery_cancelled_during_delay(
hass,
monkeypatch,
cancel_reason,
):
"""A changed off intent during recovery delay prevents adaptation."""
switch, _ = await setup_lights_and_switch(
hass,
{
CONF_TAKE_OVER_CONTROL: False,
CONF_DETECT_NON_HA_CHANGES: False,
CONF_ADAPT_DELAY: 0.1234,
CONF_INITIAL_TRANSITION: 0,
CONF_MIN_BRIGHTNESS: 50,
CONF_MAX_BRIGHTNESS: 50,
},
)
attributes = dict(hass.states.get(ENTITY_LIGHT_1).attributes)
attributes[ATTR_BRIGHTNESS] = 200
hass.states.async_set(ENTITY_LIGHT_1, STATE_UNAVAILABLE, attributes)
await hass.async_block_till_done()
entered_delay = asyncio.Event()
release_delay = asyncio.Event()
original_sleep = asyncio.sleep
async def controlled_sleep(delay, *args, **kwargs):
if delay == 0.1234:
entered_delay.set()
await release_delay.wait()
else:
await original_sleep(delay, *args, **kwargs)
monkeypatch.setattr(asyncio, "sleep", controlled_sleep)
hass.states.async_set(ENTITY_LIGHT_1, STATE_ON, attributes)
await entered_delay.wait()
if cancel_reason == "light_turn_off":
hass.bus.async_fire(
EVENT_CALL_SERVICE,
{
"domain": LIGHT_DOMAIN,
"service": SERVICE_TURN_OFF,
"service_data": {
ATTR_ENTITY_ID: ENTITY_LIGHT_1,
ATTR_TRANSITION: 10,
},
},
context=Context(id="turn_off_during_recovery_delay"),
)
await original_sleep(0)
assert hass.states.get(ENTITY_LIGHT_1).state == STATE_ON
assert switch.manager.last_service_call_was_turn_off(ENTITY_LIGHT_1)
else:
await switch.async_turn_off()
assert not switch.is_on
release_delay.set()
await hass.async_block_till_done()
assert hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS] == 200
@pytest.mark.parametrize("intercept", [True, False])
async def test_manual_control_on_external_turn_on_keeps_non_ha_change_detection(
hass,