Fix a number of issues with camera_stage_mapping

src/openflexure_microscope_server/things/camera_stage_mapping.py

@@ -10,16 +10,9 @@ This module is only intended to be called from the OpenFlexure Microscope
 server, and depends on that server and its underlying LabThings library.
 """
 
+import json
 import time
-from typing import (
+from typing import Any, List, Mapping, NamedTuple, Optional, Tuple, cast
-    Any,
-    Dict,
-    List,
-    Mapping,
-    NamedTuple,
-    Optional,
-    Tuple,
-)
 
 import numpy as np
 
@@ -35,9 +28,6 @@ from labthings_fastapi.types.numpy import DenumpifyingDict
 from .camera import BaseCamera
 from .stage import BaseStage
 
-CoordinateType = Tuple[float, float, float]
-XYCoordinateType = Tuple[float, float]
-
 
 class MoveHistory(NamedTuple):
     """A named tuple containing the position over time for a single move.
@@ -49,7 +39,27 @@ class MoveHistory(NamedTuple):
     """
 
     times: List[float]
-    stage_positions: List[CoordinateType]
+    stage_positions: List[tuple[int, int, int]]
+
+
+def _array_to_stage_tuple(pos: np.ndarray) -> tuple[int, int, int]:
+    """Convert a numpy array into a tuple of ints.
+
+    :param pos: Input position array must be 3 elements long.
+    :return: a tuple of 3 integers
+    :raises ValueError: If the array is not of length 3.
+    """
+    pos_tuple = tuple(int(i) for i in pos)
+    if len(pos_tuple) == 3:
+        return cast(tuple[int, int, int], pos_tuple)
+    raise ValueError("Input array was not 3 elements long.")
+
+
+def _serialise_numpy_in_dict(dict_with_numpy: dict) -> dict:
+    serialised = json.loads(DenumpifyingDict(dict_with_numpy).model_dump_json())
+    if not isinstance(serialised, dict):
+        raise TypeError(f"Expecting a dictionary to serialise not a {type(serialised)}")
+    return serialised
 
 
 class RecordedMove:
@@ -68,21 +78,24 @@ class RecordedMove:
             be called whenever the instance is called.
         """
         self._stage = stage
-        self._current_position: Optional[CoordinateType] = None
+        self._current_position: Optional[tuple[int, int, int]] = None
-        self._history: List[Tuple[float, Optional[CoordinateType]]] = []
+        self._history: List[Tuple[float, tuple[int, int, int]]] = []
 
-    def __call__(self, new_position: CoordinateType) -> None:
+    def __call__(self, new_position: np.ndarray) -> None:
         """Move to a new position, and record it."""
-        self._history.append((time.time(), self._current_position))
+        new_stage_pos = _array_to_stage_tuple(new_position)
-        self._stage.move_to_xyz_position(xyz_pos=new_position)
+        starting_pos = self._current_position
-        self._current_position = new_position
+        if starting_pos is not None:
-        self._history.append((time.time(), self._current_position))
+            self._history.append((time.time(), starting_pos))
+        self._stage.move_to_xyz_position(xyz_pos=new_stage_pos)
+        self._current_position = new_stage_pos
+        self._history.append((time.time(), new_stage_pos))
 
     @property
     def history(self) -> MoveHistory:
         """The history, as a numpy array of times and another of positions."""
-        times: List[float] = [t for t, p in self._history if p is not None]
+        times: List[float] = [t for t, p in self._history]
-        positions: List[CoordinateType] = [p for t, p in self._history if p is not None]
+        positions: List[tuple[int, int, int]] = [p for t, p in self._history]
         return MoveHistory(times, positions)
 
     def clear_history(self) -> None:
@@ -109,8 +122,7 @@ class CameraStageMapper(lt.Thing):
     _cam: BaseCamera = lt.thing_slot()
     _stage: BaseStage = lt.thing_slot()
 
