From 54ed78c832ec529b722394a0ce4264c670aba81d Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Thu, 15 Jan 2026 14:53:47 +0000 Subject: [PATCH] Create aquisition routines for scan workflows, remove background detector from base ScanWorkflow class --- .../things/scan_workflows.py | 78 ++++++++++++++++--- .../things/smart_scan.py | 60 +++++--------- 2 files changed, 86 insertions(+), 52 deletions(-) diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index dea895b2..161caef9 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -7,7 +7,6 @@ from openflexure_microscope_server.scan_planners import ScanPlanner, SmartSpiral from openflexure_microscope_server.stitching import STITCHING_RESOLUTION from openflexure_microscope_server.things.autofocus import AutofocusThing from openflexure_microscope_server.things.background_detect import ( - BackgroundDetectAlgorithm, ChannelDeviationLUV, ) from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper @@ -20,14 +19,10 @@ class ScanWorkflow(lt.Thing): scan planning, aquisition routine. """ - # All workdlows must have at least one background detector and a set class for scan - # planning - _background_detector: Optional[ - BackgroundDetectAlgorithm | Mapping[str, BackgroundDetectAlgorithm] - ] = lt.thing_slot() + # All workdlows must have a set class for scan planning _planner_cls: type[ScanPlanner] - # All workdlows set a save resolution + # All workflows set a save resolution save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232)) """A tuple of the image resolution to capture.""" @@ -56,15 +51,26 @@ class ScanWorkflow(lt.Thing): is returned if it is not possible to stitch the scan. """ raise NotImplementedError( - "Each specific ScanWorkflow must implement a `all_settings`. " - "method." + "Each specific ScanWorkflow must implement a `all_settings`. method." ) - def pre_scan_routine(self) -> None: + def pre_scan_routine(self, settings: dict) -> None: raise NotImplementedError( "Each specific ScanWorkflow must implement a pre-scan routine." ) + def new_scan_planner(self, settings: dict, position: Mapping[str, int]) -> None: + raise NotImplementedError( + "Each specific ScanWorkflow must implement a ``new_scan_planner`` method." + ) + + def aquisition_routine( + self, settings: dict, xyz_pos: tuple[int, int, int] + ) -> tuple[bool, Optional[int]]: + raise NotImplementedError( + "Each specific ScanWorkflow must implement an aquisition routine" + ) + class HistoScanWorkflow(ScanWorkflow): # Thing Slots @@ -184,5 +190,53 @@ class HistoScanWorkflow(ScanWorkflow): # If not use the other stage axes return x_move_stage["y"], y_move_stage["x"] - def pre_scan_routine(self) -> None: - self._autofocus.looping_autofocus(dz=self.autofocus_dz, start="centre") + def pre_scan_routine(self, settings: dict) -> None: + self._autofocus.looping_autofocus(dz=settings["autofocus_dz"], start="centre") + + def new_scan_planner(self, settings: dict, position: Mapping[str, int]) -> None: + # 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"], + "max_dist": settings["max_dist"], + } + return self._planner_cls( + initial_position=(position["x"], position["y"]), + planner_settings=planner_settings, + ) + + def aquisition_routine( + self, settings: dict, xyz_pos: tuple[int, int, int] + ) -> tuple[bool, Optional[int]]: + """Perform aquisition routine. This is run at each scan location. + + :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. + """ + # If skipping background, take an image to check if current field of view is background + if settings["skip_background"]: + capture_image, bg_message = self._cam.image_is_sample() + + if not capture_image: + msg = f"Skipping {xyz_pos} as it is {bg_message}." + self.logger.info(msg) + return False, None + + save_on_failure = settings["skip_background"] + + focused, focused_height = self._autofocus.run_smart_stack( + stack_parameters=self.stack_params, + save_on_failure=save_on_failure, + ) + # An image was captured if we are focussed or we are not skipping background. + imaged = focused or save_on_failure + + # run_smart_stage always returns a focus height for the sharpest image even + # if it failed to find a good focus. Set to None if not focussed. + if not focused: + focused_height = None + + return imaged, focused_height diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 7278d4f3..4752f8e4 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -22,19 +22,18 @@ from typing import ( TypeVar, ) -import numpy as np from fastapi import HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel import labthings_fastapi as lt -from openflexure_microscope_server import scan_directories, scan_planners, stitching +from openflexure_microscope_server import scan_directories, stitching # Things from .autofocus import AutofocusThing, StackParams from .camera import BaseCamera -from .camera_stage_mapping import CameraStageMapper, CSMUncalibratedError +from .camera_stage_mapping import CameraStageMapper from .scan_workflows import HistoScanWorkflow, ScanWorkflow from .stage import BaseStage @@ -284,6 +283,9 @@ class SmartScanThing(lt.Thing): @_scan_running def _manage_stitching_threads(self) -> None: """Manage the stitching threads, starting them if needed and not already running.""" + if self.scan_data.stitching_settings is None: + # This scan can't stitch. + return # Assume 4 images means at least one offset in x and y, making the stitching # well constrained. if self.scan_data.image_count > 3 and not self.preview_stitcher.running: @@ -354,18 +356,9 @@ class SmartScanThing(lt.Thing): This loop runs during a scan, until no more scan x,y positions are remaining. """ - # 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": self.scan_data.dx, - "dy": self.scan_data.dy, - "max_dist": self.scan_data.max_dist, - } - route_planner = scan_planners.SmartSpiral( - initial_position=(self._stage.position["x"], self._stage.position["y"]), - planner_settings=planner_settings, + workflow_settings = self.scan_data.workflow_settings + route_planner = workflow.new_scan_planner( + workflow_settings, self._stage.position ) # The loop tests if the scan should continue, moves to the next position, @@ -383,43 +376,30 @@ class SmartScanThing(lt.Thing): new_pos_xyz[1], self._stage.position["z"], ) - - capture_image = True - # If skipping background, take an image to check if current field of view is background - if self.scan_data.skip_background: - capture_image, bg_message = self._cam.image_is_sample() - - if not capture_image: - route_planner.mark_location_visited( - new_pos_xyz, imaged=False, focused=False - ) - msg = f"Skipping {new_pos_xyz} as it is {bg_message}." - self.logger.info(msg) - continue - - focused, focused_height = self._autofocus.run_smart_stack( - stack_parameters=self.stack_params, - save_on_failure=not self.scan_data.skip_background, + imaged, focus_height = workflow.aquisition_routine( + workflow_settings, current_pos_xyz ) - current_pos_xyz = (new_pos_xyz[0], new_pos_xyz[1], focused_height) - - # An image was captured if we are focussed or we are not skipping background. - imaged = focused or not self.scan_data.skip_background + if focus_height is None: + focused = False + else: + current_pos_xyz = (new_pos_xyz[0], new_pos_xyz[1], focus_height) route_planner.mark_location_visited( current_pos_xyz, imaged=imaged, focused=focused ) - # increment capture counter as thread has completed - self.scan_data.image_count += 1 - # Add it to the incremental zip - self.ongoing_scan.zip_files() + if imaged: + # increment capture counter as thread has completed + self.scan_data.image_count += 1 + # Add it to the incremental zip + self.ongoing_scan.zip_files() @_scan_running def _return_to_starting_position(self) -> None: """Return to the initial scan position, if set.""" self.logger.info("Returning to starting position.") + if self._scan_data is not None: self._stage.move_absolute( **self.scan_data.starting_position, block_cancellation=True