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