fix: handle animation for lights

Author: muhlba91

custom_components/hella_onyx/sensors/light.py

@@ -1,6 +1,8 @@
 """The ONYX light entity."""
 import asyncio
 import logging
+import time
+from datetime import timedelta
 from math import ceil
 from typing import Any
 
@@ -10,12 +12,18 @@
     LightEntityFeature,
     ATTR_BRIGHTNESS,
 )
+from homeassistant.helpers.event import (
+    track_point_in_utc_time,
+)
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.util import utcnow
+from onyx_client.data.animation_value import AnimationValue
 from onyx_client.enum.action import Action
 from onyx_client.enum.device_type import DeviceType
 from onyx_client.data.numeric_value import NumericValue
 
 from custom_components.hella_onyx.api_connector import APIConnector
+from custom_components.hella_onyx.const import INCREASED_INTERVAL_DELTA
 from custom_components.hella_onyx.sensors.onyx_entity import OnyxEntity
 
 _LOGGER = logging.getLogger(__name__)
@@ -82,6 +90,8 @@ def brightness(self) -> int | None:
             self._uuid,
             brightness,
         )
+        if brightness.animation is not None and len(brightness.animation.keyframes) > 0:
+            self._start_update_device(brightness.animation)
         return brightness.value / brightness.maximum * 255
 
     @property
@@ -137,6 +147,64 @@ def turn_off(self, **kwargs: Any) -> None:
             self.hass.loop,
         )
 
+    def _start_update_device(self, animation: AnimationValue):
+        """Start the update loop."""
+        keyframes = len(animation.keyframes)
+        keyframe = animation.keyframes[keyframes - 1]
+
+        current_time = time.time()
+        end_time = animation.start + keyframe.duration + keyframe.delay
+        delta = end_time - current_time
+        moving = current_time < end_time
+
+        _LOGGER.debug(
+            "updating device %s with current_time %s and end_time %s: %s",
+            self._uuid,
+            current_time,
+            end_time,
+            moving,
+        )
+
+        if moving:
+            track_point_in_utc_time(
+                self.hass,
+                self._end_update_device,
+                utcnow() + timedelta(seconds=delta + INCREASED_INTERVAL_DELTA),
+            )
+        else:
+            _LOGGER.debug("end update device %s due to too old data", self._uuid)
+            self._end_update_device()
+
+    def _end_update_device(self, *args: Any):
+        """Call STOP to update the device values on ONYX."""
+        animation = self._actual_brightness.animation
+        keyframe = (
+            animation.keyframes[len(animation.keyframes) - 1]
+            if animation is not None and len(animation.keyframes) > 0
+            else None
+        )
+        end_time = (
+            (animation.start + keyframe.duration + keyframe.delay)
+            if keyframe is not None
+            else None
+        )
+
+        current_time = time.time()
+
+        if current_time > end_time:
+            _LOGGER.debug(
+                "calling stop to force update device %s",
+                self._uuid,
+            )
+            asyncio.run_coroutine_threadsafe(
+                self.api.send_device_command_action(
+                    self._uuid,
+                    Action.STOP,
+                ),
+                self.hass.loop,
+            )
+        self.async_write_ha_state()
+
     @property
     def _actual_brightness(self) -> int:
         """Get the actual brightness."""
@@ -146,6 +214,7 @@ def _actual_brightness(self) -> int:
             brightness.minimum,
             brightness.maximum,
             brightness.read_only,
+            brightness.animation,
         )
 
     def _get_dim_duration(self, target) -> int:

tests/sensors/test_light.py

@@ -2,12 +2,17 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+import time
+import pytz
+from datetime import datetime
 from homeassistant.components.light import (
     ColorMode,
     brightness_supported,
 )
 from homeassistant.core import HomeAssistant
 from onyx_client.data.numeric_value import NumericValue
+from onyx_client.data.animation_value import AnimationValue
+from onyx_client.data.animation_keyframe import AnimationKeyframe
 from onyx_client.device.light import Light
 from onyx_client.enum.action import Action
 from onyx_client.enum.device_type import DeviceType
@@ -92,6 +97,29 @@ def test_brightness(self, api, entity, device):
         assert entity.brightness == 25.5
         assert api.device.called
 
+    def test_brightness_with_animation(self, api, entity, device):
+        animation = AnimationValue(
+            start=0,
+            current_value=0,
+            keyframes=[
+                AnimationKeyframe(
+                    interpolation="linear", duration=10, delay=0, value=10
+                )
+            ],
+        )
+        device.actual_brightness = NumericValue(
+            value=10,
+            minimum=0,
+            maximum=100,
+            read_only=False,
+            animation=animation,
+        )
+        api.device.return_value = device
+        with patch.object(entity, "_start_update_device") as mock_start_update_device:
+            assert entity.brightness == 25.5
+            mock_start_update_device.assert_called_with(animation)
+            assert api.device.called
+
     def test_is_on(self, api, entity, device):
         device.actual_brightness = NumericValue(
             value=10, minimum=0, maximum=100, read_only=False
@@ -198,3 +226,57 @@ def test__get_dim_duration_invalid_value(self, api, entity, device):
         api.device.return_value = device
         assert entity._get_dim_duration(90) == 5450
         assert api.device.called
+
+    @patch("asyncio.run_coroutine_threadsafe")
+    def test_start_update_device_end(self, api, entity, device):
+        current_time = time.mktime(datetime.now(pytz.timezone("UTC")).timetuple())
+        animation = AnimationValue(
+            start=current_time - 100,
+            current_value=0,
+            keyframes=[
+                AnimationKeyframe(
+                    interpolation="linear",
+                    value=0,
+                    duration=10,
+                    delay=0,
+                )
+            ],
+        )
+        with patch.object(entity, "_end_update_device") as mock_end_update_device:
+            entity._start_update_device(animation)
+            mock_end_update_device.assert_called()
+
+    @patch("asyncio.run_coroutine_threadsafe")
+    def test__end_update_device(self, mock_run_coroutine_threadsafe, api, entity):
+        entity._device.actual_brightness.animation = AnimationValue(
+            time.time() - 1000, 10, [AnimationKeyframe("linear", 0, 100, 90)]
+        )
+        with patch.object(entity, "async_write_ha_state") as mock_async_write_ha_state:
+            entity._end_update_device()
+            assert mock_async_write_ha_state.called
+            assert mock_run_coroutine_threadsafe.called
+            api.send_device_command_action.assert_called_with("uuid", Action.STOP)
+
+    @patch("asyncio.run_coroutine_threadsafe")
+    def test__end_update_device_within_time(
+        self, mock_run_coroutine_threadsafe, api, entity
+    ):
+        entity._device.actual_brightness.animation = AnimationValue(
+            time.time(), 10, [AnimationKeyframe("linear", 0, 20000, 90)]
+        )
+        with patch.object(entity, "async_write_ha_state") as mock_async_write_ha_state:
+            entity._end_update_device()
+            assert mock_async_write_ha_state.called
+            assert not mock_run_coroutine_threadsafe.called
+
+    @patch("asyncio.run_coroutine_threadsafe")
+    def test__end_update_device_within_time_using_delay(
+        self, mock_run_coroutine_threadsafe, api, entity
+    ):
+        entity._device.actual_brightness.animation = AnimationValue(
+            time.time() - 100, 10, [AnimationKeyframe("linear", 100000, 10, 90)]
+        )
+        with patch.object(entity, "async_write_ha_state") as mock_async_write_ha_state:
+            entity._end_update_device()
+            assert mock_async_write_ha_state.called
+            assert not mock_run_coroutine_threadsafe.called