Interrupt movement routine when desk stops moving

idasen/__init__.py

@@ -11,6 +11,7 @@
 import asyncio
 import logging
 import sys
+import time
 
 
 _UUID_HEIGHT: str = "99fa0021-338a-1024-8a49-009c0215f78a"
@@ -299,6 +300,7 @@ async def move_to_target(self, target: float):
         async def do_move():
             previous_height = await self.get_height()
             will_move_up = target > previous_height
+            last_move_time: Optional[float] = None
             while True:
                 height = await self.get_height()
                 difference = target - height
@@ -310,6 +312,20 @@ async def do_move():
                         "stopped moving because desk safety feature kicked in"
                     )
                     return
+
+                if height == previous_height:
+                    if (
+                        last_move_time is not None
+                        and time.time() - last_move_time > 0.5
+                    ):
+                        self._logger.warning(
+                            "desk is not moving anymore. physical button probably "
+                            "pressed"
+                        )
+                        return
+                else:
+                    last_move_time = time.time()
+
                 if not self._moving:
                     return
 

tests/test_idasen.py

@@ -151,6 +151,26 @@ async def test_move_to_target(desk: IdasenDesk, target: float):
     assert abs(await desk.get_height() - target) < 0.005
 
 
+async def test_move_abort_when_no_movement():
+    desk = IdasenDesk(mac=desk_mac)
+    client = MockBleakClient()
+    desk._client = client
+    client.write_gatt_char = mock.AsyncMock()
+
+    async def write_gatt_char_mock(
+        uuid: str, command: bytearray, response: bool = False
+    ):
+        if client.write_gatt_char.call_count == 1:
+            assert desk.is_moving
+            client._height -= 0.001
+
+    client.write_gatt_char.side_effect = write_gatt_char_mock
+
+    async with desk:
+        await desk.move_to_target(0.7)
+        assert not desk.is_moving
+
+
 async def test_move_stop():
     desk = IdasenDesk(mac=desk_mac)
     client = MockBleakClient()