Split out matix calculations from CSM, and continue testing recentre

src/openflexure_microscope_server/things/camera_stage_mapping.py

@@ -31,8 +31,9 @@ from camera_stage_mapping.exceptions import MappingError
 
 import labthings_fastapi as lt
 from labthings_fastapi.types.numpy import DenumpifyingDict
-
 from camera_stage_mapping.camera_stage_tracker import Tracker
+
+from openflexure_microscope_server.utilities import apply_2d_matrix_to_mapping
 from .camera import CameraDependency as CameraClient
 from .stage import StageDependency as Stage
 
@@ -218,8 +219,8 @@ class CameraStageMapper(lt.Thing):
         .. code-block:: python
 
             stage_disp = np.dot(
-                np.array(image_to_stage_displacement_matrix),
                 np.array([dy,dx]),
+                np.array(image_to_stage_displacement_matrix),
             )
 
         """
@@ -281,22 +282,20 @@ class CameraStageMapper(lt.Thing):
     ) -> Mapping[str, int]:
         """Convert image coordinates to stage coordinates. Only x and y are returned."""
         self.assert_calibrated()
-        relative_move: np.ndarray = np.dot(
+        return apply_2d_matrix_to_mapping(
-            np.array([y, x]), np.array(self.image_to_stage_displacement_matrix)
+            self.image_to_stage_displacement_matrix, x=x, y=y, to_int=True
         )
-        return {"x": int(relative_move[0]), "y": int(relative_move[1])}
 
     @lt.thing_action
     def convert_stage_to_image_coordinates(
-        self, x: float, y: float, **_kwargs: float
+        self, x: int, y: int, **_kwargs: int
-    ) -> Mapping[str, int]:
+    ) -> Mapping[str, float]:
         """Convert stage coordinates to image coordinates. Only x and y are returned."""
         self.assert_calibrated()
         inverse_matrix = np.linalg.inv(
             np.array(self.image_to_stage_displacement_matrix)
         )
-        relative_move = np.dot(np.array([x, y]), inverse_matrix)
+        return apply_2d_matrix_to_mapping(inverse_matrix, x=x, y=y, to_int=False)
-        return {"x": int(relative_move[1]), "y": int(relative_move[0])}
 
     @lt.thing_property
     def thing_state(self) -> Mapping[str, Any]:

src/openflexure_microscope_server/utilities.py

@@ -11,6 +11,7 @@ from typing import (
     overload,
     TypeAlias,
     Literal,
+    Mapping,
 )
 import os
 import re
@@ -22,6 +23,7 @@ import tomllib
 from functools import wraps
 import json
 
+import numpy as np
 from pydantic import BaseModel
 
 T = TypeVar("T")
@@ -475,3 +477,54 @@ def resolve_path_from_dir(path: str, directory: str) -> str:
     if not os.path.isabs(path):
         path = os.path.join(directory, path)
     return os.path.normpath(path)
+
+
+# Use overload to that the dict is either of integers or of float depending on the
+# to_int argument
+@overload
+def apply_2d_matrix_to_mapping(
+    matrix: np.ndarray,
+    *,
+    x: float | int,
+    y: float | int,
+    to_int: Literal[False],
+    **_kwargs: int,
+) -> Mapping[str, float]: ...
+@overload
+def apply_2d_matrix_to_mapping(
+    matrix: np.ndarray,
+    *,
+    x: float | int,
+    y: float | int,
+    to_int: Literal[True],
+    **_kwargs: int,
+) -> Mapping[str, int]: ...
+@overload
+def apply_2d_matrix_to_mapping(
+    matrix: np.ndarray, *, x: float | int, y: float | int, **_kwargs: int
+) -> Mapping[str, float]: ...
+
+
+def apply_2d_matrix_to_mapping(
+    matrix: np.ndarray,
+    *,
+    x: float | int,
+    y: float | int,
+    to_int: bool = False,
+    **_kwargs: int,
+) -> Mapping[str, int] | Mapping[str, float]:
+    """Perform a dot product with a (x,y) vector and a matrix. The result is a mapping.
+
+    This is designed x and y must be kwargs and extra kwargs are ignored allowing:
+    ``apply_2d_matrix_to_mapping(matrix, **position)`` to run for a mapping position.
+
+    :param matrix: The matrix to use in the calculation
+    :param x: the x coordinate (keyword only)
+    :param y: the y coordinate (keyword only)
+    :param to_int: Whether to convert the resulting coordinates to integers
+    :return: The resulting coordinates as a mapping.
+    """
+    relative_move: np.ndarray = np.dot(np.array([y, x]), np.array(matrix))
+    if to_int:
+        return {"x": round(relative_move[0]), "y": round(relative_move[1])}
+    return {"x": float(relative_move[0]), "y": float(relative_move[1])}

tests/test_stage_measure.py

@@ -12,6 +12,7 @@ import pytest
 import labthings_fastapi as lt
 
 from openflexure_microscope_server.things import stage_measure
+from openflexure_microscope_server.utilities import apply_2d_matrix_to_mapping
 
 LOGGER = logging.getLogger("mock-invocation_logger")
 
@@ -162,11 +163,15 @@ def rom_thing(example_rom_data) -> stage_measure.RangeofMotionThing:
 @pytest.fixture
 def mock_rom_deps(csm_matrix, mocker) -> stage_measure.RomDeps:
     """Return a RomDeps object full of mocks, except the logger which is LOGGER."""
+    inverse_matrix = np.linalg.inv(csm_matrix)
 
-    def apply_csm(x: float, y: float) -> dict[str, int]:
+    def apply_csm(x: float, y: float, **_kwargs: float) -> dict[str, int]:
         """Convert image coordinates to stage coordinates."""
-        vec = np.dot(np.array([y, x]), np.array(csm_matrix))
+        return apply_2d_matrix_to_mapping(csm_matrix, x=x, y=y, to_int=True)
-        return {"x": int(vec[0]), "y": int(vec[1])}
+
+    def un_apply_csm(x: float, y: float, **_kwargs: float) -> dict[str, float]:
+        """Convert stage coordinates to image coordinates."""
+        return apply_2d_matrix_to_mapping(inverse_matrix, x=x, y=y, to_int=True)
 
     mock_cam = mocker.Mock()
     mock_cam.image_is_sample.return_value = (True, "Mocked not measured.")
@@ -175,6 +180,7 @@ def mock_rom_deps(csm_matrix, mocker) -> stage_measure.RomDeps:
     # Set up mock csm to return a CSM matrix
     mock_csm.image_to_stage_displacement_matrix = csm_matrix
     mock_csm.convert_image_to_stage_coordinates.side_effect = apply_csm
+    mock_csm.convert_stage_to_image_coordinates.side_effect = un_apply_csm
 
     return stage_measure.RomDeps(
         autofocus=mocker.Mock(),
@@ -185,6 +191,53 @@ def mock_rom_deps(csm_matrix, mocker) -> stage_measure.RomDeps:
     )
 
 
+@pytest.mark.parametrize(
+    ("pos", "target_pos", "axis", "expected_dir"),
+    [
+        ({"x": 5000, "y": 0, "z": 0}, {"x": 0, "y": 0, "z": 0}, "x", 1),
+        ({"x": -5000, "y": 0, "z": 0}, {"x": 0, "y": 0, "z": 0}, "x", -1),
+        ({"x": 0, "y": 0, "z": 0}, {"x": 5000, "y": 0, "z": 0}, "x", -1),
+        # Y is flipped due to CSM sign
+        ({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", -1),
+    ],
+)
+def test_img_dir_from_stage_coords(
+    pos, target_pos, axis, expected_dir, rom_thing, mock_rom_deps
+):
+    """Check image direction is correctly generated."""
+    mock_rom_deps.stage.position = pos
+    direction = rom_thing._img_dir_from_stage_coords(target_pos, axis, mock_rom_deps)
+    assert direction == expected_dir
+
+
+def test_distance_in_img_percentage_err(rom_thing, mock_rom_deps):
+    """Check that _distance_in_img_percentage throws error if stream res is not set."""
+    rom_thing._stream_resolution = None
+    with pytest.raises(RuntimeError):
+        rom_thing._distance_in_img_percentage(
+            {"x": 0, "y": 0, "z": 0}, "x", mock_rom_deps
+        )
+
+
+@pytest.mark.parametrize(
+    ("pos", "target_pos", "axis", "expected_perc"),
+    [
+        ({"x": 5000, "y": 0, "z": 0}, {"x": 0, "y": 0, "z": 0}, "x", 346.625),
+        ({"x": -5000, "y": 0, "z": 0}, {"x": 0, "y": 0, "z": 0}, "x", -346.625),
+        ({"x": 0, "y": 0, "z": 0}, {"x": 5000, "y": 0, "z": 0}, "x", -346.625),
+        # Y is flipped due to CSM sign
+        ({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", -470.0),
+    ],
+)
+def test_distance_in_img_percentage(
+    pos, target_pos, axis, expected_perc, rom_thing, mock_rom_deps
+):
+    """Check _distance_in_img_percentage calculates expected values based on mock CSM."""
+    mock_rom_deps.stage.position = pos
+    img_perc = rom_thing._distance_in_img_percentage(target_pos, axis, mock_rom_deps)
+    assert expected_perc == img_perc
+
+
 def test_offset_from(rom_thing, mock_rom_deps, mocker):
     """Check the calls and returns for RangeofMotionThing._offset_from."""
     # Set up mock for the FFT displacement
@@ -551,13 +604,16 @@ def test_move_until_edge(rom_thing, mock_rom_deps, mocker):
     ]
 
 
-def test_perform_rom_test_locked(rom_thing, mock_rom_deps):
+def test_perform_rom_actions_locked(rom_thing, mock_rom_deps):
-    """Check the error if the Rom test is locked."""
+    """Check the error if running one of the calibration actions while locked."""
     # Not an RLock so no need to thread.
     rom_thing._lock.acquire()
     err_msg = "Trying to run ROM test when a test is already running."
     with pytest.raises(RuntimeError, match=err_msg):
         rom_thing.perform_rom_test(**dataclasses.asdict(mock_rom_deps))
+    err_msg = "Trying to run recentre when a test is already running."
+    with pytest.raises(RuntimeError, match=err_msg):
+        rom_thing.perform_recentre(**dataclasses.asdict(mock_rom_deps))
 
 
 def test_perform_rom_test_(rom_thing, mock_rom_deps, mocker):