fix: interpolate updates instead of dropping old ones

custom_components/hella_onyx/sensors/light.py

@@ -82,7 +82,7 @@ def supported_color_modes(self) -> set[ColorMode] | set[str] | None:
         return [self.color_mode]
 
     @property
-    def brightness(self) -> int | None:
+    def brightness(self) -> int:
         """Return the brightness of this light between 0..255."""
         brightness = self._actual_brightness
         _LOGGER.debug(
@@ -155,17 +155,17 @@ def _start_update_device(self, animation: AnimationValue):
         current_time = time.time()
         end_time = animation.start + keyframe.duration + keyframe.delay
         delta = end_time - current_time
-        moving = current_time < end_time
+        updating = 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,
+            updating,
         )
 
-        if moving:
+        if updating:
             track_point_in_utc_time(
                 self.hass,
                 self._end_update_device,
@@ -183,11 +183,10 @@ def _end_update_device(self, *args: Any):
             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
+        start_time = (
+            (animation.start + keyframe.delay) if keyframe is not None else None
         )
+        end_time = (start_time + keyframe.duration) if keyframe is not None else None
 
         current_time = time.time()
 
@@ -203,10 +202,22 @@ def _end_update_device(self, *args: Any):
                 ),
                 self.hass.loop,
             )
-        self.async_write_ha_state()
+        elif current_time > start_time:
+            delta = current_time - (animation.start + keyframe.delay)
+            delta_per_unit = (
+                self._device.target_brightness.value - animation.current_value
+            ) / keyframe.duration
+            update = ceil(animation.current_value + delta_per_unit * delta)
+            _LOGGER.debug(
+                "interpolating brightness update for device %s: %d",
+                self._uuid,
+                update,
+            )
+            self._device.actual_brightness.value = update
+            self.async_write_ha_state()
 
     @property
-    def _actual_brightness(self) -> int:
+    def _actual_brightness(self) -> NumericValue:
         """Get the actual brightness."""
         brightness = self._device.actual_brightness
         return NumericValue(

custom_components/hella_onyx/sensors/shutter.py

@@ -250,13 +250,14 @@ def _end_moving_device(self, *args: Any):
             if position_animation is not None and len(position_animation.keyframes) > 0
             else None
         )
+        position_start_time = (
+            (position_animation.start + position_keyframe.delay)
+            if position_keyframe is not None and position_animation is not None
+            else None
+        )
         position_end_time = (
-            (
-                position_animation.start
-                + position_keyframe.duration
-                + position_keyframe.delay
-            )
-            if position_keyframe is not None
+            (position_start_time + position_keyframe.duration)
+            if position_keyframe is not None and position_start_time is not None
             else None
         )
 
@@ -266,9 +267,14 @@ def _end_moving_device(self, *args: Any):
             if angle_animation is not None and len(angle_animation.keyframes) > 0
             else None
         )
+        angle_start_time = (
+            (angle_animation.start + angle_keyframe.delay)
+            if angle_keyframe is not None and angle_animation is not None
+            else None
+        )
         angle_end_time = (
-            (angle_animation.start + angle_keyframe.duration + angle_keyframe.delay)
-            if angle_keyframe is not None
+            (angle_start_time + angle_keyframe.duration)
+            if angle_keyframe is not None and angle_start_time is not None
             else None
         )
 
@@ -278,14 +284,50 @@ def _end_moving_device(self, *args: Any):
             if (
                 (angle_end_time is None and current_time > position_end_time)
                 or (position_end_time is None and current_time > angle_end_time)
-                or (current_time > position_end_time and current_time > angle_end_time)
+                or (
+                    position_end_time is not None
+                    and current_time > position_end_time
+                    and angle_end_time is not None
+                    and current_time > angle_end_time
+                )
             ):
                 self.stop_cover()
