Change RangeofMotionThing to StageMeasureThing
Also changing the data tracker and name.
This commit is contained in:
parent
3e38722d45
commit
6a571616c9
3 changed files with 209 additions and 164 deletions
|
|
@ -16,7 +16,7 @@
|
|||
"histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow",
|
||||
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
|
||||
"raster_workflow": "openflexure_microscope_server.things.scan_workflows:RasterWorkflow",
|
||||
"stage_measure": "openflexure_microscope_server.things.stage_measure:RangeofMotionThing",
|
||||
"stage_measure": "openflexure_microscope_server.things.stage_measure:StageMeasurementThing",
|
||||
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
|
||||
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"
|
||||
},
|
||||
|
|
|
|||
|
|
@ -1,16 +1,10 @@
|
|||
"""File contains all the functions used to measure the range of motion.
|
||||
"""File contains all the functions used to measure the stage against the camera.
|
||||
|
||||
The range of motion is measured by first taking 5 'medium' sized steps which gives
|
||||
enough positions to predict future z positions. Next, one 'big' step is taken followed
|
||||
by 3 'small' steps to test that the stage is still moving as expected and has not
|
||||
reached the edge. Once the edge has been found and to account for the possibility that
|
||||
the edge was reached during a "big" step, the stage is moved in a sequence of steps
|
||||
in the opposite direction until motion is detected. This is position is taken as the
|
||||
true final position.
|
||||
This includes:
|
||||
|
||||
Throughout the test, parasitic motion (motion in the axis not being measured)
|
||||
is tracked and an error is raised if it exceeds a minimum amount. Currently
|
||||
this is 10% of the expected motion is the measured axis.
|
||||
* Calibrating the relationship between camera and stage coordinates.
|
||||
* Finding the centre of motion
|
||||
* Measuring the range of motion
|
||||
"""
|
||||
|
||||
import json
|
||||
|
|
@ -43,8 +37,8 @@ PARASITIC_MOTION_TOL = 0.1
|
|||
DETECT_MOTION_TOL = 0.65
|
||||
|
||||
|
||||
class RomDataTracker:
|
||||
"""Class for tracking range of motion data.
|
||||
class CamStageTracker:
|
||||
"""Class for tracking camera and stage movement together.
|
||||
|
||||
The attributes storing data are:
|
||||
|
||||
|
|
@ -119,7 +113,7 @@ class ParasiticMotionError(lt.exceptions.InvocationError):
|
|||
"""
|
||||
|
||||
|
||||
class RangeofMotionThing(OFMThing):
|
||||
class StageMeasurementThing(OFMThing):
|
||||
"""A class used to measure the range of motion of the stage in X and Y."""
|
||||
|
||||
_class_settings = {"validate_properties_on_set": True}
|
||||
|
|
@ -138,12 +132,24 @@ class RangeofMotionThing(OFMThing):
|
|||
super().__init__(thing_server_interface)
|
||||
self._lock = Lock()
|
||||
self._stream_resolution: Optional[tuple[int, int]] = None
|
||||
self._rom_data = RomDataTracker()
|
||||
self._cam_stage_data = CamStageTracker()
|
||||
|
||||
@lt.action
|
||||
def perform_rom_test(self) -> Mapping[str, Any]:
|
||||
"""Measures the range of motion of the stage across the x and y axes.
|
||||
|
||||
The range of motion is measured by first taking 5 'medium' sized steps which
|
||||
gives enough positions to predict future z positions. Next, one 'big' step is
|
||||
taken followed by 3 'small' steps to test that the stage is still moving as
|
||||
expected and has not reached the edge. Once the edge has been found and to
|
||||
account for the possibility that the edge was reached during a "big" step, the
|
||||
stage is moved in a sequence of steps in the opposite direction until motion is
|
||||
detected. This is position is taken as the final position.
|
||||
|
||||
Throughout the test, parasitic motion (motion in the axis not being measured)
|
||||
is tracked and an error is raised if it exceeds a minimum amount. Currently
|
||||
this is 10% of the expected motion is the measured axis.
|
||||
|
||||
:return: Results dictionary separated into keys of each axis and direction.
|
||||
"""
|
||||
got_lock = self._lock.acquire(blocking=False)
|
||||
|
|
@ -175,13 +181,15 @@ class RangeofMotionThing(OFMThing):
|
|||
axis_limits = []
|
||||
for axis_dir in directions:
|
||||
# Create a new tracker at start of measurement.
|
||||
self._rom_data = RomDataTracker()
|
||||
self._cam_stage_data = CamStageTracker()
|
||||
self._move_until_edge(axis=axis, direction=axis_dir)
|
||||
# Append final position from tracker
|
||||
axis_limits.append(self._rom_data.final_position[axis])
|
||||
axis_limits.append(self._cam_stage_data.final_position[axis])
|
||||
axis_string = "positive" if axis_dir == 1 else "negative"
|
||||
stage_dic[f"{axis}_{axis_string}"] = self._rom_data.stage_coords
|
||||
cor_dic[f"{axis}_{axis_string}"] = self._rom_data.offsets
|
||||
stage_dic[f"{axis}_{axis_string}"] = (
|
||||
self._cam_stage_data.stage_coords
|
||||
)
|
||||
cor_dic[f"{axis}_{axis_string}"] = self._cam_stage_data.offsets
|
||||
# Calculate step range.
|
||||
step_range.append(abs(axis_limits[0] - axis_limits[1]))
|
||||
|
||||
|
|
@ -261,7 +269,7 @@ class RangeofMotionThing(OFMThing):
|
|||
# Start by performing an autofocus and recording starting position
|
||||
self._autofocus.looping_autofocus(dz=1000)
|
||||
starting_position = self._stage.position
|
||||
self._rom_data.stage_coords.append(starting_position)
|
||||
self._cam_stage_data.stage_coords.append(starting_position)
|
||||
|
||||
try:
|
||||
dir_str = "positive" if direction == 1 else "negative"
|
||||
|
|
@ -278,7 +286,7 @@ class RangeofMotionThing(OFMThing):
|
|||
self._big_z_corrected_movement(axis=axis, direction=direction)
|
||||
# Autofocus and record position
|
||||
self._autofocus.looping_autofocus(dz=800)
|
||||
self._rom_data.stage_coords.append(self._stage.position)
|
||||
self._cam_stage_data.stage_coords.append(self._stage.position)
|
||||
|
||||
# Perform small moves to check stage is still moving.
|
||||
still_moving = self._stage_still_moves(axis=axis, direction=direction)
|
||||
|
|
@ -289,7 +297,7 @@ class RangeofMotionThing(OFMThing):
|
|||
self._move_back_until_motion_detected(axis=axis, direction=direction)
|
||||
|
||||
# Replace final position.
|
||||
self._rom_data.stage_coords[-1] = self._stage.position
|
||||
self._cam_stage_data.stage_coords[-1] = self._stage.position
|
||||
|
||||
finally:
|
||||
self._stage.move_absolute(
|
||||
|
|
@ -303,7 +311,7 @@ class RangeofMotionThing(OFMThing):
|
|||
"""
|
||||
self.logger.info(f"Finding centre in {axis}.")
|
||||
# A new tracker for this axis.
|
||||
self._rom_data = RomDataTracker()
|
||||
self._cam_stage_data = CamStageTracker()
|
||||
# Direction is in image coords, initial assumption is that centre is at (0,0).
|
||||
direction = self._img_dir_from_stage_coords({"x": 0, "y": 0, "z": 0}, axis)
|
||||
|
||||
|
|
@ -313,7 +321,7 @@ class RangeofMotionThing(OFMThing):
|
|||
# Find z then make a number of moves (autofocussing and logging position)
|
||||
# 5 moves initially (2 subsequently).
|
||||
self._autofocus.looping_autofocus(dz=1000)
|
||||
self._rom_data.stage_coords.append(self._stage.position)
|
||||
self._cam_stage_data.stage_coords.append(self._stage.position)
|
||||
self._moves_for_z_prediction(
|
||||
axis=axis,
|
||||
direction=direction,
|
||||
|
|
@ -348,7 +356,7 @@ class RangeofMotionThing(OFMThing):
|
|||
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.
|
||||
"""
|
||||
estimate = self._rom_data.find_turning_point(axis=axis)
|
||||
estimate = self._cam_stage_data.find_turning_point(axis=axis)
|
||||
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
|
||||
|
|
@ -462,7 +470,7 @@ class RangeofMotionThing(OFMThing):
|
|||
for _loop in range(n_moves):
|
||||
offset = self._move_and_measure(movement=movement)
|
||||
|
||||
self._rom_data.record_movement(self._stage.position, offset)
|
||||
self._cam_stage_data.record_movement(self._stage.position, offset)
|
||||
if _parasitic_motion_detected(movement, offset):
|
||||
raise ParasiticMotionError(
|
||||
"Parasitic motion detected during images to calculate z-curvature. "
|
||||
|
|
@ -486,7 +494,7 @@ class RangeofMotionThing(OFMThing):
|
|||
# Convert to stage coordinates
|
||||
stage_movement = self._csm.convert_image_to_stage_coordinates(**movement)
|
||||
|
||||
z_disp = self._rom_data.predict_z_displacement(
|
||||
z_disp = self._cam_stage_data.predict_z_displacement(
|
||||
axis=axis,
|
||||
stage_movement=stage_movement,
|
||||
stage_position=self._stage.position,
|
||||
|
|
@ -527,7 +535,7 @@ class RangeofMotionThing(OFMThing):
|
|||
)
|
||||
parasitic_motion = _parasitic_motion_detected(movement, offset)
|
||||
self.logger.info(f"Offset measured as {int(offset[axis])}")
|
||||
self._rom_data.record_movement(self._stage.position, offset)
|
||||
self._cam_stage_data.record_movement(self._stage.position, offset)
|
||||
|
||||
if abs(offset[axis]) < abs_min_offset or parasitic_motion:
|
||||
# If the offset is below the abs min offset, or significant
|
||||
|
|
|
|||
|
|
@ -51,8 +51,8 @@ def csm_matrix():
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def example_rom_data():
|
||||
"""Return some example data in a RomDataTracker."""
|
||||
def example_cam_stage_data():
|
||||
"""Return some example data in a CamStageTracker."""
|
||||
mock_positions = [
|
||||
{"x": 0, "y": 0, "z": 42},
|
||||
{"x": 727, "y": 2, "z": 154},
|
||||
|
|
@ -61,18 +61,18 @@ def example_rom_data():
|
|||
{"x": 2908, "y": 8, "z": 430},
|
||||
{"x": 3635, "y": 10, "z": 490},
|
||||
]
|
||||
rom_data = stage_measure.RomDataTracker()
|
||||
cam_stage_data = stage_measure.CamStageTracker()
|
||||
|
||||
# loop through mock positions recording them with an offset.
|
||||
for position in mock_positions:
|
||||
offset = {"x": 54.4, "y": 0}
|
||||
rom_data.record_movement(position, offset)
|
||||
return rom_data
|
||||
cam_stage_data.record_movement(position, offset)
|
||||
return cam_stage_data
|
||||
|
||||
|
||||
def test_predict_z(example_rom_data):
|
||||
def test_predict_z(example_cam_stage_data):
|
||||
"""Check that the prediction for the next z position is correct."""
|
||||
mock_z_diff = example_rom_data.predict_z_displacement(
|
||||
mock_z_diff = example_cam_stage_data.predict_z_displacement(
|
||||
axis="x",
|
||||
stage_movement={"x": 5243, "y": 0},
|
||||
stage_position={"x": 3635, "y": 10, "z": 490},
|
||||
|
|
@ -81,9 +81,9 @@ def test_predict_z(example_rom_data):
|
|||
assert mock_z_diff == expected_z_diff
|
||||
|
||||
|
||||
def test_find_tuning_point(example_rom_data):
|
||||
def test_find_tuning_point(example_cam_stage_data):
|
||||
"""Check that the prediction for the tuning point and z position."""
|
||||
mock_turn = example_rom_data.find_turning_point(axis="x")
|
||||
mock_turn = example_cam_stage_data.find_turning_point(axis="x")
|
||||
expected_turning = {"x": 7580, "y": 10, "z": 661}
|
||||
assert mock_turn == expected_turning
|
||||
|
||||
|
|
@ -141,24 +141,28 @@ def test_parasitic_detect(par_fraction, too_high):
|
|||
|
||||
def test_error_if_no_stream_res_set_when_requesting_img_coords():
|
||||
"""Check a RuntimeError thrown when requesting image coordinates if resolution unset."""
|
||||
rom_thing = create_thing_without_server(
|
||||
stage_measure.RangeofMotionThing, mock_all_slots=True
|
||||
stage_measure_thing = create_thing_without_server(
|
||||
stage_measure.StageMeasurementThing, mock_all_slots=True
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="Stream resolution must be set"):
|
||||
rom_thing._img_percentage_to_img_coords(20, "x")
|
||||
stage_measure_thing._img_percentage_to_img_coords(20, "x")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rom_thing(example_rom_data, csm_matrix, mocker) -> stage_measure.RangeofMotionThing:
|
||||
"""Yield a RangeofMotionThing already populated with some example rom_data."""
|
||||
rom_thing = create_thing_without_server(
|
||||
stage_measure.RangeofMotionThing, mock_all_slots=True
|
||||
def stage_measure_thing(
|
||||
example_cam_stage_data, csm_matrix, mocker
|
||||
) -> stage_measure.StageMeasurementThing:
|
||||
"""Yield a StageMeasurementThing already populated with some example cam_stage_data."""
|
||||
stage_measure_thing = create_thing_without_server(
|
||||
stage_measure.StageMeasurementThing, mock_all_slots=True
|
||||
)
|
||||
type(rom_thing._thing_server_interface).application_config = mocker.PropertyMock(
|
||||
type(
|
||||
stage_measure_thing._thing_server_interface
|
||||
).application_config = mocker.PropertyMock(
|
||||
return_value={"data_folder": tempfile.gettempdir()}
|
||||
)
|
||||
rom_thing._stream_resolution = [800, 600]
|
||||
rom_thing._rom_data = example_rom_data
|
||||
stage_measure_thing._stream_resolution = [800, 600]
|
||||
stage_measure_thing._cam_stage_data = example_cam_stage_data
|
||||
|
||||
def apply_csm(x: float, y: float, **_kwargs: float) -> dict[str, int]:
|
||||
"""Convert image coordinates to stage coordinates."""
|
||||
|
|
@ -168,13 +172,18 @@ def rom_thing(example_rom_data, csm_matrix, mocker) -> stage_measure.RangeofMoti
|
|||
"""Convert stage coordinates to image coordinates."""
|
||||
return csm_stage_to_img(csm_matrix, x=x, y=y)
|
||||
|
||||
rom_thing._cam.image_is_sample.return_value = (True, "Mocked not measured.")
|
||||
stage_measure_thing._cam.image_is_sample.return_value = (
|
||||
True,
|
||||
"Mocked not measured.",
|
||||
)
|
||||
|
||||
# Set up mock csm to return a CSM matrix
|
||||
rom_thing._csm.image_to_stage_displacement_matrix = csm_matrix
|
||||
rom_thing._csm.convert_image_to_stage_coordinates.side_effect = apply_csm
|
||||
rom_thing._csm.convert_stage_to_image_coordinates.side_effect = un_apply_csm
|
||||
return rom_thing
|
||||
stage_measure_thing._csm.image_to_stage_displacement_matrix = csm_matrix
|
||||
stage_measure_thing._csm.convert_image_to_stage_coordinates.side_effect = apply_csm
|
||||
stage_measure_thing._csm.convert_stage_to_image_coordinates.side_effect = (
|
||||
un_apply_csm
|
||||
)
|
||||
return stage_measure_thing
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -187,18 +196,20 @@ def rom_thing(example_rom_data, csm_matrix, mocker) -> stage_measure.RangeofMoti
|
|||
({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 1),
|
||||
],
|
||||
)
|
||||
def test_img_dir_from_stage_coords(pos, target_pos, axis, expected_dir, rom_thing):
|
||||
def test_img_dir_from_stage_coords(
|
||||
pos, target_pos, axis, expected_dir, stage_measure_thing
|
||||
):
|
||||
"""Check image direction is correctly generated."""
|
||||
rom_thing._stage.position = pos
|
||||
direction = rom_thing._img_dir_from_stage_coords(target_pos, axis)
|
||||
stage_measure_thing._stage.position = pos
|
||||
direction = stage_measure_thing._img_dir_from_stage_coords(target_pos, axis)
|
||||
assert direction == expected_dir
|
||||
|
||||
|
||||
def test_distance_in_img_percentage_err(rom_thing):
|
||||
def test_distance_in_img_percentage_err(stage_measure_thing):
|
||||
"""Check that _distance_in_img_percentage throws error if stream res is not set."""
|
||||
rom_thing._stream_resolution = None
|
||||
stage_measure_thing._stream_resolution = None
|
||||
with pytest.raises(RuntimeError):
|
||||
rom_thing._distance_in_img_percentage({"x": 0, "y": 0, "z": 0}, "x")
|
||||
stage_measure_thing._distance_in_img_percentage({"x": 0, "y": 0, "z": 0}, "x")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -211,15 +222,17 @@ def test_distance_in_img_percentage_err(rom_thing):
|
|||
({"x": 0, "y": 5000, "z": 0}, {"x": 0, "y": 0, "z": 0}, "y", 462.131),
|
||||
],
|
||||
)
|
||||
def test_distance_in_img_percentage(pos, target_pos, axis, expected_perc, rom_thing):
|
||||
def test_distance_in_img_percentage(
|
||||
pos, target_pos, axis, expected_perc, stage_measure_thing
|
||||
):
|
||||
"""Check _distance_in_img_percentage calculates expected values based on mock CSM."""
|
||||
rom_thing._stage.position = pos
|
||||
img_perc = rom_thing._distance_in_img_percentage(target_pos, axis)
|
||||
stage_measure_thing._stage.position = pos
|
||||
img_perc = stage_measure_thing._distance_in_img_percentage(target_pos, axis)
|
||||
assert round(img_perc, 3) == expected_perc
|
||||
|
||||
|
||||
def test_offset_from(rom_thing, mocker):
|
||||
"""Check the calls and returns for RangeofMotionThing._offset_from."""
|
||||
def test_offset_from(stage_measure_thing, mocker):
|
||||
"""Check the calls and returns for StageMeasurementThing._offset_from."""
|
||||
# Set up mock for the FFT displacement
|
||||
disp_between_route = (
|
||||
"openflexure_microscope_server.things.stage_measure."
|
||||
|
|
@ -231,14 +244,14 @@ def test_offset_from(rom_thing, mocker):
|
|||
)
|
||||
|
||||
# Run it
|
||||
offset = rom_thing._offset_from(before_img="MOCK_IMAGE")
|
||||
offset = stage_measure_thing._offset_from(before_img="MOCK_IMAGE")
|
||||
|
||||
# Check the offset is a dictionary with the correct values for the axes
|
||||
assert offset["x"] == 456
|
||||
assert offset["y"] == 123
|
||||
# Check 1 image was taken
|
||||
assert rom_thing._cam.grab_as_array.call_count == 1
|
||||
mock_after_image = rom_thing._cam.grab_as_array.return_value
|
||||
assert stage_measure_thing._cam.grab_as_array.call_count == 1
|
||||
mock_after_image = stage_measure_thing._cam.grab_as_array.return_value
|
||||
# Check the FFT displacement was called once
|
||||
assert mock_disp_between.call_count == 1
|
||||
displacement_kwargs = mock_disp_between.call_args.kwargs
|
||||
|
|
@ -248,25 +261,31 @@ def test_offset_from(rom_thing, mocker):
|
|||
|
||||
|
||||
@pytest.mark.parametrize("perform_autofocus", [True, False])
|
||||
def test_move_and_measure(perform_autofocus, rom_thing, mocker):
|
||||
def test_move_and_measure(perform_autofocus, stage_measure_thing, mocker):
|
||||
"""Test _move_and_measure with and without initial autofocus checking call counts.
|
||||
|
||||
This doesn't test with the repeated autofocus if motion isn't detected
|
||||
"""
|
||||
mock_offset_value = {"x": 100, "y": 3}
|
||||
mocker.patch.object(rom_thing, "_offset_from", return_value=mock_offset_value)
|
||||
mocker.patch.object(
|
||||
stage_measure_thing, "_offset_from", return_value=mock_offset_value
|
||||
)
|
||||
movement = {"x": 100, "y": 0}
|
||||
offset = rom_thing._move_and_measure(
|
||||
offset = stage_measure_thing._move_and_measure(
|
||||
movement=movement, perform_autofocus=perform_autofocus
|
||||
)
|
||||
|
||||
# Check exactly 1 move
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
assert stage_measure_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
# The kwargs of the call should movement dict
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_args.kwargs == movement
|
||||
assert (
|
||||
stage_measure_thing._csm.move_in_image_coordinates.call_args.kwargs == movement
|
||||
)
|
||||
# Check autofocus call count
|
||||
expected_af_count = 1 if perform_autofocus else 0
|
||||
assert rom_thing._autofocus.looping_autofocus.call_count == expected_af_count
|
||||
assert (
|
||||
stage_measure_thing._autofocus.looping_autofocus.call_count == expected_af_count
|
||||
)
|
||||
# And check final return
|
||||
assert offset == mock_offset_value
|
||||
|
||||
|
|
@ -284,43 +303,47 @@ def test_move_and_measure(perform_autofocus, rom_thing, mocker):
|
|||
],
|
||||
)
|
||||
def test_move_and_measure_with_refocus(
|
||||
x_offsets, n_offset_measures, expected_return, rom_thing, mocker
|
||||
x_offsets, n_offset_measures, expected_return, stage_measure_thing, mocker
|
||||
):
|
||||
"""Test _move_and_measure with final refocus if offset is too small."""
|
||||
return_dicts = tuple({"x": x, "y": 0} for x in x_offsets)
|
||||
offset_from_mock = mocker.patch.object(
|
||||
rom_thing, "_offset_from", side_effect=return_dicts
|
||||
stage_measure_thing, "_offset_from", side_effect=return_dicts
|
||||
)
|
||||
movement = {"x": 10, "y": 0}
|
||||
offset = rom_thing._move_and_measure(
|
||||
offset = stage_measure_thing._move_and_measure(
|
||||
movement=movement,
|
||||
perform_autofocus=False,
|
||||
max_autofocus_repeats=3,
|
||||
abs_min_offset=5,
|
||||
)
|
||||
# Check exactly 1 move
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
assert stage_measure_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
# Check expected _offset_from calls
|
||||
assert offset_from_mock.call_count == n_offset_measures
|
||||
# The kwargs of the call should movement dict
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_args.kwargs == movement
|
||||
assert (
|
||||
stage_measure_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
|
||||
# is performed before the first one
|
||||
expected_af_count = n_offset_measures - 1
|
||||
assert rom_thing._autofocus.looping_autofocus.call_count == expected_af_count
|
||||
assert (
|
||||
stage_measure_thing._autofocus.looping_autofocus.call_count == expected_af_count
|
||||
)
|
||||
# And check final return
|
||||
assert offset == expected_return
|
||||
|
||||
|
||||
def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
|
||||
def test_move_and_measure_with_bad_refocus_args(stage_measure_thing, mocker):
|
||||
"""Check error if abs_min_offset is not positive when using it to determine if to autofocus."""
|
||||
offset_from_mock = mocker.patch.object(
|
||||
rom_thing, "_offset_from", return_value={"x": 0, "y": 0}
|
||||
stage_measure_thing, "_offset_from", return_value={"x": 0, "y": 0}
|
||||
)
|
||||
|
||||
# First check with abs_min_offset not set. The default should be zero.
|
||||
with pytest.raises(ValueError, match="abs_min_offset must be positive"):
|
||||
rom_thing._move_and_measure(
|
||||
stage_measure_thing._move_and_measure(
|
||||
movement={"x": 10, "y": 0},
|
||||
perform_autofocus=False,
|
||||
max_autofocus_repeats=3,
|
||||
|
|
@ -328,7 +351,7 @@ def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
|
|||
# Then check with abs_min_offset negative, this might happen if the the expected
|
||||
# move calculation is not made absolute.
|
||||
with pytest.raises(ValueError, match="abs_min_offset must be positive"):
|
||||
rom_thing._move_and_measure(
|
||||
stage_measure_thing._move_and_measure(
|
||||
movement={"x": 10, "y": 0},
|
||||
perform_autofocus=False,
|
||||
max_autofocus_repeats=3,
|
||||
|
|
@ -338,18 +361,18 @@ def test_move_and_measure_with_bad_refocus_args(rom_thing, mocker):
|
|||
assert offset_from_mock.call_count == 0
|
||||
|
||||
|
||||
def test_move_back_until_motion_detected(rom_thing, mocker):
|
||||
def test_move_back_until_motion_detected(stage_measure_thing, mocker):
|
||||
"""Check that _move_back_until_motion_detected is making increasing negative moves.
|
||||
|
||||
The moves for this method should be in opposite direction to the direction
|
||||
specified as this is moving back after the stage reaches end of its movement.
|
||||
"""
|
||||
mock_move_n_meas = mocker.patch.object(
|
||||
rom_thing, "_move_and_measure", return_value={"x": 0, "y": 0}
|
||||
stage_measure_thing, "_move_and_measure", return_value={"x": 0, "y": 0}
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="Cannot detect motion again"):
|
||||
rom_thing._move_back_until_motion_detected("y", -1)
|
||||
stage_measure_thing._move_back_until_motion_detected("y", -1)
|
||||
|
||||
max_tries = int(1.5 * stage_measure.BIG_STEP / stage_measure.SMALL_STEP)
|
||||
expected_step = 800 * stage_measure.SMALL_STEP / 100
|
||||
|
|
@ -367,7 +390,7 @@ def test_move_back_until_motion_detected(rom_thing, mocker):
|
|||
)
|
||||
|
||||
# Other axis and direction this time
|
||||
rom_thing._move_back_until_motion_detected("x", 1)
|
||||
stage_measure_thing._move_back_until_motion_detected("x", 1)
|
||||
|
||||
# Should only be called 3 times
|
||||
assert mock_move_n_meas.call_count == 3
|
||||
|
|
@ -397,7 +420,7 @@ def test_stage_still_moves(
|
|||
good_moves,
|
||||
expected_to_detect_motion,
|
||||
offset_calls,
|
||||
rom_thing,
|
||||
stage_measure_thing,
|
||||
mocker,
|
||||
):
|
||||
"""Test _stage_still_moves correctly detects stage movement."""
|
||||
|
|
@ -412,141 +435,147 @@ def test_stage_still_moves(
|
|||
i += 1
|
||||
|
||||
mock_offset_from = mocker.patch.object(
|
||||
rom_thing, "_offset_from", side_effect=gen_offsets()
|
||||
stage_measure_thing, "_offset_from", side_effect=gen_offsets()
|
||||
)
|
||||
|
||||
still_moves = rom_thing._stage_still_moves(axis="x", direction=1)
|
||||
still_moves = stage_measure_thing._stage_still_moves(axis="x", direction=1)
|
||||
assert still_moves is expected_to_detect_motion
|
||||
assert mock_offset_from.call_count == offset_calls
|
||||
|
||||
|
||||
def test_big_z_corrected_movement(rom_thing):
|
||||
def test_big_z_corrected_movement(stage_measure_thing):
|
||||
"""Check big z corrected move moves in x/y and z the expected distances."""
|
||||
rom_thing._stage.position = {"x": 5000, "y": 30, "z": 500}
|
||||
stage_measure_thing._stage.position = {"x": 5000, "y": 30, "z": 500}
|
||||
|
||||
rom_thing._big_z_corrected_movement("x", direction=1)
|
||||
stage_measure_thing._big_z_corrected_movement("x", direction=1)
|
||||
|
||||
expected_movement = {"x": 800 * stage_measure.BIG_STEP / 100, "y": 0}
|
||||
|
||||
# Check there is one z move in steps
|
||||
assert rom_thing._stage.move_relative.call_count == 1
|
||||
move_kwargs = rom_thing._stage.move_relative.call_args.kwargs
|
||||
assert stage_measure_thing._stage.move_relative.call_count == 1
|
||||
move_kwargs = stage_measure_thing._stage.move_relative.call_args.kwargs
|
||||
assert "x" not in move_kwargs
|
||||
assert "y" not in move_kwargs
|
||||
assert "z" in move_kwargs
|
||||
assert move_kwargs["z"] == 160
|
||||
|
||||
# And one move in image coordinates
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
lat_mov_kwargs = rom_thing._csm.move_in_image_coordinates.call_args.kwargs
|
||||
assert stage_measure_thing._csm.move_in_image_coordinates.call_count == 1
|
||||
lat_mov_kwargs = stage_measure_thing._csm.move_in_image_coordinates.call_args.kwargs
|
||||
assert lat_mov_kwargs == expected_movement
|
||||
|
||||
|
||||
def test_moves_for_z_prediction(rom_thing, mocker):
|
||||
def test_moves_for_z_prediction(stage_measure_thing, mocker):
|
||||
"""Check the initial moves are of the correct size and are recorded."""
|
||||
# Mock the _offset_from and stage.position to return generated dictionaries that
|
||||
# increment each time they are called. (All values 0 the first time, all values 1
|
||||
# the second time ...)
|
||||
mocker.patch.object(
|
||||
rom_thing, "_offset_from", side_effect=increasing_xy_dict_generator()
|
||||
stage_measure_thing, "_offset_from", side_effect=increasing_xy_dict_generator()
|
||||
)
|
||||
type(rom_thing._stage).position = mocker.PropertyMock(
|
||||
type(stage_measure_thing._stage).position = mocker.PropertyMock(
|
||||
side_effect=increasing_xyz_dict_generator()
|
||||
)
|
||||
|
||||
expected_movement = {"x": -800 * stage_measure.MEDIUM_STEP / 100, "y": 0}
|
||||
|
||||
# Remove the mock RomData before starting
|
||||
rom_thing._rom_data = stage_measure.RomDataTracker()
|
||||
stage_measure_thing._cam_stage_data = stage_measure.CamStageTracker()
|
||||
|
||||
# Run it!
|
||||
rom_thing._moves_for_z_prediction("x", direction=-1)
|
||||
stage_measure_thing._moves_for_z_prediction("x", direction=-1)
|
||||
|
||||
# Check that _rom_data now contains the 5 mocked returns in order.
|
||||
assert rom_thing._rom_data.offsets == [{"x": i, "y": i} for i in range(5)]
|
||||
assert rom_thing._rom_data.stage_coords == [
|
||||
# Check that _cam_stage_data now contains the 5 mocked returns in order.
|
||||
assert stage_measure_thing._cam_stage_data.offsets == [
|
||||
{"x": i, "y": i} for i in range(5)
|
||||
]
|
||||
assert stage_measure_thing._cam_stage_data.stage_coords == [
|
||||
{"x": i, "y": i, "z": i} for i in range(5)
|
||||
]
|
||||
|
||||
# Check that the csm movement function is called 5 times
|
||||
assert rom_thing._csm.move_in_image_coordinates.call_count == 5
|
||||
assert stage_measure_thing._csm.move_in_image_coordinates.call_count == 5
|
||||
# Each time with the expected movement
|
||||
for arg_list in rom_thing._csm.move_in_image_coordinates.call_args_list:
|
||||
for arg_list in stage_measure_thing._csm.move_in_image_coordinates.call_args_list:
|
||||
assert arg_list.kwargs == expected_movement
|
||||
|
||||
|
||||
def test_move_until_edge_error(rom_thing, mocker):
|
||||
def test_move_until_edge_error(stage_measure_thing, mocker):
|
||||
"""Check that if there is an error while moving to the edge the stage returns to start."""
|
||||
mock_position_dict = {"x": 123, "y": 456, "z": 789}
|
||||
type(rom_thing._stage).position = mocker.PropertyMock(
|
||||
type(stage_measure_thing._stage).position = mocker.PropertyMock(
|
||||
return_value=mock_position_dict
|
||||
)
|
||||
mocker.patch.object(
|
||||
rom_thing, "_moves_for_z_prediction", side_effect=RuntimeError("Mock")
|
||||
stage_measure_thing, "_moves_for_z_prediction", side_effect=RuntimeError("Mock")
|
||||
)
|
||||
|
||||
# Remove the mock RomData before starting
|
||||
rom_thing._rom_data = stage_measure.RomDataTracker()
|
||||
stage_measure_thing._cam_stage_data = stage_measure.CamStageTracker()
|
||||
|
||||
# Error should be raised even though it is in the Try:
|
||||
with pytest.raises(RuntimeError, match="Mock"):
|
||||
rom_thing._move_until_edge("y", direction=-1)
|
||||
stage_measure_thing._move_until_edge("y", direction=-1)
|
||||
# However the "finally" should have executed, returning to the starting position
|
||||
assert rom_thing._stage.move_absolute.call_count == 1
|
||||
abs_move_kwargs = rom_thing._stage.move_absolute.call_args.kwargs
|
||||
assert stage_measure_thing._stage.move_absolute.call_count == 1
|
||||
abs_move_kwargs = stage_measure_thing._stage.move_absolute.call_args.kwargs
|
||||
expected_abs_move_kwargs = dict(
|
||||
**mock_position_dict, block_cancellation=True, backlash_compensation=None
|
||||
)
|
||||
assert abs_move_kwargs == expected_abs_move_kwargs
|
||||
|
||||
|
||||
def test_move_until_edge(rom_thing, mocker):
|
||||
def test_move_until_edge(stage_measure_thing, mocker):
|
||||
"""Check move until edge runs the correct movement sequence."""
|
||||
# Remove the mock RomData before starting
|
||||
rom_thing._rom_data = stage_measure.RomDataTracker()
|
||||
stage_measure_thing._cam_stage_data = stage_measure.CamStageTracker()
|
||||
|
||||
rom_thing._stage.position = {"x": "mock", "y": "starting", "z": "pos"}
|
||||
stage_measure_thing._stage.position = {"x": "mock", "y": "starting", "z": "pos"}
|
||||
|
||||
def add_fake_initial_positions(*_args, **_kwargs):
|
||||
"""Rather than run initial moves just add some fake data."""
|
||||
for i in range(5):
|
||||
rom_thing._rom_data.record_movement(f"mock-init-pos{i + 1}", "mock-offset")
|
||||
stage_measure_thing._cam_stage_data.record_movement(
|
||||
f"mock-init-pos{i + 1}", "mock-offset"
|
||||
)
|
||||
|
||||
def update_stage_pos_on_big_move(*_args, **_kwargs):
|
||||
"""With big moves update the stage position."""
|
||||
big_move_count = 1
|
||||
while True:
|
||||
rom_thing._stage.position = f"mocked-big-move-pos{big_move_count}"
|
||||
stage_measure_thing._stage.position = f"mocked-big-move-pos{big_move_count}"
|
||||
yield
|
||||
big_move_count += 1
|
||||
|
||||
def set_final_pos(*_args, **_kwargs):
|
||||
"""When move_back_until_motion_detected is run set a final stage position."""
|
||||
rom_thing._stage.position = "mock-final-pos"
|
||||
stage_measure_thing._stage.position = "mock-final-pos"
|
||||
|
||||
# Mock the main movement functions
|
||||
mock_init_moves = mocker.patch.object(
|
||||
rom_thing,
|
||||
stage_measure_thing,
|
||||
"_moves_for_z_prediction",
|
||||
side_effect=add_fake_initial_positions,
|
||||
)
|
||||
mock_big_moves = mocker.patch.object(
|
||||
rom_thing,
|
||||
stage_measure_thing,
|
||||
"_big_z_corrected_movement",
|
||||
side_effect=update_stage_pos_on_big_move(),
|
||||
)
|
||||
mock_move_check = mocker.patch.object(
|
||||
rom_thing, "_stage_still_moves", side_effect=[True] * 9 + [False]
|
||||
stage_measure_thing, "_stage_still_moves", side_effect=[True] * 9 + [False]
|
||||
)
|
||||
mock_move_back = mocker.patch.object(
|
||||
rom_thing, "_move_back_until_motion_detected", side_effect=set_final_pos
|
||||
stage_measure_thing,
|
||||
"_move_back_until_motion_detected",
|
||||
side_effect=set_final_pos,
|
||||
)
|
||||
|
||||
# Remove the mock RomData before starting
|
||||
rom_thing._rom_data = stage_measure.RomDataTracker()
|
||||
stage_measure_thing._cam_stage_data = stage_measure.CamStageTracker()
|
||||
|
||||
# Run function
|
||||
rom_thing._move_until_edge("y", direction=-1)
|
||||
stage_measure_thing._move_until_edge("y", direction=-1)
|
||||
|
||||
# Check the call counts are as expected
|
||||
# One call of initial moves
|
||||
|
|
@ -558,7 +587,7 @@ def test_move_until_edge(rom_thing, mocker):
|
|||
# And one call of _move_back_until_motion_detected
|
||||
assert mock_move_back.call_count == 1
|
||||
|
||||
assert rom_thing._rom_data.stage_coords == [
|
||||
assert stage_measure_thing._cam_stage_data.stage_coords == [
|
||||
{"x": "mock", "y": "starting", "z": "pos"},
|
||||
"mock-init-pos1",
|
||||
"mock-init-pos2",
|
||||
|
|
@ -578,41 +607,43 @@ def test_move_until_edge(rom_thing, mocker):
|
|||
]
|
||||
|
||||
|
||||
def test_perform_rom_actions_locked(rom_thing):
|
||||
def test_perform_rom_actions_locked(stage_measure_thing):
|
||||
"""Check the error if running one of the calibration actions while locked."""
|
||||
# Not an RLock so no need to thread.
|
||||
rom_thing._lock.acquire()
|
||||
stage_measure_thing._lock.acquire()
|
||||
err_msg = "Trying to run ROM test when a test is already running."
|
||||
with pytest.raises(RuntimeError, match=err_msg):
|
||||
rom_thing.perform_rom_test()
|
||||
stage_measure_thing.perform_rom_test()
|
||||
err_msg = "Trying to run recentre when a test is already running."
|
||||
with pytest.raises(RuntimeError, match=err_msg):
|
||||
rom_thing.perform_recentre()
|
||||
stage_measure_thing.perform_recentre()
|
||||
|
||||
|
||||
def test_perform_rom_test(rom_thing, mocker):
|
||||
def test_perform_rom_test(stage_measure_thing, mocker):
|
||||
"""Check that perform Rom Test runs the expected high level algorithm."""
|
||||
|
||||
def check_and_modify_rom_data(axis: str, direction: int, **_kwargs):
|
||||
"""Check the rom_data is empty each time, and add a final position."""
|
||||
# check rom_data was cleared at the start of this run
|
||||
assert len(rom_thing._rom_data.stage_coords) == 0
|
||||
def check_and_modify_cam_stage_data(axis: str, direction: int, **_kwargs):
|
||||
"""Check the cam_stage_data is empty each time, and add a final position."""
|
||||
# check cam_stage_data was cleared at the start of this run
|
||||
assert len(stage_measure_thing._cam_stage_data.stage_coords) == 0
|
||||
dist = 1111 if axis == "x" else 2222
|
||||
final_pos = {"x": 0, "y": 0}
|
||||
final_pos[axis] = dist * direction
|
||||
rom_thing._rom_data.stage_coords.append(final_pos)
|
||||
stage_measure_thing._cam_stage_data.stage_coords.append(final_pos)
|
||||
|
||||
mock_move_until_edge = mocker.patch.object(
|
||||
rom_thing, "_move_until_edge", side_effect=check_and_modify_rom_data
|
||||
stage_measure_thing,
|
||||
"_move_until_edge",
|
||||
side_effect=check_and_modify_cam_stage_data,
|
||||
)
|
||||
|
||||
mocker.patch.object(
|
||||
rom_thing._cam, "grab_as_array", return_value=np.zeros([123, 456, 3])
|
||||
stage_measure_thing._cam, "grab_as_array", return_value=np.zeros([123, 456, 3])
|
||||
)
|
||||
|
||||
# Enter the ROM thing to get the data directory.
|
||||
with rom_thing as running_rom_thing:
|
||||
final_dict = running_rom_thing.perform_rom_test()
|
||||
# Enter the StageMeasureThing to get the data directory.
|
||||
with stage_measure_thing as running_stage_measure_thing:
|
||||
final_dict = running_stage_measure_thing.perform_rom_test()
|
||||
|
||||
# This should take less than 1 sec
|
||||
assert final_dict["Time"] <= 1
|
||||
|
|
@ -640,24 +671,24 @@ def test_perform_rom_test(rom_thing, mocker):
|
|||
assert mock_move_until_edge.call_args_list[3].kwargs["direction"] == -1
|
||||
|
||||
|
||||
def test_perform_recenter(rom_thing, mocker, caplog):
|
||||
def test_perform_recenter(stage_measure_thing, mocker, caplog):
|
||||
"""Check that performing the recentre runs through expected operations."""
|
||||
|
||||
def check_lock(*_args, **_kwargs):
|
||||
"""Check the thing is locked."""
|
||||
assert not rom_thing._lock.acquire(blocking=False)
|
||||
assert not stage_measure_thing._lock.acquire(blocking=False)
|
||||
|
||||
mock_set_stream_res = mocker.patch.object(
|
||||
rom_thing, "_set_stream_resolution", side_effect=check_lock
|
||||
stage_measure_thing, "_set_stream_resolution", side_effect=check_lock
|
||||
)
|
||||
mock_recentre_axis = mocker.patch.object(rom_thing, "_recentre_axis")
|
||||
mock_recentre_axis = mocker.patch.object(stage_measure_thing, "_recentre_axis")
|
||||
|
||||
# Set a mock stage position, as we patch _recentre_axis this should be logged
|
||||
# as the centre.
|
||||
rom_thing._stage.position = {"x": 4321, "y": 1234, "z": 0}
|
||||
stage_measure_thing._stage.position = {"x": 4321, "y": 1234, "z": 0}
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
rom_thing.perform_recentre()
|
||||
stage_measure_thing.perform_recentre()
|
||||
|
||||
assert len(caplog.messages) == 3
|
||||
assert caplog.messages[0] == "Recentring the stage."
|
||||
|
|
@ -665,8 +696,8 @@ def test_perform_recenter(rom_thing, mocker, caplog):
|
|||
assert caplog.messages[2] == "Position reset to (0, 0, 0)."
|
||||
|
||||
# Check the lock was freed
|
||||
assert rom_thing._lock.acquire(blocking=False)
|
||||
rom_thing._lock.release()
|
||||
assert stage_measure_thing._lock.acquire(blocking=False)
|
||||
stage_measure_thing._lock.release()
|
||||
|
||||
assert mock_set_stream_res.call_count == 1
|
||||
|
||||
|
|
@ -676,31 +707,31 @@ def test_perform_recenter(rom_thing, mocker, caplog):
|
|||
assert mock_recentre_axis.call_args_list[1].args[0] == "y"
|
||||
|
||||
# check that the position set to zero once
|
||||
assert rom_thing._stage.set_zero_position.call_count == 1
|
||||
assert stage_measure_thing._stage.set_zero_position.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("true_on", [1, 4, 10, 11])
|
||||
def test_recentre_axis(true_on, rom_thing, mocker):
|
||||
def test_recentre_axis(true_on, stage_measure_thing, mocker):
|
||||
"""Test the high level algorithm of recentring an axis.
|
||||
|
||||
This doesn't include deciding if we are centred or choosing the direction to move.
|
||||
"""
|
||||
## Start such that movement starts negative
|
||||
rom_thing._stage.position = {"x": 10000, "y": 10000, "z": 0}
|
||||
mock_moves = mocker.patch.object(rom_thing, "_moves_for_z_prediction")
|
||||
stage_measure_thing._stage.position = {"x": 10000, "y": 10000, "z": 0}
|
||||
mock_moves = mocker.patch.object(stage_measure_thing, "_moves_for_z_prediction")
|
||||
|
||||
# Mock _recentre_decision so it returns (False, 1) a number of times then finally
|
||||
# (True, 1).
|
||||
decisions = [(False, 1)] * (true_on - 1) + [(True, 1)]
|
||||
mock_recentre_decision = mocker.patch.object(
|
||||
rom_thing, "_recentre_decision", side_effect=decisions
|
||||
stage_measure_thing, "_recentre_decision", side_effect=decisions
|
||||
)
|
||||
|
||||
if true_on > 10:
|
||||
with pytest.raises(RuntimeError, match="Couldn't find centre"):
|
||||
rom_thing._recentre_axis("x")
|
||||
stage_measure_thing._recentre_axis("x")
|
||||
else:
|
||||
rom_thing._recentre_axis("x")
|
||||
stage_measure_thing._recentre_axis("x")
|
||||
|
||||
# _recentre_decision is called until True or exits after 10 attempts
|
||||
assert mock_recentre_decision.call_count == min(true_on, 10)
|
||||
|
|
@ -725,18 +756,24 @@ def test_recentre_axis(true_on, rom_thing, mocker):
|
|||
(-199, -1, (True, 1)), # Always return 1 when c
|
||||
],
|
||||
)
|
||||
def test_recentre_decision(dist, direction, expected_decision, rom_thing, mocker):
|
||||
def test_recentre_decision(
|
||||
dist, direction, expected_decision, stage_measure_thing, mocker
|
||||
):
|
||||
"""What the algorithm decides to do in different situations."""
|
||||
# Mock find turning point just because there is no _rom_data
|
||||
mocker.patch.object(rom_thing._rom_data, "find_turning_point")
|
||||
# Mock find turning point just because there is no _cam_stage_data
|
||||
mocker.patch.object(stage_measure_thing._cam_stage_data, "find_turning_point")
|
||||
|
||||
# As _distance_in_img_percentage and _img_dir_from_stage_coords are tested
|
||||
# this test mocks their values and checks the return is as expected
|
||||
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(
|
||||
stage_measure_thing, "_distance_in_img_percentage", return_value=dist
|
||||
)
|
||||
mocker.patch.object(
|
||||
stage_measure_thing, "_img_dir_from_stage_coords", return_value=direction
|
||||
)
|
||||
|
||||
assert rom_thing._recentre_decision("x") == expected_decision
|
||||
assert stage_measure_thing._recentre_decision("x") == expected_decision
|
||||
|
||||
# Check that we make an absolute move (to the centre) before returning centred.
|
||||
centred = expected_decision[0]
|
||||
rom_thing._stage.move_absolute.call_count == 1 if centred else 0
|
||||
stage_measure_thing._stage.move_absolute.call_count == 1 if centred else 0
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue