rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'continue-removing-deps' into 'v3'

Browse source 
Removing dependencies from CSM, autofocus, and stage_measure

See merge request openflexure/openflexure-microscope-server!452
This commit is contained in:
Julian Stirling 2025-12-17 11:57:07 +00:00
commit 150e69d235
7 changed files with 279 additions and 493 deletions
Download patch file Download diff file Expand all files Collapse all files

149
src/openflexure_microscope_server/things/autofocus.py
View file

@ -10,20 +10,18 @@ See repository root for licensing information.
import logging import logging
import os import os
import time import time
from contextlib import contextmanager
from dataclasses import dataclass from dataclasses import dataclass
from typing import Annotated, Literal, Mapping, Optional, Sequence from types import TracebackType
from typing import Literal, Mapping, Optional, Self, Sequence
import numpy as np import numpy as np
from fastapi import Depends
from pydantic import BaseModel, computed_field, field_validator, model_validator from pydantic import BaseModel, computed_field, field_validator, model_validator
import labthings_fastapi as lt import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray from labthings_fastapi.types.numpy import NDArray
from .camera import CameraDependency as CameraClient from .camera import BaseCamera
from .camera import RawCameraDependency as RawCamera from .stage import BaseStage
from .stage import StageDependency as Stage
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
MIN_TEST_IMAGE_COUNT = 3 MIN_TEST_IMAGE_COUNT = 3
@ -220,24 +218,18 @@ class JPEGSharpnessMonitor:
``data_dict`` and data with interpolated ``z`` positions can be retrieved with ``data_dict`` and data with interpolated ``z`` positions can be retrieved with
move_data. move_data.
A new JPEGSharpnessMonitor instance is created each time an action with the
SharpnessMonitorDep as an argument is called.
""" """
def __init__( def __init__(self, stage: BaseStage, camera: BaseCamera) -> None:
self, stage: Stage, camera: RawCamera, portal: lt.deps.BlockingPortal
) -> None:
"""Initialise a new JPEGSharpnessMonitor. The args are injected automatically. """Initialise a new JPEGSharpnessMonitor. The args are injected automatically.
:param stage: A direct_thing_client dependency for the the microscope stage. :param stage: A direct_thing_client dependency for the the microscope stage.
:param camera: A raw_thing_client depeendency for the camera. This is a raw :param camera: A raw_thing_client depeendency for the camera. This is a raw
dependency as the underlying class needs to be dependency as the underlying class needs to be
:param portal: The asyncio blocking portal for asynchronous task scheduling.
""" """
self.camera = camera self.camera = camera
self.stage = stage self.stage = stage
self.portal = portal LOGGER.debug(f"Created sharpness monitor with {stage}, {camera}")
LOGGER.debug(f"Created sharpness monitor with {stage}, {camera}, {portal}")
self.stage_positions: list[Mapping[str, int]] = [] self.stage_positions: list[Mapping[str, int]] = []
self.stage_times: list[float] = [] self.stage_times: list[float] = []
self.jpeg_times: list[float] = [] self.jpeg_times: list[float] = []
@ -254,14 +246,23 @@ class JPEGSharpnessMonitor:
if not self.running: if not self.running:
break break
@contextmanager def __enter__(self) -> Self:
def run(self) -> None: """Start context manager, during which sharpness from the camera is monitored."""
"""Context manager, during which we will monitor sharpness from the camera.""" # Use the cameras _thing_server_interface to get access to the server event loop
self.portal.start_task_soon(self.monitor_sharpness) # and start a task.
try: self.camera._thing_server_interface.start_async_task_soon(
yield self.monitor_sharpness
finally: )
self.running = False return self
def __exit__(
self,
_exc_type: type[BaseException],
_exc_value: Optional[BaseException],
_traceback: Optional[TracebackType],
) -> None:
"""Clean up after context manager is closed."""
self.running = False
def focus_rel(self, dz: int, block_cancellation: int = False) -> tuple[int, int]: def focus_rel(self, dz: int, block_cancellation: int = False) -> tuple[int, int]:
"""Move the stage by dz, monitoring the position over time. """Move the stage by dz, monitoring the position over time.
@ -339,9 +340,6 @@ class JPEGSharpnessMonitor:
return SharpnessDataArrays(**data) return SharpnessDataArrays(**data)
SharpnessMonitorDep = Annotated[JPEGSharpnessMonitor, Depends()]
class AutofocusThing(lt.Thing): class AutofocusThing(lt.Thing):
"""The Thing concerned with combinations of z axis movements and the camera. """The Thing concerned with combinations of z axis movements and the camera.
@ -350,10 +348,12 @@ class AutofocusThing(lt.Thing):
field of view to assess focus (autofocus and testing the success of a z-stack) field of view to assess focus (autofocus and testing the success of a z-stack)
""" """
_cam: BaseCamera = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
@lt.action @lt.action
def fast_autofocus( def fast_autofocus(
self, self,
sharpness_monitor: SharpnessMonitorDep,
dz: int = 2000, dz: int = 2000,
start: Literal["centre", "base"] = "centre", start: Literal["centre", "base"] = "centre",
) -> SharpnessDataArrays: ) -> SharpnessDataArrays:
@ -363,7 +363,7 @@ class AutofocusThing(lt.Thing):
the position where the image was sharpest. We'll then move back down, and the position where the image was sharpest. We'll then move back down, and
finally up to the sharpest point. finally up to the sharpest point.
""" """
with sharpness_monitor.run(): with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
# Move to (-dz / 2) # Move to (-dz / 2)
if start == "centre": if start == "centre":
sharpness_monitor.focus_rel(-dz // 2) sharpness_monitor.focus_rel(-dz // 2)
@ -384,7 +384,6 @@ class AutofocusThing(lt.Thing):
@lt.action @lt.action
def z_move_and_measure_sharpness( def z_move_and_measure_sharpness(
self, self,
sharpness_monitor: SharpnessMonitorDep,
dz: Sequence[int], dz: Sequence[int],
wait: float = 0, wait: float = 0,
) -> SharpnessDataArrays: ) -> SharpnessDataArrays:
@ -400,7 +399,7 @@ class AutofocusThing(lt.Thing):
If ``wait`` is specified, we will wait for that many seconds If ``wait`` is specified, we will wait for that many seconds
between moves. between moves.
""" """
with sharpness_monitor.run(): with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
for move_index, current_dz in enumerate(dz): for move_index, current_dz in enumerate(dz):
if move_index > 0 and wait > 0: if move_index > 0 and wait > 0:
time.sleep(wait) time.sleep(wait)
@ -410,8 +409,6 @@ class AutofocusThing(lt.Thing):
@lt.action @lt.action
def looping_autofocus( def looping_autofocus(
self, self,
stage: Stage,
sharpness_monitor: SharpnessMonitorDep,
dz: int = 2000, dz: int = 2000,
start: Literal["centre", "base"] = "centre", start: Literal["centre", "base"] = "centre",
) -> tuple[list[float], list[float]]: ) -> tuple[list[float], list[float]]:
@ -425,12 +422,12 @@ class AutofocusThing(lt.Thing):
attempt = 0 attempt = 0
backlash = 200 backlash = 200
with sharpness_monitor.run(): with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
while attempt < 10: while attempt < 10:
attempt += 1 attempt += 1
if start == "centre": if start == "centre":
stage.move_relative(x=0, y=0, z=-(backlash + dz / 2)) self._stage.move_relative(x=0, y=0, z=-(backlash + dz / 2))
stage.move_relative(x=0, y=0, z=backlash) self._stage.move_relative(x=0, y=0, z=backlash)
# Always start centrally for future runs # Always start centrally for future runs
start = "centre" start = "centre"
@ -445,8 +442,8 @@ class AutofocusThing(lt.Thing):
target_max = np.max(heights) - dz / 5 target_max = np.max(heights) - dz / 5
# move to the peak # move to the peak
stage.move_absolute(z=peak_height - backlash) self._stage.move_absolute(z=peak_height - backlash)
stage.move_absolute(z=peak_height) self._stage.move_absolute(z=peak_height)
if target_min < peak_height < target_max: if target_min < peak_height < target_max:
# If it is within the target range then return # If it is within the target range then return
@ -487,28 +484,26 @@ class AutofocusThing(lt.Thing):
images_dir: str, images_dir: str,
autofocus_dz: int, autofocus_dz: int,
save_resolution: tuple[int, int], save_resolution: tuple[int, int],
logger: lt.deps.InvocationLogger,
) -> StackParams: ) -> StackParams:
"""Set up the parameters used for all stacks in a scan. """Set up the parameters used for all stacks in a scan.
:param images_dir: the folder to save all images :param images_dir: the folder to save all images
:param autofocus_dz: the range to autofocus over if a stack fails :param autofocus_dz: the range to autofocus over if a stack fails
:param save_resolution: The resolution to save the captures to disk with :param save_resolution: The resolution to save the captures to disk with
:param logger: A logger passed from the calling Thing
:returns: A StackParams object with the required parameters. :returns: A StackParams object with the required parameters.
""" """
# Coerce min_images_to_test parameter # Coerce min_images_to_test parameter
min_images_to_test = self.stack_min_images_to_test min_images_to_test = self.stack_min_images_to_test
if min_images_to_test < MIN_TEST_IMAGE_COUNT: if min_images_to_test < MIN_TEST_IMAGE_COUNT:
logger.warning( self.logger.warning(
f"Cannot test only {min_images_to_test} image(s) as this will fail. " f"Cannot test only {min_images_to_test} image(s) as this will fail. "
"Setting min images to test to lowest possible value of" "Setting min images to test to lowest possible value of"
f"{MIN_TEST_IMAGE_COUNT}." f"{MIN_TEST_IMAGE_COUNT}."
) )
min_images_to_test = MIN_TEST_IMAGE_COUNT min_images_to_test = MIN_TEST_IMAGE_COUNT
elif min_images_to_test > MAX_TEST_IMAGE_COUNT: elif min_images_to_test > MAX_TEST_IMAGE_COUNT:
logger.warning( self.logger.warning(
f"Testing {min_images_to_test} images will cause defocus. " f"Testing {min_images_to_test} images will cause defocus. "
"Setting min images to test to highest possible value of " "Setting min images to test to highest possible value of "
f"{MAX_TEST_IMAGE_COUNT}." f"{MAX_TEST_IMAGE_COUNT}."
@ -516,7 +511,7 @@ class AutofocusThing(lt.Thing):
min_images_to_test = MAX_TEST_IMAGE_COUNT min_images_to_test = MAX_TEST_IMAGE_COUNT
elif min_images_to_test % 2 == 0: elif min_images_to_test % 2 == 0:
min_images_to_test += 1 min_images_to_test += 1
logger.warning( self.logger.warning(
"Minimum number of images to test should be odd, setting to " "Minimum number of images to test should be odd, setting to "
f"{min_images_to_test}." f"{min_images_to_test}."
) )
@ -527,19 +522,19 @@ class AutofocusThing(lt.Thing):
# min_images_to_save # min_images_to_save
images_to_save = self.stack_images_to_save images_to_save = self.stack_images_to_save
if images_to_save <= 0: if images_to_save <= 0:
logger.warning( self.logger.warning(
"At least 1 images must be saved. Setting images to save to 1." "At least 1 images must be saved. Setting images to save to 1."
) )
images_to_save = 1 images_to_save = 1
elif images_to_save > min_images_to_test: elif images_to_save > min_images_to_test:
logger.warning( self.logger.warning(
f"Cannot save {images_to_save} images as this above the minimum " f"Cannot save {images_to_save} images as this above the minimum "
f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}." f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}."
) )
images_to_save = min_images_to_test images_to_save = min_images_to_test
elif images_to_save % 2 == 0: elif images_to_save % 2 == 0:
images_to_save += 1 images_to_save += 1
logger.warning( self.logger.warning(
f"Images to save should be odd, setting to {images_to_save}." f"Images to save should be odd, setting to {images_to_save}."
) )
# Set the Thing property to the coerced value # Set the Thing property to the coerced value
@ -557,9 +552,6 @@ class AutofocusThing(lt.Thing):
@lt.action @lt.action
def run_smart_stack( def run_smart_stack(
self, self,
cam: CameraClient,
stage: Stage,
sharpness_monitor: SharpnessMonitorDep,
stack_parameters: StackParams, stack_parameters: StackParams,
save_on_failure: bool = False, save_on_failure: bool = False,
check_turning_points: bool = True, check_turning_points: bool = True,
@ -572,11 +564,6 @@ class AutofocusThing(lt.Thing):
The sharpest image, and optionally images around the sharpest, will be saved The sharpest image, and optionally images around the sharpest, will be saved
to the images_dir with their coordinates in the filename. to the images_dir with their coordinates in the filename.
:param cam: Camera Dependency supplied by LabThings dependency injection
:param stage: Stage Dependency supplied by LabThings dependency injection
:param sharpness_monitor: Sharpness Monitor Dependency (for focus detection)
supplied by LabThings dependency injection
:param stack_parameters: A StackParams object containing the required :param stack_parameters: A StackParams object containing the required
parameters to run a stack. parameters to run a stack.
:param save_on_failure: Whether to save an image even if no focus was found. :param save_on_failure: Whether to save an image even if no focus was found.
@ -595,8 +582,6 @@ class AutofocusThing(lt.Thing):
success, captures, sharpest_id = self.z_stack( success, captures, sharpest_id = self.z_stack(
stack_parameters=stack_parameters, stack_parameters=stack_parameters,
check_turning_points=check_turning_points, check_turning_points=check_turning_points,
cam=cam,
stage=stage,
) )
if success: if success:
@ -609,12 +594,7 @@ class AutofocusThing(lt.Thing):
initial_z_pos = captures[0].position["z"] initial_z_pos = captures[0].position["z"]
# If a stack is not successful, move to the start and autofocus # If a stack is not successful, move to the start and autofocus
try: try:
self.reset_stack( self.reset_stack(initial_z_pos, stack_parameters.autofocus_dz)
initial_z_pos,
stack_parameters.autofocus_dz,
stage,
sharpness_monitor,
)
except NoFocusFoundError: except NoFocusFoundError:
break break
@ -626,7 +606,6 @@ class AutofocusThing(lt.Thing):
sharpest_id=sharpest_id, sharpest_id=sharpest_id,
captures=captures, captures=captures,
stack_parameters=stack_parameters, stack_parameters=stack_parameters,
cam=cam,
) )
return success, _get_capture_by_id(captures, sharpest_id).position["z"] return success, _get_capture_by_id(captures, sharpest_id).position["z"]
@ -635,8 +614,6 @@ class AutofocusThing(lt.Thing):
self, self,
initial_z_pos: list[int], initial_z_pos: list[int],
autofocus_dz: int, autofocus_dz: int,
stage: Stage,
sharpness_monitor: SharpnessMonitorDep,
) -> None: ) -> None:
"""Return to the initial z position and run a looping autofocus. """Return to the initial z position and run a looping autofocus.
@ -646,19 +623,14 @@ class AutofocusThing(lt.Thing):
``stage`` and ``sharpness_monitor`` are Thing dependencies passed through ``stage`` and ``sharpness_monitor`` are Thing dependencies passed through
from the calling action. from the calling action.
""" """
stage.move_absolute(z=initial_z_pos) self._stage.move_absolute(z=initial_z_pos)
self.looping_autofocus( self.looping_autofocus(dz=autofocus_dz)
stage=stage,
sharpness_monitor=sharpness_monitor,
dz=autofocus_dz,
)
def save_stack( def save_stack(
self, self,
sharpest_id: int, sharpest_id: int,
captures: list[list], captures: list[list],
stack_parameters: StackParams, stack_parameters: StackParams,
cam: CameraClient,
) -> int: ) -> int:
"""Save the required captures to disk. """Save the required captures to disk.
@ -669,28 +641,24 @@ class AutofocusThing(lt.Thing):
:param captures: a list of captures, including file name, image data and :param captures: a list of captures, including file name, image data and
metadata metadata
:param stack_parameters: a StackParams object holding stack parameters :param stack_parameters: a StackParams object holding stack parameters
:param cam: is a Thing dependency passed through from the calling action
""" """
sharpest_index = _get_capture_index_by_id(captures, sharpest_id) sharpest_index = _get_capture_index_by_id(captures, sharpest_id)
slice_to_save = stack_parameters.slice_to_save(sharpest_index) slice_to_save = stack_parameters.slice_to_save(sharpest_index)
# Loop through the range, saving each capture to disk # Loop through the range, saving each capture to disk
for capture in captures[slice_to_save]: for capture in captures[slice_to_save]:
cam.save_from_memory( self._cam.save_from_memory(
jpeg_path=os.path.join(stack_parameters.images_dir, capture.filename), jpeg_path=os.path.join(stack_parameters.images_dir, capture.filename),
save_resolution=stack_parameters.save_resolution, save_resolution=stack_parameters.save_resolution,
buffer_id=capture.buffer_id, buffer_id=capture.buffer_id,
) )
cam.clear_buffers() self._cam.clear_buffers()
return sharpest_index return sharpest_index
def z_stack( def z_stack(
self, self,
stack_parameters: StackParams, stack_parameters: StackParams,
check_turning_points: bool, check_turning_points: bool,
cam: CameraClient,
stage: Stage,
) -> tuple[bool, list[CaptureInfo], int]: ) -> tuple[bool, list[CaptureInfo], int]:
"""Capture a series of images checking that sharpest image central. """Capture a series of images checking that sharpest image central.
@ -703,8 +671,6 @@ class AutofocusThing(lt.Thing):
:param check_turning_points: Whether to check the number of turning points in :param check_turning_points: Whether to check the number of turning points in
the sharpnesses of the images in the stack is exactly 1. (May fail with the sharpnesses of the images in the stack is exactly 1. (May fail with
thick samples) thick samples)
:param cam: Camera Dependency to be passed through from the calling action
:param stage: Stage Dependency to be passed through from the calling action
:returns: A tuple of :returns: A tuple of
@ -714,14 +680,14 @@ class AutofocusThing(lt.Thing):
""" """
# Move down by the height of the z stack, plus an overshoot # Move down by the height of the z stack, plus an overshoot
# Better to start too low and take too many images than too high and need to refocus # Better to start too low and take too many images than too high and need to refocus
stage.move_relative( self._stage.move_relative(
z=-( z=-(
stack_parameters.steps_undershoot stack_parameters.steps_undershoot
+ stack_parameters.backlash_correction + stack_parameters.backlash_correction
+ stack_parameters.stack_z_range / 2 + stack_parameters.stack_z_range / 2
) )
) )
stage.move_relative(z=stack_parameters.backlash_correction) self._stage.move_relative(z=stack_parameters.backlash_correction)
captures = [] captures = []
# Always check for focus using the the last `min_images_to_test` in the # Always check for focus using the the last `min_images_to_test` in the
@ -735,11 +701,7 @@ class AutofocusThing(lt.Thing):
# Append a new image to the stack # Append a new image to the stack
captures.append( captures.append(
self.capture_stack_image( self.capture_stack_image(buffer_max=stack_parameters.min_images_to_test)
cam,
stage,
buffer_max=stack_parameters.min_images_to_test,
)
) )
# If the number of images is enough to test, test them # If the number of images is enough to test, test them
@ -754,33 +716,26 @@ class AutofocusThing(lt.Thing):
if result == "restart": if result == "restart":
return False, captures, capture_id return False, captures, capture_id
# If reached here the result was "continue" # If reached here the result was "continue"
stage.move_relative(z=stack_parameters.stack_dz) self._stage.move_relative(z=stack_parameters.stack_dz)
return False, captures, capture_id return False, captures, capture_id
def capture_stack_image( def capture_stack_image(self, buffer_max: int) -> CaptureInfo:
self,
cam: CameraClient,
stage: Stage,
buffer_max: int,
) -> CaptureInfo:
"""Capture another image and return the capture information. """Capture another image and return the capture information.
The capture is stored by the camera Thing, and can be saved by ID. The capture is stored by the camera Thing, and can be saved by ID.
:param cam: Camera Dependency to be passed through from the calling action
:param stage: Stage Dependency to be passed through from the calling action
:param buffer_max: The maximum number of images to tell the camera to keep in memory :param buffer_max: The maximum number of images to tell the camera to keep in memory
for saving once the stack is complete for saving once the stack is complete
:returns: A CaptureInfo object containing the capture information including its :returns: A CaptureInfo object containing the capture information including its
camera buffer_id needed for saving. camera buffer_id needed for saving.
""" """
stage_location = stage.position stage_location = self._stage.position
buffer_id = cam.capture_to_memory(buffer_max=buffer_max) buffer_id = self._cam.capture_to_memory(buffer_max=buffer_max)
return CaptureInfo( return CaptureInfo(
buffer_id=buffer_id, buffer_id=buffer_id,
position=stage_location, position=stage_location,
sharpness=cam.grab_jpeg_size(stream_name="lores"), sharpness=self._cam.grab_jpeg_size(stream_name="lores"),
) )
# Silence too many returns in this situation as refactoring to reduce returns is # Silence too many returns in this situation as refactoring to reduce returns is

