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

Remove deps from stage_measure

Browse source
This commit is contained in:
Julian Stirling 2025-12-16 16:37:25 +00:00
parent 189e82e8dc
commit 7f8fb623ef
1 changed files with 86 additions and 179 deletions
Download patch file Download diff file Expand all files Collapse all files

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,