-            else:
-                _LOGGER.debug(
-                    "not ending moving device %s. overlapping angle and positioning",
-                    self._uuid,
-                )
+            elif (
+                position_start_time is not None and current_time > position_start_time
+            ) or (angle_start_time is not None and current_time > angle_start_time):
+                if position_animation is not None:
+                    delta = current_time - (
+                        position_animation.start + position_keyframe.delay
+                    )
+                    delta_per_unit = (
+                        self._device.target_position.value
+                        - position_animation.current_value
+                    ) / position_keyframe.duration
+                    update = ceil(
+                        position_animation.current_value + delta_per_unit * delta
+                    )
+                    _LOGGER.debug(
+                        "interpolating actual_position update for device %s: %d",
+                        self._uuid,
+                        update,
+                    )
+                    self._device.actual_position.value = update
+                if angle_animation is not None:
+                    delta = current_time - (
+                        angle_animation.start + angle_keyframe.delay
+                    )
+                    delta_per_unit = (
+                        self._device.target_angle.value - angle_animation.current_value
+                    ) / angle_keyframe.duration
+                    update = ceil(
+                        angle_animation.current_value + delta_per_unit * delta
+                    )
+                    _LOGGER.debug(
+                        "interpolating actual_angle update for device %s: %d",
+                        self._uuid,
+                        update,
+                    )
+                    self._device.actual_angle.value = update
 
         self.async_write_ha_state()
 

tests/sensors/test_light.py

@@ -3,6 +3,7 @@
 
 import pytest
 import time
+from math import ceil
 import pytz
 from datetime import datetime
 from homeassistant.components.light import (
@@ -189,10 +190,10 @@ def test_turn_on_no_brightness(
 
     def test__get_dim_duration(self, api, entity, device):
         device.actual_brightness = NumericValue(
-            value=100, maximum=100, minimum=0, read_only=False
+            value=14645, maximum=65535, minimum=0, read_only=False
         )
         api.device.return_value = device
-        assert entity._get_dim_duration(90) == 50
+        assert entity._get_dim_duration(31) == 726
         assert api.device.called
 
     def test__actual_brightness_no_value(self, api, entity, device):
@@ -247,36 +248,75 @@ def test_start_update_device_end(self, api, entity, device):
             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)]
+    def test__end_update_device(
+        self, mock_run_coroutine_threadsafe, api, entity, device
+    ):
+        device.actual_brightness = NumericValue(
+            value=None,
+            maximum=100,
+            minimum=0,
+            read_only=False,
+            animation=AnimationValue(
+                time.time() - 1000, 10, [AnimationKeyframe("linear", 0, 100, 90)]
+            ),
         )
+        api.device.return_value = device
         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 api.device.called
+            assert not 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
+        self, mock_run_coroutine_threadsafe, api, entity, device
     ):
-        entity._device.actual_brightness.animation = AnimationValue(
-            time.time(), 10, [AnimationKeyframe("linear", 0, 20000, 90)]
+        device.actual_brightness = NumericValue(
+            value=0,
+            maximum=100,
+            minimum=0,
+            read_only=False,
+            animation=AnimationValue(
+                time.time(), 0, [AnimationKeyframe("linear", 0, 20000, 50)]
+            ),
         )
+        device.target_brightness = NumericValue(
+            value=50,
+            maximum=100,
+            minimum=0,
+            read_only=False,
+        )
+        api.device.return_value = device
         with patch.object(entity, "async_write_ha_state") as mock_async_write_ha_state:
             entity._end_update_device()
+            assert api.device.called
             assert mock_async_write_ha_state.called
             assert not mock_run_coroutine_threadsafe.called
+            assert entity._device.actual_brightness.value == 1
 
     @patch("asyncio.run_coroutine_threadsafe")
     def test__end_update_device_within_time_using_delay(
-        self, mock_run_coroutine_threadsafe, api, entity
+        self, mock_run_coroutine_threadsafe, api, entity, device
     ):
-        entity._device.actual_brightness.animation = AnimationValue(
-            time.time() - 100, 10, [AnimationKeyframe("linear", 100000, 10, 90)]
+        device.actual_brightness = NumericValue(
+            value=0,
+            maximum=100,
+            minimum=0,
+            read_only=False,
+            animation=AnimationValue(
+                time.time() - 100, 0, [AnimationKeyframe("linear", 100000, 10, 50)]
+            ),
+        )
+        device.target_brightness = NumericValue(
+            value=50,
+            maximum=100,
+            minimum=0,
+            read_only=False,
         )
+        api.device.return_value = device
         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 api.device.called
+            assert not mock_async_write_ha_state.called
             assert not mock_run_coroutine_threadsafe.called