-    @lt.action
+    def calibrate_1d(self, direction: Tuple[int, int, int]) -> dict:
-    def calibrate_1d(self, direction: Tuple[float, float, float]) -> DenumpifyingDict:
         """Move a microscope's stage in 1D, and figure out the relationship with the camera."""
         # Record positions and times for stage calibration
         recorded_move = RecordedMove(self._stage)
@@ -140,7 +152,7 @@ class CameraStageMapper(lt.Thing):
         return result
 
     @lt.action
-    def calibrate_xy(self) -> DenumpifyingDict:
+    def calibrate_xy(self) -> dict:
         """Move the microscope's stage in X and Y, to calibrate its relationship to the camera.
 
         This performs two 1d calibrations in x and y, then combines their results.
@@ -170,7 +182,7 @@ class CameraStageMapper(lt.Thing):
         )
         self.logger.info(f"CSM matrix is {csm_as_string}.")
 
-        data: Dict[str, dict] = {
+        data = {
             "camera_stage_mapping_calibration": cal_xy,
             "linear_calibration_x": cal_x,
             "linear_calibration_y": cal_y,
@@ -179,7 +191,8 @@ class CameraStageMapper(lt.Thing):
             "downsampling": downsampling_factor,
         }
 
-        self.last_calibration = DenumpifyingDict(data).model_dump()
+        data = _serialise_numpy_in_dict(data)
+        self.last_calibration = data
 
         return data
 
@@ -226,13 +239,14 @@ class CameraStageMapper(lt.Thing):
         """Whether the camera stage mapper needs calibrating."""
         return self.image_to_stage_displacement_matrix is None
 
-    def assert_calibrated(self) -> None:
+    def assert_calibration(self) -> List[List[float]]:
-        """Raise an exception if the image_to_stage_displacement matrix is not set."""
+        """Return image_to_stage_displacement matrix or raise error if it's not set."""
         if self.image_to_stage_displacement_matrix is None:
             raise CSMUncalibratedError(
                 "The camera_stage_mapping calibration is not yet available. "
                 "This probably means you need to run the calibration routine."
             )
+        return self.image_to_stage_displacement_matrix
 
     @lt.action
     def move_in_image_coordinates(self, x: float, y: float) -> None:
@@ -247,24 +261,26 @@ class CameraStageMapper(lt.Thing):
         and ``y`` to the shorter one. Checking what shape your chosen toolkit reports for
         an image usually helps resolve any ambiguity.
         """
-        self.assert_calibrated()
+        self._stage.move_relative(
-        self._stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y))
+            **self.convert_image_to_stage_coordinates(x=x, y=y),
+            block_cancellation=False,
+        )
 
     @lt.action
     def convert_image_to_stage_coordinates(
         self, x: float, y: float, **_kwargs: float
     ) -> Mapping[str, int]:
         """Convert image coordinates to stage coordinates. Only x and y are returned."""
-        self.assert_calibrated()
+        csm_matrix = self.assert_calibration()
-        return csm_img_to_stage(self.image_to_stage_displacement_matrix, x=x, y=y)
+        return csm_img_to_stage(csm_matrix, x=x, y=y)
 
     @lt.action
     def convert_stage_to_image_coordinates(
         self, x: int, y: int, **_kwargs: int
     ) -> Mapping[str, float]:
         """Convert stage coordinates to image coordinates. Only x and y are returned."""
-        self.assert_calibrated()
+        csm_matrix = self.assert_calibration()
-        return csm_stage_to_img(self.image_to_stage_displacement_matrix, x=x, y=y)
+        return csm_stage_to_img(csm_matrix, x=x, y=y)
 
     @lt.property
     def thing_state(self) -> Mapping[str, Any]:

src/openflexure_microscope_server/things/smart_scan.py

@@ -231,7 +231,7 @@ class SmartScanThing(lt.Thing):
 
         Raise warning if not using background detect that scan will go on until max steps reached
         """
-        self._csm.assert_calibrated()
+        self._csm.assert_calibration()
 
         if self.skip_background:
             if not self._cam.background_detector_status.ready: