diff --git a/ofm_config_full.json b/ofm_config_full.json index 60b85a66..2be08676 100644 --- a/ofm_config_full.json +++ b/ofm_config_full.json @@ -15,7 +15,8 @@ "kwargs": { "scans_folder": "/var/openflexure/scans/" } - } + }, + "/stage_measure/":"openflexure_microscope_server.things.stage_measure:RangeofMotionThing" }, "settings_folder": "/var/openflexure/settings/", "log_folder": "/var/openflexure/logs/" diff --git a/src/openflexure_microscope_server/things/camera_stage_mapping.py b/src/openflexure_microscope_server/things/camera_stage_mapping.py index a05a3385..57f3e742 100644 --- a/src/openflexure_microscope_server/things/camera_stage_mapping.py +++ b/src/openflexure_microscope_server/things/camera_stage_mapping.py @@ -268,10 +268,18 @@ class CameraStageMapper(lt.Thing): an image usually helps resolve any ambiguity. """ self.assert_calibrated() + stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y)) + + @lt.thing_action + def convert_image_to_stage_coordinates( + self, x: float, y: float + ) -> Mapping[str, int]: + """Convert image coordinates to stage coordinates.""" + self.assert_calibrated() relative_move: np.ndarray = np.dot( np.array([y, x]), np.array(self.image_to_stage_displacement_matrix) ) - stage.move_relative(x=relative_move[0], y=relative_move[1]) + return {"x": int(relative_move[0]), "y": int(relative_move[1])} @lt.thing_property def thing_state(self) -> Mapping[str, Any]: diff --git a/src/openflexure_microscope_server/things/stage_measure.py b/src/openflexure_microscope_server/things/stage_measure.py new file mode 100644 index 00000000..45175c2c --- /dev/null +++ b/src/openflexure_microscope_server/things/stage_measure.py @@ -0,0 +1,595 @@ +"""File contains all the functions used to measure the range of motion. + +The range of motion is measured by first taking 5 'medium' sized steps which gives +enough positions to predict future z positions. Next, one 'big' step is taken followed +by 3 'small' steps to test that the stage is still moving as expected and has not +reached the edge. Once the edge has been found and to account for the possibility that +the edge was reached during a "big" step, the stage is moved in a sequence of steps +in the opposite direction until motion is detected. This is position is taken as the +true final position. + +Throughout the test, parasitic motion (motion in the axis not being measured) +is tracked and an error is raised if it exceeds a minimum amount. Currently +this is 10% of the expected motion is the measured axis. +""" + +from typing import Literal, Any, Optional, overload +import time +from dataclasses import dataclass +from threading import Lock + +from scipy.optimize import curve_fit +import numpy as np + +from camera_stage_mapping import fft_image_tracking +import labthings_fastapi as lt + +from openflexure_microscope_server.utilities import quadratic + +# Things +from .autofocus import AutofocusThing +from .camera_stage_mapping import CameraStageMapper +from .camera import CameraDependency as CamDep +from .stage import StageDependency as StageDep + +CSMDep = lt.deps.direct_thing_client_dependency( + CameraStageMapper, "/camera_stage_mapping/" +) +AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/") + +## Size of movement in percentage of field of view +SMALL_STEP = 20 +MEDIUM_STEP = 50 +BIG_STEP = 200 + +PARASITIC_MOTION_TOL = 0.1 +DETECT_MOTION_TOL = 0.65 + + +class RomDataTracker: + """Class for tracking range of motion data. + + The attributes storing data are: + + * ``stage_coords`` A list of the the stage coordinates at each measurement location + * ``displacements`` A list of the displacement calculated after each movement + """ + + def __init__(self) -> None: + """Define useful data tracked throughout test.""" + self.stage_coords: list[dict[str, int]] = [] + self.offsets: list[dict[str, float]] = [] + + def record_movement( + self, current_pos: dict[str, int], offset: dict[str, float] + ) -> None: + """Record the current position and the measured offset of the last move.""" + self.stage_coords.append(current_pos) + self.offsets.append(offset) + + @property + def final_position(self) -> dict[str, int]: + """The last stage coordinate recorded.""" + return self.stage_coords[-1].copy() + + def predict_z_displacement( + self, + axis: Literal["x", "y"], + stage_movement: dict[str, int], + stage_position: dict[str, int], + ) -> int: + """Predict the z-displacement needed for a given movement. + + :param axis: The axis which is being measured. This must be 'x' or 'y'. + :param stage_movement: The movement to be performed in stage coordinates. + :param stage_position: The current stage position in stage coordinates. + :return: The predicted relative z displacement needed to stay in focus. + """ + # x or y positions + lateral_positions = [i[axis] for i in self.stage_coords] + z_positions = [i["z"] for i in self.stage_coords] + fit_params, *_others = curve_fit(quadratic, lateral_positions, z_positions) + z_dest = quadratic(stage_position[axis] + stage_movement[axis], *fit_params) + + return int(z_dest - stage_position["z"]) + + +@dataclass +class RomDeps: + """Grouped dependencies for the Range of motion Thing. + + These are used to pass the dependencies from actions to other sub-functions. + """ + + autofocus: AutofocusDep + stage: StageDep + cam: CamDep + csm: CSMDep + logger: lt.deps.InvocationLogger + + +class ParasiticMotionError(Exception): + """Custom exception raised when parasitic motion is detected. + + Parasitic motion is when motion in the direction not being measured + is too high. + """ + + +class RangeofMotionThing(lt.Thing): + """A class used to measure the range of motion of the stage in X and Y.""" + + calibrated_range = lt.ThingSetting( + initial_value=None, model=Optional[list[int, int]], readonly=True + ) + + def __init__(self) -> None: + """Initialise and create the lock.""" + super().__init__() + self._lock = Lock() + self._stream_resolution: Optional[tuple[int, int]] = None + self._rom_data = RomDataTracker() + + @lt.thing_action + def perform_rom_test( + self, + autofocus: AutofocusDep, + stage: StageDep, + cam: CamDep, + csm: CSMDep, + logger: lt.deps.InvocationLogger, + ) -> dict[str, Any]: + """Measures the range of motion of the stage across the x and y axes. + + :param autofocus: A raw_thing_client dependency for autofocus. + :param stage: A raw_thing_client depeendency for the microscope stage. + :param cam: A raw_thing_client depeendency for the camera. + :param csm: A raw_thing_client depeendency for camera stage mapping. + :param logger: A raw_thing_client depeendency for the logger. + :return: Results dictionary separated into keys of each axis and direction. + """ + got_lock = self._lock.acquire(blocking=False) + if not got_lock: + raise RuntimeError("Trying to run ROM test when a test is already running.") + + try: + rom_deps = RomDeps( + autofocus=autofocus, stage=stage, csm=csm, cam=cam, logger=logger + ) + logger.info( + "Using the stage to measure the Range of Motion. " + "Please ensure you are using a sample that covers the whole range of " + "motion. This should be approximately 12 x 12 mm." + ) + start_time = time.time() + + self._set_stream_resolution(cam) + + # The total range for the two axes under test + step_range = [] + + # Strictly typed definitions to iterate over for MyPys sake. + axes: tuple[Literal["x"], Literal["y"]] = ("x", "y") + directions: tuple[Literal[1], Literal[-1]] = (1, -1) + + for axis in axes: + # The final position of the axis under test in the given direction + axis_limits = [] + for axis_dir in directions: + # Create a new tracker at start of measurement. + self._rom_data = RomDataTracker() + self._move_until_edge( + axis=axis, direction=axis_dir, rom_deps=rom_deps + ) + # Append final position from tracker + axis_limits.append(self._rom_data.final_position[axis]) + # Calculate step range. + step_range.append(abs(axis_limits[0] - axis_limits[1])) + + total_time = time.time() - start_time + logger.info(f"Range of motion is {step_range[0]} x {step_range[1]} steps") + self.calibrated_range = step_range + finally: + self._lock.release() + return { + "Time": total_time, + "CSM Matrix": csm.image_to_stage_displacement_matrix, + "Step Range": step_range, + } + + def _set_stream_resolution(self, cam: CamDep) -> None: + """Set the self._stream_resolution attribute by reading camera. + + :param cam: The camera dependency. + """ + stream_shape = cam.grab_as_array().shape + # Swap axes as numpy is [y, x] + self._stream_resolution = (stream_shape[1], stream_shape[0]) + + def _move_until_edge( + self, + axis: Literal["x", "y"], + direction: Literal[1, -1], + rom_deps: RomDeps, + ) -> None: + """Move in one direction until movement per step decreases significantly. + + This should move until the edge of the stage. Once the edge is reached there + will be some movement as there is no hard stop, but it will reduce + significantly. + + :param rom_deps: All dependencies that were passed to the calling Action + :param axis: The axis which is being measured. This must be 'x' or 'y'. + :param direction: The direction which is being measured. This must be 1 or -1. + :return: Results dictionary containing stage positions, + correlations and the final position. + """ + # Start by performing an autofocus and recording starting position + rom_deps.autofocus.looping_autofocus(dz=1000) + starting_position = rom_deps.stage.position + self._rom_data.stage_coords.append(starting_position) + + try: + dir_str = "positive" if direction == 1 else "negative" + rom_deps.logger.info( + f"Beginning the {axis}-axis in the {dir_str} direction" + ) + + rom_deps.logger.info("Moving the stage in 5 medium sized steps.") + self._initial_moves_for_z_prediction( + axis=axis, + direction=direction, + rom_deps=rom_deps, + ) + + still_moving = True + # Loop taking 1 big step followed by 3 small steps to check if the + # stage is moving as expected or has reached end of range of motion. + # Stop once end of range of motion is detected. + while still_moving: + self._big_z_corrected_movement( + axis=axis, direction=direction, rom_deps=rom_deps + ) + # Autofocus and record position + rom_deps.autofocus.looping_autofocus(dz=800) + self._rom_data.stage_coords.append(rom_deps.stage.position) + + # Perform small moves to check stage is still moving. + still_moving = self._stage_still_moves( + axis=axis, + direction=direction, + rom_deps=rom_deps, + ) + + # Stage may have crashed during a large move. Move stage back until motion + # is detected + rom_deps.logger.info("Moving stage back until motion is detect") + self._move_back_until_motion_detected( + axis=axis, + direction=direction, + rom_deps=rom_deps, + ) + + # Replace final position. + self._rom_data.stage_coords[-1] = rom_deps.stage.position + + finally: + rom_deps.stage.move_absolute(**starting_position, block_cancellation=True) + + def _img_percentage_to_img_coords( + self, fov_perc: int, axis: Literal["x", "y"] + ) -> float: + """For a given image percentage and axis return the distance in img coords. + + :param fov_perc: The percentage of field of view the stage should move by. + :param axis: The resolution of the stream from the camera. + :return: Distance in image coordinates (pixels) + """ + if self._stream_resolution is None: + raise RuntimeError( + "Stream resolution must be set before generating movement in coords" + ) + + img_index = 0 if axis == "x" else 1 + return (fov_perc / 100) * self._stream_resolution[img_index] + + def _movement_in_img_coords( + self, + fov_perc: int, + axis: Literal["x", "y"], + direction: Literal[1, -1], + ) -> dict[str, float]: + """Return the dictionary for a move in image coordinates. + + This dictionary can be passed directly to csm.move_in_image_coordinates + + :param fov_perc: The percentage of field of view the stage should move by. + :param axis: The resolution of the stream from the camera. + :param direction: The direction the stage moves. + + :return: The movement size in image coordinates + """ + distance = self._img_percentage_to_img_coords(fov_perc=fov_perc, axis=axis) + if axis == "x": + return {"x": distance * direction, "y": 0} + return {"x": 0, "y": distance * direction} + + def _initial_moves_for_z_prediction( + self, + axis: Literal["x", "y"], + direction: Literal[1, -1], + rom_deps: RomDeps, + ) -> None: + """Perform 5 medium sized moves with autofocus for z feed-forward. + + z-feed forward allows prediction of the z-position as the stage moves. For the + feed forward calculation to work an initial number of measurements must be + taken. This method performs these initial measurements. + + :param direction: The direction the stage moves. + :param axis: The axis which is being measured. This must be 'x' or 'y'. + :param rom_deps: All dependencies that were passed to the calling Action + """ + movement = self._movement_in_img_coords( + fov_perc=MEDIUM_STEP, axis=axis, direction=direction + ) + for _loop in range(5): + offset = self._move_and_measure(movement=movement, rom_deps=rom_deps) + rom_deps.logger.info(f"Offset measured as {offset[axis]}") + + self._rom_data.record_movement(rom_deps.stage.position, offset) + if _parasitic_motion_detected(movement, offset): + raise ParasiticMotionError( + "Parasitic motion detected during initial images to calculate " + "z-curvature. This may indicate you have started at the end of the " + "range of travel, or that camera stage mapping is poorly " + "calibrated." + ) + + def _big_z_corrected_movement( + self, axis: Literal["x", "y"], direction: Literal[1, -1], rom_deps: RomDeps + ) -> None: + """Take one big move with feed-forward z-correction. + + The size is defined by the BIG_STEP constant. + + :param axis: The axis to move in. + :param direction: The direction to move in. + :param rom_deps: All dependencies that were passed to the calling Action + """ + movement = self._movement_in_img_coords( + fov_perc=BIG_STEP, axis=axis, direction=direction + ) + # Convert to stage coordinates + stage_movemenet = rom_deps.csm.convert_image_to_stage_coordinates(**movement) + + z_disp = self._rom_data.predict_z_displacement( + axis=axis, + stage_movement=stage_movemenet, + stage_position=rom_deps.stage.position, + ) + rom_deps.stage.move_relative(z=z_disp) + rom_deps.csm.move_in_image_coordinates(**movement) + + def _stage_still_moves( + self, + axis: Literal["x", "y"], + direction: Literal[1, -1], + rom_deps: RomDeps, + ) -> bool: + """Carry out 3 small moves in a given direction and axis. + + An image is taken before and after each move to check the stage has moved as + far as it should. If the offset (calculated by cross correlation) is less than + expected, 3 attempts are made to refocus the image to ensure that image quality + is not causing the offset to mistakenly be reported as a low value. If the + calculated offset is still too low then this is taken as an indication that the + edge has been found. + + :param axis: The axis which is being measured. This must be 'x' or 'y'. + :param direction: The direction the stage moves. + :param rom_deps: All dependencies that were passed to the calling Action + """ + movement = self._movement_in_img_coords( + fov_perc=SMALL_STEP, axis=axis, direction=direction + ) + abs_min_offset = abs(movement[axis] * DETECT_MOTION_TOL) + + for _loop in range(3): + offset = self._move_and_measure( + movement=movement, + rom_deps=rom_deps, + # Don't autofocus initially + perform_autofocus=False, + # But retry focus 3 times if detected motion is too small or parasitic + max_autofocus_repeats=3, + abs_min_offset=abs_min_offset, + ) + parasitic_motion = _parasitic_motion_detected(movement, offset) + rom_deps.logger.info(f"Offset measured as {offset[axis]}") + self._rom_data.record_movement(rom_deps.stage.position, offset) + + if abs(offset[axis]) < abs_min_offset or parasitic_motion: + # If the offset is below the abs min offset, or significant + # parasitic motion is detected then the edge has been found. + rom_deps.logger.info("Edge has been found.") + return False + # If we reached here the stage is still moving fine + return True + + def _move_back_until_motion_detected( + self, + axis: Literal["x", "y"], + direction: Literal[1, -1], + rom_deps: RomDeps, + ) -> None: + """Move the stage against the direction of test unil motion is detected. + + In the case that the stage has reached the end of the motion, this method moves + the motor in the opposite direction until reliable motion is detected. + + :param axis: The axis in which the stage is moving. This must be 'x' or 'y'. + :param direction: The direction in which the stage was moving during the test, + this method will move in the opposite direction. + :param rom_deps: All dependencies that were passed to the calling Action + """ + movement = self._movement_in_img_coords( + fov_perc=SMALL_STEP, axis=axis, direction=-direction + ) + + max_moves = int(1.5 * BIG_STEP / SMALL_STEP) + # minimum number of pixels for motion to be detected as reliable + motion_minimum = abs(movement[axis] * DETECT_MOTION_TOL) + + for i in range(max_moves): + # Increment movement + rom_deps.logger.info(f"Testing with step size {movement[axis]}") + + # Run an autofocus every 3 steps + run_autofocus = i % 3 == 2 + offset = self._move_and_measure( + movement=movement, rom_deps=rom_deps, perform_autofocus=run_autofocus + ) + + rom_deps.logger.info(f"Offset measured as {abs(offset[axis])}") + if abs(offset[axis]) > motion_minimum: + rom_deps.logger.info("Motion detected.") + return + # If this point is reached then motion is never detected + raise RuntimeError("Cannot detect motion again after reaching end of range.") + + def _move_and_measure( + self, + movement: dict[str, float], + rom_deps: RomDeps, + perform_autofocus: bool = True, + max_autofocus_repeats: int = 0, + abs_min_offset: float = 0.0, + ) -> dict[str, float]: + """Move the stage and measure the offset between the two positions. + + :param movement: A dictionary containing the distance to move in image coords. + :param rom_deps: All dependencies that were passed to the calling Action + :param perform_autofocus: Set to False to disable autofocus after move. + Default is True + :param max_autofocus_repeats: The number of times to repeat the focus if the + detected (on-axis) offset is below ``abs_min_offset``. This will only work + if ``abs_min_offset`` is also set + :param abs_min_offset: The absolute minimum (on-axis) offset, under which the + autofocus is repeated. + :return: The calculated offset from cross correlation. + """ + if max_autofocus_repeats > 0 and abs_min_offset <= 0: + raise ValueError( + "abs_min_offset must be positive if max_autofocus_repeats > 0." + ) + + # Take image before move + before_img = rom_deps.cam.grab_as_array() + # Move and autofocus if required + rom_deps.csm.move_in_image_coordinates(**movement) + if perform_autofocus: + rom_deps.autofocus.looping_autofocus(dz=800) + # Take image and calculate offset + offset = self._offset_from(before_img, rom_deps=rom_deps) + + if max_autofocus_repeats > 0: + axis = _axis_from_movement_dict(movement) + af_repeats = 0 + parasitic_motion = _parasitic_motion_detected(movement, offset) + while abs(offset[axis]) < abs_min_offset or parasitic_motion: + af_repeats += 1 + rom_deps.logger.info( + f"Motion not detected. Refocusing to check. Attempt {af_repeats}/3." + ) + rom_deps.autofocus.looping_autofocus(dz=800) + # Re-take image and calculate offset + offset = self._offset_from(before_img, rom_deps=rom_deps) + parasitic_motion = _parasitic_motion_detected(movement, offset) + if af_repeats >= max_autofocus_repeats: + # break if the maximum number of tries are exceeded. + break + + return offset + + def _offset_from( + self, before_img: np.ndarray, rom_deps: RomDeps + ) -> dict[str, float]: + """Take an image and calculate the offset from an input image. + + :param before_img: The image the offset should be calculated with respect to. + :param rom_deps: All dependencies that were passed to the calling Action + :return: The calculated offset as a dictionary in pixels + """ + is_sample, _bg_message = rom_deps.cam.image_is_sample() + if not is_sample: + raise RuntimeError( + "No sample detected. Sample must be densely featured and cover the " + "whole range of motion." + ) + after_img = rom_deps.cam.grab_as_array() + offset = fft_image_tracking.displacement_between_images( + image_0=before_img, + image_1=after_img, + sigma=10, + fractional_threshold=0.1, + pad=True, + ) + return {"x": offset[1], "y": offset[0]} + + +# The number of return arguments depends on if return_other is True or False +# let MyPy know with overloads typed to Literal[False] and Literal[True]. +@overload +def _axis_from_movement_dict( + movement: dict[str, float], return_other: Literal[False] +) -> str: ... +@overload +def _axis_from_movement_dict( + movement: dict[str, float], return_other: Literal[True] +) -> tuple[str, str]: ... + + +# Overload the case where return_other is not specified, this is needed +# because MyPy doesn't see that return other's default is `False` and use +# the `Literal[False]` overload. +@overload +def _axis_from_movement_dict(movement: dict[str, float]) -> str: ... + + +# Finally The function. +def _axis_from_movement_dict( + movement: dict[str, float], return_other: bool = False +) -> str | tuple[str, str]: + """Return the axis that a given movement dictionary moves in. + + For example: ``_axis_from_movement_dict({"x": 10, "y":0})`` will return ``x``. + + :param movement: The movement dictionary. + :param return_other: If True return a tuple with the first value being the movement + axis and the second being the other axis. Default is False + :return: The movement axis (and optionally the other axis) as strings. + """ + # Not exclusive or, this errors if both axes are zero or neither axes are zero. + if not ((movement["x"] == 0) ^ (movement["y"] == 0)): + raise ValueError("Either x or y movement should be zero, but not both.") + + if return_other: + return ("y", "x") if movement["x"] == 0 else ("x", "y") + return "y" if movement["x"] == 0 else "x" + + +def _parasitic_motion_detected( + movement: dict[str, float], offset: dict[str, float] +) -> bool: + """Compare desired movement to measured offset, report if parasitic motion was detected. + + :param movement: The movement dictionary in image coordinates. + :param offset: The offset calculated from correlation. + :return: True if too much parasitic motion is detects. False if movement is as + expected. + """ + movement_axis, other_axis = _axis_from_movement_dict(movement, return_other=True) + parasitic_motion = abs(offset[other_axis]) + max_parasitic_motion = abs(movement[movement_axis] * PARASITIC_MOTION_TOL) + return parasitic_motion > max_parasitic_motion diff --git a/src/openflexure_microscope_server/utilities.py b/src/openflexure_microscope_server/utilities.py index d9debd1e..b7377f9a 100644 --- a/src/openflexure_microscope_server/utilities.py +++ b/src/openflexure_microscope_server/utilities.py @@ -1,6 +1,15 @@ """Utility functions and classes.""" -from typing import TypeVar, Callable, ParamSpec, Optional, Any, Concatenate, Self +from typing import ( + TypeVar, + Callable, + ParamSpec, + Optional, + Any, + Concatenate, + Self, + overload, +) import os import re import sys @@ -11,6 +20,7 @@ import tomllib from functools import wraps from pydantic import BaseModel +import numpy as np T = TypeVar("T") P = ParamSpec("P") @@ -327,3 +337,25 @@ def _get_version_from_toml(toml_path: str) -> str: except (IOError, ValueError, KeyError): LOGGER.error("Problem opening pyproject.toml") return "Undefined" + + +# Let MyPy know that the output type matches x. +@overload +def quadratic(x: np.ndarray, a: float, b: float, c: float) -> np.ndarray: ... +@overload +def quadratic(x: float, a: float, b: float, c: float) -> float: ... + + +def quadratic( + x: float | np.ndarray, a: float, b: float, c: float +) -> float | np.ndarray: + """Quadratic function. Used for predicting z. + + :param x: The point or points at which to evaluate the quadratic. This can be a + float or a numpy array. The return will be the same type. + :param a: The coefficient of x^2. + :param b: The coefficient of x. + :param c: The constant coefficient. + :return: The quadratic, evaluated at each point in ``x`` + """ + return a * x**2 + b * x + c diff --git a/tests/mock_things/mock_csm.py b/tests/mock_things/mock_csm.py index 20178a8b..4daa0d57 100644 --- a/tests/mock_things/mock_csm.py +++ b/tests/mock_things/mock_csm.py @@ -17,3 +17,7 @@ class MockCSMThing: """ image_resolution = (123, 456) + image_to_stage_displacement_matrix = [ + [0.03061156624485296, 1.8031242270940833], + [1.773236372778601, 0.006660431608601435], + ] diff --git a/tests/mock_things/mock_stage.py b/tests/mock_things/mock_stage.py index 9a36d044..6733986a 100644 --- a/tests/mock_things/mock_stage.py +++ b/tests/mock_things/mock_stage.py @@ -16,4 +16,4 @@ class MockStageThing: is not artificially inflated. """ - position = (111, 222, 333) + position = {"x": 3635, "y": 10, "z": 617} diff --git a/tests/test_stage_measure.py b/tests/test_stage_measure.py new file mode 100644 index 00000000..9f4e3a4e --- /dev/null +++ b/tests/test_stage_measure.py @@ -0,0 +1,600 @@ +"""File contains unit tests for stage_measure.""" + +from copy import copy +import logging +import dataclasses +import tempfile + +from fastapi.testclient import TestClient +import numpy as np +import pytest + +import labthings_fastapi as lt + +from openflexure_microscope_server.things import stage_measure + +LOGGER = logging.getLogger("mock-invocation_logger") + + +# Useful generators +def increasing_xy_dict_generator(*_args, **_kwargs): + """Generate x-y dictionaries of incrementing sizes. + + These don't simulate expected effects, but allow checking sequential reads of a + function/property were used. + """ + i = 0 + while True: + yield {"x": i, "y": i} + i += 1 + + +def increasing_xyz_dict_generator(*_args, **_kwargs): + """Generate x-y-z dictionaries of incrementing sizes. + + These don't simulate expected effects, but allow checking sequential reads of a + function/property were used. + """ + i = 0 + while True: + yield {"x": i, "y": i, "z": i} + i += 1 + + +@pytest.fixture +def csm_matrix(): + """Return an example CSM matrix.""" + return [ + [0.03061156624485296, -1.8031242270940833], + [1.773236372778601, 0.006660431608601435], + ] + + +@pytest.fixture +def example_rom_data(): + """Return some example data in a RomDataTracker.""" + mock_positions = [ + {"x": 0, "y": 0, "z": 42}, + {"x": 727, "y": 2, "z": 154}, + {"x": 1454, "y": 4, "z": 228}, + {"x": 2181, "y": 6, "z": 351}, + {"x": 2908, "y": 8, "z": 509}, + {"x": 3635, "y": 10, "z": 617}, + ] + rom_data = stage_measure.RomDataTracker() + + # loop through mock positions recording them with an offset. + for position in mock_positions: + offset = {"x": 54.4, "y": 0} + rom_data.record_movement(position, offset) + return rom_data + + +def test_predict_z(example_rom_data): + """Check that the prediction for the next z position is correct.""" + mock_z_diff = example_rom_data.predict_z_displacement( + axis="x", + stage_movement={"x": 5243, "y": 0}, + stage_position={"x": 3635, "y": 10, "z": 617}, + ) + expected_z_diff = 1343 + assert mock_z_diff == expected_z_diff + + +@pytest.mark.parametrize( + ("movement", "axis", "other_axis"), + [ + ({"x": 2908, "y": 0}, "x", "y"), + ({"x": -29, "y": 0}, "x", "y"), + ({"x": 0, "y": 123}, "y", "x"), + ({"x": 0, "y": -456}, "y", "x"), + ], +) +def test_axis_from_movement_dict(movement, axis, other_axis): + """Test that _axis_from_movement_dict identifies the correct axes.""" + assert stage_measure._axis_from_movement_dict(movement) == axis + + ret_axes = stage_measure._axis_from_movement_dict(movement, return_other=True) + assert ret_axes == (axis, other_axis) + + +@pytest.mark.parametrize("movement", [{"x": 0, "y": 0}, {"x": 10, "y": 10}]) +def test_error_on_axis_from_movement_dict(movement): + """Check _axis_from_movement_dict errors if both axes are zero, or both are non-zero.""" + with pytest.raises(ValueError, match="Either x or y movement should be zero"): + assert stage_measure._axis_from_movement_dict(movement) + + +@pytest.mark.parametrize( + ("par_fraction", "too_high"), + [ + (-0.20, True), + (-0.11, True), + (-0.09, False), + (-0.05, False), + (0.00, False), + (0.05, False), + (0.09, False), + (0.11, True), + (0.20, True), + ], +) +def test_parasitic_detect(par_fraction, too_high): + """Check parasitic motion is detected if the fraction of parasitic motion is too high.""" + movement = {"x": 2908, "y": 0} + + offset = copy(movement) + offset["y"] = movement["x"] * par_fraction + + detected = stage_measure._parasitic_motion_detected( + movement=movement, offset=offset + ) + assert detected == too_high + + +def test_error_if_no_stream_res_set_when_requesting_img_coords(): + """Check a RuntimeError thrown when requesting image coordinates if resolution unset.""" + with pytest.raises(RuntimeError, match="Stream resolution must be set"): + stage_measure.RangeofMotionThing()._img_percentage_to_img_coords(20, "x") + + +@pytest.fixture +def rom_thing(example_rom_data) -> stage_measure.RangeofMotionThing: + """Yield a RangeofMotionThing already populated with some example rom_data.""" + rom_thing = stage_measure.RangeofMotionThing() + rom_thing._stream_resolution = [800, 600] + rom_thing._rom_data = example_rom_data + + with tempfile.TemporaryDirectory() as tmpdir: + server = lt.ThingServer(settings_folder=tmpdir) + server.add_thing(rom_thing, "/rom_thing/") + with TestClient(server.app): + yield rom_thing + + +@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.""" + + def apply_csm(x: float, y: float) -> dict[str, int]: + """Convert image coordinates to stage coordinates.""" + vec = np.dot(np.array([y, x]), np.array(csm_matrix)) + return {"x": int(vec[0]), "y": int(vec[1])} + + mock_cam = mocker.Mock() + mock_cam.image_is_sample.return_value = (True, "Mocked not measured.") + + mock_csm = mocker.Mock() + # 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 + + return stage_measure.RomDeps( + autofocus=mocker.Mock(), + stage=mocker.Mock(), + cam=mock_cam, + csm=mock_csm, + logger=LOGGER, + ) + + +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 + disp_between_route = ( + "openflexure_microscope_server.things.stage_measure." + "fft_image_tracking.displacement_between_images" + ) + mock_disp_between = mocker.patch( + disp_between_route, + return_value=[123, 456], + ) + + # Run it + offset = rom_thing._offset_from(before_img="MOCK_IMAGE", rom_deps=mock_rom_deps) + + # Check the offset is a dictionary with the correct values for the axes + assert offset["x"] == 456 + assert offset["y"] == 123 + # Check 1 image was taken + assert mock_rom_deps.cam.grab_as_array.call_count == 1 + mock_after_image = mock_rom_deps.cam.grab_as_array.return_value + # Check the FFT displacement was called once + assert mock_disp_between.call_count == 1 + displacement_kwargs = mock_disp_between.call_args.kwargs + # Check the image inputs + assert displacement_kwargs["image_0"] == "MOCK_IMAGE" + assert displacement_kwargs["image_1"] == mock_after_image + + +@pytest.mark.parametrize("perform_autofocus", [True, False]) +def test_move_and_measure(perform_autofocus, rom_thing, mock_rom_deps, mocker): + """Test _move_and_measure with and without initial autofocus checking call counts. + + This doesn't test with the repeate autofocus if motion isn't detected + """ + mock_offset_value = {"x": 100, "y": 3} + mocker.patch.object(rom_thing, "_offset_from", return_value=mock_offset_value) + movement = {"x": 100, "y": 0} + offset = rom_thing._move_and_measure( + movement=movement, rom_deps=mock_rom_deps, perform_autofocus=perform_autofocus + ) + + # Check exactly 1 move + assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 + # The kwargs of the call should movement dict + assert mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs == movement + # Check autofocus call count + expected_af_count = 1 if perform_autofocus else 0 + assert mock_rom_deps.autofocus.looping_autofocus.call_count == expected_af_count + # And check final return + assert offset == mock_offset_value + + +@pytest.mark.parametrize( + ("x_offsets", "n_offset_measures", "expected_return"), + [ + ([5.1, 0.2], 1, {"x": 5.1, "y": 0}), + ([-5.1, 0.2], 1, {"x": -5.1, "y": 0}), + ([0.1, 5.2, 0.3, 0.4, 0.5], 2, {"x": 5.2, "y": 0}), + ([0.1, 0.2, 5.3, 0.4, 0.5], 3, {"x": 5.3, "y": 0}), + ([0.1, 0.2, 0.3, 5.4, 0.5], 4, {"x": 5.4, "y": 0}), + # n_offset_measures shouldn't go higher than 4 as max autofocus repeats is 3 + ([0.1, 0.2, 0.3, 0.4, 5.5], 4, {"x": 0.4, "y": 0}), + ], +) +def test_move_and_measure_with_refocus( + x_offsets, n_offset_measures, expected_return, rom_thing, mock_rom_deps, mocker +): + """Test _move_and_measure with final refocus if offset is too small.""" + return_dicts = tuple({"x": x, "y": 0} for x in x_offsets) + offset_from_mock = mocker.patch.object( + rom_thing, "_offset_from", side_effect=return_dicts + ) + movement = {"x": 10, "y": 0} + offset = rom_thing._move_and_measure( + movement=movement, + rom_deps=mock_rom_deps, + perform_autofocus=False, + max_autofocus_repeats=3, + abs_min_offset=5, + ) + # Check exactly 1 move + assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 + # Check expected _offset_from calls + assert offset_from_mock.call_count == n_offset_measures + # The kwargs of the call should movement dict + assert mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs == movement + # Check autofocus call count is 1 less than number of offset measures as no autofocus + # is performed before the first one + expected_af_count = n_offset_measures - 1 + assert mock_rom_deps.autofocus.looping_autofocus.call_count == expected_af_count + # And check final return + assert offset == expected_return + + +def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker): + """Check error if abs_min_offset is not positive when using it to determine if to autofocus.""" + offset_from_mock = mocker.patch.object( + rom_thing, "_offset_from", return_value={"x": 0, "y": 0} + ) + + # First check with abs_min_offset not set. The default should be zero. + with pytest.raises(ValueError, match="abs_min_offset must be positive"): + rom_thing._move_and_measure( + movement={"x": 10, "y": 0}, + rom_deps=mock_rom_deps, + perform_autofocus=False, + max_autofocus_repeats=3, + ) + # Then check with abs_min_offset negative, this might happen if the the expected + # move calculation is not made absolute. + with pytest.raises(ValueError, match="abs_min_offset must be positive"): + rom_thing._move_and_measure( + movement={"x": 10, "y": 0}, + rom_deps=mock_rom_deps, + perform_autofocus=False, + max_autofocus_repeats=3, + abs_min_offset=-123.456, + ) + # Should have never measured and offset. Just error straight away. + assert offset_from_mock.call_count == 0 + + +def test_move_back_until_motion_detected(rom_thing, mock_rom_deps, mocker): + """Check that _move_back_until_motion_detected is making increasing negative moves. + + The moves for this method should be in opposite direction to the direction + specified as this is moving back after the stage reaches end of its movement. + """ + mock_move_n_meas = mocker.patch.object( + rom_thing, "_move_and_measure", return_value={"x": 0, "y": 0} + ) + + with pytest.raises(RuntimeError, match="Cannot detect motion again"): + rom_thing._move_back_until_motion_detected("y", -1, rom_deps=mock_rom_deps) + + max_tries = int(1.5 * stage_measure.BIG_STEP / stage_measure.SMALL_STEP) + expected_step = 800 * stage_measure.SMALL_STEP / 100 + assert mock_move_n_meas.call_count == max_tries + for _i, call_args in enumerate(mock_move_n_meas.call_args_list): + call_args.kwargs["movement"] = {"x": 0, "y": expected_step} + call_args.kwargs["perform_autofocus"] = False + + ## Reset mock and change the side effect + mock_move_n_meas.reset_mock() + mock_move_n_meas.side_effect = ( + {"x": 0, "y": 0}, + {"x": 0, "y": 0}, + {"x": expected_step * 0.8, "y": 0}, + ) + + # Other axis and direction this time + rom_thing._move_back_until_motion_detected("x", 1, rom_deps=mock_rom_deps) + + # Should only be called 3 times + assert mock_move_n_meas.call_count == 3 + for _i, call_args in enumerate(mock_move_n_meas.call_args_list): + call_args.kwargs["movement"] = {"x": -expected_step, "y": 0} + call_args.kwargs["perform_autofocus"] = False + + +@pytest.mark.parametrize( + ("good_moves", "expected_to_detect_motion", "offset_calls"), + [ + ([0, 1, 2], True, 3), # First 3 pass all good + ([1, 2, 3], True, 4), # First is bad, will refocus, still complete + ([2, 3, 4], True, 5), # First 2 are bad, will refocus twice, still complete + ([3, 4, 5], True, 6), # First 3 are bad, will refocus 3 times, still complete + ([4, 5, 6], False, 4), # First 4 are bad, fails + # Check second and 3rd measurement can fail after first fails 3 times + ([3, 5, 7], True, 8), + ([3, 6, 9], True, 10), + ([3, 7, 11], True, 12), + # But they can't fail 4 times + ([3, 8, 9], False, 8), + ([3, 7, 12], False, 12), + ], +) +def test_stage_still_moves( + good_moves, + expected_to_detect_motion, + offset_calls, + rom_thing, + mock_rom_deps, + mocker, +): + """Test _stage_still_moves correctly detects stage movement.""" + min_offset = 800 * stage_measure.SMALL_STEP / 100 * stage_measure.DETECT_MOTION_TOL + + def gen_offsets(*_args, **_kwargs): + """Generate offset dictionaries with small moves unless count matches ``good_moves``.""" + i = 0 + while True: + x = min_offset * 1.2 if i in good_moves else 0.1 + yield {"x": x, "y": 0} + i += 1 + + mock_offset_from = mocker.patch.object( + rom_thing, "_offset_from", side_effect=gen_offsets() + ) + + still_moves = rom_thing._stage_still_moves( + axis="x", + direction=1, + rom_deps=mock_rom_deps, + ) + assert still_moves is expected_to_detect_motion + assert mock_offset_from.call_count == offset_calls + + +def test_big_z_corrected_movement(rom_thing, mock_rom_deps): + """Check big z corrected move moves in x/y and z the expected distances.""" + mock_rom_deps.stage.position = {"x": 5000, "y": 30, "z": 500} + + rom_thing._big_z_corrected_movement("x", direction=1, rom_deps=mock_rom_deps) + + expected_movement = {"x": 800 * stage_measure.BIG_STEP / 100, "y": 0} + + # Check there is one z move in steps + assert mock_rom_deps.stage.move_relative.call_count == 1 + move_kwargs = mock_rom_deps.stage.move_relative.call_args.kwargs + assert "x" not in move_kwargs + assert "y" not in move_kwargs + assert "z" in move_kwargs + assert move_kwargs["z"] == 1148 + + # And one move in image coordinates + assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 + lat_mov_kwargs = mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs + assert lat_mov_kwargs == expected_movement + + +def test_initial_moves_for_z_prediction(rom_thing, mock_rom_deps, mocker): + """Check the initial moves are of the correct size and are recorded.""" + # Mock the _offset_from and stage.position to return generated dictionaries that + # increment each time they are called. (All values 0 the first time, all values 1 + # the second time ...) + mocker.patch.object( + rom_thing, "_offset_from", side_effect=increasing_xy_dict_generator() + ) + type(mock_rom_deps.stage).position = mocker.PropertyMock( + side_effect=increasing_xyz_dict_generator() + ) + + expected_movement = {"x": -800 * stage_measure.MEDIUM_STEP / 100, "y": 0} + + # Remove the mock RomData before starting + rom_thing._rom_data = stage_measure.RomDataTracker() + + # Run it! + rom_thing._initial_moves_for_z_prediction("x", direction=-1, rom_deps=mock_rom_deps) + + # Check that _rom_data now contains the 5 mocked returns in order. + assert rom_thing._rom_data.offsets == [{"x": i, "y": i} for i in range(5)] + assert rom_thing._rom_data.stage_coords == [ + {"x": i, "y": i, "z": i} for i in range(5) + ] + + # Check that the csm movement function is called 5 times + assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 5 + # Each time with the expected movement + for arg_list in mock_rom_deps.csm.move_in_image_coordinates.call_args_list: + assert arg_list.kwargs == expected_movement + + +def test_move_until_edge_error(rom_thing, mock_rom_deps, mocker): + """Check that if there is an error while moving to the edge the stage returns to start.""" + mock_position_dict = {"x": 123, "y": 456, "z": 789} + type(mock_rom_deps.stage).position = mocker.PropertyMock( + return_value=mock_position_dict + ) + mocker.patch.object( + rom_thing, "_initial_moves_for_z_prediction", side_effect=RuntimeError("Mock") + ) + + # Remove the mock RomData before starting + rom_thing._rom_data = stage_measure.RomDataTracker() + + # Error should be raised even though it is in the Try: + with pytest.raises(RuntimeError, match="Mock"): + rom_thing._move_until_edge("y", direction=-1, rom_deps=mock_rom_deps) + # However the "finally" should have executed returning to the starting position + assert mock_rom_deps.stage.move_absolute.call_count == 1 + abs_move_kwargs = mock_rom_deps.stage.move_absolute.call_args.kwargs + expected_abs_move_kwargs = dict(**mock_position_dict, block_cancellation=True) + assert abs_move_kwargs == expected_abs_move_kwargs + + +def test_move_until_edge(rom_thing, mock_rom_deps, mocker): + """Check move until edge runs the correct movement sequence.""" + # Remove the mock RomData before starting + rom_thing._rom_data = stage_measure.RomDataTracker() + + mock_rom_deps.stage.position = {"x": "mock", "y": "starting", "z": "pos"} + + def add_fake_initial_positions(*_args, **_kwargs): + """Rather than run initial moves just add some fake data.""" + for i in range(5): + rom_thing._rom_data.record_movement(f"mock-init-pos{i + 1}", "mock-offset") + + def update_stage_pos_on_big_move(*_args, **_kwargs): + """With big moves update the stage position.""" + big_move_count = 1 + while True: + mock_rom_deps.stage.position = f"mocked-big-move-pos{big_move_count}" + yield + big_move_count += 1 + + def set_final_pos(*_args, **_kwargs): + """When move_back_until_motion_detected is run set a final stage position.""" + mock_rom_deps.stage.position = "mock-final-pos" + + # Mock the main movement functions + mock_init_moves = mocker.patch.object( + rom_thing, + "_initial_moves_for_z_prediction", + side_effect=add_fake_initial_positions, + ) + mock_big_moves = mocker.patch.object( + rom_thing, + "_big_z_corrected_movement", + side_effect=update_stage_pos_on_big_move(), + ) + mock_move_check = mocker.patch.object( + rom_thing, "_stage_still_moves", side_effect=[True] * 9 + [False] + ) + mock_move_back = mocker.patch.object( + rom_thing, "_move_back_until_motion_detected", side_effect=set_final_pos + ) + + # Remove the mock RomData before starting + rom_thing._rom_data = stage_measure.RomDataTracker() + + # Run function + rom_thing._move_until_edge("y", direction=-1, rom_deps=mock_rom_deps) + + # Check the call counts are as expected + # One call of initial moves + assert mock_init_moves.call_count == 1 + # Big moves and the check the stage is moving are each called 10 times as the mock + # for _stage_still_moves replies False on the 10th call. + assert mock_big_moves.call_count == 10 + assert mock_move_check.call_count == 10 + # And one call of _move_back_until_motion_detected + assert mock_move_back.call_count == 1 + + assert rom_thing._rom_data.stage_coords == [ + {"x": "mock", "y": "starting", "z": "pos"}, + "mock-init-pos1", + "mock-init-pos2", + "mock-init-pos3", + "mock-init-pos4", + "mock-init-pos5", + "mocked-big-move-pos1", + "mocked-big-move-pos2", + "mocked-big-move-pos3", + "mocked-big-move-pos4", + "mocked-big-move-pos5", + "mocked-big-move-pos6", + "mocked-big-move-pos7", + "mocked-big-move-pos8", + "mocked-big-move-pos9", + "mock-final-pos", # This overwrites the 10th big move position recording. + ] + + +def test_perform_rom_test_locked(rom_thing, mock_rom_deps): + """Check the error if the Rom test is 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)) + + +def test_perform_rom_test_(rom_thing, mock_rom_deps, mocker): + """Check that perform Rom Test runs the expected high level algorithm.""" + + def check_and_modify_rom_data(axis: str, direction: int, **_kwargs): + """Check the rom_data is empty each time, and add a final position.""" + # check rom_data was cleared at the start of this run + assert len(rom_thing._rom_data.stage_coords) == 0 + dist = 1111 if axis == "x" else 2222 + final_pos = {"x": 0, "y": 0} + final_pos[axis] = dist * direction + rom_thing._rom_data.stage_coords.append(final_pos) + + mock_move_until_edge = mocker.patch.object( + rom_thing, "_move_until_edge", side_effect=check_and_modify_rom_data + ) + + mocker.patch.object( + mock_rom_deps.cam, "grab_as_array", return_value=np.zeros([123, 456, 3]) + ) + final_dict = rom_thing.perform_rom_test(**dataclasses.asdict(mock_rom_deps)) + + # This should take less than 1 sec + assert final_dict["Time"] <= 1 + # CSM should be 2x2 list + assert isinstance(final_dict["CSM Matrix"], list) + assert len(final_dict["CSM Matrix"]) == 2 + assert len(final_dict["CSM Matrix"][0]) == 2 + assert len(final_dict["CSM Matrix"][1]) == 2 + # And step range should be [2222, 4444] + assert final_dict["Step Range"] == [2222, 4444] + + # Check that the axes were called in the expected order. + assert mock_move_until_edge.call_count == 4 + + assert mock_move_until_edge.call_args_list[0].kwargs["axis"] == "x" + assert mock_move_until_edge.call_args_list[0].kwargs["direction"] == 1 + + assert mock_move_until_edge.call_args_list[1].kwargs["axis"] == "x" + assert mock_move_until_edge.call_args_list[1].kwargs["direction"] == -1 + + assert mock_move_until_edge.call_args_list[2].kwargs["axis"] == "y" + assert mock_move_until_edge.call_args_list[2].kwargs["direction"] == 1 + + assert mock_move_until_edge.call_args_list[3].kwargs["axis"] == "y" + assert mock_move_until_edge.call_args_list[3].kwargs["direction"] == -1