diff --git a/ofm_config_simulation.json b/ofm_config_simulation.json index 367088de..b061dbd8 100644 --- a/ofm_config_simulation.json +++ b/ofm_config_simulation.json @@ -13,6 +13,7 @@ } }, "histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow", + "snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow", "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 930856ca..e032bf0f 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -252,41 +252,15 @@ class ScanPlanner: next_location = self._remaining_locations[0].xy_tuple - # If focussed locations exist return closest location, favouring most recent - closest_pos = self.closest_focus_site(next_location) + # Each scanner defines its own method of choosing a representative nearby site + closest_pos = self.select_nearby_focus_site(next_location) z = None if closest_pos is None else closest_pos[2] return next_location, z - def closest_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]: - """Return the xyz position of the closest site where focus was achieved. - - The most recently taken image is returned in the case of a tie. - - :param xy_pos: The xy_position which the returned position should be closest - to. - - Returns None if there if no focussed locations are present - """ - # save to variable rather than search for focussed sites each time. - focused_locations = self.focused_locations - 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] - - # Use linalg.norm to calculate the direct distance bweween the points - # Note linalg.norm always uses float64 - dists = np.linalg.norm((path_pos - current_pos), axis=1) - - # Get indices of all minima. - # Note np.where always returns a tuple of arrays, hence the trailing [0] - indices = np.where(dists == np.min(dists))[0] - - # The last index is most recent - return focused_locations[indices[-1]] + def select_nearby_focus_site(self, next_location: XYPos) -> Optional[XYZPos]: + """Return the focused site near xy_pos according to the tiebreak.""" + raise NotImplementedError("Did you call the ScanPlanner base class?") def mark_location_visited( self, xyz_pos: XYZPos, imaged: bool, focused: bool @@ -316,6 +290,20 @@ class ScanPlanner: ) ) + def _grid_to_future_locations( + self, + grid: list[list[XYPos]], + ) -> list[FutureScanLocation]: + """Flatten a 2D grid of coordinates into flat list of FutureScanLocation objects. + + :param grid: A 2D nested list of XY coordinates + """ + path = [] + for line in grid: + path += line + + return [FutureScanLocation(location) for location in path] + class SmartSpiral(ScanPlanner): """A scan planner that spirals outward from the centre, prioritising short moves. @@ -385,7 +373,7 @@ class SmartSpiral(ScanPlanner): def _initial_location_list(self) -> list[FutureScanLocation]: """Set the initial list of locations for this scan planner. - This is salled on initialisation. + This is called on initialisation. For smart spiral this is just the first point """ @@ -514,26 +502,6 @@ class SmartSpiral(ScanPlanner): self._remaining_locations.sort(key=sort_key) - def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]: - """Return the next location to scan and its estimated z-position. - - This overrides the default behaviour of ScanPlanner to take the lowest value of - nearest neighbours as this works best for smart stack. - - Note z-position may be None! This indicates that the current z position - should be used. - """ - if self.scan_complete: - raise RuntimeError("Can't get next position, scan is complete") - - next_location = self._remaining_locations[0].xy_tuple - - # If focused locations exist, return the neighbour with the lowest z position - closest_pos = self.select_nearby_focus_site(next_location) - z = None if closest_pos is None else closest_pos[2] - - return next_location, z - def select_nearby_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]: """Return the xyz position of the nearby site with the lowest z position. @@ -575,7 +543,11 @@ class SmartSpiral(ScanPlanner): # Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best # if started too low, so the lowest z will perform best - chosen_focused_site = min(focused_locations_array[indices], key=lambda x: x[-1]) + candidates = focused_locations_array[indices] + min_z = np.min(candidates[:, -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()) @@ -664,91 +636,16 @@ class SnakeScan(ScanPlanner): style="snake", ) - # create_rectangular_scan_path provides a nested list, which is flattened here - path = [] - for line in grid: - path += line + return self._grid_to_future_locations(grid) - return [FutureScanLocation(location) for location in path] - - def mark_location_visited( - self, xyz_pos: XYZPos, imaged: bool = True, focused: bool = True - ) -> None: - """Mark the location as visited. - - :param xyz_pos: the x_y_z position - :param imaged: true if an image was taken, false if not (due to background detect) - :param focused: true if autofocus completed successfully - """ - # Only call the base class to update the positions - super().mark_location_visited(xyz_pos, imaged, focused) - - def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]: - """Return the next location to scan and its estimated z-position. - - This overrides the default behaviour of ScanPlanner to take the lowest value of - nearest neighbours as this works best for smart stack. - - Note z-position may be None! This indicates that the current z position - should be used. - """ - if self.scan_complete: - raise RuntimeError("Can't get next position, scan is complete") - - next_site = self._remaining_locations[0] - next_location = next_site.xy_tuple - - # If focused locations exist, return the neighbour with the lowest z position - closest_pos = self.select_nearby_focus_site(next_location) - z = None if closest_pos is None else closest_pos[2] - - return next_location, z - - def select_nearby_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]: - """Return the xyz position of the nearby site with the lowest z position. - - Lowest position is best, as starting too high causes smart stacking to - autofocus and restart. Starting too low just requires extra movements in +z. - Nearby is defined as within 1.1 times the larger of the x and y scan offsets. - - If no focused sites are within this range, use the height of the nearest - focused site. - - Returns None if no focused locations are present - """ - # save to variable rather than search for focussed sites each time. + # The noqa statement is because next_position is unused but is needed for equivalence + # with other workflows that require the next pos to select a neighbour. + def select_nearby_focus_site(self, next_position: XYPos) -> Optional[XYZPos]: # noqa: ARG002 + """For a snake scan, use the most recent focused site to predict focus.""" focused_locations = self.focused_locations if not focused_locations: return None - - # 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] - - # 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) - - # 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] - - # 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] - - # Turning into an array allows slicing based on a list - focused_locations_array = np.array(focused_locations) - - # Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best - # if started too low, so the lowest z will perform best - chosen_focused_site = min(focused_locations_array[indices], key=lambda x: x[-1]) - - # Convert back into list so values are of type int instead of np.int32 - return tuple(chosen_focused_site.tolist()) + return focused_locations[-1] def distance_between( diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index dc6a30a0..0ec71ec0 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -6,6 +6,7 @@ as well as specific workflows. from __future__ import annotations +import os from typing import ( Generic, Mapping, @@ -17,7 +18,11 @@ from pydantic import BaseModel import labthings_fastapi as lt -from openflexure_microscope_server.scan_planners import ScanPlanner, SmartSpiral +from openflexure_microscope_server.scan_planners import ( + ScanPlanner, + SmartSpiral, + SnakeScan, +) from openflexure_microscope_server.stitching import ( STITCHING_RESOLUTION, StitchingSettings, @@ -33,6 +38,7 @@ from openflexure_microscope_server.things.background_detect import ( ) from openflexure_microscope_server.things.camera import BaseCamera from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper +from openflexure_microscope_server.things.stage import BaseStage from openflexure_microscope_server.ui import PropertyControl, property_control_for SettingModelType = TypeVar("SettingModelType", bound=BaseModel) @@ -59,6 +65,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232)) """A tuple of the image resolution to capture.""" + # CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None + _csm: Optional[CameraStageMapper] = None + def check_before_start(self, scan_name: str) -> None: """Check before the scan starts. Throw an error if the scan shouldn't start. @@ -119,6 +128,42 @@ 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 + + def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]: + """Use camera stage mapping to calculate x and y displacement from given overlap. + + :param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means + that each image should overlap its nearest neighbour by 50%. + + :returns: (dx, dy) - the x and y displacements in steps + """ + csm = self._require_csm() + + csm_image_res = csm.image_resolution + if csm_image_res is None: + raise RuntimeError("CSM not set. Scan shouldn't have progresses this far.") + + # Calculate displacements in image coordinates + dx_img = csm_image_res[1] * (1 - overlap) + dy_img = csm_image_res[0] * (1 - overlap) + + x_move_stage = csm.convert_image_to_stage_coordinates(x=dx_img, y=0) + y_move_stage = csm.convert_image_to_stage_coordinates(x=0, y=dy_img) + + # 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. + return y_move_stage["x"], x_move_stage["y"] + class HistoScanSettingsModel(BaseModel): """The settings for a scan with the HistoScanWorkflow. @@ -285,32 +330,6 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): return scan_settings, stitching_settings - def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]: - """Use camera stage mapping to calculate x and y displacement. - - :param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means - that each image should overlap its nearest neighbour by 50%. - - :returns: (dx, dy) - the x and y displacements in steps - """ - csm_image_res = self._csm.image_resolution - if csm_image_res is None: - raise RuntimeError("CSM not set. Scan shouldn't have progressed this far.") - - # Calculate displacements in image coordinates - dx_img = csm_image_res[1] * (1 - overlap) - dy_img = csm_image_res[0] * (1 - overlap) - - x_move_stage = 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. - return y_move_stage["x"], x_move_stage["y"] - def create_smart_stack_params( self, images_dir: str, @@ -478,3 +497,173 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), property_control_for(self, "max_range", label="Maximum Distance (steps)"), ] + + +class SnakeSettingsModel(BaseModel): + """The settings for a scan with the SnakeWorkflow. + + This includes settings calculated when starting. This will be held by smart scan + during a scan and serialised to disk. + """ + + overlap: float + dx: int + dy: int + x_count: int + y_count: int + images_dir: str + autofocus_dz: int + save_resolution: tuple[int, int] + + +class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): + """A workflow optimised for snaking around samples. + + 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, + ) + + _settings_model = SnakeSettingsModel + _planner_cls: type[ScanPlanner] = SnakeScan + # Thing Slots + _background_detector: ChannelDeviationLUV = lt.thing_slot() + _cam: BaseCamera = lt.thing_slot() + _csm: CameraStageMapper = lt.thing_slot() + _autofocus: AutofocusThing = lt.thing_slot() + _stage: BaseStage = lt.thing_slot() + + # Scan settings + + 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[SnakeSettingsModel, StitchingSettings]: + """Return the workflow and stitching settings. + + :param images_dir: The directory that images are to be written to. + :return: A tuple containing the settings model for this workflow and the + settings model for stitching. + """ + stitching_settings = StitchingSettings( + overlap=self.overlap, + correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0], + ) + + dx, dy = self._calc_displacement_from_overlap(self.overlap) + self.logger.info( + f"Based on an overlap of {self.overlap}, the stage will make steps of " + f"{dx}, {dy}" + ) + + scan_settings = SnakeSettingsModel( + overlap=self.overlap, + dx=dx, + dy=dy, + x_count=self.x_count, + y_count=self.y_count, + images_dir=images_dir, + autofocus_dz=self.autofocus_dz, + save_resolution=self.save_resolution, + ) + + return scan_settings, stitching_settings + + def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: + """Autofocus before starting the scan. + + :param settings: The settings for this scan as a SnakeSettingsModel + """ + self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") + + def new_scan_planner( + self, settings: SnakeSettingsModel, position: Mapping[str, int] + ) -> ScanPlanner: + """Return a new scan planner object. + + :param settings: The settings for this scan as a SnakeSettingsModel + :param position: The starting position as a mapping of axes names to int. + """ + # The initial plan for the scan should be a single x,y position. All future + # moves will be planned around this point. In future, route planner could + # have multiple starting positions, each of which will be visited before the + # scan can end. + planner_settings = { + "dx": settings.dx, + "dy": settings.dy, + "x_count": settings.x_count, + "y_count": settings.y_count, + } + return self._planner_cls( + initial_position=(position["x"], position["y"]), + planner_settings=planner_settings, + ) + + def acquisition_routine( + self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] + ) -> tuple[bool, Optional[int]]: + """Perform acquisition routine. This is run at each scan location. + + :param settings: The settings for this scan as a SnakeSettingsModel + :param xyz_pos: The current position as a tuple or 3 ints. + :return: A tuple of whether an image was taken, and the z-position for focus. + If failed to find focus, returns for the focus z-position. + """ + self._autofocus.fast_autofocus(dz=settings.autofocus_dz) + focus_height = self._stage.get_xyz_position()[2] + filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg" + self._cam.capture_and_save( + jpeg_path=os.path.join(settings.images_dir, filename), + save_resolution=settings.save_resolution, + ) + + imaged = True + return imaged, focus_height + + @lt.property + def settings_ui(self) -> list[PropertyControl]: + """A list of PropertyControl objects to create the settings in the scan tab.""" + return [ + property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"), + property_control_for(self, "x_count", label="Number of columns"), + property_control_for(self, "y_count", label="Number of rows"), + property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), + ] diff --git a/tests/unit_tests/test_scan_planners.py b/tests/unit_tests/test_scan_planners.py index fcbd9dd7..dea5c530 100644 --- a/tests/unit_tests/test_scan_planners.py +++ b/tests/unit_tests/test_scan_planners.py @@ -143,6 +143,8 @@ def test_smart_spiral_first_few_pos(): # if we mark this position as visited, imaged, and focused planner.mark_location_visited(xyz_pos1, imaged=True, focused=True) + # current visited path is [[100, 50, 10]] + # scan is not complete assert not planner.scan_complete @@ -172,6 +174,8 @@ def test_smart_spiral_first_few_pos(): # if we mark this position as visited, imaged, and NOT focused planner.mark_location_visited(xyz_pos2, imaged=True, focused=False) + # current visited path is [[100, 50, 10], [50, 50, 10]] + # Check this position remove from planned assert xy_pos2 not in planner.remaining_locations # Check original position not re-added @@ -195,19 +199,22 @@ def test_smart_spiral_first_few_pos(): assert z_pos3 is z_focus # Check that the closest focus site to pos3 is pos 1 as # pos 2 is not focussed - assert planner.closest_focus_site(xy_pos3) == xyz_pos1 + assert planner.select_nearby_focus_site(xy_pos3) == xyz_pos1 new_z_focus = 20 xyz_pos3 = (xy_pos3[0], xy_pos3[1], new_z_focus) # Finally check that if this is focused... planner.mark_location_visited(xyz_pos3, imaged=True, focused=True) + # current visited path is [[100, 50, 10], [50, 50, 10], [50, 0, 20]] + # ... then the new 4th point ... xy_pos4, z_pos4 = planner.get_next_location_and_z_estimate() # ...(100, 0)... assert xy_pos4 == (100, 0) # ... and it should get its focus from the lowest neighbouring point + # lowest neighbour to [100, 0] is [100, 50, 10] assert z_pos4 is z_focus - assert planner.closest_focus_site(xy_pos4) == xyz_pos3 + assert planner.select_nearby_focus_site(xy_pos4) == xyz_pos1 def test_smart_spiral_stops_on_max_dist(): @@ -265,20 +272,20 @@ def test_closest_focus_with_large_numbers(): scan_planners.VisitedScanLocation((1000000, 0, 0), imaged=True, focused=True), scan_planners.VisitedScanLocation((0, 1000000, 0), imaged=True, focused=True), ] - assert planner.closest_focus_site((0, 0)) == (0, 1000000, 0) + assert planner.select_nearby_focus_site((0, 0)) == (0, 1000000, 0) # Try similar planner._path_history = [ scan_planners.VisitedScanLocation((1234567, 0, 0), imaged=True, focused=True), scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True), ] - assert planner.closest_focus_site((0, 0)) == (-1234567, 0, 0) + assert planner.select_nearby_focus_site((0, 0)) == (-1234567, 0, 0) # Make the first point 1 step closer planner._path_history = [ scan_planners.VisitedScanLocation((1234566, 0, 0), imaged=True, focused=True), scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True), ] - assert planner.closest_focus_site((0, 0)) == (1234566, 0, 0) + assert planner.select_nearby_focus_site((0, 0)) == (1234566, 0, 0) def test_example_smart_spiral():