detect significant changes

custom_components/adaptive_lighting/switch.py

@@ -14,6 +14,7 @@ import astral
 import voluptuous as vol
 
 from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
     ATTR_BRIGHTNESS_PCT,
     ATTR_COLOR_TEMP,
     ATTR_RGB_COLOR,
@@ -126,6 +127,7 @@ async def handle_apply(switch: AdaptiveSwitch, service_call: ServiceCall):
                 data[CONF_ADAPT_BRIGHTNESS],
                 data[CONF_ADAPT_COLOR_TEMP],
                 data[CONF_ADAPT_RGB_COLOR],
+                force=True,
             )
 
 
@@ -211,6 +213,11 @@ def _supported_features(hass, light: str):
     return {key for key, value in _SUPPORT_OPTS.items() if supported_features & value}
 
 
+def abs_rel_diff(a, b):
+    """Absolute relative difference in %."""
+    return abs((a - b) / b) * 100
+
+
 class AdaptiveSwitch(SwitchEntity, RestoreEntity):
     """Representation of a Adaptive Lighting switch."""
 
@@ -238,7 +245,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         self._only_once = data[CONF_ONLY_ONCE]
         self._prefer_rgb_color = data[CONF_PREFER_RGB_COLOR]
         self._take_over_control = data[CONF_TAKE_OVER_CONTROL]
-        self._transition = data[CONF_TRANSITION]
+        self._transition = min(
+            data[CONF_TRANSITION], self._interval.total_seconds() // 2
+        )
 
         self._sun_light_settings = SunLightSettings(
             name=self._name,
@@ -270,7 +279,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         self.__context = Context()  # self._context will be overwritten
 
         # Set in self._update_attrs_and_maybe_adapt_lights
-        self._light_settings = {}
+        self._settings: Dict[str, Any] = {}
 
         # Set and unset tracker in async_turn_on and async_turn_off
         self.remove_listeners = []
@@ -364,12 +373,11 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
     def device_state_attributes(self) -> Dict[str, Any]:
         """Return the attributes of the switch."""
         if not self.is_on:
-            return {key: None for key in self._light_settings}
-        return self._light_settings
+            return {key: None for key in self._settings}
+        return self._settings
 
     def _reset_take_over_control(self):
-        for light in self._lights:
-            self.turn_on_off_listener.manually_controlled[light] = False
+        self.turn_on_off_listener.reset(*self._lights)
 
     async def async_turn_on(  # pylint: disable=arguments-differ
         self, adapt_lights: bool = True
@@ -406,6 +414,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         adapt_brightness: Optional[bool] = None,
         adapt_color_temp: Optional[bool] = None,
         adapt_rgb_color: Optional[bool] = None,
+        force: bool = False,
     ) -> None:
         lock = self._locks.get(light)
         if lock is not None and lock.locked():
@@ -428,7 +437,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
             service_data[ATTR_TRANSITION] = transition
 
         if "brightness" in features and adapt_brightness:
-            service_data[ATTR_BRIGHTNESS_PCT] = self._light_settings["brightness_pct"]
+            service_data[ATTR_BRIGHTNESS_PCT] = self._settings["brightness_pct"]
 
         if (
             "color_temp" in features
@@ -437,18 +446,29 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         ):
             attributes = self.hass.states.get(light).attributes
             min_mireds, max_mireds = attributes["min_mireds"], attributes["max_mireds"]
-            color_temp_mired = self._light_settings["color_temp_mired"]
+            color_temp_mired = self._settings["color_temp_mired"]
             color_temp_mired = max(min(color_temp_mired, max_mireds), min_mireds)
             service_data[ATTR_COLOR_TEMP] = color_temp_mired
         elif "color" in features and adapt_rgb_color:
-            service_data[ATTR_RGB_COLOR] = self._light_settings["rgb_color"]
+            service_data[ATTR_RGB_COLOR] = self._settings["rgb_color"]
 
+        if (
+            self._take_over_control
+            and not force
+            and self.turn_on_off_listener.significant_change(
+                light,
+                self._adapt_brightness,
+                self._adapt_color_temp,
+                self._adapt_rgb_color,
+            )
+        ):
+            return
+        self.turn_on_off_listener.last_service_data[light] = service_data
         _LOGGER.debug(
             "%s: Scheduling 'light.turn_on' with the following 'service_data': %s",
             self._name,
             service_data,
         )
-
         await self.hass.services.async_call(
             LIGHT_DOMAIN,
             SERVICE_TURN_ON,
@@ -464,7 +484,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
     ):
         _LOGGER.debug("%s: '_update_attrs_and_maybe_adapt_lights' called", self._name)
         assert self.is_on
-        self._light_settings = self._sun_light_settings.get_settings(
+        self._settings = self._sun_light_settings.get_settings(
             self.sleep_mode_switch.is_on
         )
         self.async_write_ha_state()
@@ -487,19 +507,21 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         for light in lights:
             if not is_on(self.hass, light):
                 continue
-            if self._take_over_control:
-                if self.turn_on_off_listener.is_manually_controlled(
+            if (
+                self._take_over_control
+                and self.turn_on_off_listener.is_manually_controlled(
                     light,
                     force,
                     adaptive_lighting_context=self.__context,
-                ):
-                    _LOGGER.debug(
-                        "%s: '%s' is being manually controlled, stop adapting.",
-                        self._name,
-                        light,
-                    )
-                    continue
-            await self._adapt_light(light, transition)
+                )
+            ):
+                _LOGGER.debug(
+                    "%s: '%s' is being manually controlled, stop adapting.",
+                    self._name,
+                    light,
+                )
+                continue
+            await self._adapt_light(light, transition, force=force)
 
     async def _sleep_state_event(self, event: Event):
         if not match_state_event(event, ("on", "off")):
@@ -553,7 +575,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
         ):
             # Tracks 'off' → 'on' state changes
             self._on_to_off_event[entity_id] = event
-            self.turn_on_off_listener.manually_controlled[entity_id] = False
+            self.turn_on_off_listener.reset(entity_id)
 
 
 class AdaptiveSleepModeSwitch(SwitchEntity, RestoreEntity):
@@ -762,23 +784,28 @@ class TurnOnOffListener:
         self.sleep_tasks: Dict[str, asyncio.Task] = {}
         # Tracks which lights are manually controlled
         self.manually_controlled: Dict[str, bool] = {}
+        # Track which settings were applied to a light
+        self.last_service_data: Dict[str, Dict[str, Any]] = {}
 
         self.remove_listener = self.hass.bus.async_listen(
             EVENT_CALL_SERVICE, self.turn_on_off_event_listener
         )
 
+    def reset(self, *lights):
+        """Reset the 'manually_controlled' status of the lights."""
+        for light in lights:
+            self.manually_controlled[light] = False
+            self.last_service_data.pop(light, None)
+
     async def turn_on_off_event_listener(self, event: Event):
         """Track 'light.turn_off' and 'light.turn_on' service calls."""
         domain = event.data.get(ATTR_DOMAIN)
         if domain != LIGHT_DOMAIN:
             return
 
-        service = event.data.get(ATTR_SERVICE)
-        service_data = event.data.get(ATTR_SERVICE_DATA, {})
-
-        entity_ids = service_data.get(ATTR_ENTITY_ID)
-        if isinstance(entity_ids, str):
-            entity_ids = [entity_ids]
+        service = event.data[ATTR_SERVICE]
+        service_data = event.data[ATTR_SERVICE_DATA]
+        entity_ids = cv.ensure_list(service_data[ATTR_ENTITY_ID])
 
         if not any(eid in self.lights for eid in entity_ids):
             return
@@ -792,7 +819,7 @@ class TurnOnOffListener:
             )
             for eid in entity_ids:
                 self.turn_off_event[eid] = event