74
src/openflexure_microscope_server/things/camera_stage_mapping.py
View file

@ -33,8 +33,8 @@ from camera_stage_mapping.camera_stage_tracker import Tracker
from camera_stage_mapping.exceptions import MappingError from camera_stage_mapping.exceptions import MappingError
from labthings_fastapi.types.numpy import DenumpifyingDict from labthings_fastapi.types.numpy import DenumpifyingDict
from .camera import CameraDependency as CameraClient from .camera import BaseCamera
from .stage import StageDependency as Stage from .stage import BaseStage
CoordinateType = Tuple[float, float, float] CoordinateType = Tuple[float, float, float]
XYCoordinateType = Tuple[float, float] XYCoordinateType = Tuple[float, float]
@ -62,13 +62,13 @@ class RecordedMove:
for calibrating the stage as it allows measuring how long moves take. for calibrating the stage as it allows measuring how long moves take.
""" """
def __init__(self, stage: Stage) -> None: def __init__(self, stage: BaseStage) -> None:
"""Set the stage client used for for movement. """Set the stage client used for for movement.
:param stage: the stage client to be used. ``stage.move_to_xyz_position`` will :param stage: the stage client to be used. ``stage.move_to_xyz_position`` will
be called whenever the instance is called. be called whenever the instance is called.
""" """
self._stage: Stage = stage self._stage = stage
self._current_position: Optional[CoordinateType] = None self._current_position: Optional[CoordinateType] = None
self._history: List[Tuple[float, Optional[CoordinateType]]] = [] self._history: List[Tuple[float, Optional[CoordinateType]]] = []
@ -118,58 +118,53 @@ class CameraStageMapper(lt.Thing):
override the ``get_xyz_position()`` and ``move_to_xyz_position()`` methods. override the ``get_xyz_position()`` and ``move_to_xyz_position()`` methods.
""" """
_cam: BaseCamera = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
@lt.action @lt.action
def calibrate_1d( def calibrate_1d(self, direction: Tuple[float, float, float]) -> DenumpifyingDict:
self,
camera: CameraClient,
stage: Stage,
logger: lt.deps.InvocationLogger,
direction: Tuple[float, float, float],
) -> DenumpifyingDict:
"""Move a microscope's stage in 1D, and figure out the relationship with the camera.""" """Move a microscope's stage in 1D, and figure out the relationship with the camera."""
# Record positions and times for stage calibration # Record positions and times for stage calibration
recorded_move = RecordedMove(stage) recorded_move = RecordedMove(self._stage)
tracker = Tracker( tracker = Tracker(
camera.capture_downsampled_array, self._cam.capture_downsampled_array,
stage.get_xyz_position, self._stage.get_xyz_position,
settle=camera.settle, settle=self._cam.settle,
) )
direction_array: np.ndarray = np.array(direction) direction_array: np.ndarray = np.array(direction)
starting_position = stage.position starting_position = self._stage.position
try: try:
result: dict = calibrate_backlash_1d( result: dict = calibrate_backlash_1d(
tracker, recorded_move, direction_array, logger=logger tracker, recorded_move, direction_array, logger=self.logger
) )
except lt.exceptions.InvocationCancelledError as e: except lt.exceptions.InvocationCancelledError as e:
logger.info("User cancelled the camera stage mapping calibration") self.logger.info("User cancelled the camera stage mapping calibration")
logger.info("Returning to starting position") self.logger.info("Returning to starting position")
stage.move_absolute(**starting_position, block_cancellation=True) self._stage.move_absolute(**starting_position, block_cancellation=True)
raise e raise e
except MappingError as e: except MappingError as e:
logger.info("Returning to starting position due to failed calibration") self.logger.info("Returning to starting position due to failed calibration")
stage.move_absolute(**starting_position, block_cancellation=True) self._stage.move_absolute(**starting_position, block_cancellation=True)
raise e raise e
result["move_history"] = recorded_move.history result["move_history"] = recorded_move.history
result["image_resolution"] = camera.capture_downsampled_array().shape[:2] result["image_resolution"] = self._cam.capture_downsampled_array().shape[:2]
return result return result
@lt.action @lt.action
def calibrate_xy( def calibrate_xy(self) -> DenumpifyingDict:
self, camera: CameraClient, stage: Stage, logger: lt.deps.InvocationLogger
) -> DenumpifyingDict:
"""Move the microscope's stage in X and Y, to calibrate its relationship to the camera. """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. This performs two 1d calibrations in x and y, then combines their results.
""" """
downsampling_factor = camera.downsampled_array_factor downsampling_factor = self._cam.downsampled_array_factor
# Calibrate y-axis first as it is more likely to fail. # Calibrate y-axis first as it is more likely to fail.
# The x-y difference is due to the camera aspect ratio, not the stage hardware. # The x-y difference is due to the camera aspect ratio, not the stage hardware.
logger.info("Calibrating Y axis:") self.logger.info("Calibrating Y axis:")
cal_y: dict = self.calibrate_1d(camera, stage, logger, (0, 1, 0)) cal_y: dict = self.calibrate_1d((0, 1, 0))
logger.info("Calibrating X axis:") self.logger.info("Calibrating X axis:")
cal_x: dict = self.calibrate_1d(camera, stage, logger, (1, 0, 0)) cal_x: dict = self.calibrate_1d((1, 0, 0))
logger.info("Calibration complete, updating metadata.") self.logger.info("Calibration complete, updating metadata.")
# Combine X and Y calibrations to make a 2D calibration # Combine X and Y calibrations to make a 2D calibration
cal_xy: dict = image_to_stage_displacement_from_1d([cal_x, cal_y]) cal_xy: dict = image_to_stage_displacement_from_1d([cal_x, cal_y])
@ -185,7 +180,7 @@ class CameraStageMapper(lt.Thing):
f"[{round(csm_matrix[0][0], 2)}, {round(csm_matrix[0][1], 2)},]," f"[{round(csm_matrix[0][0], 2)}, {round(csm_matrix[0][1], 2)},],"
f"[{round(csm_matrix[1][0], 2)}, {round(csm_matrix[1][1], 2)}]" f"[{round(csm_matrix[1][0], 2)}, {round(csm_matrix[1][1], 2)}]"
) )
logger.info(f"CSM matrix is {csm_as_string}.") self.logger.info(f"CSM matrix is {csm_as_string}.")
data: Dict[str, dict] = { data: Dict[str, dict] = {
"camera_stage_mapping_calibration": cal_xy, "camera_stage_mapping_calibration": cal_xy,
@ -201,9 +196,7 @@ class CameraStageMapper(lt.Thing):
return data return data
@lt.property @lt.property
def image_to_stage_displacement_matrix( def image_to_stage_displacement_matrix(self) -> Optional[List[List[float]]]:
self,
) -> Optional[List[List[float]]]: # 2x2 integer array
"""A 2x2 matrix that converts displacement in image coordinates to stage coordinates. """A 2x2 matrix that converts displacement in image coordinates to stage coordinates.
Note that this matrix is defined using "matrix coordinates", i.e. image coordinates Note that this matrix is defined using "matrix coordinates", i.e. image coordinates
@ -253,12 +246,7 @@ class CameraStageMapper(lt.Thing):
raise CSMUncalibratedError() # noqa: RSE102 raise CSMUncalibratedError() # noqa: RSE102
@lt.action @lt.action
def move_in_image_coordinates( def move_in_image_coordinates(self, x: float, y: float) -> None:
self,
stage: Stage,
x: float,
y: float,
) -> None:
"""Move by a given number of pixels on the camera. """Move by a given number of pixels on the camera.
NB x and y here refer to what is usually understood to be the horizontal and NB x and y here refer to what is usually understood to be the horizontal and
@ -271,7 +259,7 @@ class CameraStageMapper(lt.Thing):
an image usually helps resolve any ambiguity. an image usually helps resolve any ambiguity.
""" """
self.assert_calibrated() self.assert_calibrated()
stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y)) self._stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y))
@lt.action @lt.action
def convert_image_to_stage_coordinates( def convert_image_to_stage_coordinates(

1
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -378,7 +378,6 @@ class SmartScanThing(lt.Thing):
images_dir=self._ongoing_scan.images_dir, images_dir=self._ongoing_scan.images_dir,
autofocus_dz=self.autofocus_dz, autofocus_dz=self.autofocus_dz,
save_resolution=self._scan_data.save_resolution, save_resolution=self._scan_data.save_resolution,
logger=self._scan_logger,
) )
self._preview_stitcher = stitching.PreviewStitcher( self._preview_stitcher = stitching.PreviewStitcher(
self._ongoing_scan.images_dir, self._ongoing_scan.images_dir,

265
src/openflexure_microscope_server/things/stage_measure.py
View file

@ -14,7 +14,6 @@ this is 10% of the expected motion is the measured axis.
""" """
import time import time
from dataclasses import dataclass
from threading import Lock from threading import Lock
from typing import Any, Literal, Optional, overload from typing import Any, Literal, Optional, overload
@ -25,14 +24,9 @@ from camera_stage_mapping import fft_image_tracking
# Things # Things
from .autofocus import AutofocusThing from .autofocus import AutofocusThing
from .camera import CameraDependency as CamDep from .camera import BaseCamera
from .camera_stage_mapping import CameraStageMapper from .camera_stage_mapping import CameraStageMapper
from .stage import StageDependency as StageDep from .stage import BaseStage
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 ## Size of movement in percentage of field of view
SMALL_STEP = 20 SMALL_STEP = 20
@ -111,20 +105,6 @@ class RomDataTracker:
return {axis: int(turning_loc), other_axis: other_coord, "z": int(turning_z)} return {axis: int(turning_loc), other_axis: other_coord, "z": int(turning_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): class ParasiticMotionError(Exception):
"""Custom exception raised when parasitic motion is detected. """Custom exception raised when parasitic motion is detected.
@ -136,6 +116,11 @@ class ParasiticMotionError(Exception):
class RangeofMotionThing(lt.Thing): class RangeofMotionThing(lt.Thing):
"""A class used to measure the range of motion of the stage in X and Y.""" """A class used to measure the range of motion of the stage in X and Y."""
_autofocus: AutofocusThing = lt.thing_slot()
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
calibrated_range: Optional[list[int, int]] = lt.setting(default=None, readonly=True) calibrated_range: Optional[list[int, int]] = lt.setting(default=None, readonly=True)
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None: def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
@ -146,21 +131,9 @@ class RangeofMotionThing(lt.Thing):
self._rom_data = RomDataTracker() self._rom_data = RomDataTracker()
@lt.action @lt.action
def perform_rom_test( def perform_rom_test(self) -> dict[str, Any]:
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. """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. :return: Results dictionary separated into keys of each axis and direction.
""" """
got_lock = self._lock.acquire(blocking=False) got_lock = self._lock.acquire(blocking=False)
@ -168,17 +141,14 @@ class RangeofMotionThing(lt.Thing):
raise RuntimeError("Trying to run ROM test when a test is already running.") raise RuntimeError("Trying to run ROM test when a test is already running.")
try: try:
rom_deps = RomDeps( self.logger.info(
autofocus=autofocus, stage=stage, csm=csm, cam=cam, logger=logger
)
logger.info(
"Using the stage to measure the Range of Motion. " "Using the stage to measure the Range of Motion. "
"Please ensure you are using a sample that covers the whole range of " "Please ensure you are using a sample that covers the whole range of "
"motion. This should be approximately 12 x 12 mm." "motion. This should be approximately 12 x 12 mm."
) )
start_time = time.time() start_time = time.time()
self._set_stream_resolution(cam) self._set_stream_resolution()
# The total range for the two axes under test # The total range for the two axes under test
step_range = [] step_range = []
@ -193,84 +163,61 @@ class RangeofMotionThing(lt.Thing):
for axis_dir in directions: for axis_dir in directions:
# Create a new tracker at start of measurement. # Create a new tracker at start of measurement.
self._rom_data = RomDataTracker() self._rom_data = RomDataTracker()
self._move_until_edge( self._move_until_edge(axis=axis, direction=axis_dir)
axis=axis, direction=axis_dir, rom_deps=rom_deps
)
# Append final position from tracker # Append final position from tracker
axis_limits.append(self._rom_data.final_position[axis]) axis_limits.append(self._rom_data.final_position[axis])
# Calculate step range. # Calculate step range.
step_range.append(abs(axis_limits[0] - axis_limits[1])) step_range.append(abs(axis_limits[0] - axis_limits[1]))
total_time = time.time() - start_time total_time = time.time() - start_time
logger.info(f"Range of motion is {step_range[0]} x {step_range[1]} steps") self.logger.info(
f"Range of motion is {step_range[0]} x {step_range[1]} steps"
)
self.calibrated_range = step_range self.calibrated_range = step_range
finally: finally:
self._lock.release() self._lock.release()
return { return {
"Time": total_time, "Time": total_time,
"CSM Matrix": csm.image_to_stage_displacement_matrix, "CSM Matrix": self._csm.image_to_stage_displacement_matrix,
"Step Range": step_range, "Step Range": step_range,
} }
@lt.action @lt.action
def perform_recentre( def perform_recentre(self) -> None:
self, """Measures the curvature of the motion to centre x and y axes."""
autofocus: AutofocusDep,
stage: StageDep,
cam: CamDep,
csm: CSMDep,
logger: lt.deps.InvocationLogger,
) -> None:
"""Measures the curvature of the motion to centre 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.
"""
got_lock = self._lock.acquire(blocking=False) got_lock = self._lock.acquire(blocking=False)
if not got_lock: if not got_lock:
raise RuntimeError("Trying to run recentre when a test is already running.") raise RuntimeError("Trying to run recentre when a test is already running.")
try: try:
rom_deps = RomDeps( self.logger.info("Recentring the stage.")
autofocus=autofocus, stage=stage, csm=csm, cam=cam, logger=logger
)
logger.info("Recentring the stage.")
self._set_stream_resolution(cam) self._set_stream_resolution()
# Strictly typed definitions to iterate over for MyPys sake. # Strictly typed definitions to iterate over for MyPys sake.
axes: tuple[Literal["x"], Literal["y"]] = ("x", "y") axes: tuple[Literal["x"], Literal["y"]] = ("x", "y")
for axis in axes: for axis in axes:
self._recentre_axis(axis, rom_deps) self._recentre_axis(axis)
centre = rom_deps.stage.position centre = self._stage.position
centre_str = f"({centre['x']}, {centre['y']})" centre_str = f"({centre['x']}, {centre['y']})"
rom_deps.logger.info(f"Centre is estimated at {centre_str}.") self.logger.info(f"Centre is estimated at {centre_str}.")
# Set the central position to (0,0,0) # Set the central position to (0,0,0)
rom_deps.stage.set_zero_position() self._stage.set_zero_position()
rom_deps.logger.info("Position reset to (0, 0, 0).") self.logger.info("Position reset to (0, 0, 0).")
finally: finally:
self._lock.release() self._lock.release()
def _set_stream_resolution(self, cam: CamDep) -> None: def _set_stream_resolution(self) -> None:
"""Set the self._stream_resolution attribute by reading camera. """Set the self._stream_resolution attribute by reading camera."""
stream_shape = self._cam.grab_as_array().shape
:param cam: The camera dependency.
"""
stream_shape = cam.grab_as_array().shape
# Swap axes as numpy is [y, x] # Swap axes as numpy is [y, x]
self._stream_resolution = (stream_shape[1], stream_shape[0]) self._stream_resolution = (stream_shape[1], stream_shape[0])
def _move_until_edge( def _move_until_edge(
self, self, axis: Literal["x", "y"], direction: Literal[1, -1]
axis: Literal["x", "y"],
direction: Literal[1, -1],
rom_deps: RomDeps,
) -> None: ) -> None:
"""Move in one direction until movement per step decreases significantly. """Move in one direction until movement per step decreases significantly.
@ -278,93 +225,72 @@ class RangeofMotionThing(lt.Thing):
will be some movement as there is no hard stop, but it will reduce will be some movement as there is no hard stop, but it will reduce
significantly. 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 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. :param direction: The direction which is being measured. This must be 1 or -1.
:return: Results dictionary containing stage positions, :return: Results dictionary containing stage positions,
correlations and the final position. correlations and the final position.
""" """
# Start by performing an autofocus and recording starting position # Start by performing an autofocus and recording starting position
rom_deps.autofocus.looping_autofocus(dz=1000) self._autofocus.looping_autofocus(dz=1000)
starting_position = rom_deps.stage.position starting_position = self._stage.position
self._rom_data.stage_coords.append(starting_position) self._rom_data.stage_coords.append(starting_position)
try: try:
dir_str = "positive" if direction == 1 else "negative" dir_str = "positive" if direction == 1 else "negative"
rom_deps.logger.info( self.logger.info(f"Beginning the {axis}-axis in the {dir_str} direction")
f"Beginning the {axis}-axis in the {dir_str} direction"
)
rom_deps.logger.info("Moving the stage in 5 medium sized steps.") self.logger.info("Moving the stage in 5 medium sized steps.")
self._moves_for_z_prediction( self._moves_for_z_prediction(axis=axis, direction=direction)
axis=axis,
direction=direction,
rom_deps=rom_deps,
)
still_moving = True still_moving = True
# Loop taking 1 big step followed by 3 small steps to check if the # 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. # stage is moving as expected or has reached end of range of motion.
# Stop once end of range of motion is detected. # Stop once end of range of motion is detected.
while still_moving: while still_moving:
self._big_z_corrected_movement( self._big_z_corrected_movement(axis=axis, direction=direction)
axis=axis, direction=direction, rom_deps=rom_deps
)
# Autofocus and record position # Autofocus and record position
rom_deps.autofocus.looping_autofocus(dz=800) self._autofocus.looping_autofocus(dz=800)
self._rom_data.stage_coords.append(rom_deps.stage.position) self._rom_data.stage_coords.append(self._stage.position)
# Perform small moves to check stage is still moving. # Perform small moves to check stage is still moving.
still_moving = self._stage_still_moves( still_moving = self._stage_still_moves(axis=axis, direction=direction)
axis=axis,
direction=direction,
rom_deps=rom_deps,
)
# Stage may have crashed during a large move. Move stage back until motion # Stage may have crashed during a large move. Move stage back until motion
# is detected # is detected
rom_deps.logger.info("Moving stage back until motion is detect") self.logger.info("Moving stage back until motion is detect")
self._move_back_until_motion_detected( self._move_back_until_motion_detected(axis=axis, direction=direction)
axis=axis,
direction=direction,
rom_deps=rom_deps,
)
# Replace final position. # Replace final position.
self._rom_data.stage_coords[-1] = rom_deps.stage.position self._rom_data.stage_coords[-1] = self._stage.position
finally: finally:
rom_deps.stage.move_absolute(**starting_position, block_cancellation=True) self._stage.move_absolute(**starting_position, block_cancellation=True)
def _recentre_axis(self, axis: Literal["x", "y"], rom_deps: RomDeps) -> None: def _recentre_axis(self, axis: Literal["x", "y"]) -> None:
"""Recentre a single axis. """Recentre a single axis.
:param axis: The axis to recentre :param axis: The axis to recentre
:param rom_deps: All dependencies that were passed to the calling Action.
""" """
rom_deps.logger.info(f"Finding centre in {axis}.") self.logger.info(f"Finding centre in {axis}.")
# A new tracker for this axis. # A new tracker for this axis.
self._rom_data = RomDataTracker() self._rom_data = RomDataTracker()
# Direction is in image coords, initial assumption is that centre is at (0,0). # Direction is in image coords, initial assumption is that centre is at (0,0).
direction = self._img_dir_from_stage_coords( direction = self._img_dir_from_stage_coords({"x": 0, "y": 0, "z": 0}, axis)
{"x": 0, "y": 0, "z": 0}, axis, rom_deps
)
i = 0 i = 0
while True: while True:
i += 1 i += 1
# Find z then make a number of moves (autofocussing and logging position) # Find z then make a number of moves (autofocussing and logging position)
# 5 moves initially (2 subsequently). # 5 moves initially (2 subsequently).
rom_deps.autofocus.looping_autofocus(dz=1000) self._autofocus.looping_autofocus(dz=1000)
self._rom_data.stage_coords.append(rom_deps.stage.position) self._rom_data.stage_coords.append(self._stage.position)
self._moves_for_z_prediction( self._moves_for_z_prediction(
axis=axis, axis=axis,
direction=direction, direction=direction,
rom_deps=rom_deps,
n_moves=5 if i == 1 else 2, n_moves=5 if i == 1 else 2,
) )
centred, direction = self._recentre_decision(axis, rom_deps) centred, direction = self._recentre_decision(axis)
if centred: if centred:
break break
@ -373,10 +299,10 @@ class RangeofMotionThing(lt.Thing):
raise RuntimeError(f"Couldn't find centre of {axis}-axis") raise RuntimeError(f"Couldn't find centre of {axis}-axis")
# Make a big z-corrected move towards estimate of centre. # Make a big z-corrected move towards estimate of centre.
self._big_z_corrected_movement(axis, direction, rom_deps) self._big_z_corrected_movement(axis, direction)
def _recentre_decision( def _recentre_decision(
self, axis: Literal["x", "y"], rom_deps: RomDeps self, axis: Literal["x", "y"]
) -> tuple[bool, Literal[1, -1]]: ) -> tuple[bool, Literal[1, -1]]:
"""Decide what to do next during recentreing. """Decide what to do next during recentreing.
@ -388,26 +314,25 @@ class RangeofMotionThing(lt.Thing):
* If larger, return that it is not centred, and the direction to move in. * If larger, return that it is not centred, and the direction to move in.
:param axis: The axis to recentre :param axis: The axis to recentre
:param rom_deps: All dependencies that were passed to the calling Action.
:return: A tuple. The first value is True for "the stage is now centred" and :return: A tuple. The first value is True for "the stage is now centred" and
False for "the stage is not centred". The second value is the direction to False for "the stage is not centred". The second value is the direction to
move in, this only has meaning if the first value is False. move in, this only has meaning if the first value is False.
""" """
estimate = self._rom_data.find_turning_point(axis=axis) estimate = self._rom_data.find_turning_point(axis=axis)
img_perc = self._distance_in_img_percentage(estimate, axis, rom_deps) img_perc = self._distance_in_img_percentage(estimate, axis)
# if the distance is less than 1 big step away then move to it and exit # if the distance is less than 1 big step away then move to it and exit
if abs(img_perc) < BIG_STEP: if abs(img_perc) < BIG_STEP:
rom_deps.logger.info(f"Estimated centre of {axis}-axis is {estimate[axis]}") self.logger.info(f"Estimated centre of {axis}-axis is {estimate[axis]}")
rom_deps.stage.move_absolute(**estimate) self._stage.move_absolute(**estimate)
# Note the second return, the direction, is meaningless here. # Note the second return, the direction, is meaningless here.
return True, 1 return True, 1
rom_deps.logger.info( self.logger.info(
f"Estimated centre {abs(img_perc):.0f}% of a field of view away, " f"Estimated centre {abs(img_perc):.0f}% of a field of view away, "
"that is too far to move in one move." "that is too far to move in one move."
) )
# Else calculate the desired direction in image coordinates. # Else calculate the desired direction in image coordinates.
direction = self._img_dir_from_stage_coords(estimate, axis, rom_deps) direction = self._img_dir_from_stage_coords(estimate, axis)
return False, direction return False, direction
def _img_percentage_to_img_coords( def _img_percentage_to_img_coords(
@ -428,30 +353,28 @@ class RangeofMotionThing(lt.Thing):
return (fov_perc / 100) * self._stream_resolution[img_index] return (fov_perc / 100) * self._stream_resolution[img_index]
def _img_dir_from_stage_coords( def _img_dir_from_stage_coords(
self, target: dict[str, int], axis: Literal["x", "y"], rom_deps: RomDeps self, target: dict[str, int], axis: Literal["x", "y"]
) -> Literal[1, -1]: ) -> Literal[1, -1]:
"""For a target location in stage coords, return the direction in image coordinates. """For a target location in stage coords, return the direction in image coordinates.
:param target: The target poisiton in stage coordinates :param target: The target poisiton in stage coordinates
:param axis: The axis which is being measured. This must be 'x' or 'y'. :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.
:return: Direction to move in image coordinates. :return: Direction to move in image coordinates.
""" """
target_im_coords = rom_deps.csm.convert_stage_to_image_coordinates(**target) target_im_coords = self._csm.convert_stage_to_image_coordinates(**target)
current_loc = rom_deps.stage.position current_loc = self._stage.position
current_loc_im_coords = rom_deps.csm.convert_stage_to_image_coordinates( current_loc_im_coords = self._csm.convert_stage_to_image_coordinates(
**current_loc **current_loc
) )
return -1 if target_im_coords[axis] < current_loc_im_coords[axis] else 1 return -1 if target_im_coords[axis] < current_loc_im_coords[axis] else 1
def _distance_in_img_percentage( def _distance_in_img_percentage(
self, target: dict[str, int], axis: Literal["x", "y"], rom_deps: RomDeps self, target: dict[str, int], axis: Literal["x", "y"]
) -> float: ) -> float:
"""For a target location in stage coords return the distance in percentage of FOV. """For a target location in stage coords return the distance in percentage of FOV.
:param target: The target poisiton in stage coordinates :param target: The target poisiton in stage coordinates
:param axis: The axis which is being measured. This must be 'x' or 'y'. :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.
:return: Percentage of field of view the stage should move by. :return: Percentage of field of view the stage should move by.
""" """
if self._stream_resolution is None: if self._stream_resolution is None:
@ -459,9 +382,9 @@ class RangeofMotionThing(lt.Thing):
"Stream resolution must be set before converting coords to percentage" "Stream resolution must be set before converting coords to percentage"
) )
current_loc = rom_deps.stage.position current_loc = self._stage.position
move_stage = {key: target[key] - current_loc[key] for key in target} move_stage = {key: target[key] - current_loc[key] for key in target}
move_img = rom_deps.csm.convert_stage_to_image_coordinates(**move_stage) move_img = self._csm.convert_stage_to_image_coordinates(**move_stage)
img_index = 0 if axis == "x" else 1 img_index = 0 if axis == "x" else 1
return (move_img[axis] / self._stream_resolution[img_index]) * 100 return (move_img[axis] / self._stream_resolution[img_index]) * 100
@ -488,11 +411,7 @@ class RangeofMotionThing(lt.Thing):
return {"x": 0, "y": distance * direction} return {"x": 0, "y": distance * direction}
def _moves_for_z_prediction( def _moves_for_z_prediction(
self, self, axis: Literal["x", "y"], direction: Literal[1, -1], n_moves: int = 5
axis: Literal["x", "y"],
direction: Literal[1, -1],
rom_deps: RomDeps,
n_moves: int = 5,
) -> None: ) -> None:
"""Perform medium sized moves with autofocus for z feed-forward. """Perform medium sized moves with autofocus for z feed-forward.
@ -502,7 +421,6 @@ class RangeofMotionThing(lt.Thing):
:param direction: The direction the stage moves. :param direction: The direction the stage moves.
:param axis: The axis which is being measured. This must be 'x' or 'y'. :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
:param n_moves: Number of moves to make. Default is 5 which is enough for an :param n_moves: Number of moves to make. Default is 5 which is enough for an
initial z estimate. initial z estimate.
""" """
@ -510,9 +428,9 @@ class RangeofMotionThing(lt.Thing):
fov_perc=MEDIUM_STEP, axis=axis, direction=direction fov_perc=MEDIUM_STEP, axis=axis, direction=direction
) )
for _loop in range(n_moves): for _loop in range(n_moves):
offset = self._move_and_measure(movement=movement, rom_deps=rom_deps) offset = self._move_and_measure(movement=movement)
self._rom_data.record_movement(rom_deps.stage.position, offset) self._rom_data.record_movement(self._stage.position, offset)
if _parasitic_motion_detected(movement, offset): if _parasitic_motion_detected(movement, offset):
raise ParasiticMotionError( raise ParasiticMotionError(
"Parasitic motion detected during images to calculate z-curvature. " "Parasitic motion detected during images to calculate z-curvature. "
@ -521,7 +439,7 @@ class RangeofMotionThing(lt.Thing):
) )
def _big_z_corrected_movement( def _big_z_corrected_movement(
self, axis: Literal["x", "y"], direction: Literal[1, -1], rom_deps: RomDeps self, axis: Literal["x", "y"], direction: Literal[1, -1]
) -> None: ) -> None:
"""Take one big move with feed-forward z-correction. """Take one big move with feed-forward z-correction.
@ -529,27 +447,25 @@ class RangeofMotionThing(lt.Thing):
:param axis: The axis to move in. :param axis: The axis to move in.
:param direction: The direction 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( movement = self._movement_in_img_coords(
fov_perc=BIG_STEP, axis=axis, direction=direction fov_perc=BIG_STEP, axis=axis, direction=direction
) )
# Convert to stage coordinates # Convert to stage coordinates
stage_movement = rom_deps.csm.convert_image_to_stage_coordinates(**movement) stage_movement = self._csm.convert_image_to_stage_coordinates(**movement)
z_disp = self._rom_data.predict_z_displacement( z_disp = self._rom_data.predict_z_displacement(
axis=axis, axis=axis,
stage_movement=stage_movement, stage_movement=stage_movement,
stage_position=rom_deps.stage.position, stage_position=self._stage.position,
) )
rom_deps.stage.move_relative(z=z_disp) self._stage.move_relative(z=z_disp)
rom_deps.csm.move_in_image_coordinates(**movement) self._csm.move_in_image_coordinates(**movement)
def _stage_still_moves( def _stage_still_moves(
self, self,
axis: Literal["x", "y"], axis: Literal["x", "y"],
direction: Literal[1, -1], direction: Literal[1, -1],
rom_deps: RomDeps,
) -> bool: ) -> bool:
"""Carry out 3 small moves in a given direction and axis. """Carry out 3 small moves in a given direction and axis.
@ -562,7 +478,6 @@ class RangeofMotionThing(lt.Thing):
:param axis: The axis which is being measured. This must be 'x' or 'y'. :param axis: The axis which is being measured. This must be 'x' or 'y'.
:param direction: The direction the stage moves. :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( movement = self._movement_in_img_coords(
fov_perc=SMALL_STEP, axis=axis, direction=direction fov_perc=SMALL_STEP, axis=axis, direction=direction
@ -572,7 +487,6 @@ class RangeofMotionThing(lt.Thing):
for _loop in range(3): for _loop in range(3):
offset = self._move_and_measure( offset = self._move_and_measure(
movement=movement, movement=movement,
rom_deps=rom_deps,
# Don't autofocus initially # Don't autofocus initially
perform_autofocus=False, perform_autofocus=False,
# But retry focus 3 times if detected motion is too small or parasitic # But retry focus 3 times if detected motion is too small or parasitic
@ -580,13 +494,13 @@ class RangeofMotionThing(lt.Thing):
abs_min_offset=abs_min_offset, abs_min_offset=abs_min_offset,
) )
parasitic_motion = _parasitic_motion_detected(movement, offset) parasitic_motion = _parasitic_motion_detected(movement, offset)
rom_deps.logger.info(f"Offset measured as {offset[axis]}") self.logger.info(f"Offset measured as {offset[axis]}")
self._rom_data.record_movement(rom_deps.stage.position, offset) self._rom_data.record_movement(self._stage.position, offset)
if abs(offset[axis]) < abs_min_offset or parasitic_motion: if abs(offset[axis]) < abs_min_offset or parasitic_motion:
# If the offset is below the abs min offset, or significant # If the offset is below the abs min offset, or significant
# parasitic motion is detected then the edge has been found. # parasitic motion is detected then the edge has been found.
rom_deps.logger.info("Edge has been found.") self.logger.info("Edge has been found.")
return False return False
# If we reached here the stage is still moving fine # If we reached here the stage is still moving fine
return True return True
@ -595,7 +509,6 @@ class RangeofMotionThing(lt.Thing):
self, self,
axis: Literal["x", "y"], axis: Literal["x", "y"],
direction: Literal[1, -1], direction: Literal[1, -1],
rom_deps: RomDeps,
) -> None: ) -> None:
"""Move the stage against the direction of test unil motion is detected. """Move the stage against the direction of test unil motion is detected.
@ -605,7 +518,6 @@ class RangeofMotionThing(lt.Thing):
:param axis: The axis in which the stage is moving. This must be 'x' or 'y'. :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, :param direction: The direction in which the stage was moving during the test,
this method will move in the opposite direction. 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( movement = self._movement_in_img_coords(
fov_perc=SMALL_STEP, axis=axis, direction=-direction fov_perc=SMALL_STEP, axis=axis, direction=-direction
@ -617,17 +529,17 @@ class RangeofMotionThing(lt.Thing):
for i in range(max_moves): for i in range(max_moves):
# Increment movement # Increment movement
rom_deps.logger.info(f"Testing with step size {movement[axis]}") self.logger.info(f"Testing with step size {movement[axis]}")
# Run an autofocus every 3 steps # Run an autofocus every 3 steps
run_autofocus = i % 3 == 2 run_autofocus = i % 3 == 2
offset = self._move_and_measure( offset = self._move_and_measure(
movement=movement, rom_deps=rom_deps, perform_autofocus=run_autofocus movement=movement, perform_autofocus=run_autofocus
) )
rom_deps.logger.info(f"Offset measured as {abs(offset[axis])}") self.logger.info(f"Offset measured as {abs(offset[axis])}")
if abs(offset[axis]) > motion_minimum: if abs(offset[axis]) > motion_minimum:
rom_deps.logger.info("Motion detected.") self.logger.info("Motion detected.")
return return
# If this point is reached then motion is never detected # If this point is reached then motion is never detected
raise RuntimeError("Cannot detect motion again after reaching end of range.") raise RuntimeError("Cannot detect motion again after reaching end of range.")
@ -635,7 +547,6 @@ class RangeofMotionThing(lt.Thing):
def _move_and_measure( def _move_and_measure(
self, self,
movement: dict[str, float], movement: dict[str, float],
rom_deps: RomDeps,
perform_autofocus: bool = True, perform_autofocus: bool = True,
max_autofocus_repeats: int = 0, max_autofocus_repeats: int = 0,
abs_min_offset: float = 0.0, abs_min_offset: float = 0.0,
@ -643,7 +554,6 @@ class RangeofMotionThing(lt.Thing):
"""Move the stage and measure the offset between the two positions. """Move the stage and measure the offset between the two positions.
:param movement: A dictionary containing the distance to move in image coords. :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. :param perform_autofocus: Set to False to disable autofocus after move.
Default is True Default is True
:param max_autofocus_repeats: The number of times to repeat the focus if the :param max_autofocus_repeats: The number of times to repeat the focus if the
@ -659,13 +569,13 @@ class RangeofMotionThing(lt.Thing):
) )
# Take image before move # Take image before move
before_img = rom_deps.cam.grab_as_array() before_img = self._cam.grab_as_array()
# Move and autofocus if required # Move and autofocus if required
rom_deps.csm.move_in_image_coordinates(**movement) self._csm.move_in_image_coordinates(**movement)
if perform_autofocus: if perform_autofocus:
rom_deps.autofocus.looping_autofocus(dz=800) self._autofocus.looping_autofocus(dz=800)
# Take image and calculate offset # Take image and calculate offset
offset = self._offset_from(before_img, rom_deps=rom_deps) offset = self._offset_from(before_img)
if max_autofocus_repeats > 0: if max_autofocus_repeats > 0:
axis = _axis_from_movement_dict(movement) axis = _axis_from_movement_dict(movement)
@ -673,12 +583,12 @@ class RangeofMotionThing(lt.Thing):
parasitic_motion = _parasitic_motion_detected(movement, offset) parasitic_motion = _parasitic_motion_detected(movement, offset)
while abs(offset[axis]) < abs_min_offset or parasitic_motion: while abs(offset[axis]) < abs_min_offset or parasitic_motion:
af_repeats += 1 af_repeats += 1
rom_deps.logger.info( self.logger.info(
f"Motion not detected. Refocusing to check. Attempt {af_repeats}/3." f"Motion not detected. Refocusing to check. Attempt {af_repeats}/3."
) )
rom_deps.autofocus.looping_autofocus(dz=800) self._autofocus.looping_autofocus(dz=800)
# Re-take image and calculate offset # Re-take image and calculate offset
offset = self._offset_from(before_img, rom_deps=rom_deps) offset = self._offset_from(before_img)
parasitic_motion = _parasitic_motion_detected(movement, offset) parasitic_motion = _parasitic_motion_detected(movement, offset)
if af_repeats >= max_autofocus_repeats: if af_repeats >= max_autofocus_repeats:
# break if the maximum number of tries are exceeded. # break if the maximum number of tries are exceeded.
@ -686,22 +596,19 @@ class RangeofMotionThing(lt.Thing):
return offset return offset
def _offset_from( def _offset_from(self, before_img: np.ndarray) -> dict[str, float]:
self, before_img: np.ndarray, rom_deps: RomDeps
) -> dict[str, float]:
"""Take an image and calculate the offset from an input image. """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 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 :return: The calculated offset as a dictionary in pixels
""" """
is_sample, _bg_message = rom_deps.cam.image_is_sample() is_sample, _bg_message = self._cam.image_is_sample()
if not is_sample: if not is_sample:
raise RuntimeError( raise RuntimeError(
"No sample detected. Sample must be densely featured and cover the " "No sample detected. Sample must be densely featured and cover the "
"whole range of motion." "whole range of motion."
) )
after_img = rom_deps.cam.grab_as_array() after_img = self._cam.grab_as_array()
offset = fft_image_tracking.displacement_between_images( offset = fft_image_tracking.displacement_between_images(
image_0=before_img, image_0=before_img,
image_1=after_img, image_1=after_img,

37
tests/test_autofocus.py
View file

@ -48,22 +48,23 @@ def fake_sharpness_data(
def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes, mocker): def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes, mocker):
"""Test the high level looping autofocus algorithm.""" """Test the high level looping autofocus algorithm."""
dz = 2000 dz = 2000
autofocus_thing = create_thing_without_server(AutofocusThing, mock_all_slots=True)
# Make a mock stage where move_absolute abs and relative updates the position counter. # Make a mock stage where move_absolute abs and relative updates the position counter.
stage = mocker.Mock() autofocus_thing._stage.position = {"x": 0, "y": 0, "z": start_z}
stage.position = {"x": 0, "y": 0, "z": start_z}
def set_pos(**kwargs: int) -> None: def set_pos(**kwargs: int) -> None:
"""Move absolute should update position. So make a side effect for the mock.""" """Move absolute should update position. So make a side effect for the mock."""
for axis, value in kwargs.items(): for axis, value in kwargs.items():
stage.position[axis] = value autofocus_thing._stage.position[axis] = value
def adjust_pos(**kwargs: int) -> None: def adjust_pos(**kwargs: int) -> None:
"""Move relative should update position. So make a side effect for the mock.""" """Move relative should update position. So make a side effect for the mock."""
for axis, value in kwargs.items(): for axis, value in kwargs.items():
stage.position[axis] += value autofocus_thing._stage.position[axis] += value
stage.move_absolute.side_effect = set_pos autofocus_thing._stage.move_absolute.side_effect = set_pos
stage.move_relative.side_effect = adjust_pos autofocus_thing._stage.move_relative.side_effect = adjust_pos
# Make a mock sharpness monitor that can generate sharpness data. # Make a mock sharpness monitor that can generate sharpness data.
sharpness_monitor = mocker.MagicMock() sharpness_monitor = mocker.MagicMock()
@ -73,27 +74,25 @@ def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes,
"""Generate sharpnesses based on parameterised input, and mock stage position.""" """Generate sharpnesses based on parameterised input, and mock stage position."""
return fake_sharpness_data( return fake_sharpness_data(
dz=dz, dz=dz,
start_z=stage.position["z"], start_z=autofocus_thing._stage.position["z"],
max_loc=max_loc, max_loc=max_loc,
) )
sharpness_monitor.move_data.side_effect = return_sharpness sharpness_monitor.move_data.side_effect = return_sharpness
autofocus_thing = create_thing_without_server(AutofocusThing) # Mock the context manager call so out MagicMock sharpness monitor is returned.
mock_context = mocker.patch(
"openflexure_microscope_server.things.autofocus.JPEGSharpnessMonitor"
)
mock_context.return_value.__enter__.return_value = sharpness_monitor
mock_context.return_value.__exit__.return_value = None
if passes: if passes:
autofocus_thing.looping_autofocus( autofocus_thing.looping_autofocus(dz=dz, start="centre" if centre else "base")
stage=stage,
sharpness_monitor=sharpness_monitor,
dz=dz,
start="centre" if centre else "base",
)
else: else:
with pytest.raises(NoFocusFoundError): with pytest.raises(NoFocusFoundError):
autofocus_thing.looping_autofocus( autofocus_thing.looping_autofocus(
stage=stage, dz=dz, start="centre" if centre else "base"
sharpness_monitor=sharpness_monitor,
dz=dz,
start="centre" if centre else "base",
) )
assert sharpness_monitor.focus_rel.call_count == attempts_expected assert sharpness_monitor.focus_rel.call_count == attempts_expected
assert abs(max_loc - stage.position["z"]) < dz / 40 assert abs(max_loc - autofocus_thing._stage.position["z"]) < dz / 40

62
tests/test_stack.py
View file

@ -26,7 +26,6 @@ from openflexure_microscope_server.things.autofocus import (
_get_peak_turning_point, _get_peak_turning_point,
) )
LOGGER = logging.getLogger("mock-invocation_logger")
RANDOM_GENERATOR = np.random.default_rng() RANDOM_GENERATOR = np.random.default_rng()
@ -265,7 +264,7 @@ def test_retrieval_of_captures(start):
@pytest.fixture @pytest.fixture
def autofocus_thing(): def autofocus_thing():
"""Return an autofocus thing connected to a server.""" """Return an autofocus thing connected to a server."""
return create_thing_without_server(AutofocusThing) return create_thing_without_server(AutofocusThing, mock_all_slots=True)
def test_create_stack(autofocus_thing, caplog): def test_create_stack(autofocus_thing, caplog):
@ -274,10 +273,7 @@ def test_create_stack(autofocus_thing, caplog):
initial_images_to_save = autofocus_thing.stack_images_to_save initial_images_to_save = autofocus_thing.stack_images_to_save
with caplog.at_level(logging.INFO): with caplog.at_level(logging.INFO):
stack_params = autofocus_thing.create_stack_params( stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000, autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
) )
assert len(caplog.records) == 0 assert len(caplog.records) == 0
@ -304,10 +300,7 @@ def test_coercing_stack_test_ims(
with caplog.at_level(logging.WARNING): with caplog.at_level(logging.WARNING):
stack_params = autofocus_thing.create_stack_params( stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000, autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
) )
assert len(caplog.records) == 1 assert len(caplog.records) == 1
@ -335,10 +328,7 @@ def test_coercing_stack_save_ims(
with caplog.at_level(logging.WARNING): with caplog.at_level(logging.WARNING):
stack_params = autofocus_thing.create_stack_params( stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000, autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
) )
assert len(caplog.records) == 1 assert len(caplog.records) == 1
@ -352,14 +342,8 @@ def test_coercing_stack_save_ims(
@pytest.mark.parametrize("pass_on", [1, 2, 3, 4]) @pytest.mark.parametrize("pass_on", [1, 2, 3, 4])
def test_run_smart_stack(pass_on, autofocus_thing, mocker): def test_run_smart_stack(pass_on, autofocus_thing, mocker):
"""Test Running smart stack with the stack passing on different attempts.""" """Test Running smart stack with the stack passing on different attempts."""
cam = mocker.Mock()
stage = mocker.Mock()
sharpness_monitor = mocker.MagicMock()
stack_params = autofocus_thing.create_stack_params( stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000, autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
) )
assert stack_params.max_attempts == 3 assert stack_params.max_attempts == 3
@ -386,9 +370,6 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker):
# Run it # Run it
success, final_z = autofocus_thing.run_smart_stack( success, final_z = autofocus_thing.run_smart_stack(
cam=cam,
stage=stage,
sharpness_monitor=sharpness_monitor,
stack_parameters=stack_params, stack_parameters=stack_params,
save_on_failure=False, save_on_failure=False,
check_turning_points=True, check_turning_points=True,
@ -403,16 +384,16 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker):
n_stacks = min(pass_on, stack_params.max_attempts) n_stacks = min(pass_on, stack_params.max_attempts)
assert autofocus_thing.z_stack.call_count == n_stacks assert autofocus_thing.z_stack.call_count == n_stacks
# Move absolute should be 1 less time that the number of times z_stack_run # Move absolute should be 1 less time that the number of times z_stack_run
assert stage.move_absolute.call_count == n_stacks - 1 assert autofocus_thing._stage.move_absolute.call_count == n_stacks - 1
# As should looping autofocus # As should looping autofocus
assert autofocus_thing.looping_autofocus.call_count == n_stacks - 1 assert autofocus_thing.looping_autofocus.call_count == n_stacks - 1
# Check rest stack is moving to the first image in the stack. # Check rest stack is moving to the first image in the stack.
if n_stacks > 1: if n_stacks > 1:
assert stage.move_absolute.call_args.kwargs["z"] == -99 assert autofocus_thing._stage.move_absolute.call_args.kwargs["z"] == -99
# Mock called to save image # Mock called to save image
assert cam.save_from_memory.call_count == (1 if success else 0) assert autofocus_thing._cam.save_from_memory.call_count == (1 if success else 0)
def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, mocker): def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, mocker):
@ -423,15 +404,10 @@ def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing,
results). If it a tuple (or anything else), it is set as a return value. results). If it a tuple (or anything else), it is set as a return value.
""" """
stack_params = autofocus_thing.create_stack_params( stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000, autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
) )
stack_params.settling_time = 0 # Don't settle or tests take forever. stack_params.settling_time = 0 # Don't settle or tests take forever.
stage = mocker.Mock()
cam = mocker.Mock()
autofocus_thing.capture_stack_image = mocker.Mock() autofocus_thing.capture_stack_image = mocker.Mock()
if isinstance(check_returns, list): if isinstance(check_returns, list):
autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns) autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns)
@ -440,8 +416,6 @@ def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing,
return autofocus_thing.z_stack( return autofocus_thing.z_stack(
stack_parameters=stack_params, stack_parameters=stack_params,
check_turning_points=check_turning_points, check_turning_points=check_turning_points,
cam=cam,
stage=stage,
) )
@ -506,20 +480,16 @@ def test_z_stack_return(autofocus_thing, mocker):
assert ret[0] assert ret[0]
def test_capture_stack_image(autofocus_thing, mocker): def test_capture_stack_image(autofocus_thing):
"""Check that capture stack image calls the expected functions and returns the expected data.""" """Check that capture stack image calls the expected functions and returns the expected data."""
stage = mocker.Mock() autofocus_thing._stage.position = {"x": 123, "y": 456, "z": 789}
stage.position = {"x": 123, "y": 456, "z": 789} autofocus_thing._cam.capture_to_memory.return_value = "fake_buffer_id"
cam = mocker.Mock() autofocus_thing._cam.grab_jpeg_size.return_value = 54321
cam.capture_to_memory.return_value = "fake_buffer_id"
cam.grab_jpeg_size.return_value = 54321
buffer_max = 11 buffer_max = 11
info = autofocus_thing.capture_stack_image( info = autofocus_thing.capture_stack_image(buffer_max=buffer_max)
cam=cam, stage=stage, buffer_max=buffer_max assert autofocus_thing._cam.capture_to_memory.call_count == 1
) assert autofocus_thing._cam.grab_jpeg_size.call_count == 1
assert cam.capture_to_memory.call_count == 1
assert cam.grab_jpeg_size.call_count == 1
assert info.buffer_id == "fake_buffer_id" assert info.buffer_id == "fake_buffer_id"
assert info.position == {"x": 123, "y": 456, "z": 789} assert info.position == {"x": 123, "y": 456, "z": 789}
assert info.sharpness == 54321 assert info.sharpness == 54321

184
tests/test_stage_measure.py
View file

@ -1,6 +1,5 @@
"""File contains unit tests for stage_measure.""" """File contains unit tests for stage_measure."""
import dataclasses
import logging import logging
from copy import copy from copy import copy
@ -143,23 +142,21 @@ def test_parasitic_detect(par_fraction, too_high):
def test_error_if_no_stream_res_set_when_requesting_img_coords(): def test_error_if_no_stream_res_set_when_requesting_img_coords():
"""Check a RuntimeError thrown when requesting image coordinates if resolution unset.""" """Check a RuntimeError thrown when requesting image coordinates if resolution unset."""
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing) rom_thing = create_thing_without_server(
stage_measure.RangeofMotionThing, mock_all_slots=True
)
with pytest.raises(RuntimeError, match="Stream resolution must be set"): with pytest.raises(RuntimeError, match="Stream resolution must be set"):
rom_thing._img_percentage_to_img_coords(20, "x") rom_thing._img_percentage_to_img_coords(20, "x")
@pytest.fixture @pytest.fixture
def rom_thing(example_rom_data) -> stage_measure.RangeofMotionThing: def rom_thing(example_rom_data, csm_matrix) -> stage_measure.RangeofMotionThing:
"""Yield a RangeofMotionThing already populated with some example rom_data.""" """Yield a RangeofMotionThing already populated with some example rom_data."""
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing) rom_thing = create_thing_without_server(
stage_measure.RangeofMotionThing, mock_all_slots=True
)
rom_thing._stream_resolution = [800, 600] rom_thing._stream_resolution = [800, 600]
rom_thing._rom_data = example_rom_data rom_thing._rom_data = example_rom_data
return 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, **_kwargs: float) -> dict[str, int]: def apply_csm(x: float, y: float, **_kwargs: float) -> dict[str, int]:
"""Convert image coordinates to stage coordinates.""" """Convert image coordinates to stage coordinates."""
@ -169,22 +166,13 @@ def mock_rom_deps(csm_matrix, mocker) -> stage_measure.RomDeps:
"""Convert stage coordinates to image coordinates.""" """Convert stage coordinates to image coordinates."""
return csm_stage_to_img(csm_matrix, x=x, y=y) return csm_stage_to_img(csm_matrix, x=x, y=y)
mock_cam = mocker.Mock() rom_thing._cam.image_is_sample.return_value = (True, "Mocked not measured.")
mock_cam.image_is_sample.return_value = (True, "Mocked not measured.")
mock_csm = mocker.Mock()
# Set up mock csm to return a CSM matrix # Set up mock csm to return a CSM matrix
mock_csm.image_to_stage_displacement_matrix = csm_matrix rom_thing._csm.image_to_stage_displacement_matrix = csm_matrix
mock_csm.convert_image_to_stage_coordinates.side_effect = apply_csm rom_thing._csm.convert_image_to_stage_coordinates.side_effect = apply_csm
mock_csm.convert_stage_to_image_coordinates.side_effect = un_apply_csm rom_thing._csm.convert_stage_to_image_coordinates.side_effect = un_apply_csm
return rom_thing
return stage_measure.RomDeps(
autofocus=mocker.Mock(),
stage=mocker.Mock(),
cam=mock_cam,
csm=mock_csm,
logger=LOGGER,
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -197,22 +185,18 @@ def mock_rom_deps(csm_matrix, mocker) -> stage_measure.RomDeps:
({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 1), ({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 1),
], ],
) )
def test_img_dir_from_stage_coords( def test_img_dir_from_stage_coords(pos, target_pos, axis, expected_dir, rom_thing):
pos, target_pos, axis, expected_dir, rom_thing, mock_rom_deps
):
"""Check image direction is correctly generated.""" """Check image direction is correctly generated."""
mock_rom_deps.stage.position = pos rom_thing._stage.position = pos
direction = rom_thing._img_dir_from_stage_coords(target_pos, axis, mock_rom_deps) direction = rom_thing._img_dir_from_stage_coords(target_pos, axis)
assert direction == expected_dir assert direction == expected_dir
def test_distance_in_img_percentage_err(rom_thing, mock_rom_deps): def test_distance_in_img_percentage_err(rom_thing):
"""Check that _distance_in_img_percentage throws error if stream res is not set.""" """Check that _distance_in_img_percentage throws error if stream res is not set."""
rom_thing._stream_resolution = None rom_thing._stream_resolution = None
with pytest.raises(RuntimeError): with pytest.raises(RuntimeError):
rom_thing._distance_in_img_percentage( rom_thing._distance_in_img_percentage({"x": 0, "y": 0, "z": 0}, "x")
{"x": 0, "y": 0, "z": 0}, "x", mock_rom_deps
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -225,16 +209,14 @@ def test_distance_in_img_percentage_err(rom_thing, mock_rom_deps):
({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 462.131), ({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 462.131),
], ],
) )
def test_distance_in_img_percentage( def test_distance_in_img_percentage(pos, target_pos, axis, expected_perc, rom_thing):
pos, target_pos, axis, expected_perc, rom_thing, mock_rom_deps
):
"""Check _distance_in_img_percentage calculates expected values based on mock CSM.""" """Check _distance_in_img_percentage calculates expected values based on mock CSM."""
mock_rom_deps.stage.position = pos rom_thing._stage.position = pos
img_perc = rom_thing._distance_in_img_percentage(target_pos, axis, mock_rom_deps) img_perc = rom_thing._distance_in_img_percentage(target_pos, axis)
assert round(img_perc, 3) == expected_perc assert round(img_perc, 3) == expected_perc
def test_offset_from(rom_thing, mock_rom_deps, mocker): def test_offset_from(rom_thing, mocker):
"""Check the calls and returns for RangeofMotionThing._offset_from.""" """Check the calls and returns for RangeofMotionThing._offset_from."""
# Set up mock for the FFT displacement # Set up mock for the FFT displacement
disp_between_route = ( disp_between_route = (
@ -247,14 +229,14 @@ def test_offset_from(rom_thing, mock_rom_deps, mocker):
) )
# Run it # Run it
offset = rom_thing._offset_from(before_img="MOCK_IMAGE", rom_deps=mock_rom_deps) offset = rom_thing._offset_from(before_img="MOCK_IMAGE")
# Check the offset is a dictionary with the correct values for the axes # Check the offset is a dictionary with the correct values for the axes
assert offset["x"] == 456 assert offset["x"] == 456
assert offset["y"] == 123 assert offset["y"] == 123
# Check 1 image was taken # Check 1 image was taken
assert mock_rom_deps.cam.grab_as_array.call_count == 1 assert rom_thing._cam.grab_as_array.call_count == 1
mock_after_image = mock_rom_deps.cam.grab_as_array.return_value mock_after_image = rom_thing._cam.grab_as_array.return_value
# Check the FFT displacement was called once # Check the FFT displacement was called once
assert mock_disp_between.call_count == 1 assert mock_disp_between.call_count == 1
displacement_kwargs = mock_disp_between.call_args.kwargs displacement_kwargs = mock_disp_between.call_args.kwargs
@ -264,7 +246,7 @@ def test_offset_from(rom_thing, mock_rom_deps, mocker):
@pytest.mark.parametrize("perform_autofocus", [True, False]) @pytest.mark.parametrize("perform_autofocus", [True, False])
def test_move_and_measure(perform_autofocus, rom_thing, mock_rom_deps, mocker): def test_move_and_measure(perform_autofocus, rom_thing, mocker):
"""Test _move_and_measure with and without initial autofocus checking call counts. """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 This doesn't test with the repeate autofocus if motion isn't detected
@ -273,16 +255,16 @@ def test_move_and_measure(perform_autofocus, rom_thing, mock_rom_deps, mocker):
mocker.patch.object(rom_thing, "_offset_from", return_value=mock_offset_value) mocker.patch.object(rom_thing, "_offset_from", return_value=mock_offset_value)
movement = {"x": 100, "y": 0} movement = {"x": 100, "y": 0}
offset = rom_thing._move_and_measure( offset = rom_thing._move_and_measure(
movement=movement, rom_deps=mock_rom_deps, perform_autofocus=perform_autofocus movement=movement, perform_autofocus=perform_autofocus
) )
# Check exactly 1 move # Check exactly 1 move
assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 assert rom_thing._csm.move_in_image_coordinates.call_count == 1
# The kwargs of the call should movement dict # The kwargs of the call should movement dict
assert mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs == movement assert rom_thing._csm.move_in_image_coordinates.call_args.kwargs == movement
# Check autofocus call count # Check autofocus call count
expected_af_count = 1 if perform_autofocus else 0 expected_af_count = 1 if perform_autofocus else 0
assert mock_rom_deps.autofocus.looping_autofocus.call_count == expected_af_count assert rom_thing._autofocus.looping_autofocus.call_count == expected_af_count
# And check final return # And check final return
assert offset == mock_offset_value assert offset == mock_offset_value
@ -300,7 +282,7 @@ def test_move_and_measure(perform_autofocus, rom_thing, mock_rom_deps, mocker):
], ],
) )
def test_move_and_measure_with_refocus( def test_move_and_measure_with_refocus(
x_offsets, n_offset_measures, expected_return, rom_thing, mock_rom_deps, mocker x_offsets, n_offset_measures, expected_return, rom_thing, mocker
): ):
"""Test _move_and_measure with final refocus if offset is too small.""" """Test _move_and_measure with final refocus if offset is too small."""
return_dicts = tuple({"x": x, "y": 0} for x in x_offsets) return_dicts = tuple({"x": x, "y": 0} for x in x_offsets)
@ -310,21 +292,20 @@ def test_move_and_measure_with_refocus(
movement = {"x": 10, "y": 0} movement = {"x": 10, "y": 0}
offset = rom_thing._move_and_measure( offset = rom_thing._move_and_measure(
movement=movement, movement=movement,
rom_deps=mock_rom_deps,
perform_autofocus=False, perform_autofocus=False,
max_autofocus_repeats=3, max_autofocus_repeats=3,
abs_min_offset=5, abs_min_offset=5,
) )
# Check exactly 1 move # Check exactly 1 move
assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 assert rom_thing._csm.move_in_image_coordinates.call_count == 1
# Check expected _offset_from calls # Check expected _offset_from calls
assert offset_from_mock.call_count == n_offset_measures assert offset_from_mock.call_count == n_offset_measures
# The kwargs of the call should movement dict # The kwargs of the call should movement dict
assert mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs == movement assert rom_thing._csm.move_in_image_coordinates.call_args.kwargs == movement
# Check autofocus call count is 1 less than number of offset measures as no autofocus # Check autofocus call count is 1 less than number of offset measures as no autofocus
# is performed before the first one # is performed before the first one
expected_af_count = n_offset_measures - 1 expected_af_count = n_offset_measures - 1
assert mock_rom_deps.autofocus.looping_autofocus.call_count == expected_af_count assert rom_thing._autofocus.looping_autofocus.call_count == expected_af_count
# And check final return # And check final return
assert offset == expected_return assert offset == expected_return
@ -339,7 +320,6 @@ def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
with pytest.raises(ValueError, match="abs_min_offset must be positive"): with pytest.raises(ValueError, match="abs_min_offset must be positive"):
rom_thing._move_and_measure( rom_thing._move_and_measure(
movement={"x": 10, "y": 0}, movement={"x": 10, "y": 0},
rom_deps=mock_rom_deps,
perform_autofocus=False, perform_autofocus=False,
max_autofocus_repeats=3, max_autofocus_repeats=3,
) )
@ -348,7 +328,6 @@ def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
with pytest.raises(ValueError, match="abs_min_offset must be positive"): with pytest.raises(ValueError, match="abs_min_offset must be positive"):
rom_thing._move_and_measure( rom_thing._move_and_measure(
movement={"x": 10, "y": 0}, movement={"x": 10, "y": 0},
rom_deps=mock_rom_deps,
perform_autofocus=False, perform_autofocus=False,
max_autofocus_repeats=3, max_autofocus_repeats=3,
abs_min_offset=-123.456, abs_min_offset=-123.456,
@ -357,7 +336,7 @@ def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
assert offset_from_mock.call_count == 0 assert offset_from_mock.call_count == 0
def test_move_back_until_motion_detected(rom_thing, mock_rom_deps, mocker): def test_move_back_until_motion_detected(rom_thing, mocker):
"""Check that _move_back_until_motion_detected is making increasing negative moves. """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 The moves for this method should be in opposite direction to the direction
@ -368,7 +347,7 @@ def test_move_back_until_motion_detected(rom_thing, mock_rom_deps, mocker):
) )
with pytest.raises(RuntimeError, match="Cannot detect motion again"): with pytest.raises(RuntimeError, match="Cannot detect motion again"):
rom_thing._move_back_until_motion_detected("y", -1, rom_deps=mock_rom_deps) rom_thing._move_back_until_motion_detected("y", -1)
max_tries = int(1.5 * stage_measure.BIG_STEP / stage_measure.SMALL_STEP) max_tries = int(1.5 * stage_measure.BIG_STEP / stage_measure.SMALL_STEP)
expected_step = 800 * stage_measure.SMALL_STEP / 100 expected_step = 800 * stage_measure.SMALL_STEP / 100
@ -386,7 +365,7 @@ def test_move_back_until_motion_detected(rom_thing, mock_rom_deps, mocker):
) )
# Other axis and direction this time # Other axis and direction this time
rom_thing._move_back_until_motion_detected("x", 1, rom_deps=mock_rom_deps) rom_thing._move_back_until_motion_detected("x", 1)
# Should only be called 3 times # Should only be called 3 times
assert mock_move_n_meas.call_count == 3 assert mock_move_n_meas.call_count == 3
@ -417,7 +396,6 @@ def test_stage_still_moves(
expected_to_detect_motion, expected_to_detect_motion,
offset_calls, offset_calls,
rom_thing, rom_thing,
mock_rom_deps,
mocker, mocker,
): ):
"""Test _stage_still_moves correctly detects stage movement.""" """Test _stage_still_moves correctly detects stage movement."""
@ -435,38 +413,34 @@ def test_stage_still_moves(
rom_thing, "_offset_from", side_effect=gen_offsets() rom_thing, "_offset_from", side_effect=gen_offsets()
) )
still_moves = rom_thing._stage_still_moves( still_moves = rom_thing._stage_still_moves(axis="x", direction=1)
axis="x",
direction=1,
rom_deps=mock_rom_deps,
)
assert still_moves is expected_to_detect_motion assert still_moves is expected_to_detect_motion
assert mock_offset_from.call_count == offset_calls assert mock_offset_from.call_count == offset_calls
def test_big_z_corrected_movement(rom_thing, mock_rom_deps): def test_big_z_corrected_movement(rom_thing):
"""Check big z corrected move moves in x/y and z the expected distances.""" """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._stage.position = {"x": 5000, "y": 30, "z": 500}
rom_thing._big_z_corrected_movement("x", direction=1, rom_deps=mock_rom_deps) rom_thing._big_z_corrected_movement("x", direction=1)
expected_movement = {"x": 800 * stage_measure.BIG_STEP / 100, "y": 0} expected_movement = {"x": 800 * stage_measure.BIG_STEP / 100, "y": 0}
# Check there is one z move in steps # Check there is one z move in steps
assert mock_rom_deps.stage.move_relative.call_count == 1 assert rom_thing._stage.move_relative.call_count == 1
move_kwargs = mock_rom_deps.stage.move_relative.call_args.kwargs move_kwargs = rom_thing._stage.move_relative.call_args.kwargs
assert "x" not in move_kwargs assert "x" not in move_kwargs
assert "y" not in move_kwargs assert "y" not in move_kwargs
assert "z" in move_kwargs assert "z" in move_kwargs
assert move_kwargs["z"] == 160 assert move_kwargs["z"] == 160
# And one move in image coordinates # And one move in image coordinates
assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 1 assert rom_thing._csm.move_in_image_coordinates.call_count == 1
lat_mov_kwargs = mock_rom_deps.csm.move_in_image_coordinates.call_args.kwargs lat_mov_kwargs = rom_thing._csm.move_in_image_coordinates.call_args.kwargs
assert lat_mov_kwargs == expected_movement assert lat_mov_kwargs == expected_movement
def test_moves_for_z_prediction(rom_thing, mock_rom_deps, mocker): def test_moves_for_z_prediction(rom_thing, mocker):
"""Check the initial moves are of the correct size and are recorded.""" """Check the initial moves are of the correct size and are recorded."""
# Mock the _offset_from and stage.position to return generated dictionaries that # 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 # increment each time they are called. (All values 0 the first time, all values 1
@ -474,7 +448,7 @@ def test_moves_for_z_prediction(rom_thing, mock_rom_deps, mocker):
mocker.patch.object( mocker.patch.object(
rom_thing, "_offset_from", side_effect=increasing_xy_dict_generator() rom_thing, "_offset_from", side_effect=increasing_xy_dict_generator()
) )
type(mock_rom_deps.stage).position = mocker.PropertyMock( type(rom_thing._stage).position = mocker.PropertyMock(
side_effect=increasing_xyz_dict_generator() side_effect=increasing_xyz_dict_generator()
) )
@ -484,7 +458,7 @@ def test_moves_for_z_prediction(rom_thing, mock_rom_deps, mocker):
rom_thing._rom_data = stage_measure.RomDataTracker() rom_thing._rom_data = stage_measure.RomDataTracker()
# Run it! # Run it!
rom_thing._moves_for_z_prediction("x", direction=-1, rom_deps=mock_rom_deps) rom_thing._moves_for_z_prediction("x", direction=-1)
# Check that _rom_data now contains the 5 mocked returns in order. # 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.offsets == [{"x": i, "y": i} for i in range(5)]
@ -493,16 +467,16 @@ def test_moves_for_z_prediction(rom_thing, mock_rom_deps, mocker):
] ]
# Check that the csm movement function is called 5 times # Check that the csm movement function is called 5 times
assert mock_rom_deps.csm.move_in_image_coordinates.call_count == 5 assert rom_thing._csm.move_in_image_coordinates.call_count == 5
# Each time with the expected movement # Each time with the expected movement
for arg_list in mock_rom_deps.csm.move_in_image_coordinates.call_args_list: for arg_list in rom_thing._csm.move_in_image_coordinates.call_args_list:
assert arg_list.kwargs == expected_movement assert arg_list.kwargs == expected_movement
def test_move_until_edge_error(rom_thing, mock_rom_deps, mocker): def test_move_until_edge_error(rom_thing, mocker):
"""Check that if there is an error while moving to the edge the stage returns to start.""" """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} mock_position_dict = {"x": 123, "y": 456, "z": 789}
type(mock_rom_deps.stage).position = mocker.PropertyMock( type(rom_thing._stage).position = mocker.PropertyMock(
return_value=mock_position_dict return_value=mock_position_dict
) )
mocker.patch.object( mocker.patch.object(
@ -514,20 +488,20 @@ def test_move_until_edge_error(rom_thing, mock_rom_deps, mocker):
# Error should be raised even though it is in the Try: # Error should be raised even though it is in the Try:
with pytest.raises(RuntimeError, match="Mock"): with pytest.raises(RuntimeError, match="Mock"):
rom_thing._move_until_edge("y", direction=-1, rom_deps=mock_rom_deps) rom_thing._move_until_edge("y", direction=-1)
# However the "finally" should have executed returning to the starting position # However the "finally" should have executed returning to the starting position
assert mock_rom_deps.stage.move_absolute.call_count == 1 assert rom_thing._stage.move_absolute.call_count == 1
abs_move_kwargs = mock_rom_deps.stage.move_absolute.call_args.kwargs abs_move_kwargs = rom_thing._stage.move_absolute.call_args.kwargs
expected_abs_move_kwargs = dict(**mock_position_dict, block_cancellation=True) expected_abs_move_kwargs = dict(**mock_position_dict, block_cancellation=True)
assert abs_move_kwargs == expected_abs_move_kwargs assert abs_move_kwargs == expected_abs_move_kwargs
def test_move_until_edge(rom_thing, mock_rom_deps, mocker): def test_move_until_edge(rom_thing, mocker):
"""Check move until edge runs the correct movement sequence.""" """Check move until edge runs the correct movement sequence."""
# Remove the mock RomData before starting # Remove the mock RomData before starting
rom_thing._rom_data = stage_measure.RomDataTracker() rom_thing._rom_data = stage_measure.RomDataTracker()
mock_rom_deps.stage.position = {"x": "mock", "y": "starting", "z": "pos"} rom_thing._stage.position = {"x": "mock", "y": "starting", "z": "pos"}
def add_fake_initial_positions(*_args, **_kwargs): def add_fake_initial_positions(*_args, **_kwargs):
"""Rather than run initial moves just add some fake data.""" """Rather than run initial moves just add some fake data."""
@ -538,13 +512,13 @@ def test_move_until_edge(rom_thing, mock_rom_deps, mocker):
"""With big moves update the stage position.""" """With big moves update the stage position."""
big_move_count = 1 big_move_count = 1
while True: while True:
mock_rom_deps.stage.position = f"mocked-big-move-pos{big_move_count}" rom_thing._stage.position = f"mocked-big-move-pos{big_move_count}"
yield yield
big_move_count += 1 big_move_count += 1
def set_final_pos(*_args, **_kwargs): def set_final_pos(*_args, **_kwargs):
"""When move_back_until_motion_detected is run set a final stage position.""" """When move_back_until_motion_detected is run set a final stage position."""
mock_rom_deps.stage.position = "mock-final-pos" rom_thing._stage.position = "mock-final-pos"
# Mock the main movement functions # Mock the main movement functions
mock_init_moves = mocker.patch.object( mock_init_moves = mocker.patch.object(
@ -568,7 +542,7 @@ def test_move_until_edge(rom_thing, mock_rom_deps, mocker):
rom_thing._rom_data = stage_measure.RomDataTracker() rom_thing._rom_data = stage_measure.RomDataTracker()
# Run function # Run function
rom_thing._move_until_edge("y", direction=-1, rom_deps=mock_rom_deps) rom_thing._move_until_edge("y", direction=-1)
# Check the call counts are as expected # Check the call counts are as expected
# One call of initial moves # One call of initial moves
@ -600,19 +574,19 @@ def test_move_until_edge(rom_thing, mock_rom_deps, mocker):
] ]
def test_perform_rom_actions_locked(rom_thing, mock_rom_deps): def test_perform_rom_actions_locked(rom_thing):
"""Check the error if running one of the calibration actions while locked.""" """Check the error if running one of the calibration actions while locked."""
# Not an RLock so no need to thread. # Not an RLock so no need to thread.
rom_thing._lock.acquire() rom_thing._lock.acquire()
err_msg = "Trying to run ROM test when a test is already running." err_msg = "Trying to run ROM test when a test is already running."
with pytest.raises(RuntimeError, match=err_msg): with pytest.raises(RuntimeError, match=err_msg):
rom_thing.perform_rom_test(**dataclasses.asdict(mock_rom_deps)) rom_thing.perform_rom_test()
err_msg = "Trying to run recentre when a test is already running." err_msg = "Trying to run recentre when a test is already running."
with pytest.raises(RuntimeError, match=err_msg): with pytest.raises(RuntimeError, match=err_msg):
rom_thing.perform_recentre(**dataclasses.asdict(mock_rom_deps)) rom_thing.perform_recentre()
def test_perform_rom_test_(rom_thing, mock_rom_deps, mocker): def test_perform_rom_test_(rom_thing, mocker):
"""Check that perform Rom Test runs the expected high level algorithm.""" """Check that perform Rom Test runs the expected high level algorithm."""
def check_and_modify_rom_data(axis: str, direction: int, **_kwargs): def check_and_modify_rom_data(axis: str, direction: int, **_kwargs):
@ -629,9 +603,9 @@ def test_perform_rom_test_(rom_thing, mock_rom_deps, mocker):
) )
mocker.patch.object( mocker.patch.object(
mock_rom_deps.cam, "grab_as_array", return_value=np.zeros([123, 456, 3]) rom_thing._cam, "grab_as_array", return_value=np.zeros([123, 456, 3])
) )
final_dict = rom_thing.perform_rom_test(**dataclasses.asdict(mock_rom_deps)) final_dict = rom_thing.perform_rom_test()
# This should take less than 1 sec # This should take less than 1 sec
assert final_dict["Time"] <= 1 assert final_dict["Time"] <= 1
@ -659,7 +633,7 @@ def test_perform_rom_test_(rom_thing, mock_rom_deps, mocker):
assert mock_move_until_edge.call_args_list[3].kwargs["direction"] == -1 assert mock_move_until_edge.call_args_list[3].kwargs["direction"] == -1
def test_perform_recenter(rom_thing, mock_rom_deps, mocker, caplog): def test_perform_recenter(rom_thing, mocker, caplog):
"""Check that performing the recentre runs through expected operations.""" """Check that performing the recentre runs through expected operations."""
def check_lock(*_args, **_kwargs): def check_lock(*_args, **_kwargs):
@ -673,14 +647,10 @@ def test_perform_recenter(rom_thing, mock_rom_deps, mocker, caplog):
# Set a mock stage position, as we patch _recentre_axis this should be logged # Set a mock stage position, as we patch _recentre_axis this should be logged
# as the centre. # as the centre.
mock_rom_deps.stage.position = {"x": 4321, "y": 1234, "z": 0} rom_thing._stage.position = {"x": 4321, "y": 1234, "z": 0}
# Unpack the dict and save because it the dictionary is a copy so we can't check
# mocks from `mock_rom_deps`
mock_deps_dict = dataclasses.asdict(mock_rom_deps)
with caplog.at_level(logging.INFO): with caplog.at_level(logging.INFO):
rom_thing.perform_recentre(**mock_deps_dict) rom_thing.perform_recentre()
assert len(caplog.messages) == 3 assert len(caplog.messages) == 3
assert caplog.messages[0] == "Recentring the stage." assert caplog.messages[0] == "Recentring the stage."
@ -699,17 +669,17 @@ def test_perform_recenter(rom_thing, mock_rom_deps, mocker, caplog):
assert mock_recentre_axis.call_args_list[1].args[0] == "y" assert mock_recentre_axis.call_args_list[1].args[0] == "y"
# check that the position set to zero once # check that the position set to zero once
assert mock_deps_dict["stage"].set_zero_position.call_count == 1 assert rom_thing._stage.set_zero_position.call_count == 1
@pytest.mark.parametrize("true_on", [1, 4, 10, 11]) @pytest.mark.parametrize("true_on", [1, 4, 10, 11])
def test_recentre_axis(true_on, rom_thing, mock_rom_deps, mocker): def test_recentre_axis(true_on, rom_thing, mocker):
"""Test the high level algorithm of recentring an axis. """Test the high level algorithm of recentring an axis.
This doesn't include deciding if we are centred or choosing the direction to move. This doesn't include deciding if we are centred or choosing the direction to move.
""" """
## Start such that movement starts negative ## Start such that movement starts negative
mock_rom_deps.stage.position = {"x": 10000, "y": 10000, "z": 0} rom_thing._stage.position = {"x": 10000, "y": 10000, "z": 0}
mock_moves = mocker.patch.object(rom_thing, "_moves_for_z_prediction") mock_moves = mocker.patch.object(rom_thing, "_moves_for_z_prediction")
# Mock _recentre_decision so it returns (False, 1) a number of times then finally # Mock _recentre_decision so it returns (False, 1) a number of times then finally
@ -721,9 +691,9 @@ def test_recentre_axis(true_on, rom_thing, mock_rom_deps, mocker):
if true_on > 10: if true_on > 10:
with pytest.raises(RuntimeError, match="Couldn't find centre"): with pytest.raises(RuntimeError, match="Couldn't find centre"):
rom_thing._recentre_axis("x", mock_rom_deps) rom_thing._recentre_axis("x")
else: else:
rom_thing._recentre_axis("x", mock_rom_deps) rom_thing._recentre_axis("x")
# _recentre_decision is called until True or exits after 10 attempts # _recentre_decision is called until True or exits after 10 attempts
assert mock_recentre_decision.call_count == min(true_on, 10) assert mock_recentre_decision.call_count == min(true_on, 10)
@ -748,9 +718,7 @@ def test_recentre_axis(true_on, rom_thing, mock_rom_deps, mocker):
(-199, -1, (True, 1)), # Always return 1 when c (-199, -1, (True, 1)), # Always return 1 when c
], ],
) )
def test_recentre_decision( def test_recentre_decision(dist, direction, expected_decision, rom_thing, mocker):
dist, direction, expected_decision, rom_thing, mock_rom_deps, mocker
):
"""What the algorithm decides to do in different situations.""" """What the algorithm decides to do in different situations."""
# Mock find turning point just because there is no _rom_data # Mock find turning point just because there is no _rom_data
mocker.patch.object(rom_thing._rom_data, "find_turning_point") mocker.patch.object(rom_thing._rom_data, "find_turning_point")
@ -760,8 +728,8 @@ def test_recentre_decision(
mocker.patch.object(rom_thing, "_distance_in_img_percentage", return_value=dist) mocker.patch.object(rom_thing, "_distance_in_img_percentage", return_value=dist)
mocker.patch.object(rom_thing, "_img_dir_from_stage_coords", return_value=direction) mocker.patch.object(rom_thing, "_img_dir_from_stage_coords", return_value=direction)
assert rom_thing._recentre_decision("x", mock_rom_deps) == expected_decision assert rom_thing._recentre_decision("x") == expected_decision
# Check that we make an absolute move (to the centre) before returning centred. # Check that we make an absolute move (to the centre) before returning centred.
centred = expected_decision[0] centred = expected_decision[0]
mock_rom_deps.stage.move_absolute.call_count == 1 if centred else 0 rom_thing._stage.move_absolute.call_count == 1 if centred else 0