From 0177596fb65e192620ba5ed52b8c7ad814a61826 Mon Sep 17 00:00:00 2001 From: Joe Knapper Date: Sat, 7 Feb 2026 16:31:51 +0000 Subject: [PATCH 01/19] Add raster scan, with generic typing to allow subclassed settings --- ofm_config_simulation.json | 1 + .../scan_planners.py | 63 +++++++++++++++++-- .../things/scan_workflows.py | 46 +++++++++++--- 3 files changed, 97 insertions(+), 13 deletions(-) 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..76bdc1c7 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -637,14 +637,67 @@ class SnakeScan(ScanPlanner): 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.""" + 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 - return focused_locations[-1] + + next_pos_arr = np.array(next_position, dtype="float64") + path_arr = np.array(focused_locations, dtype="float64")[:, :2] + + # Find all focused positions within dx and dy of next_position + dx_ok = np.abs(path_arr[:, 0] - next_pos_arr[0]) <= self._dx + dy_ok = np.abs(path_arr[:, 1] - next_pos_arr[1]) <= self._dy + nearby_indices = np.where(dx_ok & dy_ok)[0] + + if len(nearby_indices) == 0: + return None + + # Pick the most recent nearby site + return focused_locations[nearby_indices[-1]] + + +class RasterScan(SnakeScan): + """A scan planner that performs a snake scan, always moving right and down. + + This planner starts at the corner of the region to scan, and always moves right + to the end of the row, then down to the next row (assuming positive dx, dy). + + This is subclassed from SnakeScan, as the only difference in behaviour is in + building the initial path. + """ + + def _initial_location_list(self) -> list[FutureScanLocation]: + """Set the initial list of locations for this scan planner. + + This is called on initialisation. + + For raster scan, this is the full grid, and none will be added during scanning. + """ + grid = create_rectangular_scan_path( + starting_pos=self._initial_position, + x_count=self._x_count, + y_count=self._y_count, + dx=self._dx, + dy=self._dy, + style="raster", + ) + + return self._grid_to_future_locations(grid) def distance_between( diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index a164cccc..ee499ece 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -19,6 +19,7 @@ from pydantic import BaseModel import labthings_fastapi as lt from openflexure_microscope_server.scan_planners import ( + RasterScan, ScanPlanner, SmartSpiral, SnakeScan, @@ -518,7 +519,7 @@ class SnakeSettingsModel(BaseModel): save_resolution: tuple[int, int] -class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): +class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and snakes @@ -534,10 +535,9 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): readonly=True, ) - _settings_model = SnakeSettingsModel + _settings_model: type[SettingModelType] = 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() @@ -578,7 +578,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): def all_settings( self, images_dir: str - ) -> tuple[SnakeSettingsModel, StitchingSettings]: + ) -> tuple[SettingModelType, StitchingSettings]: """Return the workflow and stitching settings. :param images_dir: The directory that images are to be written to. @@ -596,7 +596,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): f"{dx}, {dy}" ) - scan_settings = SnakeSettingsModel( + scan_settings = self._settings_model( overlap=self.overlap, dx=dx, dy=dy, @@ -609,7 +609,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): return scan_settings, stitching_settings - def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: + def pre_scan_routine(self, settings: SettingModelType) -> None: """Autofocus before starting the scan. :param settings: The settings for this scan as a SnakeSettingsModel @@ -617,7 +617,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") def new_scan_planner( - self, settings: SnakeSettingsModel, position: Mapping[str, int] + self, settings: SettingModelType, position: Mapping[str, int] ) -> ScanPlanner: """Return a new scan planner object. @@ -640,7 +640,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): ) def acquisition_routine( - self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] + self, settings: SettingModelType, xyz_pos: tuple[int, int, int] ) -> tuple[bool, Optional[int]]: """Perform acquisition routine. This is run at each scan location. @@ -669,3 +669,33 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): property_control_for(self, "y_count", label="Number of rows"), property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), ] + + +class RasterSettingsModel(SnakeSettingsModel): + """The settings for a scan with the Raster Workflow. + + This is identical to the SnakeSettings, but is subclassed for clarity. + """ + + pass + + +class RasterWorkflow(SnakeWorkflow[RasterSettingsModel]): + """A workflow optimised for snaking around samples. + + 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). + """ + + 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, + ) + + _settings_model = RasterSettingsModel + _planner_cls: type[ScanPlanner] = RasterScan From 3beb8e114762fa7b134b5240aa3705584aa9bc1d Mon Sep 17 00:00:00 2001 From: Joe Knapper Date: Sat, 7 Feb 2026 19:27:32 +0000 Subject: [PATCH 02/19] Auto-populate workflow dropdown --- .../things/smart_scan.py | 10 +++++ .../tabContentComponents/slideScanContent.vue | 45 +++++++++++++++++++ 2 files changed, 55 insertions(+) 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/webapp/src/components/tabContentComponents/slideScanContent.vue b/webapp/src/components/tabContentComponents/slideScanContent.vue index eeef6d86..b10b9d49 100644 --- a/webapp/src/components/tabContentComponents/slideScanContent.vue +++ b/webapp/src/components/tabContentComponents/slideScanContent.vue @@ -2,6 +2,24 @@
+ +
+ + + +

{{ workflowDisplayName }}

@@ -145,6 +163,7 @@ export default { workflowSettings: [], workflowDisplayName: undefined, workflowBlurb: undefined, + workflowOptions: [], }; }, @@ -159,6 +178,10 @@ export default { async created() { this.readSettings(); + this.workflowOptions = await this.readThingProperty( + "smart_scan", + "workflow_display_names", + ); }, methods: { @@ -216,6 +239,28 @@ 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`; From 1a0dd71f30d9e8eab7b3dd75485e34b78d12e567 Mon Sep 17 00:00:00 2001 From: Joe Knapper Date: Mon, 9 Feb 2026 15:45:16 +0000 Subject: [PATCH 03/19] Intermediate level RectangleScan class for all three scanners --- .../scan_planners.py | 289 ++++++++---------- 1 file changed, 134 insertions(+), 155 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 76bdc1c7..afe81a4d 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -304,7 +304,97 @@ class ScanPlanner: return [FutureScanLocation(location) for line in grid for location in line] -class SmartSpiral(ScanPlanner): +class RectangleScan(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 = "RectangleScan requires planner_settings with keys: " + if not planner_settings: + raise ValueError(invalid_msg + ",".join(expected_keys)) + if 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, + 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)) + + 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 + + next_pos_arr = np.array(next_position, dtype="float64") + path_arr = np.array(focused_locations, dtype="float64")[:, :2] + + # Find all focused positions within dx and dy of next_position. + # Don't just use _adjacent_positions as some grids might have intermediate sites. + dx_ok = np.abs(path_arr[:, 0] - next_pos_arr[0]) <= self._dx + dy_ok = np.abs(path_arr[:, 1] - next_pos_arr[1]) <= self._dy + nearby_indices = np.where(dx_ok & dy_ok)[0] + + if len(nearby_indices) == 0: + return None + + # Pick the most recent nearby site + return focused_locations[nearby_indices[-1]] + + +class SmartSpiral(RectangleScan): """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 +411,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 +431,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 "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 +541,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.""" @@ -518,104 +572,51 @@ class SmartSpiral(ScanPlanner): if not focused_locations: return None - # must be float64 (double precision) to deal with the huge numbers involved! 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 site(s) that are 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 + best_sites = nearby_sites[nearby_sites[:, 2] == min_z] + chosen_site = best_sites[-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 tuple(chosen_site.astype(int)) -class SnakeScan(ScanPlanner): +class SnakeScan(RectangleScan): """A scan planner that performs a snake scan, 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 - 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"] + invalid_msg = "SnakeScan requires planner_settings with keys: " + if 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"]) @@ -637,50 +638,28 @@ class SnakeScan(ScanPlanner): return self._grid_to_future_locations(grid) - 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 - - next_pos_arr = np.array(next_position, dtype="float64") - path_arr = np.array(focused_locations, dtype="float64")[:, :2] - - # Find all focused positions within dx and dy of next_position - dx_ok = np.abs(path_arr[:, 0] - next_pos_arr[0]) <= self._dx - dy_ok = np.abs(path_arr[:, 1] - next_pos_arr[1]) <= self._dy - nearby_indices = np.where(dx_ok & dy_ok)[0] - - if len(nearby_indices) == 0: - return None - - # Pick the most recent nearby site - return focused_locations[nearby_indices[-1]] - - -class RasterScan(SnakeScan): +class RasterScan(RectangleScan): """A scan planner that performs a snake scan, always moving right and down. This planner starts at the corner of the region to scan, and always moves right to the end of the row, then down to the next row (assuming positive dx, dy). - - This is subclassed from SnakeScan, as the only difference in behaviour is in - building the initial path. """ + _x_count: int = 0 + _y_count: int = 0 + + def _parse(self, planner_settings: Optional[dict] = None) -> None: + super()._parse(planner_settings) + + expected_keys = ["x_count", "y_count"] + invalid_msg = "SnakeScan requires planner_settings with keys: " + if not all(k in planner_settings for k in expected_keys): + raise KeyError(invalid_msg + ",".join(expected_keys)) + + self._x_count = int(planner_settings["x_count"]) + self._y_count = int(planner_settings["y_count"]) + def _initial_location_list(self) -> list[FutureScanLocation]: """Set the initial list of locations for this scan planner. From fb7a8b49599cd267a585b346f429a1daa66a9c8f Mon Sep 17 00:00:00 2001 From: Joe Knapper Date: Mon, 9 Feb 2026 17:09:38 +0000 Subject: [PATCH 04/19] Prettier on slideScanContent --- .../tabContentComponents/slideScanContent.vue | 60 +++++++------------ 1 file changed, 23 insertions(+), 37 deletions(-) diff --git a/webapp/src/components/tabContentComponents/slideScanContent.vue b/webapp/src/components/tabContentComponents/slideScanContent.vue index b10b9d49..9682f255 100644 --- a/webapp/src/components/tabContentComponents/slideScanContent.vue +++ b/webapp/src/components/tabContentComponents/slideScanContent.vue @@ -2,24 +2,20 @@
- -
- + +
+ - -
+ +