-                self.manually_controlled[eid] = False
+                self.reset(eid)
 
         elif service == SERVICE_TURN_ON:
             _LOGGER.debug("Detected an 'light.turn_on('%s')' event", entity_ids)
@@ -832,6 +859,73 @@ class TurnOnOffListener:
             )
         return manually_controlled
 
+    def significant_change(
+        self, light, adapt_brightness, adapt_color_temp, adapt_rgb_color, threshold=5
+    ):
+        """Has the light made a significant change since last update.
+
+        This method will detect changes that were made to the light without
+        calling 'light.turn_on', so outside of Home Assistant. If a change is
+        detected, we mark the light as 'manually_controlled' until the light
+        or switch is turned 'off' and 'on' again.
+        """
+        if light not in self.last_service_data:
+            return False
+        changed = False
+        service_data = self.last_service_data[light]
+        attributes = self.hass.states.get(light).attributes
+        if (
+            adapt_brightness
+            and ATTR_BRIGHTNESS_PCT in service_data
+            and ATTR_BRIGHTNESS in attributes
+        ):
+            applied_brightness = round(255 * service_data[ATTR_BRIGHTNESS_PCT] / 100)
+            current_brightness = attributes["brightness"]
+            if abs_rel_diff(current_brightness, applied_brightness) > threshold:
+                _LOGGER.debug("Brightness of '%s' significantly changed", light)
+                changed = True
+
+        if (
+            adapt_color_temp
+            and ATTR_COLOR_TEMP in service_data
+            and ATTR_COLOR_TEMP in attributes
+        ):
+            applied_color_temp = service_data[ATTR_COLOR_TEMP]
+            current_color_temp = attributes[ATTR_COLOR_TEMP]
+            if abs_rel_diff(current_color_temp, applied_color_temp) > threshold:
+                _LOGGER.debug(
+                    "Color temperature of '%s' significantly changed",
+                    light,
+                )
+                changed = True
+
+        if (
+            adapt_rgb_color
+            and ATTR_RGB_COLOR in service_data
+            and ATTR_RGB_COLOR in attributes
+        ):
+            applied_rgb_color = service_data[ATTR_RGB_COLOR]
+            current_rgb_color = attributes[ATTR_RGB_COLOR]
+            for col_applied, col_current in zip(applied_rgb_color, current_rgb_color):
+                if abs_rel_diff(col_applied, col_current) > threshold:
+                    _LOGGER.debug(
+                        "color RGB of '%s' significantly changed",
+                        light,
+                    )
+                    changed = True
+
+        if (ATTR_RGB_COLOR in service_data and ATTR_RGB_COLOR not in attributes) or (
+            ATTR_COLOR_TEMP in service_data and ATTR_COLOR_TEMP not in attributes
+        ):
+            # Light switched from RGB mode to color_temp or visa versa
+            _LOGGER.debug(
+                "'%s' switched from RGB mode to color_temp or visa versa",
+                light,
+            )
+            changed = True
+        self.manually_controlled[light] = changed
+        return changed
+
     async def maybe_cancel_adjusting(
         self, entity_id: str, off_to_on_event: Event, on_to_off_event: Optional[Event]
     ) -> bool: