Add auto_reset_manual_control with async timer (#487)

* Add auto_reset_manual_control with async timer

* Add failing test

* Debugging

* Refactor find_switch_for_lights

* Revert changes in is_manually_controlled

* style

* Fix test_manual_control

* add types

* fixes

* dict

* make all tests pass

* text

* Rework find_switch_for_lights

* Style

* Better check

* revert, do in other test!

* No need to log when raising

* Add type hint

* Suggestion https://github.com/basnijholt/adaptive-lighting/pull/488/files#r1152408541 by @th3w1zard1

* Small fixes

* document new config everywhere (#496)

* chore(docs): update TOC

* undo change

* fi

* Update README.md

* Update README.md

* chore(docs): update TOC

* Update README.md

* Use markdown-code-runner

* Remove

* Use markdown-code-runner instead of packaged solution

* fix comment

* Only commit when needed

---------

Co-authored-by: Benjamin Auquite <halomastar@gmail.com>
Co-authored-by: basnijholt <basnijholt@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.github/workflows/update-readme.yml

@@ -30,13 +30,21 @@ jobs:
         run: markdown-code-runner --debug README.md
 
       - name: Commit updated README.md
+        id: commit
         run: |
           git add README.md
           git config --local user.email "github-actions[bot]@users.noreply.github.com"
           git config --local user.name "github-actions[bot]"
-          git diff --quiet && git diff --staged --quiet || git commit -m "Update README.md"
+          if git diff --quiet && git diff --staged --quiet; then
+            echo "No changes in README.md, skipping commit."
+            echo "commit_status=skipped" >> $GITHUB_ENV
+          else
+            git commit -m "Update README.md"
+            echo "commit_status=committed" >> $GITHUB_ENV
+          fi
 
       - name: Push changes
+        if: env.commit_status == 'committed'
         uses: ad-m/github-push-action@master
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}

README.md

@@ -29,7 +29,7 @@ Adaptive Lighting provides four switches (using "living_room" as an example comp
 
 Adaptive Lighting is designed to automatically detect when you or another source (e.g., automation) manually changes light settings 🕹️.
 When this occurs, the affected light is marked as "manually controlled," and Adaptive Lighting will not make further adjustments until the light is turned off and back on or reset using the `adaptive_lighting.set_manual_control` service call.
-This feature is available when take_over_control is enabled.
+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 🤖.
@@ -47,7 +47,7 @@ The `adaptive_lighting.manual_control` event is fired when a light is marked as
       - [`adaptive_lighting.change_switch_settings`](#adaptive_lightingchange_switch_settings)
   - [:robot: Automation examples](#robot-automation-examples)
 - [Additional Information](#additional-information)
-- [Troubleshooting](#troubleshooting)
+- [:sos: Troubleshooting](#sos-troubleshooting)
   - [:exclamation: Common Problems & Solutions](#exclamation-common-problems--solutions)
     - [:bulb: Lights Not Responding or Turning On by Themselves](#bulb-lights-not-responding-or-turning-on-by-themselves)
       - [:signal_strength: WiFi Networks](#signal_strength-wifi-networks)
@@ -119,6 +119,7 @@ The YAML and frontend configuration methods support all of the options listed be
 | `separate_turn_on_commands`    | Use separate `light.turn_on` calls for color and brightness, needed for some light types. 🔀                                                                                     | `False`        | `bool`                               |
 | `send_split_delay`             | Wait time (milliseconds) between commands when using `separate_turn_on_commands`. Helps ensure correct handling. ⏲️                                                             | `0`            | `int` 0-10000                        |
 | `adapt_delay`                  | Wait time (seconds) between light turn on and Adaptive Lighting applying changes. Helps avoid flickering. ⏲️                                                                    | `0`            | `float > 0`                          |
+| `autoreset_control_seconds`    | Automatically reset the manual control after a number of seconds. Set to 0 to disable. ⏲️                                                                                       | `0`            | `int` 0-604800                       |
 
 <!-- END_OUTPUT -->
 
@@ -298,7 +299,7 @@ For more details on adding the integration and setting options, refer to the [do
 
 Adaptive Lighting was initially inspired by @claytonjn's [hass-circadian\_lighting](https://github.com/claytonjn/hass-circadian_lighting), but has since been entirely rewritten and expanded with new features.
 
-# Troubleshooting
+# :sos: Troubleshooting
 
 Encountering issues? Enable debug logging in your `configuration.yaml`:
 

custom_components/adaptive_lighting/const.py

@@ -152,6 +152,11 @@ DOCS[CONF_SEND_SPLIT_DELAY] = (
     "Helps ensure correct handling. ⏲️"
 )
 
+CONF_AUTORESET_CONTROL, DEFAULT_AUTORESET_CONTROL = "autoreset_control_seconds", 0
+DOCS[CONF_AUTORESET_CONTROL] = (
+    "Automatically reset the manual control after a number of seconds. "
+    "Set to 0 to disable. ⏲️"
+)
 
 SLEEP_MODE_SWITCH = "sleep_mode_switch"
 ADAPT_COLOR_SWITCH = "adapt_color_switch"
@@ -220,6 +225,11 @@ VALIDATION_TUPLES = [
     (CONF_SEPARATE_TURN_ON_COMMANDS, DEFAULT_SEPARATE_TURN_ON_COMMANDS, bool),
     (CONF_SEND_SPLIT_DELAY, DEFAULT_SEND_SPLIT_DELAY, int_between(0, 10000)),
     (CONF_ADAPT_DELAY, DEFAULT_ADAPT_DELAY, cv.positive_float),
+    (
+        CONF_AUTORESET_CONTROL,
+        DEFAULT_AUTORESET_CONTROL,
+        int_between(0, 7 * 24 * 60 * 60),  # 7 days max
+    ),
 ]
 
 

custom_components/adaptive_lighting/services.yaml

@@ -248,3 +248,9 @@ change_switch_settings:
       example: 0
       selector:
         text:
+    autoreset_control_seconds:
+      description: "autoreset_control_seconds: wait time (seconds) before Adaptive Lighting resets `manual_control` status of any light (default: 0)"
+      required: false
+      example: 0
+      selector:
+        text:

custom_components/adaptive_lighting/strings.json

@@ -45,7 +45,8 @@
           "take_over_control": "take_over_control: If anything but Adaptive Lighting calls 'light.turn_on' when a light is already on, stop adapting that light until it (or the switch) toggles off -> on.",
           "detect_non_ha_changes": "detect_non_ha_changes: detects all >10% changes made to the lights (also outside of HA), requires 'take_over_control' to be enabled (calls 'homeassistant.update_entity' every 'interval'!)",
           "transition": "Transition time when applying a change to the lights (seconds)",
-          "adapt_delay": "adapt_delay: wait time between light turn on (seconds), and Adaptive Lights applying changes to the light state. May avoid flickering."
+          "adapt_delay": "adapt_delay: wait time between light turn on (seconds), and Adaptive Lights applying changes to the light state. May avoid flickering.",
+          "autoreset_control_seconds": "autoreset_control_seconds: wait time (seconds) before Adaptive Lighting resets `manual_control` status of any light (default: 0)"
         }
       }
     },

custom_components/adaptive_lighting/switch.py

@@ -97,6 +97,7 @@ from .const import (
     ATTR_ADAPT_COLOR,
     ATTR_TURN_ON_OFF_LISTENER,
     CONF_ADAPT_DELAY,
+    CONF_AUTORESET_CONTROL,
     CONF_DETECT_NON_HA_CHANGES,
     CONF_INCLUDE_CONFIG_IN_ATTRIBUTES,
     CONF_INITIAL_TRANSITION,
@@ -265,7 +266,7 @@ def find_switch_for_lights(
     is_on: bool = False,
 ) -> AdaptiveSwitch:
     """Find the switch that controls the lights in 'lights'."""
-    switches = _get_switches_with_lights(hass, lights, is_on)
+    switches = _get_switches_with_lights(hass, lights)
     if len(switches) == 1:
         return switches[0]
     elif len(switches) > 1:
@@ -330,7 +331,7 @@ def _get_switches_from_service_call(
 
 async def handle_change_switch_settings(
     switch: AdaptiveSwitch, service_call: ServiceCall
-):
+) -> None:
     """Allows HASS to change config values via a service call."""
     data = service_call.data
 
@@ -473,7 +474,7 @@ async def async_setup_entry(
                 all_lights = _expand_light_groups(switch.hass, lights)
             if service_call.data[CONF_MANUAL_CONTROL]:
                 for light in all_lights:
-                    switch.turn_on_off_listener.manual_control[light] = True
+                    switch.turn_on_off_listener.mark_as_manual_control(light)
                     _fire_manual_control_event(switch, light, service_call.context)
             else:
                 switch.turn_on_off_listener.reset(*all_lights)
@@ -820,6 +821,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         self._transition = data[CONF_TRANSITION]
         self._adapt_delay = data[CONF_ADAPT_DELAY]
         self._send_split_delay = data[CONF_SEND_SPLIT_DELAY]
+        self._auto_reset_manual_control_time = data[CONF_AUTORESET_CONTROL]
         _loc = get_astral_location(self.hass)
         if isinstance(_loc, tuple):
             # Astral v2.2
@@ -893,6 +895,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
     def _expand_light_groups(self) -> None:
         all_lights = _expand_light_groups(self.hass, self._lights)
         self.turn_on_off_listener.lights.update(all_lights)
+        self.turn_on_off_listener.set_auto_reset_manual_control_times(
+            all_lights, self._auto_reset_manual_control_time
+        )
         self._lights = list(all_lights)
 
     async def _setup_listeners(self, _=None) -> None:
@@ -1526,6 +1531,10 @@ class TurnOnOffListener:
         # Track last 'service_data' to 'light.turn_on' resulting from this integration
         self.last_service_data: dict[str, dict[str, Any]] = {}
 
+        # Track auto reset of manual_control
+        self.auto_reset_manual_control_timers: dict[str, _AsyncSingleShotTimer] = {}
+        self.auto_reset_manual_control_times: dict[str, float] = {}
+
         # When a state is different `max_cnt_significant_changes` times in a row,
         # mark it as manually_controlled.
         self.max_cnt_significant_changes = 2
@@ -1537,11 +1546,71 @@ class TurnOnOffListener:
             EVENT_STATE_CHANGED, self.state_changed_event_listener
         )
 
+    def set_auto_reset_manual_control_times(self, lights: list[str], time: float):
+        """Set the time after which the lights are automatically reset."""
+        if time == 0:
+            return
+        for light in lights:
+            old_time = self.auto_reset_manual_control_times.get(light)
+            if (old_time is not None) and (old_time != time):
+                _LOGGER.info(
+                    "Setting auto_reset_manual_control for '%s' from %s seconds to %s seconds."
+                    " This might happen because the light is in multiple swiches"
+                    " or because of a config change.",
+                    light,
+                    old_time,
+                    time,
+                )
+            self.auto_reset_manual_control_times[light] = time
+
+    def mark_as_manual_control(self, light: str) -> None:
+        """Mark a light as manually controlled."""
+        _LOGGER.debug("Marking '%s' as manually controlled.", light)
+        self.manual_control[light] = True
+        delay = self.auto_reset_manual_control_times.get(light)
+        timer = self.auto_reset_manual_control_timers.get(light)
+        if timer is not None:
+            if delay is None:  # Timer object exists, but should not anymore
+                timer.cancel()
+                self.auto_reset_manual_control_timers.pop(light)
+            else:  # Timer object already exists, just update the delay and restart it
+                timer.delay = delay
+                timer.start()
+        elif delay is not None:  # Timer object does not exist, create it
+
+            async def reset():
+                self.reset(light)
+                switches = _get_switches_with_lights(self.hass, [light])
+                for switch in switches:
+                    if not switch.is_on:
+                        continue
+                    # pylint: disable=protected-access
+                    await switch._update_attrs_and_maybe_adapt_lights(
+                        [light],
+                        transition=switch._initial_transition,
+                        force=True,
+                        context=switch.create_context("autoreset"),
+                    )
+                _LOGGER.debug(
+                    "Auto resetting 'manual_control' status of '%s' because"
+                    " it was not manually controlled for %s seconds.",
+                    light,
+                    delay,
+                )
+                assert not self.manual_control[light]
+
+            timer = _AsyncSingleShotTimer(delay, reset)
+            self.auto_reset_manual_control_timers[light] = timer
+            timer.start()
+
     def reset(self, *lights, reset_manual_control=True) -> None:
         """Reset the 'manual_control' status of the lights."""
         for light in lights:
             if reset_manual_control:
                 self.manual_control[light] = False
+                timer = self.auto_reset_manual_control_timers.pop(light, None)
+                if timer is not None:
+                    timer.cancel()
             self.last_state_change.pop(light, None)
             self.last_service_data.pop(light, None)
             self.cnt_significant_changes[light] = 0
@@ -1599,6 +1668,14 @@ class TurnOnOffListener:
                 if task is not None:
                     task.cancel()
                 self.turn_on_event[eid] = event
+                timer = self.auto_reset_manual_control_timers.get(eid)
+                if (
+                    timer is not None
+                    and timer.is_running()
+                    and event.time_fired > timer.start_time
+                ):
+                    # Restart the auto reset timer
+                    timer.start()
 
     async def state_changed_event_listener(self, event: Event) -> None:
         """Track 'state_changed' events."""
@@ -1674,7 +1751,7 @@ class TurnOnOffListener:
             ):
                 # Light was already on and 'light.turn_on' was not called by
                 # the adaptive_lighting integration.
-                manual_control = self.manual_control[light] = True
+                manual_control = self.mark_as_manual_control(light)
                 _fire_manual_control_event(switch, light, turn_on_event.context)
                 _LOGGER.debug(
                     "'%s' was already on and 'light.turn_on' was not called by the"
@@ -1746,7 +1823,7 @@ class TurnOnOffListener:
                 # Only mark a light as significantly changing, if changed==True
                 # N times in a row. We do this because sometimes a state changes
                 # happens only *after* a new update interval has already started.
-                self.manual_control[light] = True
+                self.mark_as_manual_control(light)
                 _fire_manual_control_event(switch, light, context, is_async=False)
         else:
             if n_changes > 1:
@@ -1857,3 +1934,38 @@ class TurnOnOffListener:
         # other 'off' → 'on' state switches resulting from polling. That
         # would mean we 'return True' here.
         return False
+
+
+class _AsyncSingleShotTimer:
+    def __init__(self, delay, callback):
+        """Initialize the timer."""
+        self.delay = delay
+        self.callback = callback
+        self.task = None
+        self.start_time: int | None = None
+
+    async def _run(self):
+        """Run the timer. Don't call this directly, use start() instead."""
+        self.start_time = dt_util.utcnow()
+        await asyncio.sleep(self.delay)
+        if self.callback:
+            if asyncio.iscoroutinefunction(self.callback):
+                await self.callback()
+            else:
+                self.callback()
+
+    def is_running(self):
+        """Return whether the timer is running."""
+        return self.task is not None and not self.task.done()
+
+    def start(self):
+        """Start the timer."""
+        if self.task is not None and not self.task.done():
+            self.task.cancel()
+        self.task = asyncio.create_task(self._run())
+
+    def cancel(self):
+        """Cancel the timer."""
+        if self.task:
+            self.task.cancel()
+            self.callback = None

tests/test_switch.py

@@ -10,6 +10,7 @@ from homeassistant.components.adaptive_lighting.const import (
     ADAPT_BRIGHTNESS_SWITCH,
     ADAPT_COLOR_SWITCH,
     ATTR_TURN_ON_OFF_LISTENER,
+    CONF_AUTORESET_CONTROL,
     CONF_DETECT_NON_HA_CHANGES,
     CONF_INITIAL_TRANSITION,
     CONF_MANUAL_CONTROL,
@@ -184,7 +185,7 @@ async def setup_lights_and_switch(hass, extra_conf=None):
 
     # Setup switch
     lights = [
-        "light.bed_light",
+        ENTITY_LIGHT,
         "light.ceiling_lights",
     ]
     assert all(hass.states.get(light) is not None for light in lights)
@@ -584,6 +585,50 @@ async def test_manual_control(hass):
     assert all([not manual_control[eid] for eid in switch._lights])
 
 
+async def test_auto_reset_manual_control(hass):
+    switch, (light, *_) = await setup_lights_and_switch(
+        hass, {CONF_AUTORESET_CONTROL: 0.1}
+    )
+    context = switch.create_context("test")  # needs to be passed to update method
+    manual_control = switch.turn_on_off_listener.manual_control
+
+    async def update():
+        await switch._update_attrs_and_maybe_adapt_lights(transition=0, context=context)
+        await hass.async_block_till_done()
+
+    async def turn_light(state, **kwargs):
+        await hass.services.async_call(
+            LIGHT_DOMAIN,
+            SERVICE_TURN_ON if state else SERVICE_TURN_OFF,
+            {ATTR_ENTITY_ID: light.entity_id, **kwargs},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        await update()
+        _LOGGER.debug(
+            "Turn light %s to state %s, to %s", light.entity_id, state, kwargs
+        )
+
+    _LOGGER.debug("Start test auto reset manual control")
+    await turn_light(True, brightness=1)
+    await turn_light(True, brightness=10)
+    assert manual_control[light.entity_id]
+    await asyncio.sleep(0.3)  # Should be enough time for auto reset
+    await update()
+    assert not manual_control[light.entity_id], (light, manual_control)
+
+    # Do a couple of quick changes and check that light is not reset
+    for i in range(3):
+        _LOGGER.debug("Quick change %s", i)
+        await turn_light(True, brightness=(i + 1) * 20)
+        await asyncio.sleep(0.05)  # Less than 0.1
+        assert manual_control[light.entity_id]
+
+    await asyncio.sleep(0.3)  # Wait the auto reset time
+    await update()
+    assert not manual_control[light.entity_id]
+
+
 async def test_apply_service(hass):
     """Test adaptive_lighting.apply service."""
     switch, (_, _, light) = await setup_lights_and_switch(hass)
@@ -734,6 +779,7 @@ async def test_significant_change(hass):
         assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]
     # On next update the light should be marked as manually controlled
     await update(force=False)
+    # TODO: the state should be `bool(manual_control) is True`
     assert not switch.turn_on_off_listener.manual_control[ENTITY_LIGHT]