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