diff --git a/ofm_config_full.json b/ofm_config_full.json index 75534b10..25ddc96a 100644 --- a/ofm_config_full.json +++ b/ofm_config_full.json @@ -19,6 +19,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", "bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV", "bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV" diff --git a/ofm_config_simulation.json b/ofm_config_simulation.json index b061dbd8..57070a66 100644 --- a/ofm_config_simulation.json +++ b/ofm_config_simulation.json @@ -14,6 +14,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", "bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV", "bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV" }, diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 0e2f1dd3..1830d1dc 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -1,14 +1,20 @@ """Functionality for planning scan routes. -A scan route can be planned by a ScanPlanner class currently there -is only one type the SmartSpiral. More can be added using by -subclassing the ScanPlanner +A scan route can be planned by a ScanPlanner. There is a base class ``ScanPlanner``, +and then a child class that is still generic called RectGridPlanner that helps planning +anything where the movements are on a regtangular grid. RectGridPlanner has two usable +child classes: + +* SmartSpiral - For spiralling around a samples but adjusting when background is + detected +* RegularGridPlanner - For Raster and Snake scanning. """ # Future annotations needed for typhinting same class in __eq__ method. Other option # would be to import Union and use a string. from __future__ import annotations +import enum import logging from copy import copy from typing import Any, Literal, Optional, TypeAlias @@ -23,6 +29,23 @@ XYPosList: TypeAlias = list[XYPos] XYZPosList: TypeAlias = list[XYZPos] +class DistanceMetric(enum.Enum): + """An enum for selecting distance metrics for grids. + + Grid distance metrics are: + + * Chebyshev (``CHEBYSHEV``) which is the larger of the number of x or y moves + in the grid. + * Manhattan (``MANHATTAN``) which is the number of moves between the two points + following the grid. Or + * Euclidean (``EUCLIDEAN``) which is the length of the direct route. + """ + + CHEBYSHEV = enum.auto() + MANHATTAN = enum.auto() + EUCLIDEAN = enum.auto() + + def enforce_xy_tuple(value: XYPos) -> XYPos: """Check input is a tuple and is of length 2. @@ -197,7 +220,12 @@ class ScanPlanner: return [loc.xyz_tuple for loc in self._path_history if loc.imaged] @property - def focused_locations(self) -> XYZPosList: + def focused_locations(self) -> list[VisitedScanLocation]: + """Property to access a copy of the focused_locations.""" + return [loc for loc in self._path_history if loc.focused] + + @property + def focused_locations_xyz(self) -> XYZPosList: """Property to access a copy of the focused_locations.""" return [loc.xyz_tuple for loc in self._path_history if loc.focused] @@ -304,7 +332,97 @@ class ScanPlanner: return [FutureScanLocation(location) for line in grid for location in line] -class SmartSpiral(ScanPlanner): +class RectGridPlanner(ScanPlanner): + """Base class for planners that operate on a rectangular grid.""" + + _dx: int = 0 + _dy: int = 0 + + def _parse(self, planner_settings: Optional[dict] = None) -> None: + expected_keys = ["dx", "dy"] + invalid_msg = "RectGridPlanner requires planner_settings with keys: " + if not planner_settings or not all( + k in planner_settings for k in expected_keys + ): + raise KeyError(invalid_msg + ",".join(expected_keys)) + + self._dx = int(planner_settings["dx"]) + self._dy = int(planner_settings["dy"]) + + def _adjacent_positions(self, xy_pos: XYPos) -> XYPosList: + return [ + (xy_pos[0] - self._dx, xy_pos[1]), + (xy_pos[0] + self._dx, xy_pos[1]), + (xy_pos[0], xy_pos[1] - self._dy), + (xy_pos[0], xy_pos[1] + self._dy), + ] + + def moves_between( + self, + starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + metric: DistanceMetric, + ) -> float: + """Return displacement in grid-move units as a numpy array [dx_moves, dy_moves]. + + :param starting_pos: the position to measure from + :param ending_pos: the position to measure to + :param metric: How the distance is calculated. See `DistanceMetric` + """ + if isinstance(starting_pos, (FutureScanLocation, VisitedScanLocation)): + starting_pos = starting_pos.xy_tuple + if isinstance(ending_pos, (FutureScanLocation, VisitedScanLocation)): + ending_pos = ending_pos.xy_tuple + + move_size = np.array([self._dx, self._dy], dtype="float64") + + starting_pos = np.array(starting_pos, dtype="float64") + ending_pos = np.array(ending_pos, dtype="float64") + + displacement = (ending_pos - starting_pos) / move_size + if metric == DistanceMetric.CHEBYSHEV: + return float(np.max(np.abs(displacement))) + if metric == DistanceMetric.MANHATTAN: + return float(np.sum(np.abs(displacement))) + return float(np.linalg.norm(displacement)) + + def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos: + """Return an (x,y) position halfway between two input positions.""" + x = (xy_pos1[0] + xy_pos2[0]) // 2 + y = (xy_pos1[1] + xy_pos2[1]) // 2 + return (x, y) + + def select_nearby_focus_site(self, next_position: XYPos) -> Optional[XYZPos]: + """Return a focused site near the given position to estimate Z for the next move. + + Looks for all previously focused locations that are within the scan + step size (self._dx, self._dy) of ``next_position``. Among these nearby focused + sites, it returns the most recently imaged one. + + This is suitable for raster or snake scans, where the scan may move along a row + or column and then jump to a new row/column. If no nearby focused sites exist, + returns None. + + :param next_position: The XY position where the next image will be taken. + :return: The XYZ tuple of the closest and most recent focused site, or None if + no focused locations exist. + """ + focused_locations = self.focused_locations + if not focused_locations: + return None + + def sort_key(pos: VisitedScanLocation) -> float: + return self.moves_between(next_position, pos, DistanceMetric.MANHATTAN) + + # Sort by the total number of dx and dy moves between sites, then by most + # recent. Using reverse=True puts the most recent, nearest at the end + nearby_focus_locations = sorted(focused_locations, key=sort_key, reverse=True) + + # Pick the most recent nearby site + return nearby_focus_locations[-1].xyz_tuple + + +class SmartSpiral(RectGridPlanner): """A scan planner that spirals outward from the centre, prioritising short moves. This planner spirals out from the centre, but prioritises short moves over rigidly @@ -321,20 +439,9 @@ class SmartSpiral(ScanPlanner): dx and dy. """ + # The maximum distance for the scan to run in any direction. + # Any future moves which would move beyond this distance are not appended. _max_dist: int = 0 - _dx: int = 0 - _dy: int = 0 - - def __init__( - self, initial_position: XYPos, planner_settings: Optional[dict] = None - ) -> None: - """Set up the lists inherited from ScanPlanner, plus a distance cutoff. - - Use the supplied _dx and _dy to set a distance cutoff for an image to be - considered neighbouring another - """ - super().__init__(initial_position, planner_settings) - self._distance_cutoff: float = max([self._dx, self._dy]) * 1.1 def _is_primary_location( self, location: FutureScanLocation | VisitedScanLocation @@ -352,21 +459,11 @@ class SmartSpiral(ScanPlanner): ] def _parse(self, planner_settings: Optional[dict] = None) -> None: - """Parse SmartSpiral Settings dictionary. + super()._parse(planner_settings) - * ``dx`` - the movement size in x - * ``dy`` - the movement size in y - * ``max_dist`` - The maximum distance to a location can be from the centre. - """ - expected_keys = ["max_dist", "dx", "dy"] - invalid_msg = "SmartSpiral requires a planner_settings dictionary with keys: " - if not planner_settings: - raise ValueError(invalid_msg + ",".join(expected_keys)) - if not all(keys in planner_settings for keys in expected_keys): - raise KeyError(invalid_msg + ",".join(expected_keys)) + if not planner_settings or "max_dist" not in planner_settings: + raise KeyError("SmartSpiral requires max_dist") - self._dx = int(planner_settings["dx"]) - self._dy = int(planner_settings["dy"]) self._max_dist = int(planner_settings["max_dist"]) def _initial_location_list(self) -> list[FutureScanLocation]: @@ -472,21 +569,6 @@ class SmartSpiral(ScanPlanner): # imaged points. self._remaining_locations.append(i_loc) - def _adjacent_positions(self, xy_pos: XYPos) -> XYPosList: - """Return 4 points +/-dx and +/-dy from the input location.""" - return [ - (xy_pos[0] - self._dx, xy_pos[1]), - (xy_pos[0] + self._dx, xy_pos[1]), - (xy_pos[0], xy_pos[1] - self._dy), - (xy_pos[0], xy_pos[1] + self._dy), - ] - - def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos: - """Return an (x,y) position halfway between two input positions.""" - x = (xy_pos1[0] + xy_pos2[0]) // 2 - y = (xy_pos1[1] + xy_pos2[1]) // 2 - return (x, y) - def _re_sort_remaining_locations(self, current_pos: XYPos) -> None: """Sort the remaining positions based on the current location.""" @@ -494,8 +576,10 @@ class SmartSpiral(ScanPlanner): def sort_key(pos: FutureScanLocation) -> tuple[bool, float, float, float]: return ( self._is_primary_location(pos), # False sorts low - self.moves_between(current_pos, pos), - self.moves_between(self._initial_position, pos), + self.moves_between(current_pos, pos, DistanceMetric.CHEBYSHEV), + self.moves_between( + self._initial_position, pos, DistanceMetric.CHEBYSHEV + ), distance_between(current_pos, pos), ) @@ -514,110 +598,72 @@ class SmartSpiral(ScanPlanner): Returns None if no focused locations are present """ # save to variable rather than search for focussed sites each time. - focused_locations = self.focused_locations + focused_locations = self.focused_locations_xyz if not focused_locations: return None - # must be float64 (double precision) to deal with the huge numbers involved! + # must be float64 to deal with large coordinates current_pos = np.array(xy_pos, dtype="float64") - path_pos = np.array(focused_locations, dtype="float64")[:, :2] + focused_arr = np.array(focused_locations, dtype="float64") - # Use linalg.norm to calculate the direct distance between the points - # Note linalg.norm always uses float64 - dists = np.linalg.norm((path_pos - current_pos), axis=1) + # Find focused sites within dx and dy + dx_ok = np.abs(focused_arr[:, 0] - current_pos[0]) <= self._dx + dy_ok = np.abs(focused_arr[:, 1] - current_pos[1]) <= self._dy + nearby_indices = np.where(dx_ok & dy_ok)[0] - # Get indices of all focused sites within distance_cutoff. - # Note np.where always returns a tuple of arrays, hence the trailing [0] - indices = np.where(dists <= self._distance_cutoff)[0] + # If no neighbouring sites were focused, choose the closest + if len(nearby_indices) == 0: + deltas = focused_arr[:, :2] - current_pos + dists = np.linalg.norm(deltas, axis=1) + min_dist = np.min(dists) + nearby_indices = np.where(dists == min_dist)[0] - # Handle the case that no focused positions are within this range, and - # instead use the nearest focused position. This will always return a - # height, due to the check that self._focused_locations exists. - if len(indices) == 0: - distance_cutoff = min(dists) - indices = np.where(dists <= distance_cutoff)[0] + nearby_sites = focused_arr[nearby_indices] - # Turning into an array allows slicing based on a list - focused_locations_array = np.array(focused_locations) + # Choose the lowest z + min_z = np.min(nearby_sites[:, 2]) - # Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best - # if started too low, so the lowest z will perform best - candidates = focused_locations_array[indices] - min_z = np.min(candidates[:, -1]) + # Among those with min z, choose the most recent + chosen_site = nearby_sites[nearby_sites[:, 2] == min_z][-1] - # Find all with the minimum z, and select the latest - chosen_focused_site = candidates[candidates[:, -1] == min_z][-1] - - # Convert back into list so values are of type int instead of np.int32 - return tuple(chosen_focused_site.tolist()) - - def moves_between( - self, - starting_pos: XYPos | np.ndarray | FutureScanLocation, - ending_pos: XYPos | np.ndarray | FutureScanLocation, - ) -> float: - """Return the larger of x moves or y moves between two xy positions. - - :param starting_pos: the position to measure from - :param ending_pos: the position to measure to - - """ - if isinstance(starting_pos, FutureScanLocation): - starting_pos = starting_pos.xy_tuple - if isinstance(ending_pos, FutureScanLocation): - ending_pos = ending_pos.xy_tuple - move_size = np.array([self._dx, self._dy]) - - starting_pos = np.array(starting_pos, dtype="float64") - ending_pos = np.array(ending_pos, dtype="float64") - - displacement_in_moves = (ending_pos - starting_pos) / move_size - - return np.max(np.abs(displacement_in_moves)) + return ( + int(chosen_site[0]), + int(chosen_site[1]), + int(chosen_site[2]), + ) -class SnakeScan(ScanPlanner): - """A scan planner that performs a snake scan, right and down from a corner. +class RegularGridPlanner(RectGridPlanner): + """A scan planner that performs a snake or a raster scan. + + Direction cannot yet be set it always scans, right and down from a corner. This planner starts at the corner of the region to scan, snaking back and forth, starting moving right and down (assuming positive dx and dy.) """ - _dx: int = 0 - _dy: int = 0 _x_count: int = 0 _y_count: int = 0 - - def __init__( - self, initial_position: XYPos, planner_settings: Optional[dict] = None - ) -> None: - """Set up the lists inherited from ScanPlanner, plus a distance cutoff. - - Use the supplied _dx and _dy to set a distance cutoff for an image to be - considered neighbouring another - """ - super().__init__(initial_position, planner_settings) - self._distance_cutoff: float = max([self._dx, self._dy]) * 1.1 + _style: Literal["snake", "raster"] def _parse(self, planner_settings: Optional[dict] = None) -> None: - """Parse SnakeScan Settings dictionary. + super()._parse(planner_settings) - * ``dx`` - the movement size in x - * ``dy`` - the movement size in y - * ``x_count`` - The number of columns in the scan. - * ``y_count`` - The number of rows in the scan. - """ - expected_keys = ["x_count", "y_count", "dx", "dy"] - invalid_msg = "SnakeScan requires a planner_settings dictionary with keys: " - if not planner_settings: - raise ValueError(invalid_msg + ",".join(expected_keys)) - if not all(keys in planner_settings for keys in expected_keys): + expected_keys = ["x_count", "y_count", "style"] + invalid_msg = "RegularGrid requires planner_settings with keys: " + if not planner_settings or not all( + k in planner_settings for k in expected_keys + ): raise KeyError(invalid_msg + ",".join(expected_keys)) - self._dx = int(planner_settings["dx"]) - self._dy = int(planner_settings["dy"]) self._x_count = int(planner_settings["x_count"]) self._y_count = int(planner_settings["y_count"]) + style = planner_settings["style"] + if style not in ("snake", "raster"): + raise ValueError( + f"Unknown regular grid style {style}. Use snake or raster." + ) + self._style = style def _initial_location_list(self) -> list[FutureScanLocation]: """Set the initial list of locations for this scan planner. @@ -632,20 +678,11 @@ class SnakeScan(ScanPlanner): y_count=self._y_count, dx=self._dx, dy=self._dy, - style="snake", + style=self._style, ) return self._grid_to_future_locations(grid) - # The noqa statement is because next_position is unused but is needed for equivalence - # with other workflows that require the next pos to select a neighbour. - def select_nearby_focus_site(self, next_position: XYPos) -> Optional[XYZPos]: # noqa: ARG002 - """For a snake scan, use the most recent focused site to predict focus.""" - focused_locations = self.focused_locations - if not focused_locations: - return None - return focused_locations[-1] - def distance_between( current_pos: XYPos | np.ndarray | FutureScanLocation, diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index a164cccc..e62608af 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -9,6 +9,7 @@ from __future__ import annotations import os from typing import ( Generic, + Literal, Mapping, Optional, TypeVar, @@ -19,9 +20,9 @@ from pydantic import BaseModel import labthings_fastapi as lt from openflexure_microscope_server.scan_planners import ( + RegularGridPlanner, ScanPlanner, SmartSpiral, - SnakeScan, ) from openflexure_microscope_server.stitching import ( STITCHING_RESOLUTION, @@ -67,6 +68,10 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): # CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None _csm: Optional[CameraStageMapper] = lt.thing_slot() + # Camera, stage and autofocus are all required by any scan workflow + _cam: BaseCamera = lt.thing_slot() + _stage: BaseStage = lt.thing_slot() + _autofocus: AutofocusThing = lt.thing_slot() def check_before_start(self, scan_name: str) -> None: """Check before the scan starts. Throw an error if the scan shouldn't start. @@ -116,11 +121,44 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): def acquisition_routine( self, settings: SettingModelType, xyz_pos: tuple[int, int, int] ) -> tuple[bool, Optional[int]]: - """Overload to set the acquisition routine that happens at each scan site.""" + """Overload to set the acquisition routine that happens at each scan site. + + :param settings: The settings for this scan, which should be a SettingModelType + :param xyz_pos: The current position as a tuple or 3 ints. + :return: A tuple of whether an image was taken, and the z-position for focus. + If failed to find focus, returns for the focus z-position. + """ raise NotImplementedError( "Each specific ScanWorkflow must implement an acquisition routine" ) + def _autofocus_and_capture( + self, + xyz_pos: tuple[int, int, int], + dz: int, + images_dir: str, + save_resolution: tuple[int, int], + ) -> tuple[bool, Optional[int]]: + """Autofocus and then capture, this can be used as an acquisition routine. + + :param dz: The dz for autofocus. + :param images_dir: The path to the directory for saving images.. + :param save_resolution: The resolution to save images at. + + :return: A tuple ready to pass out of acquisition routine. In this method, + image is always taken, so first return is True. + + """ + self._autofocus.fast_autofocus(dz=dz) + focus_height = self._stage.get_xyz_position()[2] + filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg" + self._cam.capture_and_save( + jpeg_path=os.path.join(images_dir, filename), + save_resolution=save_resolution, + ) + + return True, focus_height + @lt.property def settings_ui(self) -> list[PropertyControl]: """A list of PropertyControl objects to create the settings in the scan tab.""" @@ -128,13 +166,35 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): "Each scan workflow must implement a settings_ui method." ) - def _require_csm(self) -> CameraStageMapper: - """Give each model the option to require CSM. Return it if present.""" - if self._csm is None: - raise RuntimeError( - "CameraStageMapping not set, and is required for this workflow." - ) - return self._csm + +class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): + """A generic workflow for any scan that captures images on a rectilinear grid.""" + + # Redefine _csm Thing Slot, as CSM is required for any RectGridWorkflow + _csm: CameraStageMapper = lt.thing_slot() + + overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7) + """The fraction that adjacent images should overlap in x and y. + + This must be between 0.1 and 0.7. + """ + + autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000) + """The z distance to perform an autofocus in steps. + + Must be greater than or equal to 200, and less than or equal to 2000. + """ + + # The noqa statement is because scan_name is unused but is needed for equivalence + # with other workflows that may want to validate the scan name. + def check_before_start(self, scan_name: str) -> None: # noqa: ARG002 + """Before starting a scan, check that camera-stage-mapping is set. + + Raise error if: + - camera stage mapping is not set + """ + if self._csm.calibration_required: + raise RuntimeError("Camera Stage Mapping is not calibrated.") def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]: """Use camera stage mapping to calculate x and y displacement from given overlap. @@ -146,26 +206,41 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): :raises RuntimeError: If there is no camera stage mapper Thing available or if CMS isn't calibrated. """ - csm = self._require_csm() - - csm_image_res = csm.image_resolution + csm_image_res = self._csm.image_resolution if csm_image_res is None: - raise RuntimeError("CSM not set. Scan shouldn't have progresses this far.") + raise RuntimeError("CSM not set. Scan shouldn't have progressed this far.") # Calculate displacements in image coordinates dx_img = csm_image_res[1] * (1 - overlap) dy_img = csm_image_res[0] * (1 - overlap) - x_move_stage = csm.convert_image_to_stage_coordinates(x=dx_img, y=0) - y_move_stage = csm.convert_image_to_stage_coordinates(x=0, y=dy_img) + x_move_stage = self._csm.convert_image_to_stage_coordinates(x=dx_img, y=0) + y_move_stage = self._csm.convert_image_to_stage_coordinates(x=0, y=dy_img) # Assume no rotation or skew and take only the aligned axis of vector. # Coerce to positive integer, but correct if x and y are flipped if abs(x_move_stage["x"]) > abs(x_move_stage["y"]): return x_move_stage["x"], y_move_stage["y"] # If not use the other stage axes. Note "dx" will be the movement in camera y. + + self.logger.info( + f"Based on an overlap of {self.overlap}, the stage will make steps of " + f"{y_move_stage['x']}, {x_move_stage['y']}" + ) return y_move_stage["x"], x_move_stage["y"] + def _get_stitching_settings_model(self) -> StitchingSettings: + """Return a stitching settings model based on current settings.""" + return StitchingSettings( + overlap=self.overlap, + correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0], + ) + + @lt.property + def ready(self) -> bool: + """Whether this scanworkflow is ready to start.""" + return not self._csm.calibration_required + class HistoScanSettingsModel(BaseModel): """The settings for a scan with the HistoScanWorkflow. @@ -175,14 +250,14 @@ class HistoScanSettingsModel(BaseModel): """ overlap: float - max_dist: int dx: int dy: int + max_dist: int skip_background: bool smart_stack_params: SmartStackParams -class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): +class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]): """A workflow optimised for scanning Histopathology samples. This workflow automatically plans its own path around a sample spiralling out from @@ -200,12 +275,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): ) _settings_model = HistoScanSettingsModel - _planner_cls: type[ScanPlanner] = SmartSpiral + _planner_cls = SmartSpiral # Thing Slots _background_detector: ChannelDeviationLUV = lt.thing_slot() - _cam: BaseCamera = lt.thing_slot() - _csm: CameraStageMapper = lt.thing_slot() - _autofocus: AutofocusThing = lt.thing_slot() # Scan settings @@ -215,21 +287,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): This uses the settings from the ``BackgroundDetectThing``. """ - autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000) - """The z distance to perform an autofocus in steps. - - Must be greater than or equal to 200, and less than or equal to 2000. - """ - max_range: int = lt.setting(default=45000) """The maximum distance in steps from the centre of the scan.""" - overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7) - """The fraction that adjacent images should overlap in x or y. - - This must be between 0.1 and 0.7. - """ - # Stacking settings stack_images_to_save: int = lt.setting(default=1) @@ -304,16 +364,8 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): :return: A tuple containing the settings model for this workflow and the settings model for stitching. """ - stitching_settings = StitchingSettings( - overlap=self.overlap, - correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0], - ) - + stitching_settings = self._get_stitching_settings_model() dx, dy = self._calc_displacement_from_overlap(self.overlap) - self.logger.info( - f"Based on an overlap of {self.overlap}, the stage will make steps of " - f"{dx}, {dy}" - ) smart_stack_params = self.create_smart_stack_params( images_dir=images_dir, @@ -484,9 +536,6 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): """A list of PropertyControl objects to create the settings in the scan tab.""" return [ property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"), - property_control_for( - self, "skip_background", label="Detect and Skip Empty Fields " - ), property_control_for( self, "stack_images_to_save", label="Images in Stack to Save" ), @@ -498,11 +547,14 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): property_control_for(self, "stack_dz", label="Stack dz (steps)"), property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), property_control_for(self, "max_range", label="Maximum Distance (steps)"), + property_control_for( + self, "skip_background", label="Detect and Skip Empty Fields " + ), ] -class SnakeSettingsModel(BaseModel): - """The settings for a scan with the SnakeWorkflow. +class RegularGridSettingsModel(BaseModel): + """The settings for a scan with a regular grid of dx and dy for x_count, y_count steps. This includes settings calculated when starting. This will be held by smart scan during a scan and serialised to disk. @@ -513,12 +565,108 @@ class SnakeSettingsModel(BaseModel): dy: int x_count: int y_count: int + style: Literal["snake", "raster"] images_dir: str autofocus_dz: int save_resolution: tuple[int, int] -class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): +class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]): + """A base workflow for any workflow that uses a regular rectangular grid.""" + + x_count: int = lt.setting(default=3, ge=1) + """The number of columns in the scan.""" + y_count: int = lt.setting(default=2, ge=1) + """The number of rows in the scan.""" + + _settings_model = RegularGridSettingsModel + _planner_cls = RegularGridPlanner + _grid_style: Literal["snake", "raster"] + + def all_settings( + self, images_dir: str + ) -> tuple[RegularGridSettingsModel, Optional[StitchingSettings]]: + """Return the workflow and stitching settings. + + :param images_dir: The directory that images are to be written to. + :return: A tuple containing the settings model for this workflow and the + settings model for stitching. + """ + stitching_settings = self._get_stitching_settings_model() + dx, dy = self._calc_displacement_from_overlap(self.overlap) + + scan_settings = self._settings_model( + overlap=self.overlap, + dx=dx, + dy=dy, + x_count=self.x_count, + y_count=self.y_count, + style=self._grid_style, + images_dir=images_dir, + autofocus_dz=self.autofocus_dz, + save_resolution=self.save_resolution, + ) + + return scan_settings, stitching_settings + + def pre_scan_routine(self, settings: RegularGridSettingsModel) -> None: + """Perform these steps before starting the scan. + + In this case, only autofocus. + + :param settings: The settings for this scan as as the relevant SettingsModel type. + """ + self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") + + def new_scan_planner( + self, settings: RegularGridSettingsModel, position: Mapping[str, int] + ) -> ScanPlanner: + """Return a new scan planner object. + + :param settings: The settings for this scan as a SnakeSettingsModel + :param position: The starting position as a mapping of axes names to int. + """ + planner_settings = { + "dx": settings.dx, + "dy": settings.dy, + "x_count": settings.x_count, + "y_count": settings.y_count, + "style": settings.style, + } + return self._planner_cls( + initial_position=(position["x"], position["y"]), + planner_settings=planner_settings, + ) + + def acquisition_routine( + self, settings: RegularGridSettingsModel, xyz_pos: tuple[int, int, int] + ) -> tuple[bool, Optional[int]]: + """Autofocus and capture. + + :param settings: The settings for this scan as a RegularGridSettingsModel + :param xyz_pos: The current position as a tuple or 3 ints. + :return: A tuple of whether an image was taken, and the z-position for focus. + If failed to find focus, returns for the focus z-position. + """ + return self._autofocus_and_capture( + xyz_pos=xyz_pos, + dz=settings.autofocus_dz, + images_dir=settings.images_dir, + save_resolution=settings.save_resolution, + ) + + @lt.property + def settings_ui(self) -> list[PropertyControl]: + """A list of PropertyControl objects to create the settings in the scan tab.""" + return [ + property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"), + property_control_for(self, "x_count", label="Number of columns"), + property_control_for(self, "y_count", label="Number of rows"), + property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), + ] + + +class SnakeWorkflow(RegularGridWorkflow): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and snakes @@ -533,139 +681,24 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): ), readonly=True, ) + _grid_style = "snake" - _settings_model = SnakeSettingsModel - _planner_cls: type[ScanPlanner] = SnakeScan - # Thing Slots - _background_detector: ChannelDeviationLUV = lt.thing_slot() - _cam: BaseCamera = lt.thing_slot() - _csm: CameraStageMapper = lt.thing_slot() - _autofocus: AutofocusThing = lt.thing_slot() - _stage: BaseStage = lt.thing_slot() - # Scan settings +class RasterWorkflow(RegularGridWorkflow): + """A workflow optimised for snaking around samples. - autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000) - """The z distance to perform an autofocus in steps. - - Must be greater than or equal to 200, and less than or equal to 2000. + This workflow generates a list of coordinates in a rectangle, and always + moves right across a row, then moves down a row while moving to the starting + column (assuming positive dx and dy). """ - overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7) - """The fraction that adjacent images should overlap in x or y. + display_name: str = lt.property(default="Raster Scan", readonly=True) + ui_blurb: str = lt.property( + default=( + "This scan workflow is optimised for performing a raster scan over a rectangle. It " + "always moves down and right from the starting point, over a defined grid." + ), + readonly=True, + ) - This must be between 0.1 and 0.7. - """ - - x_count: int = lt.setting(default=3) - y_count: int = lt.setting(default=2) - - # The noqa statement is because scan_name is unused but is needed for equivalence - # with other workflows that may want to validate the scan name. - def check_before_start(self, scan_name: str) -> None: # noqa: ARG002 - """Before starting a scan, check that camera-stage-mapping is set. - - Raise error if: - - camera stage mapping is not set - """ - if self._csm.calibration_required: - raise RuntimeError("Camera Stage Mapping is not calibrated.") - - @lt.property - def ready(self) -> bool: - """Whether this scanworkflow is ready to start.""" - return not self._csm.calibration_required - - def all_settings( - self, images_dir: str - ) -> tuple[SnakeSettingsModel, StitchingSettings]: - """Return the workflow and stitching settings. - - :param images_dir: The directory that images are to be written to. - :return: A tuple containing the settings model for this workflow and the - settings model for stitching. - """ - stitching_settings = StitchingSettings( - overlap=self.overlap, - correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0], - ) - - dx, dy = self._calc_displacement_from_overlap(self.overlap) - self.logger.info( - f"Based on an overlap of {self.overlap}, the stage will make steps of " - f"{dx}, {dy}" - ) - - scan_settings = SnakeSettingsModel( - overlap=self.overlap, - dx=dx, - dy=dy, - x_count=self.x_count, - y_count=self.y_count, - images_dir=images_dir, - autofocus_dz=self.autofocus_dz, - save_resolution=self.save_resolution, - ) - - return scan_settings, stitching_settings - - def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: - """Autofocus before starting the scan. - - :param settings: The settings for this scan as a SnakeSettingsModel - """ - self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") - - def new_scan_planner( - self, settings: SnakeSettingsModel, position: Mapping[str, int] - ) -> ScanPlanner: - """Return a new scan planner object. - - :param settings: The settings for this scan as a SnakeSettingsModel - :param position: The starting position as a mapping of axes names to int. - """ - # The initial plan for the scan should be a single x,y position. All future - # moves will be planned around this point. In future, route planner could - # have multiple starting positions, each of which will be visited before the - # scan can end. - planner_settings = { - "dx": settings.dx, - "dy": settings.dy, - "x_count": settings.x_count, - "y_count": settings.y_count, - } - return self._planner_cls( - initial_position=(position["x"], position["y"]), - planner_settings=planner_settings, - ) - - def acquisition_routine( - self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] - ) -> tuple[bool, Optional[int]]: - """Perform acquisition routine. This is run at each scan location. - - :param settings: The settings for this scan as a SnakeSettingsModel - :param xyz_pos: The current position as a tuple or 3 ints. - :return: A tuple of whether an image was taken, and the z-position for focus. - If failed to find focus, returns for the focus z-position. - """ - self._autofocus.fast_autofocus(dz=settings.autofocus_dz) - focus_height = self._stage.get_xyz_position()[2] - filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg" - self._cam.capture_and_save( - jpeg_path=os.path.join(settings.images_dir, filename), - save_resolution=settings.save_resolution, - ) - - imaged = True - return imaged, focus_height - - @lt.property - def settings_ui(self) -> list[PropertyControl]: - """A list of PropertyControl objects to create the settings in the scan tab.""" - return [ - property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"), - property_control_for(self, "x_count", label="Number of columns"), - property_control_for(self, "y_count", label="Number of rows"), - property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), - ] + _grid_style = "raster" diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 3f8b36e0..55e6443c 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -208,6 +208,16 @@ class SmartScanThing(lt.Thing): raise ScanNotRunningError("Cannot get ongoing scan if scan is not running.") return self._ongoing_scan + @lt.property + def all_workflow_names(self) -> list[str]: + """Return a list of all available Scan Workflows.""" + return list(self._all_workflows.keys()) + + @lt.property + def workflow_display_names(self) -> dict[str, str]: + """Return a list of the display names of all available Scan Workflows.""" + return {name: wf.display_name for name, wf in self._all_workflows.items()} + _scan_data: Optional[ActiveScanData] = None @property diff --git a/tests/unit_tests/test_scan_planners.py b/tests/unit_tests/test_scan_planners.py index dea5c530..995f33e7 100644 --- a/tests/unit_tests/test_scan_planners.py +++ b/tests/unit_tests/test_scan_planners.py @@ -89,8 +89,8 @@ def test_bad_smart_spiral_settings(): initial_position = (100, 50) # Class init should raise error if no planner_settings dictionary set - msg = "SmartSpiral requires a planner_settings dictionary with keys" - with pytest.raises(ValueError, match=msg): + msg = "RectGridPlanner requires planner_settings with keys" + with pytest.raises(KeyError, match=msg): scan_planners.SmartSpiral(initial_position=initial_position) planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000} @@ -105,7 +105,7 @@ def test_bad_smart_spiral_settings(): initial_position=initial_position, planner_settings=bad_planner_settings ) - # Class init should raise error if planner_settings if any value can't be cast + # Class init should raise error if any value in planner_settings can't be cast # to int keys = ["dx", "dy", "max_dist"] for badkey in keys: @@ -317,12 +317,18 @@ def test_example_smart_spiral(): assert planner.imaged_locations == expected_planner.imaged_locations -def test_snake_scan_basic_grid(): - """Check that SnakeScan generates a single point for a 1x1 scan.""" +def test_snake_planner_basic_grid(): + """Check that snake scan planner generates a single point for a 1x1 scan.""" initial_position = (100, 50) - planner_settings = {"dx": 100, "dy": 100, "x_count": 1, "y_count": 1} + planner_settings = { + "dx": 100, + "dy": 100, + "x_count": 1, + "y_count": 1, + "style": "snake", + } - planner = scan_planners.SnakeScan( + planner = scan_planners.RegularGridPlanner( initial_position=initial_position, planner_settings=planner_settings, ) @@ -352,11 +358,17 @@ def test_snake_scan_basic_grid(): def test_snake_scan_basic_length(): - """SnakeScan should generate the correct number of locations.""" + """Snake scan planner should generate the correct number of locations.""" initial_position = (100, 50) - planner_settings = {"dx": 100, "dy": 100, "x_count": 3, "y_count": 4} + planner_settings = { + "dx": 100, + "dy": 100, + "x_count": 3, + "y_count": 4, + "style": "snake", + } - planner = scan_planners.SnakeScan( + planner = scan_planners.RegularGridPlanner( initial_position=initial_position, planner_settings=planner_settings, ) @@ -369,9 +381,15 @@ def test_snake_scan_basic_length(): def test_snake_scan_ordering(): """Test that snake scan returns a path in the right order.""" initial_position = (0, 0) - planner_settings = {"dx": 10, "dy": 10, "x_count": 4, "y_count": 3} + planner_settings = { + "dx": 10, + "dy": 10, + "x_count": 4, + "y_count": 3, + "style": "snake", + } - planner = scan_planners.SnakeScan( + planner = scan_planners.RegularGridPlanner( initial_position=initial_position, planner_settings=planner_settings, ) @@ -399,9 +417,9 @@ def test_snake_scan_ordering(): def test_snake_scan_single_row(): """Test edge case of a single row scan.""" initial_position = (0, 0) - planner_settings = {"dx": 5, "dy": 5, "x_count": 4, "y_count": 1} + planner_settings = {"dx": 5, "dy": 5, "x_count": 4, "y_count": 1, "style": "snake"} - planner = scan_planners.SnakeScan( + planner = scan_planners.RegularGridPlanner( initial_position=initial_position, planner_settings=planner_settings, ) @@ -412,9 +430,9 @@ def test_snake_scan_single_row(): def test_snake_scan_single_column(): """Test edge case of a single column scan.""" initial_position = (0, 0) - planner_settings = {"dx": 5, "dy": 5, "x_count": 1, "y_count": 4} + planner_settings = {"dx": 5, "dy": 5, "x_count": 1, "y_count": 4, "style": "snake"} - planner = scan_planners.SnakeScan( + planner = scan_planners.RegularGridPlanner( initial_position=initial_position, planner_settings=planner_settings, ) @@ -425,3 +443,229 @@ def test_snake_scan_single_column(): (0, 10), (0, 15), ] + + +def test_snake_scan_z_propagation(): + """Test that snake planner selects correct previous focus height.""" + initial_position = (0, 0) + planner_settings = { + "dx": 50, + "dy": 50, + "x_count": 5, + "y_count": 5, + "style": "snake", + } + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + expected_z = 1 + + while not planner.scan_complete: + xy_pos, z_est = planner.get_next_location_and_z_estimate() + print(xy_pos, z_est) + + # First position should have no estimate + if xy_pos == (0, 0): + assert z_est is None + else: + # Should estimate from previous focused point + assert z_est == expected_z - 1 + + xyz_pos = (xy_pos[0], xy_pos[1], expected_z) + + planner.mark_location_visited( + xyz_pos, + imaged=True, + focused=True, + ) + + expected_z += 1 + + +def test_raster_planner_basic_grid(): + """Check that raster scan planner generates a single point for a 1x1 scan.""" + initial_position = (100, 50) + planner_settings = { + "dx": 100, + "dy": 100, + "x_count": 1, + "y_count": 1, + "style": "raster", + } + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + assert not planner.scan_complete + # When we start it should want to stay in the initial pos and have + # no z_estimate + xy_pos, z_pos = planner.get_next_location_and_z_estimate() + assert xy_pos == initial_position + assert z_pos is None + + # Try to mark location as imaged with only xy_position + with pytest.raises(ValueError, match="3 value tuple expected"): + planner.mark_location_visited(xy_pos, imaged=False, focused=False) + # scan still not complete + assert not planner.scan_complete + # if we mark this position as visited but not imaged + planner.mark_location_visited( + (xy_pos[0], xy_pos[1], 10), imaged=False, focused=False + ) + # scan is now complete + assert planner.scan_complete + + # if scan is complete, asking for the next location returns an error + with pytest.raises(RuntimeError): + planner.get_next_location_and_z_estimate() + + +def test_raster_scan_basic_length(): + """Raster scan planner should generate the correct number of locations.""" + initial_position = (100, 50) + planner_settings = { + "dx": 100, + "dy": 100, + "x_count": 3, + "y_count": 4, + "style": "raster", + } + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + coords = planner.remaining_locations + + assert len(coords) == 3 * 4 + + +def test_raster_scan_ordering(): + """Test that raster scan returns a path in the right order.""" + initial_position = (0, 0) + planner_settings = { + "dx": 10, + "dy": 10, + "x_count": 4, + "y_count": 3, + "style": "raster", + } + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + coords = planner.remaining_locations + + expected = [ + (0, 0), + (10, 0), + (20, 0), + (30, 0), + (0, 10), + (10, 10), + (20, 10), + (30, 10), + (0, 20), + (10, 20), + (20, 20), + (30, 20), + ] + + assert coords == expected + + +def test_raster_scan_single_row(): + """Test edge case of a single row scan.""" + initial_position = (0, 0) + planner_settings = {"dx": 5, "dy": 5, "x_count": 4, "y_count": 1, "style": "raster"} + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + assert planner.remaining_locations == [(0, 0), (5, 0), (10, 0), (15, 0)] + + +def test_raster_scan_single_column(): + """Test edge case of a single column scan.""" + initial_position = (0, 0) + planner_settings = {"dx": 5, "dy": 5, "x_count": 1, "y_count": 4, "style": "raster"} + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings=planner_settings, + ) + + assert planner.remaining_locations == [ + (0, 0), + (0, 5), + (0, 10), + (0, 15), + ] + + +def test_raster_z_propagation(): + """Test that snake planner selects correct previous focus height. + + Constructs a 5x5 grid in a raster pattern, and test that for each movement, + the chosen next z position is either + - None, for the first point + - the start of the previous row, for the first point in a row + - the previous site otherwise + """ + initial_position = (0, 0) + x_count = 5 + y_count = 5 + dx = 50 + dy = 50 + + planner = scan_planners.RegularGridPlanner( + initial_position=initial_position, + planner_settings={ + "dx": dx, + "dy": dy, + "x_count": x_count, + "y_count": y_count, + "style": "raster", + }, + ) + + visited_positions = [] + current_z = 1 + + while not planner.scan_complete: + visited_count = len(visited_positions) + xy_pos, z_est = planner.get_next_location_and_z_estimate() + print(visited_count) + + if visited_count == 0: + assert z_est is None + else: + # check whether this site is at the start of a new row + if visited_count % x_count == 0: + # if so, get the z position from the start of the previous row + expected_z = visited_positions[-x_count][2] + else: + expected_z = visited_positions[-1][2] + + assert z_est == expected_z + + xyz_pos = (xy_pos[0], xy_pos[1], current_z) + + planner.mark_location_visited( + xyz_pos, + imaged=True, + focused=True, + ) + + visited_positions.append(xyz_pos) + current_z += 1 diff --git a/tests/unit_tests/test_scan_workflows.py b/tests/unit_tests/test_scan_workflows.py index 6406d5ab..b408db06 100644 --- a/tests/unit_tests/test_scan_workflows.py +++ b/tests/unit_tests/test_scan_workflows.py @@ -342,11 +342,11 @@ def test_histo_workflow_settings_ui(histo_workflow): names = [el.property_name for el in ui] expected_names = [ "overlap", - "skip_background", "stack_images_to_save", "stack_min_images_to_test", "stack_dz", "autofocus_dz", "max_range", + "skip_background", ] assert names == expected_names diff --git a/webapp/src/components/tabContentComponents/slideScanContent.vue b/webapp/src/components/tabContentComponents/slideScanContent.vue index eeef6d86..9682f255 100644 --- a/webapp/src/components/tabContentComponents/slideScanContent.vue +++ b/webapp/src/components/tabContentComponents/slideScanContent.vue @@ -2,6 +2,20 @@
+ +
+ + + +

{{ workflowDisplayName }}

@@ -145,6 +159,7 @@ export default { workflowSettings: [], workflowDisplayName: undefined, workflowBlurb: undefined, + workflowOptions: [], }; }, @@ -159,6 +174,7 @@ export default { async created() { this.readSettings(); + this.workflowOptions = await this.readThingProperty("smart_scan", "workflow_display_names"); }, methods: { @@ -216,6 +232,21 @@ export default { setTimeout(this.pollScan, 1000); // keep rescheduling until it's stopped } }, + async setWorkflow(name) { + try { + this.workflowName = name; + + await this.writeThingProperty("smart_scan", "workflow_name", name); + + // refresh UI + await this.readSettings(); + } catch (err) { + this.modalError(err); + + // revert if server rejected + this.workflowName = await this.readThingProperty("smart_scan", "workflow_name"); + } + }, async downloadZipFile(response) { const scan_name = response.input.scan_name; const filename = `${scan_name}_images.zip`;