Allow changing level or tilt while blind is moving (#1549)

* Allow changing level or tilt while blind is moving

* Rename _transmit_position to _send_level

* Do not send use_command_queue explicitly

* Fix empty_queue

* Reproduce race condition better in test case

Previously, the test case would succeed even when a command queue was
being used. Since that leads to bugs on a real CCU (where the time for
sending a command is not negligible), this commit introduces a similar
communication delay to the test.

At the same time, this allows us to reduce the number of iterations that
this test needs to run in order to be reasonably sure that a race
condition is absent.

* Pass collector argument

---------

Co-authored-by: SukramJ <[email protected]>

hahomematic/platforms/custom/cover.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Mapping
 from enum import IntEnum, StrEnum
 import logging
@@ -96,6 +97,7 @@ class CeCover(CustomEntity):
     def _init_entity_fields(self) -> None:
         """Init the entity fields."""
         super()._init_entity_fields()
+        self._command_processing_lock = asyncio.Lock()
         self._e_direction: HmSensor = self._get_entity(field=Field.DIRECTION, entity_type=HmSensor)
         self._e_level: HmFloat = self._get_entity(field=Field.LEVEL, entity_type=HmFloat)
         self._e_stop: HmAction = self._get_entity(field=Field.STOP, entity_type=HmAction)
@@ -256,7 +258,20 @@ def current_tilt_position(self) -> int:
         """Return current tilt position of cover."""
         return int(self._channel_tilt_level * 100)
 
-    @bind_collector(use_command_queue=False)
+    @property
+    def _target_level(self) -> float | None:
+        """Return the level of last service call."""
+        if (last_value_send := self._e_level.unconfirmed_last_value_send) is not None:
+            return float(last_value_send)
+        return None
+
+    @property
+    def _target_tilt_level(self) -> float | None:
+        """Return the tilt level of last service call."""
+        if (last_value_send := self._e_level_2.unconfirmed_last_value_send) is not None:
+            return float(last_value_send)
+        return None
+
     async def set_position(
         self,
         position: int | None = None,
@@ -279,22 +294,54 @@ async def _set_level(
         collector: CallParameterCollector | None = None,
     ) -> None:
         """
-        Move the cover to a specific tilt level. Value range is 0.0 to 1.01.
+        Move the cover to a specific tilt level. Value range is 0.0 to 1.00.
 
-        1.01 means no change.
+        level or tilt_level may be set to None for no change.
         """
-        _level = level if level is not None else self.current_position / 100.0
-        _tilt_level = tilt_level if tilt_level is not None else self.current_tilt_position / 100.0
+        currently_moving = False
+
+        async with self._command_processing_lock:
+            if level is not None:
+                _level = level
+            elif self._target_level is not None:
+                # The blind moves and the target blind height is known
+                currently_moving = True
+                _level = self._target_level
+            else:  # The blind is at a standstill and no level is explicitly requested => we remain at the current level
+                _level = self._channel_level
+
+            if tilt_level is not None:
+                _tilt_level = tilt_level
+            elif self._target_tilt_level is not None:
+                # The blind moves and the target slat position is known
+                currently_moving = True
+                _tilt_level = self._target_tilt_level
+            else:  # The blind is at a standstill and no tilt is explicitly desired => we remain at the current angle
+                _tilt_level = self._channel_tilt_level
+
+            if currently_moving:
+                # Blind actors are buggy when sending new coordinates while they are moving. So we stop them first.
+                await self._stop()
+
+            await self._send_level(level=_level, tilt_level=_tilt_level, collector=collector)
+
+    @bind_collector()
+    async def _send_level(
+        self,
+        level: float,
+        tilt_level: float,
+        collector: CallParameterCollector | None = None,
+    ) -> None:
+        """Transmit a new target level to the device."""
         if self._e_combined.is_hmtype and (
-            combined_parameter := self._get_combined_value(level=_level, tilt_level=_tilt_level)
+            combined_parameter := self._get_combined_value(level=level, tilt_level=tilt_level)
         ):
             await self._e_combined.send_value(value=combined_parameter, collector=collector)
             return
 
-        await self._e_level_2.send_value(value=_tilt_level, collector=collector)
-        await super()._set_level(level=_level, collector=collector)
+        await self._e_level_2.send_value(value=tilt_level, collector=collector)
+        await super()._set_level(level=level, collector=collector)
 
-    @bind_collector(use_command_queue=False)
     async def open(self, collector: CallParameterCollector | None = None) -> None:
         """Open the cover and open the tilt."""
         if not self.is_state_change(open=True, tilt_open=True):
@@ -305,7 +352,6 @@ async def open(self, collector: CallParameterCollector | None = None) -> None:
             collector=collector,
         )
 
-    @bind_collector(use_command_queue=False)
     async def close(self, collector: CallParameterCollector | None = None) -> None:
         """Close the cover and close the tilt."""
         if not self.is_state_change(close=True, tilt_close=True):
@@ -316,24 +362,32 @@ async def close(self, collector: CallParameterCollector | None = None) -> None:
             collector=collector,
         )
 
