Replace LoggingMoveWrapper with RecordedMove accepts a stage rather than an arbitrary move function

This was needed to get the correct key word arguments
This commit is contained in:
Julian Stirling 2025-07-13 23:26:26 +01:00
parent b89449be77
commit 531e5c27b2
2 changed files with 15 additions and 19 deletions

View file

@ -13,7 +13,6 @@ server, and depends on that server and its underlying LabThings library.
import time import time
from typing import ( from typing import (
Any, Any,
Callable,
Dict, Dict,
List, List,
NamedTuple, NamedTuple,
@ -53,30 +52,28 @@ class MoveHistory(NamedTuple):
stage_positions: List[CoordinateType] stage_positions: List[CoordinateType]
class LoggingMoveWrapper: class RecordedMove:
"""Wrap a move function, and maintain a log position/time. """Call stage movement and maintain a record of position and time.
This class is callable, so it doesn't change the signature This class is callable, the callable wraps stage.move_to_xyz_position.
of the function it wraps - it just makes it possible to get
a list of all the moves we've made, and how long they took.
Said list is intended to be useful for calibrating the stage The class records a list of all moves made and how long they took. This is useful
so we can estimate how long moves will take. for calibrating the stage as it allows measuring how long moves take.
""" """
def __init__(self, move_function: Callable): def __init__(self, stage: Stage):
"""Set the movement function to be wrapped. """Set the movement function to be wrapped.
:param move_function: the movement function to be wrapped :param move_function: the movement function to be wrapped
""" """
self._move_function: Callable = move_function self._stage: 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]]] = []
def __call__(self, new_position: CoordinateType, *args, **kwargs): def __call__(self, new_position: CoordinateType):
"""Move to a new position, and record it.""" """Move to a new position, and record it."""
self._history.append((time.time(), self._current_position)) self._history.append((time.time(), self._current_position))
self._move_function(new_position, *args, **kwargs) self._stage.move_to_xyz_position(xyz_pos=new_position)
self._current_position = new_position self._current_position = new_position
self._history.append((time.time(), self._current_position)) self._history.append((time.time(), self._current_position))
@ -128,8 +125,8 @@ class CameraStageMapper(lt.Thing):
direction: Tuple[float, float, float], direction: Tuple[float, float, float],
) -> DenumpifyingDict: ) -> 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."""
# log positions and times for stage calibration # Record positions and times for stage calibration
wrapped_move = LoggingMoveWrapper(stage.move_to_xyz_position) recorded_move = RecordedMove(stage)
tracker = Tracker( tracker = Tracker(
camera.capture_downsampled_array, camera.capture_downsampled_array,
stage.get_xyz_position, stage.get_xyz_position,
@ -140,7 +137,7 @@ class CameraStageMapper(lt.Thing):
starting_position = stage.position starting_position = stage.position
try: try:
result: dict = calibrate_backlash_1d( result: dict = calibrate_backlash_1d(
tracker, wrapped_move, direction_array, logger=logger tracker, recorded_move, direction_array, logger=logger
) )
except lt.exceptions.InvocationCancelledError as e: except lt.exceptions.InvocationCancelledError as e:
logger.info("User cancelled the camera stage mapping calibration") logger.info("User cancelled the camera stage mapping calibration")
@ -151,7 +148,7 @@ class CameraStageMapper(lt.Thing):
logger.info("Returning to starting position due to failed calibration") logger.info("Returning to starting position due to failed calibration")
stage.move_absolute(**starting_position, block_cancellation=True) stage.move_absolute(**starting_position, block_cancellation=True)
raise e raise e
result["move_history"] = wrapped_move.history result["move_history"] = recorded_move.history
result["image_resolution"] = camera.capture_downsampled_array().shape[:2] result["image_resolution"] = camera.capture_downsampled_array().shape[:2]
return result return result

View file

@ -24,9 +24,6 @@ from openflexure_microscope_server.things.camera.simulation import SimulatedCame
from openflexure_microscope_server.things.stage.dummy import DummyStage from openflexure_microscope_server.things.stage.dummy import DummyStage
from openflexure_microscope_server.things.autofocus import AutofocusThing from openflexure_microscope_server.things.autofocus import AutofocusThing
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
from openflexure_microscope_server.things import camera_stage_mapping
camera_stage_mapping.DEFAULT_SETTLING_TIME = 0 # skip the settling time for tests
@pytest.fixture @pytest.fixture
@ -117,5 +114,7 @@ def test_capture_array(client):
def test_camera_stage_mapping_calibration(client): def test_camera_stage_mapping_calibration(client):
"""Check that camera stage mapping can run without an exception.""" """Check that camera stage mapping can run without an exception."""
camera = lt.ThingClient.from_url("/camera/", client)
camera.settling_time = 0
camera_stage_mapping = lt.ThingClient.from_url("/camera_stage_mapping/", client) camera_stage_mapping = lt.ThingClient.from_url("/camera_stage_mapping/", client)
camera_stage_mapping.calibrate_xy() camera_stage_mapping.calibrate_xy()