From 648c4fb66caf85e82eedc50036cd5d35196918df Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sun, 22 Feb 2026 02:45:40 +0100 Subject: [PATCH 1/5] WIP: Clean up light transitioning flag on task cancellation --- zha/application/platforms/light/__init__.py | 466 ++++++++++---------- 1 file changed, 245 insertions(+), 221 deletions(-) diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index dcf4393b4..7fe3475c7 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -372,210 +372,224 @@ async def async_turn_on( if set_transition_flag: self.async_transition_set_flag() - # If the light is currently off but a turn_on call with a color/temperature is - # sent, the light needs to be turned on first at a low brightness level where - # the light is immediately transitioned to the correct color. Afterwards, the - # transition is only from the low brightness to the new brightness. - # Otherwise, the transition is from the color the light had before being turned - # on to the new color. This can look especially bad with transitions longer than - # a second. We do not want to do this for devices that need to be forced to use - # the on command because we would end up with 4 commands sent: - # move to level, on, color, move to level... We also will not set this - # if the bulb is already in the desired color mode with the desired color - # or color temperature. - new_color_provided_while_off = ( - self._zha_config_enhanced_light_transition - and not self._FORCE_ON - and not self._state - and ( - ( - color_temp is not None - and ( - self._color_temp != color_temp - or self._color_mode != ColorMode.COLOR_TEMP + try: + # If the light is currently off but a turn_on call with a color/temperature is + # sent, the light needs to be turned on first at a low brightness level where + # the light is immediately transitioned to the correct color. Afterwards, the + # transition is only from the low brightness to the new brightness. + # Otherwise, the transition is from the color the light had before being turned + # on to the new color. This can look especially bad with transitions longer than + # a second. We do not want to do this for devices that need to be forced to use + # the on command because we would end up with 4 commands sent: + # move to level, on, color, move to level... We also will not set this + # if the bulb is already in the desired color mode with the desired color + # or color temperature. + new_color_provided_while_off = ( + self._zha_config_enhanced_light_transition + and not self._FORCE_ON + and not self._state + and ( + ( + color_temp is not None + and ( + self._color_temp != color_temp + or self._color_mode != ColorMode.COLOR_TEMP + ) + ) + or ( + xy_color is not None + and ( + self._xy_color != xy_color + or self._color_mode != ColorMode.XY + ) ) ) - or ( - xy_color is not None - and (self._xy_color != xy_color or self._color_mode != ColorMode.XY) - ) + and brightness_supported + and not execute_if_off_supported ) - and brightness_supported - and not execute_if_off_supported - ) - if ( - brightness is None - and (self._off_with_transition or new_color_provided_while_off) - and self._off_brightness is not None - ): - brightness = self._off_brightness - - if brightness is not None: - level = min(254, brightness) - else: - level = self._brightness or 254 + if ( + brightness is None + and (self._off_with_transition or new_color_provided_while_off) + and self._off_brightness is not None + ): + brightness = self._off_brightness - t_log = {} + if brightness is not None: + level = min(254, brightness) + else: + level = self._brightness or 254 - if new_color_provided_while_off: - assert self._level_cluster_handler is not None + t_log = {} - # If the light is currently off, we first need to turn it on at a low - # brightness level with no transition. - # After that, we set it to the desired color/temperature with no transition. - result = await self._level_cluster_handler.move_to_level_with_on_off( - level=DEFAULT_MIN_BRIGHTNESS, - transition_time=int(10 * self._DEFAULT_MIN_TRANSITION_TIME), - ) - t_log["move_to_level_with_on_off"] = result - if result[1] is not Status.SUCCESS: - # First 'move to level' call failed, so if the transitioning delay - # isn't running from a previous call, - # the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return - # Currently only setting it to "on", as the correct level state will - # be set at the second move_to_level call - self._state = True - - if execute_if_off_supported: - self.debug("handling color commands before turning on/level") - if not await self.async_handle_color_commands( - color_temp, - duration, # duration is ignored by lights when off - xy_color, - new_color_provided_while_off, - t_log, - ): - # Color calls before on/level calls failed, - # so if the transitioning delay isn't running from a previous call, - # the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return + if new_color_provided_while_off: + assert self._level_cluster_handler is not None - if ( - (brightness is not None or transition is not None) - and not new_color_provided_while_off - and brightness_supported - ): - assert self._level_cluster_handler is not None + # If the light is currently off, we first need to turn it on at a low + # brightness level with no transition. + # After that, we set it to the desired color/temperature with no transition. + result = await self._level_cluster_handler.move_to_level_with_on_off( + level=DEFAULT_MIN_BRIGHTNESS, + transition_time=int(10 * self._DEFAULT_MIN_TRANSITION_TIME), + ) + t_log["move_to_level_with_on_off"] = result + if result[1] is not Status.SUCCESS: + # First 'move to level' call failed, so if the transitioning delay + # isn't running from a previous call, + # the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return + # Currently only setting it to "on", as the correct level state will + # be set at the second move_to_level call + self._state = True + + if execute_if_off_supported: + self.debug("handling color commands before turning on/level") + if not await self.async_handle_color_commands( + color_temp, + duration, # duration is ignored by lights when off + xy_color, + new_color_provided_while_off, + t_log, + ): + # Color calls before on/level calls failed, + # so if the transitioning delay isn't running from a previous call, + # the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return - result = await self._level_cluster_handler.move_to_level_with_on_off( - level=level, - transition_time=int(10 * duration), - ) - t_log["move_to_level_with_on_off"] = result - if result[1] is not Status.SUCCESS: - # First 'move to level' call failed, so if the transitioning delay - # isn't running from a previous call, the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return - self._state = bool(level) - if level: - self._brightness = level + if ( + (brightness is not None or transition is not None) + and not new_color_provided_while_off + and brightness_supported + ): + assert self._level_cluster_handler is not None - if ( - (brightness is None and transition is None) - and not new_color_provided_while_off - or (self._FORCE_ON and brightness != 0) - ): - assert self._on_off_cluster_handler is not None + result = await self._level_cluster_handler.move_to_level_with_on_off( + level=level, + transition_time=int(10 * duration), + ) + t_log["move_to_level_with_on_off"] = result + if result[1] is not Status.SUCCESS: + # First 'move to level' call failed, so if the transitioning delay + # isn't running from a previous call, the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return + self._state = bool(level) + if level: + self._brightness = level - # since FORCE_ON lights don't turn on with move_to_level_with_on_off, - # we should call the on command on the on_off cluster - # if brightness is not 0. - result = await self._on_off_cluster_handler.on() - t_log["on_off"] = result - if result[1] is not Status.SUCCESS: - # 'On' call failed, but as brightness may still transition - # (for FORCE_ON lights), we start the timer to unset the flag after - # the transition_time if necessary. - self.async_transition_start_timer(transition_time) - self.debug("turned on: %s", t_log) - return - self._state = True - - if not execute_if_off_supported: - self.debug("handling color commands after turning on/level") - if not await self.async_handle_color_commands( - color_temp, - duration, - xy_color, - new_color_provided_while_off, - t_log, + if ( + (brightness is None and transition is None) + and not new_color_provided_while_off + or (self._FORCE_ON and brightness != 0) ): - # Color calls failed, but as brightness may still transition, - # we start the timer to unset the flag - self.async_transition_start_timer(transition_time) - self.debug("turned on: %s", t_log) - return - - if new_color_provided_while_off: - assert self._level_cluster_handler is not None + assert self._on_off_cluster_handler is not None + + # since FORCE_ON lights don't turn on with move_to_level_with_on_off, + # we should call the on command on the on_off cluster + # if brightness is not 0. + result = await self._on_off_cluster_handler.on() + t_log["on_off"] = result + if result[1] is not Status.SUCCESS: + # 'On' call failed, but as brightness may still transition + # (for FORCE_ON lights), we start the timer to unset the flag after + # the transition_time if necessary. + self.async_transition_start_timer(transition_time) + self.debug("turned on: %s", t_log) + return + self._state = True + + if not execute_if_off_supported: + self.debug("handling color commands after turning on/level") + if not await self.async_handle_color_commands( + color_temp, + duration, + xy_color, + new_color_provided_while_off, + t_log, + ): + # Color calls failed, but as brightness may still transition, + # we start the timer to unset the flag + self.async_transition_start_timer(transition_time) + self.debug("turned on: %s", t_log) + return + + if new_color_provided_while_off: + assert self._level_cluster_handler is not None + + # The light has the correct color, so we can now transition + # it to the correct brightness level. + result = await self._level_cluster_handler.move_to_level( + level=level, transition_time=int(10 * duration) + ) + t_log["move_to_level_if_color"] = result + if result[1] is not Status.SUCCESS: + self.debug("turned on: %s", t_log) + return + self._state = bool(level) + if level: + self._brightness = level + + # Our light is guaranteed to have just started the transitioning process + # if necessary, so we start the delay for the transition (to stop parsing + # attribute reports after the completed transition). + self.async_transition_start_timer(transition_time) - # The light has the correct color, so we can now transition - # it to the correct brightness level. - result = await self._level_cluster_handler.move_to_level( - level=level, transition_time=int(10 * duration) - ) - t_log["move_to_level_if_color"] = result - if result[1] is not Status.SUCCESS: - self.debug("turned on: %s", t_log) - return - self._state = bool(level) - if level: - self._brightness = level + if self._color_cluster_handler is not None: + if effect == EFFECT_COLORLOOP: + result = await self._color_cluster_handler.color_loop_set( + update_flags=( + Color.ColorLoopUpdateFlags.Action + | Color.ColorLoopUpdateFlags.Direction + | Color.ColorLoopUpdateFlags.Time + ), + action=Color.ColorLoopAction.Activate_from_current_hue, + direction=Color.ColorLoopDirection.Increment, + time=transition if transition else 7, + start_hue=0, + ) + t_log["color_loop_set"] = result + self._effect = EFFECT_COLORLOOP + elif self._effect == EFFECT_COLORLOOP and effect != EFFECT_COLORLOOP: + result = await self._color_cluster_handler.color_loop_set( + update_flags=Color.ColorLoopUpdateFlags.Action, + action=Color.ColorLoopAction.Deactivate, + direction=Color.ColorLoopDirection.Decrement, + time=0, + start_hue=0, + ) + t_log["color_loop_set"] = result + self._effect = EFFECT_OFF - # Our light is guaranteed to have just started the transitioning process - # if necessary, so we start the delay for the transition (to stop parsing - # attribute reports after the completed transition). - self.async_transition_start_timer(transition_time) - - if self._color_cluster_handler is not None: - if effect == EFFECT_COLORLOOP: - result = await self._color_cluster_handler.color_loop_set( - update_flags=( - Color.ColorLoopUpdateFlags.Action - | Color.ColorLoopUpdateFlags.Direction - | Color.ColorLoopUpdateFlags.Time - ), - action=Color.ColorLoopAction.Activate_from_current_hue, - direction=Color.ColorLoopDirection.Increment, - time=transition if transition else 7, - start_hue=0, - ) - t_log["color_loop_set"] = result - self._effect = EFFECT_COLORLOOP - elif self._effect == EFFECT_COLORLOOP and effect != EFFECT_COLORLOOP: - result = await self._color_cluster_handler.color_loop_set( - update_flags=Color.ColorLoopUpdateFlags.Action, - action=Color.ColorLoopAction.Deactivate, - direction=Color.ColorLoopDirection.Decrement, - time=0, - start_hue=0, + if flash is not None: + assert self._identify_cluster_handler is not None + result = await self._identify_cluster_handler.trigger_effect( + effect_id=FLASH_EFFECTS[flash], + effect_variant=Identify.EffectVariant.Default, ) - t_log["color_loop_set"] = result - self._effect = EFFECT_OFF - - if flash is not None: - assert self._identify_cluster_handler is not None - result = await self._identify_cluster_handler.trigger_effect( - effect_id=FLASH_EFFECTS[flash], - effect_variant=Identify.EffectVariant.Default, - ) - t_log["trigger_effect"] = result + t_log["trigger_effect"] = result - self._off_with_transition = False - self._off_brightness = None - self.debug("turned on: %s", t_log) - self.maybe_emit_state_changed_event() + self._off_with_transition = False + self._off_brightness = None + self.debug("turned on: %s", t_log) + self.maybe_emit_state_changed_event() + finally: + # If the task was cancelled (e.g. by a mode: restart automation) before + # the transition timer was started, clean up the transitioning flag so + # the light does not get stuck in a transitioning state indefinitely. + if ( + set_transition_flag + and self._transitioning_individual + and not self._transition_listener + ): + self.async_transition_complete() async def async_turn_off(self, *, transition: float | None = None) -> None: """Turn the entity off.""" @@ -594,41 +608,51 @@ async def async_turn_off(self, *, transition: float | None = None) -> None: if self._zha_config_enable_light_transitioning_flag: self.async_transition_set_flag() - # is not none looks odd here, but it will override built in bulb - # transition times if we pass 0 in here - if transition is not None and brightness_supported: - assert self._level_cluster_handler is not None - - result = await self._level_cluster_handler.move_to_level_with_on_off( - level=0, - transition_time=int( - 10 * (transition or self._DEFAULT_MIN_TRANSITION_TIME) - ), - ) - else: - assert self._on_off_cluster_handler is not None - result = await self._on_off_cluster_handler.off() + try: + # is not none looks odd here, but it will override built in bulb + # transition times if we pass 0 in here + if transition is not None and brightness_supported: + assert self._level_cluster_handler is not None - # Pause parsing attribute reports until transition is complete - if self._zha_config_enable_light_transitioning_flag: - self.async_transition_start_timer(transition_time) - self.debug("turned off: %s", result) - if result[1] is not Status.SUCCESS: - return - self._state = False - - if brightness_supported and not self._off_with_transition: - # store current brightness so that the next turn_on uses it: - # when using "enhanced turn on" - self._off_brightness = self._brightness - if transition is not None: - # save for when calling turn_on without a brightness: - # current_level is set to 1 after transitioning to level 0, - # needed for correct state with light groups - self._brightness = 1 - self._off_with_transition = transition is not None + result = await self._level_cluster_handler.move_to_level_with_on_off( + level=0, + transition_time=int( + 10 * (transition or self._DEFAULT_MIN_TRANSITION_TIME) + ), + ) + else: + assert self._on_off_cluster_handler is not None + result = await self._on_off_cluster_handler.off() - self.maybe_emit_state_changed_event() + # Pause parsing attribute reports until transition is complete + if self._zha_config_enable_light_transitioning_flag: + self.async_transition_start_timer(transition_time) + self.debug("turned off: %s", result) + if result[1] is not Status.SUCCESS: + return + self._state = False + + if brightness_supported and not self._off_with_transition: + # store current brightness so that the next turn_on uses it: + # when using "enhanced turn on" + self._off_brightness = self._brightness + if transition is not None: + # save for when calling turn_on without a brightness: + # current_level is set to 1 after transitioning to level 0, + # needed for correct state with light groups + self._brightness = 1 + self._off_with_transition = transition is not None + + self.maybe_emit_state_changed_event() + finally: + # If the task was cancelled before the transition timer was started, + # clean up the transitioning flag so the light does not get stuck. + if ( + self._zha_config_enable_light_transitioning_flag + and self._transitioning_individual + and not self._transition_listener + ): + self.async_transition_complete() async def async_handle_color_commands( self, From c7a14b2479674962d49b0c2b9640be00b824e948 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sun, 22 Feb 2026 02:46:30 +0100 Subject: [PATCH 2/5] Add `_async_cleanup_transition_if_stuck` method --- zha/application/platforms/light/__init__.py | 22 ++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index 7fe3475c7..38d71a3bd 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -584,12 +584,7 @@ async def async_turn_on( # If the task was cancelled (e.g. by a mode: restart automation) before # the transition timer was started, clean up the transitioning flag so # the light does not get stuck in a transitioning state indefinitely. - if ( - set_transition_flag - and self._transitioning_individual - and not self._transition_listener - ): - self.async_transition_complete() + self._async_cleanup_transition_if_stuck(set_transition_flag) async def async_turn_off(self, *, transition: float | None = None) -> None: """Turn the entity off.""" @@ -647,12 +642,9 @@ async def async_turn_off(self, *, transition: float | None = None) -> None: finally: # If the task was cancelled before the transition timer was started, # clean up the transitioning flag so the light does not get stuck. - if ( + self._async_cleanup_transition_if_stuck( self._zha_config_enable_light_transitioning_flag - and self._transitioning_individual - and not self._transition_listener - ): - self.async_transition_complete() + ) async def async_handle_color_commands( self, @@ -743,6 +735,14 @@ def _async_unsub_transition_listener(self) -> None: with contextlib.suppress(ValueError): self._tracked_handles.remove(self._transition_listener) + def _async_cleanup_transition_if_stuck(self, guarded: bool) -> None: + """Call async_transition_complete if the flag is set but no timer is running. + + Used in finally blocks to handle task cancellation gracefully. + """ + if guarded and self._transitioning_individual and not self._transition_listener: + self.async_transition_complete() + def async_transition_complete(self, _=None) -> None: """Set _transitioning_individual to False and write HA state.""" self.debug("transition complete - future attribute reports will write HA state") From eddd4442a43b5f3ca226f3f1afd5cb7f1bbeeb99 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sun, 22 Feb 2026 02:46:35 +0100 Subject: [PATCH 3/5] Add tests --- tests/test_light.py | 79 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/tests/test_light.py b/tests/test_light.py index 449471b61..19177b3b0 100644 --- a/tests/test_light.py +++ b/tests/test_light.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio +import contextlib import logging from typing import Any from unittest.mock import AsyncMock, call, patch, sentinel @@ -1976,3 +1977,81 @@ async def test_light_state_restoration(zha_gateway: Gateway) -> None: assert entity.state["xy_color"] == (1, 2) assert entity.state["color_mode"] == ColorMode.XY assert entity.state["effect"] == "colorloop" + + +async def test_turn_on_cancellation_cleans_up_transition_flag( + zha_gateway: Gateway, +) -> None: + """Test that task cancellation resets the transitioning flag. + + When a mode:restart automation cancels the task, the light must not be + stuck ignoring attribute reports indefinitely. + """ + device = await device_light_1_mock(zha_gateway) + entity = get_entity(device, platform=Platform.LIGHT) + + cluster_level = device.device.endpoints[1].level + + # Make the level cluster block indefinitely so we can cancel the task while + # it is suspended at the first await inside async_turn_on. + blocked: asyncio.Future[None] = asyncio.get_running_loop().create_future() + original_request = cluster_level.request + + async def blocking_request(*args, **kwargs): + await blocked + return await original_request(*args, **kwargs) + + cluster_level.request = AsyncMock(side_effect=blocking_request) + + # Start turn_on as a separate task, mirroring what HA does for automations. + task = asyncio.ensure_future(entity.async_turn_on(brightness=200, transition=1)) + + # Yield control so the task can run until it suspends on the cluster call. + await asyncio.sleep(0) + await asyncio.sleep(0) + + # The transitioning flag must be set now, with no timer running yet. + assert entity.is_transitioning is True + assert entity._transition_listener is None + + # Cancel the task (what mode:restart does to the running automation task). + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + # The finally block must have cleared the flag. + assert entity.is_transitioning is False + + +async def test_turn_off_cancellation_cleans_up_transition_flag( + zha_gateway: Gateway, +) -> None: + """Test that task cancellation during async_turn_off resets the transitioning flag.""" + device = await device_light_1_mock(zha_gateway) + entity = get_entity(device, platform=Platform.LIGHT) + + cluster_on_off = device.device.endpoints[1].on_off + + # Make the on/off cluster block indefinitely so we can cancel mid-turn-off. + blocked: asyncio.Future[None] = asyncio.get_running_loop().create_future() + original_request = cluster_on_off.request + + async def blocking_request(*args, **kwargs): + await blocked + return await original_request(*args, **kwargs) + + cluster_on_off.request = AsyncMock(side_effect=blocking_request) + + task = asyncio.ensure_future(entity.async_turn_off()) + + await asyncio.sleep(0) + await asyncio.sleep(0) + + assert entity.is_transitioning is True + assert entity._transition_listener is None + + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + assert entity.is_transitioning is False From 59611bf18c2ed308c45273248d85a39b6d08d8ce Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Sun, 22 Feb 2026 02:58:18 +0100 Subject: [PATCH 4/5] Add missing guard for "move_to_level_if_color" step --- zha/application/platforms/light/__init__.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index 38d71a3bd..8fef27e5b 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -311,7 +311,7 @@ def recompute_capabilities(self) -> None: light_options.enable_light_transitioning_flag ) - async def async_turn_on( + async def async_turn_on( # noqa: C901 self, *, transition: float | None = None, @@ -531,6 +531,11 @@ async def async_turn_on( ) t_log["move_to_level_if_color"] = result if result[1] is not Status.SUCCESS: + # Second 'move to level' call failed; the light is on but at the + # wrong brightness. If no previous timer is running, unset the flag + # immediately so attribute reports are not ignored indefinitely. + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() self.debug("turned on: %s", t_log) return self._state = bool(level) From 34e38c4ad6665e8c3b9f094b9976a5d112340d57 Mon Sep 17 00:00:00 2001 From: TheJulianJES Date: Wed, 25 Feb 2026 01:21:33 +0100 Subject: [PATCH 5/5] Reduce diff by introducing new method for implementation --- zha/application/platforms/light/__init__.py | 429 +++++++++++--------- 1 file changed, 228 insertions(+), 201 deletions(-) diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index 8fef27e5b..dc3d145a8 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -311,7 +311,7 @@ def recompute_capabilities(self) -> None: light_options.enable_light_transitioning_flag ) - async def async_turn_on( # noqa: C901 + async def async_turn_on( self, *, transition: float | None = None, @@ -373,223 +373,250 @@ async def async_turn_on( # noqa: C901 self.async_transition_set_flag() try: - # If the light is currently off but a turn_on call with a color/temperature is - # sent, the light needs to be turned on first at a low brightness level where - # the light is immediately transitioned to the correct color. Afterwards, the - # transition is only from the low brightness to the new brightness. - # Otherwise, the transition is from the color the light had before being turned - # on to the new color. This can look especially bad with transitions longer than - # a second. We do not want to do this for devices that need to be forced to use - # the on command because we would end up with 4 commands sent: - # move to level, on, color, move to level... We also will not set this - # if the bulb is already in the desired color mode with the desired color - # or color temperature. - new_color_provided_while_off = ( - self._zha_config_enhanced_light_transition - and not self._FORCE_ON - and not self._state - and ( - ( - color_temp is not None - and ( - self._color_temp != color_temp - or self._color_mode != ColorMode.COLOR_TEMP - ) - ) - or ( - xy_color is not None - and ( - self._xy_color != xy_color - or self._color_mode != ColorMode.XY - ) + await self._async_turn_on_impl( + transition=transition, + brightness=brightness, + effect=effect, + flash=flash, + color_temp=color_temp, + xy_color=xy_color, + duration=duration, + execute_if_off_supported=execute_if_off_supported, + brightness_supported=brightness_supported, + set_transition_flag=set_transition_flag, + transition_time=transition_time, + ) + finally: + # If the task was cancelled (e.g. by a mode: restart automation) before + # the transition timer was started, clean up the transitioning flag so + # the light does not get stuck in a transitioning state indefinitely. + self._async_cleanup_transition_if_stuck(set_transition_flag) + + async def _async_turn_on_impl( # noqa: C901 + self, + *, + transition: float | None, + brightness: int | None, + effect: str | None, + flash: FlashMode | None, + color_temp: int | None, + xy_color: tuple[int, int] | None, + duration: float, + execute_if_off_supported: bool, + brightness_supported: bool, + set_transition_flag: bool, + transition_time: float, + ) -> None: + """Implement the turn on logic.""" + # If the light is currently off but a turn_on call with a color/temperature is + # sent, the light needs to be turned on first at a low brightness level where + # the light is immediately transitioned to the correct color. Afterwards, the + # transition is only from the low brightness to the new brightness. + # Otherwise, the transition is from the color the light had before being turned + # on to the new color. This can look especially bad with transitions longer than + # a second. We do not want to do this for devices that need to be forced to use + # the on command because we would end up with 4 commands sent: + # move to level, on, color, move to level... We also will not set this + # if the bulb is already in the desired color mode with the desired color + # or color temperature. + new_color_provided_while_off = ( + self._zha_config_enhanced_light_transition + and not self._FORCE_ON + and not self._state + and ( + ( + color_temp is not None + and ( + self._color_temp != color_temp + or self._color_mode != ColorMode.COLOR_TEMP ) ) - and brightness_supported - and not execute_if_off_supported + or ( + xy_color is not None + and (self._xy_color != xy_color or self._color_mode != ColorMode.XY) + ) ) + and brightness_supported + and not execute_if_off_supported + ) - if ( - brightness is None - and (self._off_with_transition or new_color_provided_while_off) - and self._off_brightness is not None - ): - brightness = self._off_brightness - - if brightness is not None: - level = min(254, brightness) - else: - level = self._brightness or 254 + if ( + brightness is None + and (self._off_with_transition or new_color_provided_while_off) + and self._off_brightness is not None + ): + brightness = self._off_brightness - t_log = {} + if brightness is not None: + level = min(254, brightness) + else: + level = self._brightness or 254 - if new_color_provided_while_off: - assert self._level_cluster_handler is not None + t_log = {} - # If the light is currently off, we first need to turn it on at a low - # brightness level with no transition. - # After that, we set it to the desired color/temperature with no transition. - result = await self._level_cluster_handler.move_to_level_with_on_off( - level=DEFAULT_MIN_BRIGHTNESS, - transition_time=int(10 * self._DEFAULT_MIN_TRANSITION_TIME), - ) - t_log["move_to_level_with_on_off"] = result - if result[1] is not Status.SUCCESS: - # First 'move to level' call failed, so if the transitioning delay - # isn't running from a previous call, - # the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return - # Currently only setting it to "on", as the correct level state will - # be set at the second move_to_level call - self._state = True - - if execute_if_off_supported: - self.debug("handling color commands before turning on/level") - if not await self.async_handle_color_commands( - color_temp, - duration, # duration is ignored by lights when off - xy_color, - new_color_provided_while_off, - t_log, - ): - # Color calls before on/level calls failed, - # so if the transitioning delay isn't running from a previous call, - # the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return + if new_color_provided_while_off: + assert self._level_cluster_handler is not None - if ( - (brightness is not None or transition is not None) - and not new_color_provided_while_off - and brightness_supported + # If the light is currently off, we first need to turn it on at a low + # brightness level with no transition. + # After that, we set it to the desired color/temperature with no transition. + result = await self._level_cluster_handler.move_to_level_with_on_off( + level=DEFAULT_MIN_BRIGHTNESS, + transition_time=int(10 * self._DEFAULT_MIN_TRANSITION_TIME), + ) + t_log["move_to_level_with_on_off"] = result + if result[1] is not Status.SUCCESS: + # First 'move to level' call failed, so if the transitioning delay + # isn't running from a previous call, + # the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return + # Currently only setting it to "on", as the correct level state will + # be set at the second move_to_level call + self._state = True + + if execute_if_off_supported: + self.debug("handling color commands before turning on/level") + if not await self.async_handle_color_commands( + color_temp, + duration, # duration is ignored by lights when off + xy_color, + new_color_provided_while_off, + t_log, ): - assert self._level_cluster_handler is not None + # Color calls before on/level calls failed, + # so if the transitioning delay isn't running from a previous call, + # the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return - result = await self._level_cluster_handler.move_to_level_with_on_off( - level=level, - transition_time=int(10 * duration), - ) - t_log["move_to_level_with_on_off"] = result - if result[1] is not Status.SUCCESS: - # First 'move to level' call failed, so if the transitioning delay - # isn't running from a previous call, the flag can be unset immediately - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return - self._state = bool(level) - if level: - self._brightness = level + if ( + (brightness is not None or transition is not None) + and not new_color_provided_while_off + and brightness_supported + ): + assert self._level_cluster_handler is not None - if ( - (brightness is None and transition is None) - and not new_color_provided_while_off - or (self._FORCE_ON and brightness != 0) + result = await self._level_cluster_handler.move_to_level_with_on_off( + level=level, + transition_time=int(10 * duration), + ) + t_log["move_to_level_with_on_off"] = result + if result[1] is not Status.SUCCESS: + # First 'move to level' call failed, so if the transitioning delay + # isn't running from a previous call, the flag can be unset immediately + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return + self._state = bool(level) + if level: + self._brightness = level + + if ( + (brightness is None and transition is None) + and not new_color_provided_while_off + or (self._FORCE_ON and brightness != 0) + ): + assert self._on_off_cluster_handler is not None + + # since FORCE_ON lights don't turn on with move_to_level_with_on_off, + # we should call the on command on the on_off cluster + # if brightness is not 0. + result = await self._on_off_cluster_handler.on() + t_log["on_off"] = result + if result[1] is not Status.SUCCESS: + # 'On' call failed, but as brightness may still transition + # (for FORCE_ON lights), we start the timer to unset the flag after + # the transition_time if necessary. + self.async_transition_start_timer(transition_time) + self.debug("turned on: %s", t_log) + return + self._state = True + + if not execute_if_off_supported: + self.debug("handling color commands after turning on/level") + if not await self.async_handle_color_commands( + color_temp, + duration, + xy_color, + new_color_provided_while_off, + t_log, ): - assert self._on_off_cluster_handler is not None + # Color calls failed, but as brightness may still transition, + # we start the timer to unset the flag + self.async_transition_start_timer(transition_time) + self.debug("turned on: %s", t_log) + return - # since FORCE_ON lights don't turn on with move_to_level_with_on_off, - # we should call the on command on the on_off cluster - # if brightness is not 0. - result = await self._on_off_cluster_handler.on() - t_log["on_off"] = result - if result[1] is not Status.SUCCESS: - # 'On' call failed, but as brightness may still transition - # (for FORCE_ON lights), we start the timer to unset the flag after - # the transition_time if necessary. - self.async_transition_start_timer(transition_time) - self.debug("turned on: %s", t_log) - return - self._state = True - - if not execute_if_off_supported: - self.debug("handling color commands after turning on/level") - if not await self.async_handle_color_commands( - color_temp, - duration, - xy_color, - new_color_provided_while_off, - t_log, - ): - # Color calls failed, but as brightness may still transition, - # we start the timer to unset the flag - self.async_transition_start_timer(transition_time) - self.debug("turned on: %s", t_log) - return - - if new_color_provided_while_off: - assert self._level_cluster_handler is not None + if new_color_provided_while_off: + assert self._level_cluster_handler is not None - # The light has the correct color, so we can now transition - # it to the correct brightness level. - result = await self._level_cluster_handler.move_to_level( - level=level, transition_time=int(10 * duration) - ) - t_log["move_to_level_if_color"] = result - if result[1] is not Status.SUCCESS: - # Second 'move to level' call failed; the light is on but at the - # wrong brightness. If no previous timer is running, unset the flag - # immediately so attribute reports are not ignored indefinitely. - if set_transition_flag and not self._transition_listener: - self.async_transition_complete() - self.debug("turned on: %s", t_log) - return - self._state = bool(level) - if level: - self._brightness = level - - # Our light is guaranteed to have just started the transitioning process - # if necessary, so we start the delay for the transition (to stop parsing - # attribute reports after the completed transition). - self.async_transition_start_timer(transition_time) - - if self._color_cluster_handler is not None: - if effect == EFFECT_COLORLOOP: - result = await self._color_cluster_handler.color_loop_set( - update_flags=( - Color.ColorLoopUpdateFlags.Action - | Color.ColorLoopUpdateFlags.Direction - | Color.ColorLoopUpdateFlags.Time - ), - action=Color.ColorLoopAction.Activate_from_current_hue, - direction=Color.ColorLoopDirection.Increment, - time=transition if transition else 7, - start_hue=0, - ) - t_log["color_loop_set"] = result - self._effect = EFFECT_COLORLOOP - elif self._effect == EFFECT_COLORLOOP and effect != EFFECT_COLORLOOP: - result = await self._color_cluster_handler.color_loop_set( - update_flags=Color.ColorLoopUpdateFlags.Action, - action=Color.ColorLoopAction.Deactivate, - direction=Color.ColorLoopDirection.Decrement, - time=0, - start_hue=0, - ) - t_log["color_loop_set"] = result - self._effect = EFFECT_OFF + # The light has the correct color, so we can now transition + # it to the correct brightness level. + result = await self._level_cluster_handler.move_to_level( + level=level, transition_time=int(10 * duration) + ) + t_log["move_to_level_if_color"] = result + if result[1] is not Status.SUCCESS: + # Second 'move to level' call failed; the light is on but at the + # wrong brightness. If no previous timer is running, unset the flag + # immediately so attribute reports are not ignored indefinitely. + if set_transition_flag and not self._transition_listener: + self.async_transition_complete() + self.debug("turned on: %s", t_log) + return + self._state = bool(level) + if level: + self._brightness = level - if flash is not None: - assert self._identify_cluster_handler is not None - result = await self._identify_cluster_handler.trigger_effect( - effect_id=FLASH_EFFECTS[flash], - effect_variant=Identify.EffectVariant.Default, + # Our light is guaranteed to have just started the transitioning process + # if necessary, so we start the delay for the transition (to stop parsing + # attribute reports after the completed transition). + self.async_transition_start_timer(transition_time) + + if self._color_cluster_handler is not None: + if effect == EFFECT_COLORLOOP: + result = await self._color_cluster_handler.color_loop_set( + update_flags=( + Color.ColorLoopUpdateFlags.Action + | Color.ColorLoopUpdateFlags.Direction + | Color.ColorLoopUpdateFlags.Time + ), + action=Color.ColorLoopAction.Activate_from_current_hue, + direction=Color.ColorLoopDirection.Increment, + time=transition if transition else 7, + start_hue=0, ) - t_log["trigger_effect"] = result + t_log["color_loop_set"] = result + self._effect = EFFECT_COLORLOOP + elif self._effect == EFFECT_COLORLOOP and effect != EFFECT_COLORLOOP: + result = await self._color_cluster_handler.color_loop_set( + update_flags=Color.ColorLoopUpdateFlags.Action, + action=Color.ColorLoopAction.Deactivate, + direction=Color.ColorLoopDirection.Decrement, + time=0, + start_hue=0, + ) + t_log["color_loop_set"] = result + self._effect = EFFECT_OFF + + if flash is not None: + assert self._identify_cluster_handler is not None + result = await self._identify_cluster_handler.trigger_effect( + effect_id=FLASH_EFFECTS[flash], + effect_variant=Identify.EffectVariant.Default, + ) + t_log["trigger_effect"] = result - self._off_with_transition = False - self._off_brightness = None - self.debug("turned on: %s", t_log) - self.maybe_emit_state_changed_event() - finally: - # If the task was cancelled (e.g. by a mode: restart automation) before - # the transition timer was started, clean up the transitioning flag so - # the light does not get stuck in a transitioning state indefinitely. - self._async_cleanup_transition_if_stuck(set_transition_flag) + self._off_with_transition = False + self._off_brightness = None + self.debug("turned on: %s", t_log) + self.maybe_emit_state_changed_event() async def async_turn_off(self, *, transition: float | None = None) -> None: """Turn the entity off."""