-    @bind_collector(use_command_queue=False)
+    async def stop(self, collector: CallParameterCollector | None = None) -> None:
+        """Stop the device if in motion."""
+        async with self._command_processing_lock:
+            await self._stop(collector=collector)
+
+    @bind_collector()
+    async def _stop(self, collector: CallParameterCollector | None = None) -> None:
+        """Stop the device if in motion. Do only call with _command_processing_lock held."""
+        self.central.command_queue_handler.empty_queue(address=self._channel_address)
+        await super().stop(collector=collector)
+
     async def open_tilt(self, collector: CallParameterCollector | None = None) -> None:
         """Open the tilt."""
         if not self.is_state_change(tilt_open=True):
             return
         await self._set_level(tilt_level=self._open_tilt_level, collector=collector)
 
-    @bind_collector(use_command_queue=False)
     async def close_tilt(self, collector: CallParameterCollector | None = None) -> None:
         """Close the tilt."""
         if not self.is_state_change(tilt_close=True):
             return
         await self._set_level(tilt_level=self._closed_level, collector=collector)
 
-    @bind_collector()
     async def stop_tilt(self, collector: CallParameterCollector | None = None) -> None:
-        """Stop the device if in motion."""
-        await self._e_stop.send_value(value=True, collector=collector)
+        """Stop the device if in motion. Use only when command_processing_lock is held."""
+        await self.stop(collector=collector)
 
     def is_state_change(self, **kwargs: Any) -> bool:
         """Check if the state changes due to kwargs."""

tests/test_cover.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import cast
-from unittest.mock import Mock, call
+from unittest.mock import DEFAULT, Mock, call
 
 import pytest
 
@@ -412,7 +413,7 @@ async def test_ceblind(
     call_count = len(mock_client.method_calls)
     await cover.open_tilt()
     await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL_SLATS", _OPEN_TILT_LEVEL)
-    assert call_count == len(mock_client.method_calls) - 2
+    assert call_count == len(mock_client.method_calls) - 5
 
     await cover.close_tilt()
     await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL_SLATS", _CLOSED_LEVEL)
@@ -427,6 +428,61 @@ async def test_ceblind(
     assert call_count == len(mock_client.method_calls)
 
 
+@pytest.mark.asyncio()
+@pytest.mark.parametrize(
+    (
+        "address_device_translation",
+        "do_mock_client",
+        "add_sysvars",
+        "add_programs",
+        "ignore_devices_on_create",
+        "un_ignore_list",
+    ),
+    [
+        (TEST_DEVICES, True, False, False, None, None),
+    ],
+)
+async def test_ceblind_separate_level_and_tilt_change(
+    central_client_factory: tuple[CentralUnit, Client | Mock, helper.Factory],
+) -> None:
+    """Test if CeBlind sends correct commands even when rapidly changing level and tilt via separate service calls."""
+    central, mock_client, _ = central_client_factory
+    cover: CeBlind = cast(CeBlind, helper.get_prepared_custom_entity(central, "VCU0000145", 1))
+
+    # In order for this test to make sense, communication with CCU must take some amount of time.
+    # This is not the case with the default local client used during testing, so we add a slight delay.
+    async def delay_communication(*args, **kwargs):
+        await asyncio.sleep(0.1)
+        return DEFAULT
+
+    mock_client.set_value.side_effect = delay_communication
+
+    # We test for the absence of race conditions.
+    # We repeat the test a few times so that it becomes unlikely for the race condition to remain undetected.
+    for _ in range(10):
+        await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL", 0)
+        await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL_SLATS", 0)
+        assert cover.current_position == 0
+        assert cover.current_tilt_position == 0
+
+        await asyncio.gather(
+            cover.set_position(position=81),
+            cover.set_position(tilt_position=19),
+        )
+
+        assert mock_client.method_calls[-1] == call.set_value(
+            channel_address="VCU0000145:1",
+            paramset_key="VALUES",
+            parameter="LEVEL_COMBINED",
+            value="0xa2,0x26",
+            wait_for_callback=WAIT_FOR_CALLBACK,
+        )
+        await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL", 0.81)
+        await central.event(const.INTERFACE_ID, "VCU0000145:1", "LEVEL_SLATS", 0.19)
+        assert cover.current_position == 81
+        assert cover.current_tilt_position == 19
+
+
 @pytest.mark.asyncio()
 @pytest.mark.parametrize(
     (