Add raster scan, with generic typing to allow subclassed settings

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"
     },

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(

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