Move all dummy stage movement into a thread

src/openflexure_microscope_server/things/stage/__init__.py

@@ -100,7 +100,7 @@ class BaseStage(lt.Thing):
         self._jog_lock = threading.Lock()
         self._jog_queue = JogQueue()
         self._jog_thread: Optional[threading.Thread] = None
-        self._hardware_position = dict.fromkeys(self._axis_names, 0)
+        self._hardware_position: Mapping[str, int] = dict.fromkeys(self._axis_names, 0)
 
         # This must be the last thing the function does in case it is caught in a try.
         if (

src/openflexure_microscope_server/things/stage/dummy.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import queue
 import threading
 import time
 from collections.abc import Sequence
@@ -13,6 +14,18 @@ import labthings_fastapi as lt
 from . import BaseStage
 
 
+class DummyStageMovement:
+    """A class used internally to send movements to the dummy stages movement thread."""
+
+    def __init__(self, displacement: Optional[Sequence[int]]) -> None:
+        """Initialise with a displacement.
+
+        :param displacement: A sequence of integers or None to stop the motion.
+        """
+        self.started = threading.Event()
+        self.displacement = displacement
+
+
 class DummyStage(BaseStage):
     """A dummy stage for testing purposes.
 
@@ -34,15 +47,22 @@ class DummyStage(BaseStage):
             also doing computationally heavy tasks like simulated image blurring.
         """
         super().__init__(thing_server_interface, **kwargs)
+        self._movement_enabled = False
         self._move_thread: Optional[threading.Thread] = None
+        self._move_queue = queue.Queue[DummyStageMovement]()
+        self._movement_ongoing = False
+
         self.step_time = step_time
         self.instantaneous_position: Mapping[str, int] = self._hardware_position
         self._inst_pos_lock = threading.Lock()
         self._abort_move = threading.Event()
 
     def __enter__(self) -> Self:
-        """Register the stage position when the Thing context manager is opened."""
+        """Register the stage position and start move thread running."""
         self.instantaneous_position = self._hardware_position
+        self._movement_enabled = True
+        self._move_thread = threading.Thread(target=self._move_loop)
+        self._move_thread.start()
         return self
 
     def __exit__(
@@ -51,7 +71,10 @@ class DummyStage(BaseStage):
         _exc_value: Optional[BaseException],
         _traceback: Optional[TracebackType],
     ) -> None:
-        """Nothing to do when the Thing context manager is closed."""
+        """Stop the move thread."""
+        if self._move_thread is not None and self._move_thread.is_alive():
+            self._movement_enabled = False
+            self._move_thread.join()
 
     axis_inverted: dict[str, bool] = lt.setting(
         default={"x": True, "y": False, "z": False}, readonly=True
@@ -62,45 +85,73 @@ class DummyStage(BaseStage):
         """Read position from the stage and set the corresponding property."""
         pass
 
-    def _calc_fractional_pos(
+    def _set_pos_during_move(
         self, displacement: Sequence[int], fraction_complete: float
-    ) -> dict[str, int]:
+    ) -> None:
-        """Calculate the position a fraction through a move."""
+        """Set the instantaneous position from the fraction through a move.
-        return {
+
+        If this is the final set, also update hardware position
+        """
+        self.instantaneous_position = {
             ax: self._hardware_position[ax] + int(fraction_complete * disp)
             for ax, disp in zip(self.axis_names, displacement, strict=True)
         }
 
-    def _apply_move(self, displacement: Sequence[int]) -> None:
+    def _move_loop(self) -> None:
-        """Make a move, this function is designed to be run in a thread by jogging.
+        """Run the move loop. This should be run in a thread on enter.
 
-        This can't be used for normal movements as the delays setting up the extra
+        This controls all movement of the dummy stage.
-        threads causes autofucs to fail. As such `_hardware_move_relative` is similarly
-        structured.
-
-        Interrupt this function with the ``self._abort_move`` event.
-
-        :param displacement: The (x, y, z) position.
         """
+        # Start with no move request
+        movement_request: Optional[DummyStageMovement] = None
+        # Loop until the server ends.
+        while self._movement_enabled:
+            # First check for a new movement request. These will interrupt ongoing moves
+            movement_request = self._check_for_new_move_request()
+
+            # If there is a new movement.
+            if movement_request is not None:
+                # Set the hardware position from instantaneous before continuing.
+                self._hardware_position = self.instantaneous_position
+
+                if movement_request.displacement is None:
+                    # If it is a stop command, stop moving
+                    self._movement_ongoing = False
+                    movement_request.started.set()
+                else:
+                    # Record the displacement for future iterations
+                    displacement = movement_request.displacement
+                    # If it is a move command, set up the variables for the move
+                    self._movement_ongoing = True
+                    movement_request.started.set()
+                    fraction_complete = 0.0
+                    dt = self.step_time
+                    max_displacement = max(abs(v) for v in displacement)
+                    # Always wait at least dt
+                    total_time = max((dt * max_displacement), dt)
+                    start_time = time.time()
+                    time.sleep(dt)
+
+            # If a movement is ongoing
+            if self._movement_ongoing:
+                fraction_complete = (time.time() - start_time) / total_time
+                if fraction_complete < 1:
+                    self._set_pos_during_move(displacement, fraction_complete)
+                else:
+                    # move is complete
+                    fraction_complete = 1.0
+                    self._set_pos_during_move(displacement, fraction_complete)
+                    self._hardware_position = self.instantaneous_position
+                    self._movement_ongoing = False
+
+    def _check_for_new_move_request(self) -> Optional[DummyStageMovement]:
+        """Check for new move request to the move_loop."""
         try:
-            fraction_complete = 0.0
+            # Timeout sets the motor speed time
-            dt = self.step_time
+            timeout = self.step_time * 10 if self._movement_ongoing else 0.1
-            max_displacement = max(abs(v) for v in displacement)
+            return self._move_queue.get(timeout=timeout)
-            start_time = time.time()
+        except queue.Empty:
-            while time.time() - start_time < dt * max_displacement:
+            return None
-                if self._abort_move.is_set():
-                    break
-                fraction_complete = (time.time() - start_time) / (dt * max_displacement)
-                self.instantaneous_position = self._calc_fractional_pos(
-                    displacement, fraction_complete
-                )
-            if not self._abort_move.is_set():
-                fraction_complete = 1.0
-        finally:
-            self._hardware_position = self._calc_fractional_pos(
-                displacement, fraction_complete
-            )
-            self.instantaneous_position = self._hardware_position
 
     def _hardware_start_move_relative(self, displacement: Sequence[int]) -> None:
         """Start a relative move.
@@ -111,22 +162,16 @@ class DummyStage(BaseStage):
         """
         with self._hardware_lock:
             self.moving = True
-            self._abort_move.clear()
+            self._move_queue.put(DummyStageMovement(displacement))
-            self._move_thread = threading.Thread(
-                target=self._apply_move, args=(displacement,)
-            )
-            self._move_thread.start()
 
     def _hardware_stop(self) -> None:
         with self._hardware_lock:
-            self._abort_move.set()
+            self._move_queue.put(DummyStageMovement(None))
             self.moving = False
 
     def _poll_moving(self) -> bool:
         """Determine if the stage is still moving."""
-        if self._move_thread is None:
+        moving = self._movement_ongoing
-            return False
-        moving = self._move_thread.is_alive()
         if self.moving != moving:
             self.moving = moving
         return moving
@@ -146,38 +191,26 @@ class DummyStage(BaseStage):
         with self._hardware_lock:
             displacement = [kwargs.get(k, 0) for k in self.axis_names]
             self.moving = True
-            # This follows a similar structure to _apply_move but need a different
+            move_request = DummyStageMovement(displacement)
-            # breakning mechanism. Running apply move in a thread and breaking here
+            self._move_queue.put(move_request)
-            # results in significantly worse autofocus
+
-            try:
+            move_request.started.wait(timeout=0.1)
-                fraction_complete = 0.0
+            while self._movement_ongoing:
-                dt = self.step_time
+                try:
-                max_displacement = max(abs(v) for v in displacement)
-                start_time = time.time()
-                while time.time() - start_time < dt * max_displacement:
                     if block_cancellation:
-                        time.sleep(self.step_time)
+                        time.sleep(self.step_time * 10)
                     else:
-                        lt.cancellable_sleep(self.step_time)
+                        lt.cancellable_sleep(self.step_time * 10)
-                    fraction_complete = (time.time() - start_time) / (
+
-                        dt * max_displacement
+                except lt.exceptions.InvocationCancelledError as e:
-                    )
+                    # If the move has been cancelled, stop it but don't handle the
-                    self.instantaneous_position = self._calc_fractional_pos(
+                    # exception. We need the exception to propagate in order to stop
-                        displacement, fraction_complete
+                    # any calling tasks, and to mark the invocation as "cancelled"
-                    )
+                    # rather than stopped.
-                fraction_complete = 1.0
+                    self._move_queue.put(DummyStageMovement(None))
-            except lt.exceptions.InvocationCancelledError as e:
+                    raise e
-                # If the move has been cancelled, stop it but don't handle the
+                finally:
-                # exception. We need the exception to propagate in order to stop
+                    self.moving = False
-                # any calling tasks, and to mark the invocation as "cancelled"
-                # rather than stopped.
-                raise e
-            finally:
-                self._hardware_position = self._calc_fractional_pos(
-                    displacement, fraction_complete
-                )
-                self.instantaneous_position = self._hardware_position
-                self.moving = False
 
     def _hardware_move_absolute(
         self,

tests/unit_tests/test_stage.py

@@ -37,7 +37,7 @@ path3d = st.lists(point3d, min_size=5, max_size=10)
 @pytest.fixture
 def dummy_stage():
     """Return a dummy stage with a very low step time."""
-    return create_thing_without_server(DummyStage, step_time=0.000001)
+    return create_thing_without_server(DummyStage, step_time=0.0001)
 
 
 def test_not_implemented_methods():
@@ -206,15 +206,20 @@ def _test_move_relative(dummy_stage, axis_inverted, path):
     y_dir = -1 if axis_inverted["y"] else 1
     z_dir = -1 if axis_inverted["z"] else 1
 
-    position = list(dummy_stage.position.values())
+    # Enter the stage to start the movement thread.
-    for movement in path:
+    with dummy_stage:
-        dummy_stage.move_relative(x=movement[0], y=movement[1], z=movement[2])
+        assert dummy_stage._move_thread.is_alive()
-        position = [pos + move for pos, move in zip(position, movement, strict=True)]
+        position = list(dummy_stage.position.values())
-        stage_pos = dummy_stage.get_xyz_position()
+        for movement in path:
-        hw_pos = dummy_stage._hardware_position
+            dummy_stage.move_relative(x=movement[0], y=movement[1], z=movement[2])
-        assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir
+            position = [
-        assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir
+                pos + move for pos, move in zip(position, movement, strict=True)
-        assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir
+            ]
+            stage_pos = dummy_stage.get_xyz_position()
+            hw_pos = dummy_stage._hardware_position
+            assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir
+            assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir
+            assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir
 
 
 @given(path=path3d)
@@ -260,14 +265,18 @@ def _test_move_absolute(dummy_stage, axis_inverted, path):
     y_dir = -1 if axis_inverted["y"] else 1
     z_dir = -1 if axis_inverted["z"] else 1
     position = list(dummy_stage.position.values())
-    for move_to in path:
+
-        dummy_stage.move_absolute(x=move_to[0], y=move_to[1], z=move_to[2])
+    # Enter the stage to start the movement thread.
-        position = move_to
+    with dummy_stage:
-        stage_pos = dummy_stage.get_xyz_position()
+        assert dummy_stage._move_thread.is_alive()
-        hw_pos = dummy_stage._hardware_position
+        for move_to in path:
-        assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir
+            dummy_stage.move_absolute(x=move_to[0], y=move_to[1], z=move_to[2])
-        assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir
+            position = move_to
-        assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir
+            stage_pos = dummy_stage.get_xyz_position()
+            hw_pos = dummy_stage._hardware_position
+            assert position[0] == stage_pos[0] == hw_pos["x"] * x_dir
+            assert position[1] == stage_pos[1] == hw_pos["y"] * y_dir
+            assert position[2] == stage_pos[2] == hw_pos["z"] * z_dir
 
 
 @given(path=path3d)