{{ workflowDisplayName }}

@@ -178,10 +174,7 @@ export default { async created() { this.readSettings(); - this.workflowOptions = await this.readThingProperty( - "smart_scan", - "workflow_display_names", - ); + this.workflowOptions = await this.readThingProperty("smart_scan", "workflow_display_names"); }, methods: { @@ -240,25 +233,18 @@ export default { } }, async setWorkflow(name) { - try { - this.workflowName = name; + try { + this.workflowName = name; - await this.writeThingProperty( - "smart_scan", - "workflow_name", - name - ); + await this.writeThingProperty("smart_scan", "workflow_name", name); - // refresh UI - await this.readSettings(); - } catch (err) { - this.modalError(err); + // refresh UI + await this.readSettings(); + } catch (err) { + this.modalError(err); - // revert if server rejected - this.workflowName = await this.readThingProperty( - "smart_scan", - "workflow_name" - ); + // revert if server rejected + this.workflowName = await this.readThingProperty("smart_scan", "workflow_name"); } }, async downloadZipFile(response) { From 6294cde47176fda4f66ccf0ca5f7fa429fda7712 Mon Sep 17 00:00:00 2001 From: jaknapper Date: Mon, 9 Feb 2026 19:35:19 +0000 Subject: [PATCH 05/19] Fully subclass Rectangle basescanner for raster and snake --- ofm_config_full.json | 1 + .../scan_planners.py | 10 +- .../things/scan_workflows.py | 157 +++++++++++++++++- 3 files changed, 156 insertions(+), 12 deletions(-) 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/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index afe81a4d..24d42d9c 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -433,7 +433,7 @@ class SmartSpiral(RectangleScan): def _parse(self, planner_settings: Optional[dict] = None) -> None: super()._parse(planner_settings) - if "max_dist" not in planner_settings: + if not planner_settings or "max_dist" not in planner_settings: raise KeyError("SmartSpiral requires max_dist") self._max_dist = int(planner_settings["max_dist"]) @@ -614,7 +614,9 @@ class SnakeScan(RectangleScan): expected_keys = ["x_count", "y_count"] invalid_msg = "SnakeScan requires planner_settings with keys: " - if not all(k in planner_settings for k in expected_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._x_count = int(planner_settings["x_count"]) @@ -654,7 +656,9 @@ class RasterScan(RectangleScan): expected_keys = ["x_count", "y_count"] invalid_msg = "SnakeScan requires planner_settings with keys: " - if not all(k in planner_settings for k in expected_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._x_count = int(planner_settings["x_count"]) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index ee499ece..a2f0825d 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -519,7 +519,7 @@ class SnakeSettingsModel(BaseModel): save_resolution: tuple[int, int] -class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): +class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and snakes @@ -535,7 +535,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): readonly=True, ) - _settings_model: type[SettingModelType] = SnakeSettingsModel + _settings_model = SnakeSettingsModel _planner_cls: type[ScanPlanner] = SnakeScan # Thing Slots _cam: BaseCamera = lt.thing_slot() @@ -578,7 +578,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): def all_settings( self, images_dir: str - ) -> tuple[SettingModelType, StitchingSettings]: + ) -> tuple[SnakeSettingsModel, StitchingSettings]: """Return the workflow and stitching settings. :param images_dir: The directory that images are to be written to. @@ -609,7 +609,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): return scan_settings, stitching_settings - def pre_scan_routine(self, settings: SettingModelType) -> None: + def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: """Autofocus before starting the scan. :param settings: The settings for this scan as a SnakeSettingsModel @@ -617,7 +617,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") def new_scan_planner( - self, settings: SettingModelType, position: Mapping[str, int] + self, settings: SnakeSettingsModel, position: Mapping[str, int] ) -> ScanPlanner: """Return a new scan planner object. @@ -640,7 +640,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): ) def acquisition_routine( - self, settings: SettingModelType, xyz_pos: tuple[int, int, int] + self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] ) -> tuple[bool, Optional[int]]: """Perform acquisition routine. This is run at each scan location. @@ -671,16 +671,23 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): ] -class RasterSettingsModel(SnakeSettingsModel): +class RasterSettingsModel(BaseModel): """The settings for a scan with the Raster Workflow. This is identical to the SnakeSettings, but is subclassed for clarity. """ - pass + overlap: float + dx: int + dy: int + x_count: int + y_count: int + images_dir: str + autofocus_dz: int + save_resolution: tuple[int, int] -class RasterWorkflow(SnakeWorkflow[RasterSettingsModel]): +class RasterWorkflow(ScanWorkflow[RasterSettingsModel]): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and always @@ -699,3 +706,135 @@ class RasterWorkflow(SnakeWorkflow[RasterSettingsModel]): _settings_model = RasterSettingsModel _planner_cls: type[ScanPlanner] = RasterScan + # Thing Slots + _cam: BaseCamera = lt.thing_slot() + _csm: CameraStageMapper = lt.thing_slot() + _autofocus: AutofocusThing = lt.thing_slot() + _stage: BaseStage = lt.thing_slot() + + # Scan settings + + 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. + """ + + 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. + """ + + 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[RasterSettingsModel, 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 = self._settings_model( + 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: RasterSettingsModel) -> 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: RasterSettingsModel, 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: RasterSettingsModel, 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 RasterSettingsModel + :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)"), + ] From f1845c303922886964e886f108495200eef11c01 Mon Sep 17 00:00:00 2001 From: jaknapper Date: Tue, 10 Feb 2026 14:04:01 +0000 Subject: [PATCH 06/19] RectangleWorkflow as a baseclass --- .../scan_planners.py | 12 +- .../things/scan_workflows.py | 243 ++++-------------- tests/unit_tests/test_scan_planners.py | 4 +- 3 files changed, 59 insertions(+), 200 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 24d42d9c..aeadb113 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -572,6 +572,7 @@ class SmartSpiral(RectangleScan): if not focused_locations: return None + # must be float64 to deal with large coordinates current_pos = np.array(xy_pos, dtype="float64") focused_arr = np.array(focused_locations, dtype="float64") @@ -580,7 +581,7 @@ class SmartSpiral(RectangleScan): dy_ok = np.abs(focused_arr[:, 1] - current_pos[1]) <= self._dy nearby_indices = np.where(dx_ok & dy_ok)[0] - # If no neighbouring sites were focused, choose the site(s) that are closest + # 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) @@ -593,10 +594,13 @@ class SmartSpiral(RectangleScan): min_z = np.min(nearby_sites[:, 2]) # Among those with min z, choose the most recent - best_sites = nearby_sites[nearby_sites[:, 2] == min_z] - chosen_site = best_sites[-1] + chosen_site = nearby_sites[nearby_sites[:, 2] == min_z][-1] - return tuple(chosen_site.astype(int)) + return ( + int(chosen_site[0]), + int(chosen_site[1]), + int(chosen_site[2]), + ) class SnakeScan(RectangleScan): diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index a2f0825d..96b75af2 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -485,9 +485,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" ), @@ -499,11 +496,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 RectangleSettingsModel(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. @@ -519,49 +519,39 @@ class SnakeSettingsModel(BaseModel): save_resolution: tuple[int, int] -class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): - """A workflow optimised for snaking around samples. +TypeSettings = TypeVar("TypeSettings", bound=RectangleSettingsModel) - This workflow generates a list of coordinates in a rectangle, and snakes - around them from the top left (assuming positive dx and dy). - """ - display_name: str = lt.property(default="Snake Scan", readonly=True) - ui_blurb: str = lt.property( - default=( - "This scan workflow is optimised for scanning over a rectangle. It " - "snakes down and right from the starting point, over a defined grid." - ), - readonly=True, - ) +class SnakeSettingsModel(RectangleSettingsModel): + """Settings for Snake scan.""" - _settings_model = SnakeSettingsModel - _planner_cls: type[ScanPlanner] = SnakeScan + pass + + +class RasterSettingsModel(RectangleSettingsModel): + """Settings for Raster scan.""" + + pass + + +class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): + """Abstract workflow for rectangular scans.""" + + # Settings and planner must be set by subclass + _settings_model: type[TypeSettings] + _planner_cls: type[ScanPlanner] # Thing Slots _cam: BaseCamera = lt.thing_slot() _csm: CameraStageMapper = lt.thing_slot() _autofocus: AutofocusThing = lt.thing_slot() _stage: BaseStage = lt.thing_slot() - # Scan settings - + # Shared scan settings 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. - """ - 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. - """ - 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. @@ -576,9 +566,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): """Whether this scanworkflow is ready to start.""" return not self._csm.calibration_required - def all_settings( - self, images_dir: str - ) -> tuple[SnakeSettingsModel, StitchingSettings]: + def all_settings(self, images_dir: str) -> tuple[TypeSettings, StitchingSettings]: """Return the workflow and stitching settings. :param images_dir: The directory that images are to be written to. @@ -609,25 +597,21 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): return scan_settings, stitching_settings - def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: - """Autofocus before starting the scan. + def pre_scan_routine(self, settings: TypeSettings) -> None: + """Perform these steps before starting the scan. - :param settings: The settings for this scan as a SnakeSettingsModel + In this case, only autofocus. """ self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") def new_scan_planner( - self, settings: SnakeSettingsModel, position: Mapping[str, int] + self, settings: TypeSettings, 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, @@ -640,13 +624,14 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): ) def acquisition_routine( - self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] + self, settings: TypeSettings, 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 settings: The settings for this scan as the correct setting model :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. + In this method, image is always taken, so first return is True. If failed to find focus, returns for the focus z-position. """ self._autofocus.fast_autofocus(dz=settings.autofocus_dz) @@ -657,8 +642,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): save_resolution=settings.save_resolution, ) - imaged = True - return imaged, focus_height + return True, focus_height @lt.property def settings_ui(self) -> list[PropertyControl]: @@ -671,23 +655,26 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): ] -class RasterSettingsModel(BaseModel): - """The settings for a scan with the Raster Workflow. +class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]): + """A workflow optimised for snaking around samples. - This is identical to the SnakeSettings, but is subclassed for clarity. + This workflow generates a list of coordinates in a rectangle, and snakes + around them from the top left (assuming positive dx and dy). """ - overlap: float - dx: int - dy: int - x_count: int - y_count: int - images_dir: str - autofocus_dz: int - save_resolution: tuple[int, int] + display_name: str = lt.property(default="Snake Scan", readonly=True) + ui_blurb: str = lt.property( + default=( + "This scan workflow is optimised for scanning over a rectangle. It " + "snakes down and right from the starting point, over a defined grid." + ), + readonly=True, + ) + _settings_model = SnakeSettingsModel + _planner_cls = SnakeScan -class RasterWorkflow(ScanWorkflow[RasterSettingsModel]): +class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and always @@ -705,136 +692,4 @@ class RasterWorkflow(ScanWorkflow[RasterSettingsModel]): ) _settings_model = RasterSettingsModel - _planner_cls: type[ScanPlanner] = RasterScan - # Thing Slots - _cam: BaseCamera = lt.thing_slot() - _csm: CameraStageMapper = lt.thing_slot() - _autofocus: AutofocusThing = lt.thing_slot() - _stage: BaseStage = lt.thing_slot() - - # Scan settings - - 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. - """ - - 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. - """ - - 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[RasterSettingsModel, 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 = self._settings_model( - 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: RasterSettingsModel) -> 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: RasterSettingsModel, 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: RasterSettingsModel, 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 RasterSettingsModel - :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)"), - ] + _planner_cls = RasterScan diff --git a/tests/unit_tests/test_scan_planners.py b/tests/unit_tests/test_scan_planners.py index dea5c530..cf4c8acc 100644 --- a/tests/unit_tests/test_scan_planners.py +++ b/tests/unit_tests/test_scan_planners.py @@ -89,7 +89,7 @@ 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" + msg = "RectangleScan requires planner_settings with keys" with pytest.raises(ValueError, match=msg): scan_planners.SmartSpiral(initial_position=initial_position) @@ -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: From 1f01717a1b9c4f46e15181543cb45bdf412f0f2d Mon Sep 17 00:00:00 2001 From: jaknapper Date: Tue, 10 Feb 2026 14:24:41 +0000 Subject: [PATCH 07/19] Fix UI workflow test as order changed --- .../things/scan_workflows.py | 24 ++++++++++++++++--- tests/unit_tests/test_scan_workflows.py | 2 +- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index 96b75af2..b3fb40a4 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -548,10 +548,26 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): # Shared scan settings autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000) - overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7) - x_count: int = lt.setting(default=3) - y_count: int = lt.setting(default=2) + """The z distance to perform an autofocus in steps. + + Must be greater than or equal to 200, and less than or equal to 2000. + """ + + 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. + """ + + x_count: int = lt.setting(default=3, ge=1) + y_count: int = lt.setting(default=2, ge=1) + """The number of columns and rows in the scan. + Must be at least 1. + """ + + # 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. @@ -601,6 +617,8 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): """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") 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 From 15798d9b7dbbe1e80c58911cd5e41a9276fe8857 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 10 Feb 2026 16:02:18 +0000 Subject: [PATCH 08/19] Further layering of the classes for scan workflows --- .../things/scan_workflows.py | 236 ++++++++---------- 1 file changed, 110 insertions(+), 126 deletions(-) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index b3fb40a4..7dac4c50 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -68,6 +68,9 @@ 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() + _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. @@ -117,11 +120,42 @@ 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 as a HistoScanSettingsModel + :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]]: + """Atuofocus and then capture, this can be used as an acquisition routine. + + :param dz: The dz for autofoucs. + :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. + """ + 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.""" @@ -129,13 +163,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.""" + + # Thing Slots + _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 or 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. @@ -147,9 +203,7 @@ 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.") @@ -157,16 +211,33 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): 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. @@ -176,14 +247,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 @@ -201,12 +272,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 @@ -216,21 +284,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) @@ -305,16 +361,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, @@ -502,7 +550,7 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): ] -class RectangleSettingsModel(BaseModel): +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 @@ -519,86 +567,28 @@ class RectangleSettingsModel(BaseModel): save_resolution: tuple[int, int] -TypeSettings = TypeVar("TypeSettings", bound=RectangleSettingsModel) - - -class SnakeSettingsModel(RectangleSettingsModel): - """Settings for Snake scan.""" - - pass - - -class RasterSettingsModel(RectangleSettingsModel): - """Settings for Raster scan.""" - - pass - - -class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): - """Abstract workflow for rectangular scans.""" - - # Settings and planner must be set by subclass - _settings_model: type[TypeSettings] - _planner_cls: type[ScanPlanner] - # Thing Slots - _cam: BaseCamera = lt.thing_slot() - _csm: CameraStageMapper = lt.thing_slot() - _autofocus: AutofocusThing = lt.thing_slot() - _stage: BaseStage = lt.thing_slot() - - # Shared scan settings - 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. - """ - - 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. - """ +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 columns and rows in the scan. - Must be at least 1. - """ + """The number of rows in the scan.""" - # 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. + _settings_model = RegularGridSettingsModel + _planner_cls: type[SnakeScan] | type[RasterScan] - 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[TypeSettings, StitchingSettings]: + 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 = 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}" - ) scan_settings = self._settings_model( overlap=self.overlap, @@ -613,7 +603,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): return scan_settings, stitching_settings - def pre_scan_routine(self, settings: TypeSettings) -> None: + def pre_scan_routine(self, settings: RegularGridSettingsModel) -> None: """Perform these steps before starting the scan. In this case, only autofocus. @@ -623,7 +613,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") def new_scan_planner( - self, settings: TypeSettings, position: Mapping[str, int] + self, settings: RegularGridSettingsModel, position: Mapping[str, int] ) -> ScanPlanner: """Return a new scan planner object. @@ -642,26 +632,22 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): ) def acquisition_routine( - self, settings: TypeSettings, xyz_pos: tuple[int, int, int] + self, settings: RegularGridSettingsModel, xyz_pos: tuple[int, int, int] ) -> tuple[bool, Optional[int]]: - """Perform acquisition routine. This is run at each scan location. + """Autofocus and capture. - :param settings: The settings for this scan as the correct setting model + :param settings: The settings for this scan as a HistoScanSettingsModel :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. - In this method, image is always taken, so first return is True. 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), + return self._autofocus_and_capture( + xyz_pos=xyz_pos, + dz=settings.autofocus_dz, + images_dir=settings.images_dir, save_resolution=settings.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.""" @@ -673,7 +659,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]): ] -class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]): +class SnakeWorkflow(RegularGridWorkflow): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and snakes @@ -688,11 +674,10 @@ class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]): ), readonly=True, ) - _settings_model = SnakeSettingsModel _planner_cls = SnakeScan -class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]): +class RasterWorkflow(RegularGridWorkflow): """A workflow optimised for snaking around samples. This workflow generates a list of coordinates in a rectangle, and always @@ -709,5 +694,4 @@ class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]): readonly=True, ) - _settings_model = RasterSettingsModel _planner_cls = RasterScan From 0e5f9f46ca4affac8b38481e2aa6de2817e2264d Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 10 Feb 2026 16:37:17 +0000 Subject: [PATCH 09/19] Fix docstring error --- src/openflexure_microscope_server/scan_planners.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index aeadb113..e08ec62f 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -363,7 +363,7 @@ class RectangleScan(ScanPlanner): """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 + 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 From 1221b6ca02bece4f4f280aa43b9cdf892cd139a6 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 10 Feb 2026 16:40:11 +0000 Subject: [PATCH 10/19] Apply suggestions from code review of branch More-workflow-layers Co-authored-by: Joe Knapper --- src/openflexure_microscope_server/things/scan_workflows.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index 7dac4c50..1afac9b8 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -68,6 +68,7 @@ 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() From 9e477a26d85afcbde085b682cbd6495b2446fb7e Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 10 Feb 2026 17:33:31 +0000 Subject: [PATCH 11/19] Simplify planners a little to consolidate snake and raster --- .../scan_planners.py | 65 +++++-------------- .../things/scan_workflows.py | 14 ++-- 2 files changed, 25 insertions(+), 54 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index e08ec62f..d684634f 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -304,7 +304,7 @@ class ScanPlanner: return [FutureScanLocation(location) for line in grid for location in line] -class RectangleScan(ScanPlanner): +class RectGridPlanner(ScanPlanner): """Base class for planners that operate on a rectangular grid.""" _dx: int = 0 @@ -312,7 +312,7 @@ class RectangleScan(ScanPlanner): def _parse(self, planner_settings: Optional[dict] = None) -> None: expected_keys = ["dx", "dy"] - invalid_msg = "RectangleScan requires planner_settings with keys: " + invalid_msg = "RectGridPlanner requires planner_settings with keys: " if not planner_settings: raise ValueError(invalid_msg + ",".join(expected_keys)) if not all(k in planner_settings for k in expected_keys): @@ -394,7 +394,7 @@ class RectangleScan(ScanPlanner): return focused_locations[nearby_indices[-1]] -class SmartSpiral(RectangleScan): +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 @@ -603,8 +603,10 @@ class SmartSpiral(RectangleScan): ) -class SnakeScan(RectangleScan): - """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.) @@ -612,11 +614,12 @@ class SnakeScan(RectangleScan): _x_count: int = 0 _y_count: int = 0 + _style: Literal["snake", "raster"] def _parse(self, planner_settings: Optional[dict] = None) -> None: super()._parse(planner_settings) - expected_keys = ["x_count", "y_count"] + expected_keys = ["x_count", "y_count", "style"] invalid_msg = "SnakeScan requires planner_settings with keys: " if not planner_settings or not all( k in planner_settings for k in expected_keys @@ -625,6 +628,12 @@ class SnakeScan(RectangleScan): 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. @@ -639,49 +648,7 @@ class SnakeScan(RectangleScan): y_count=self._y_count, dx=self._dx, dy=self._dy, - style="snake", - ) - - return self._grid_to_future_locations(grid) - - -class RasterScan(RectangleScan): - """A scan planner that performs a snake scan, always moving right and down. - - This planner starts at the corner of the region to scan, and always moves right - to the end of the row, then down to the next row (assuming positive dx, dy). - """ - - _x_count: int = 0 - _y_count: int = 0 - - def _parse(self, planner_settings: Optional[dict] = None) -> None: - super()._parse(planner_settings) - - expected_keys = ["x_count", "y_count"] - invalid_msg = "SnakeScan 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._x_count = int(planner_settings["x_count"]) - self._y_count = int(planner_settings["y_count"]) - - def _initial_location_list(self) -> list[FutureScanLocation]: - """Set the initial list of locations for this scan planner. - - This is called on initialisation. - - For raster scan, this is the full grid, and none will be added during scanning. - """ - grid = create_rectangular_scan_path( - starting_pos=self._initial_position, - x_count=self._x_count, - y_count=self._y_count, - dx=self._dx, - dy=self._dy, - style="raster", + style=self._style, ) return self._grid_to_future_locations(grid) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index 1afac9b8..9c538bb4 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,10 +20,9 @@ from pydantic import BaseModel import labthings_fastapi as lt from openflexure_microscope_server.scan_planners import ( - RasterScan, + RegularGridPlanner, ScanPlanner, SmartSpiral, - SnakeScan, ) from openflexure_microscope_server.stitching import ( STITCHING_RESOLUTION, @@ -563,6 +563,7 @@ class RegularGridSettingsModel(BaseModel): dy: int x_count: int y_count: int + style: Literal["snake", "raster"] images_dir: str autofocus_dz: int save_resolution: tuple[int, int] @@ -577,7 +578,8 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]): """The number of rows in the scan.""" _settings_model = RegularGridSettingsModel - _planner_cls: type[SnakeScan] | type[RasterScan] + _planner_cls = RegularGridPlanner + _grid_style: Literal["snake", "raster"] def all_settings( self, images_dir: str @@ -597,6 +599,7 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]): 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, @@ -626,6 +629,7 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]): "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"]), @@ -675,7 +679,7 @@ class SnakeWorkflow(RegularGridWorkflow): ), readonly=True, ) - _planner_cls = SnakeScan + _grid_style = "snake" class RasterWorkflow(RegularGridWorkflow): @@ -695,4 +699,4 @@ class RasterWorkflow(RegularGridWorkflow): readonly=True, ) - _planner_cls = RasterScan + _grid_style = "raster" From 86fa48fb9b8cdf0da889270c73d1a5c87f2395e6 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 11 Feb 2026 09:53:44 +0000 Subject: [PATCH 12/19] Update select_nearby_focus_site to use moves between --- .../scan_planners.py | 37 ++++++++++--------- 1 file changed, 20 insertions(+), 17 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index d684634f..c6975226 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -197,7 +197,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] @@ -331,8 +336,8 @@ class RectGridPlanner(ScanPlanner): def moves_between( self, - starting_pos: XYPos | np.ndarray | FutureScanLocation, - ending_pos: XYPos | np.ndarray | FutureScanLocation, + starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, ) -> float: """Return the larger of x moves or y moves between two xy positions. @@ -340,9 +345,9 @@ class RectGridPlanner(ScanPlanner): :param ending_pos: the position to measure to """ - if isinstance(starting_pos, FutureScanLocation): + if isinstance(starting_pos, (FutureScanLocation, VisitedScanLocation)): starting_pos = starting_pos.xy_tuple - if isinstance(ending_pos, FutureScanLocation): + if isinstance(ending_pos, (FutureScanLocation, VisitedScanLocation)): ending_pos = ending_pos.xy_tuple move_size = np.array([self._dx, self._dy]) @@ -378,20 +383,18 @@ class RectGridPlanner(ScanPlanner): if not focused_locations: return None - next_pos_arr = np.array(next_position, dtype="float64") - path_arr = np.array(focused_locations, dtype="float64")[:, :2] + def sort_key(pos: VisitedScanLocation) -> float: + return self.moves_between(next_position, pos) - # Find all focused positions within dx and dy of next_position. - # Don't just use _adjacent_positions as some grids might have intermediate sites. - dx_ok = np.abs(path_arr[:, 0] - next_pos_arr[0]) <= self._dx - dy_ok = np.abs(path_arr[:, 1] - next_pos_arr[1]) <= self._dy - nearby_indices = np.where(dx_ok & dy_ok)[0] - - if len(nearby_indices) == 0: - return None + # Sort by negative number of moves between so smallest moves are at the end. + # The most recent once should be the last point of multiple have the same + # number of moves. + nearby_focus_locations = sorted( + focused_locations, key=lambda p: -self.moves_between(next_position, p) + ) # Pick the most recent nearby site - return focused_locations[nearby_indices[-1]] + return nearby_focus_locations[-1].xyz_tuple class SmartSpiral(RectGridPlanner): @@ -568,7 +571,7 @@ class SmartSpiral(RectGridPlanner): 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 From 351a66a42cc3afe0bd3aea38b686b62cd99337bf Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 11 Feb 2026 10:05:31 +0000 Subject: [PATCH 13/19] Cleaning up workflows and fixing tests. --- .../scan_planners.py | 8 ++-- tests/unit_tests/test_scan_planners.py | 48 +++++++++++++------ 2 files changed, 37 insertions(+), 19 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index c6975226..2ba4d6e9 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -318,9 +318,9 @@ class RectGridPlanner(ScanPlanner): 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: - raise ValueError(invalid_msg + ",".join(expected_keys)) - if not all(k in planner_settings for k in expected_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"]) @@ -623,7 +623,7 @@ class RegularGridPlanner(RectGridPlanner): super()._parse(planner_settings) expected_keys = ["x_count", "y_count", "style"] - invalid_msg = "SnakeScan requires planner_settings with keys: " + invalid_msg = "RegularGrid requires planner_settings with keys: " if not planner_settings or not all( k in planner_settings for k in expected_keys ): diff --git a/tests/unit_tests/test_scan_planners.py b/tests/unit_tests/test_scan_planners.py index cf4c8acc..995aa761 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 = "RectangleScan requires planner_settings 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} @@ -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, ) From af5c8f01da5109cbf28a47abdb26750dd8a5cdb5 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 11 Feb 2026 10:37:34 +0000 Subject: [PATCH 14/19] Update scan_planner module docstring --- src/openflexure_microscope_server/scan_planners.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 2ba4d6e9..11174d91 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -1,8 +1,13 @@ """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 From 5d5bae7f79d5135cc5c3f75655e60f399b797c3d Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 11 Feb 2026 10:46:40 +0000 Subject: [PATCH 15/19] Apply suggestions from code review of branch More-workflow-layers Co-authored-by: Joe Knapper --- .../things/scan_workflows.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index 9c538bb4..13f30e9b 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -123,7 +123,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): ) -> tuple[bool, Optional[int]]: """Overload to set the acquisition routine that happens at each scan site. - :param settings: The settings for this scan as a HistoScanSettingsModel + :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. @@ -139,9 +139,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): images_dir: str, save_resolution: tuple[int, int], ) -> tuple[bool, Optional[int]]: - """Atuofocus and then capture, this can be used as an acquisition routine. + """Autofocus and then capture, this can be used as an acquisition routine. - :param dz: The dz for autofoucs. + :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. @@ -168,11 +168,11 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): """A generic workflow for any scan that captures images on a rectilinear grid.""" - # Thing Slots + # 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 or y. + """The fraction that adjacent images should overlap in x and y. This must be between 0.1 and 0.7. """ @@ -206,7 +206,7 @@ class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType] """ 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) @@ -641,7 +641,7 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]): ) -> tuple[bool, Optional[int]]: """Autofocus and capture. - :param settings: The settings for this scan as a HistoScanSettingsModel + :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. From 511b37176e2e75162226b5fc3975986667b41edc Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 11 Feb 2026 10:47:37 +0000 Subject: [PATCH 16/19] Add clarifying docstring --- src/openflexure_microscope_server/things/scan_workflows.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index 13f30e9b..e62608af 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -145,7 +145,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): :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. + :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] From d0f55786ceb758d5e4f67fa2765a400f1d2fbb9d Mon Sep 17 00:00:00 2001 From: jaknapper Date: Wed, 11 Feb 2026 15:13:17 +0000 Subject: [PATCH 17/19] Test raster scan planner and z estimate from raster and snake --- tests/unit_tests/test_scan_planners.py | 226 +++++++++++++++++++++++++ 1 file changed, 226 insertions(+) diff --git a/tests/unit_tests/test_scan_planners.py b/tests/unit_tests/test_scan_planners.py index 995aa761..995f33e7 100644 --- a/tests/unit_tests/test_scan_planners.py +++ b/tests/unit_tests/test_scan_planners.py @@ -443,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 From 8081dc94266fb697f4ca191415278cbe99d3d6ef Mon Sep 17 00:00:00 2001 From: jaknapper Date: Wed, 11 Feb 2026 15:13:47 +0000 Subject: [PATCH 18/19] Split moves_between to support Manhattan or Chebyshev distance --- .../scan_planners.py | 57 +++++++++++++------ 1 file changed, 39 insertions(+), 18 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 11174d91..41e9b0f2 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -339,6 +339,28 @@ class RectGridPlanner(ScanPlanner): (xy_pos[0], xy_pos[1] + self._dy), ] + def _displacement_in_moves( + self, + starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ) -> np.ndarray: + """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 + """ + 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") + + return (ending_pos - starting_pos) / move_size + def moves_between( self, starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, @@ -348,20 +370,22 @@ class RectGridPlanner(ScanPlanner): :param starting_pos: the position to measure from :param ending_pos: the position to measure to - """ - 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]) + displacement = self._displacement_in_moves(starting_pos, ending_pos) + return float(np.max(np.abs(displacement))) - starting_pos = np.array(starting_pos, dtype="float64") - ending_pos = np.array(ending_pos, dtype="float64") + def total_moves_between( + self, + starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, + ) -> float: + """Return the total x and y moves between two xy positions. - displacement_in_moves = (ending_pos - starting_pos) / move_size - - return np.max(np.abs(displacement_in_moves)) + :param starting_pos: the position to measure from + :param ending_pos: the position to measure to + """ + displacement = self._displacement_in_moves(starting_pos, ending_pos) + return float(np.sum(np.abs(displacement))) def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos: """Return an (x,y) position halfway between two input positions.""" @@ -389,14 +413,11 @@ class RectGridPlanner(ScanPlanner): return None def sort_key(pos: VisitedScanLocation) -> float: - return self.moves_between(next_position, pos) + return self.total_moves_between(next_position, pos) - # Sort by negative number of moves between so smallest moves are at the end. - # The most recent once should be the last point of multiple have the same - # number of moves. - nearby_focus_locations = sorted( - focused_locations, key=lambda p: -self.moves_between(next_position, p) - ) + # 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 From 2a3837504025554307c7301ae706543e96158493 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Thu, 12 Feb 2026 11:00:03 +0000 Subject: [PATCH 19/19] Consolidate moves_between and total_moves_between into 1 function with switchable metrics --- .../scan_planners.py | 65 ++++++++++--------- 1 file changed, 33 insertions(+), 32 deletions(-) diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 41e9b0f2..1830d1dc 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -14,6 +14,7 @@ child classes: # 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 @@ -28,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. @@ -339,15 +357,17 @@ class RectGridPlanner(ScanPlanner): (xy_pos[0], xy_pos[1] + self._dy), ] - def _displacement_in_moves( + def moves_between( self, starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, - ) -> np.ndarray: + 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 @@ -359,33 +379,12 @@ class RectGridPlanner(ScanPlanner): starting_pos = np.array(starting_pos, dtype="float64") ending_pos = np.array(ending_pos, dtype="float64") - return (ending_pos - starting_pos) / move_size - - def moves_between( - self, - starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, - ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, - ) -> 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 - """ - displacement = self._displacement_in_moves(starting_pos, ending_pos) - return float(np.max(np.abs(displacement))) - - def total_moves_between( - self, - starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, - ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation, - ) -> float: - """Return the total x and y moves between two xy positions. - - :param starting_pos: the position to measure from - :param ending_pos: the position to measure to - """ - displacement = self._displacement_in_moves(starting_pos, ending_pos) - return float(np.sum(np.abs(displacement))) + 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.""" @@ -413,7 +412,7 @@ class RectGridPlanner(ScanPlanner): return None def sort_key(pos: VisitedScanLocation) -> float: - return self.total_moves_between(next_position, pos) + 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 @@ -577,8 +576,10 @@ class SmartSpiral(RectGridPlanner): 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), )