From c938560f00dd053e6d1c5b5b4648d45ee60a654f Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Thu, 15 Jan 2026 21:59:53 +0000 Subject: [PATCH] Docstrings and final tweaks of ScanWorflow refactor --- .../scan_directories.py | 36 ++++---- .../stitching.py | 88 ++++--------------- .../things/scan_workflows.py | 75 +++++++++++----- .../things/smart_scan.py | 34 +++---- 4 files changed, 103 insertions(+), 130 deletions(-) diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 86ab0455..1a2d25ff 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -19,6 +19,7 @@ from pydantic import ( model_validator, ) +from openflexure_microscope_server.stitching import StitchingSettings from openflexure_microscope_server.utilities import make_name_safe, requires_lock LOGGER = logging.getLogger(__name__) @@ -49,16 +50,6 @@ class ScanInfo(BaseModel): dzi: Optional[str] -class StitchingData(BaseModel): - """The data needed to stitch a scan.""" - - correlation_resize: float - """The resize factor applied to images when the stitching program is correlating.""" - - overlap: float - """The overlap between adjacent images as a fraction of the image size.""" - - class BaseScanData(BaseModel): """Data about a scan not including workflow specific data. @@ -116,7 +107,7 @@ class BaseScanData(BaseModel): This should be set with ``set_final_data()`` to ensure duration is set. """ - stitching_settings: Optional[StitchingData] + stitching_settings: Optional[StitchingSettings] """The data needed to stitch a scan. Set to None for types of scan that cannot be stitched. @@ -185,6 +176,13 @@ class BaseScanData(BaseModel): class HistoricScanData(BaseScanData): + """A Model for the ScanData that has been loaded from disk. + + Any workflow specific settings are loaded as an arbitrary dictionary. Other + settings such as those which are needed for the UI or stitching are loaded and + validated by the parent class ``BaseScanData``. + """ + workflow_settings: dict """A dictionary of the settings for the workflow that was used workflow.""" @@ -204,7 +202,7 @@ class HistoricScanData(BaseScanData): # This is done because in future workflows the stitching overlap may be a # directly set setting or something that is calculated from other settings. overlap = data["overlap"] - data["stitching_settings"] = StitchingData( + data["stitching_settings"] = StitchingSettings( correlation_resize=correlation_resize, overlap=overlap, ) @@ -320,13 +318,9 @@ class ScanDirectoryManager: return None return scan_data_path - def get_scan_data_dict(self, scan_name: str) -> Optional[dict[str, Any]]: - """Return the scan data read from a JSON file as a dict. - - This is a dictionary not a base model as the data format has changed - somewhat over time. - """ - return ScanDirectory(scan_name, self.base_dir).get_scan_data_dict() + def get_scan_data(self, scan_name: str) -> Optional[HistoricScanData]: + """Return the scan data read from a JSON file as a dict.""" + return ScanDirectory(scan_name, self.base_dir).get_scan_data() @property @requires_lock @@ -542,7 +536,7 @@ class ScanDirectory: """Return the modified time of the directory.""" return max(os.stat(root).st_mtime for root, _, _ in os.walk(self.dir_path)) - def get_scan_data_dict(self) -> Optional[dict[str, Any]]: + def _get_scan_data_dict(self) -> Optional[dict[str, Any]]: """Return the scan data from the json file as a dictionary. This is safer than get_scan_data for older scans before a defined model was @@ -564,7 +558,7 @@ class ScanDirectory: :return: The data as a HistoricScanData model or None if it couldn't be loaded or valdiated. """ - data_dict = self.get_scan_data_dict() + data_dict = self._get_scan_data_dict() if data_dict is None: LOGGER.warning(f"Could not load scan data for {self.name}.") return None diff --git a/src/openflexure_microscope_server/stitching.py b/src/openflexure_microscope_server/stitching.py index 88e06df6..2d68736b 100644 --- a/src/openflexure_microscope_server/stitching.py +++ b/src/openflexure_microscope_server/stitching.py @@ -12,7 +12,9 @@ import shlex import signal import subprocess import threading -from typing import IO, Any, Optional +from typing import IO, Optional + +from pydantic import BaseModel import labthings_fastapi as lt @@ -28,6 +30,16 @@ DEFAULT_OVERLAP = 0.1 DEFAULT_RESIZE = 0.5 +class StitchingSettings(BaseModel): + """The data needed to stitch a scan.""" + + correlation_resize: float + """The resize factor applied to images when the stitching program is correlating.""" + + overlap: float + """The overlap between adjacent images as a fraction of the image size.""" + + class ExternalSigkillError(ChildProcessError): """Exception called when stitch is killed by an external process calling Sigkill.""" @@ -200,10 +212,8 @@ class FinalStitcher(BaseStitcher): images_dir: str, *, logger: logging.Logger, - overlap: Optional[float] = None, - correlation_resize: Optional[float] = None, + stitching_settings: StitchingSettings, stitch_tiff: bool = False, - scan_data_dict: Optional[dict[str, Any]] = None, ) -> None: """Initialise a final stitcher, this has more args than the base class. @@ -211,24 +221,13 @@ class FinalStitcher(BaseStitcher): :param images_dir: The images directory of the scan to stitch. :param logger: The logger from the Thing that created this stitcher. - :param overlap: The scan overlap, if not known enter None. A value will be - chosen from the scan_data_dict, or set to a default value if no value is - available in the scan_data. - :param correlation_resize: The fraction to resize images by when correlating, - if not known enter None. A value will be chosen from the scan_data_dict, or - set to a default value if no value is available in the scan_data. + :param stitching_settings: A StitchingSettings model this can be loaded from a + HistoricScanData for this scan as a dictionary. :param stitch_tiff: Whether to stitch a pyramidal TIFF. - :param scan_data_dict: The HistoricScanData for this scan as a dictionary. This - is used to read/calculate overlap and correlation_resize if they are not - provided. """ - # TODO ^fix the above historic data self.logger = logger - overlap, correlation_resize = self._process_inputs( - overlap=overlap, - correlation_resize=correlation_resize, - scan_data_dict=scan_data_dict, - ) + overlap = stitching_settings.overlap + correlation_resize = stitching_settings.correlation_resize super().__init__( images_dir, overlap=overlap, correlation_resize=correlation_resize ) @@ -242,57 +241,6 @@ class FinalStitcher(BaseStitcher): str(STITCH_TILE_SIZE), ] - def _process_inputs( - self, - overlap: Optional[float], - correlation_resize: Optional[float], - scan_data_dict: Optional[dict[str, Any]], - ) -> tuple[float, float]: - """Process inputs to ensure ``overlap`` and ``correlation_resize`` have values. - - First the scan_data_dict is inspected for values to allow ``overlap`` and - ``correlation_resize`` to be set correctly, if these values are not available - then default values are used, and a warning is logged to the thing logger. - - :param overlap: overlap as input to __init__ - :param correlation_resize: correlation_resize as input to __init__ - :param scan_data_dict: scan_data_dict as input to __init__ - - :returns: overlap and correlation_resize as floats. - """ - if overlap is None: - if scan_data_dict is not None and "overlap" in scan_data_dict: - overlap = scan_data_dict["overlap"] - - # Warn if still None and set to default. - if overlap is None: - overlap = DEFAULT_OVERLAP - self.logger.warning( - "No value set for overlap. Attempting stitch with overlap " - f"value of {DEFAULT_OVERLAP}" - ) - - if correlation_resize is None: - if scan_data_dict is not None: - # Handle "capture resolution" being used to store the save resolution - # in old scans. - key = ( - "capture resolution" - if "capture resolution" in scan_data_dict - else "save_resolution" - ) - if key in scan_data_dict: - save_resolution = scan_data_dict[key] - correlation_resize = STITCHING_RESOLUTION[0] / save_resolution[0] - # Warn if still None and set to default. - if correlation_resize is None: - correlation_resize = DEFAULT_RESIZE - self.logger.warning( - "No information available to calculate stitch resize. Attempting " - f"stitch with resize value of {DEFAULT_RESIZE}" - ) - return overlap, correlation_resize - def run(self) -> None: """Run the final stitch logging any output. diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py index feb81653..17e03147 100644 --- a/src/openflexure_microscope_server/things/scan_workflows.py +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -1,3 +1,9 @@ +"""Scan workflows set different ways that smart scan can behave. + +This module contains the base ``ScanWorkflow`` class that all workflows should subclass, +as well as specific workflows. +""" + from __future__ import annotations from typing import ( @@ -11,9 +17,11 @@ from pydantic import BaseModel import labthings_fastapi as lt -from openflexure_microscope_server.scan_directories import StitchingData from openflexure_microscope_server.scan_planners import ScanPlanner, SmartSpiral -from openflexure_microscope_server.stitching import STITCHING_RESOLUTION +from openflexure_microscope_server.stitching import ( + STITCHING_RESOLUTION, + StitchingSettings, +) from openflexure_microscope_server.things.autofocus import ( MAX_TEST_IMAGE_COUNT, MIN_TEST_IMAGE_COUNT, @@ -64,12 +72,12 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): def all_settings( self, images_dir: str - ) -> tuple[SettingModelType, Optional[StitchingData]]: + ) -> tuple[SettingModelType, Optional[StitchingSettings]]: """Return the scan settings and the stitching settings. - The specific settings for this scan workflow are returned as a Base Model of the type set when defining the class. - - Stitiching settings are returned either as a StitchingData object or None + - Stitiching settings are returned either as a StitchingSettings object or None is returned if it is not possible to stitch the scan. """ raise NotImplementedError( @@ -77,6 +85,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): ) def pre_scan_routine(self, settings: SettingModelType) -> None: + """Overload to set the routine that happens before each scan.""" raise NotImplementedError( "Each specific ScanWorkflow must implement a pre-scan routine." ) @@ -84,6 +93,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): def new_scan_planner( self, settings: SettingModelType, position: Mapping[str, int] ) -> ScanPlanner: + """Return the a new scan planner object for a scan.""" raise NotImplementedError( "Each specific ScanWorkflow must implement a ``new_scan_planner`` method." ) @@ -91,13 +101,14 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing): def aquisition_routine( self, settings: SettingModelType, xyz_pos: tuple[int, int, int] ) -> tuple[bool, Optional[int]]: + """Overload to set the aquisition routine that happens at each scan site.""" raise NotImplementedError( "Each specific ScanWorkflow must implement an aquisition routine" ) class HistoScanSettingsModel(BaseModel): - """The settings including needed for running a HistoScan. + """The settings for a scan with the HistoScanWorkflow. This includes settings caluclated when starting. This will be held by smart scan during a scan and serialised to disk. @@ -112,6 +123,12 @@ class HistoScanSettingsModel(BaseModel): class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): + """A workflow optimised for scanning Histopathology samples. + + This workflow automatically plans its own path around a sample spiralling out from + the centre position, scanning only where it detects sample. + """ + _settings_model = HistoScanSettingsModel _planner_cls: type[ScanPlanner] = SmartSpiral # Thing Slots @@ -128,14 +145,20 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): This uses the settings from the ``BackgroundDetectThing``. """ - autofocus_dz: int = lt.setting(default=1000) - """The z distance to perform an autofocus in steps.""" + 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) - """The fraction (0-1) that adjacent images should overlap in x or y.""" + 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 @@ -204,8 +227,14 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): def all_settings( self, images_dir: str - ) -> tuple[HistoScanSettingsModel, StitchingData]: - stitching_settings = StitchingData( + ) -> tuple[HistoScanSettingsModel, 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], ) @@ -216,20 +245,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): f"{dx}, {dy}" ) - # TODO: set a min on the property - autofocus_dz = self.autofocus_dz - if autofocus_dz == 0: - self.logger.info("Running scan without autofocus") - elif autofocus_dz <= 200: - self.logger.warning( - f"Your autofocus range is {autofocus_dz} steps, which is too short to " - "attempt to focus. Running without autofocus" - ) - autofocus_dz = 0 - smart_stack_params = self.create_smart_stack_params( images_dir=images_dir, - autofocus_dz=autofocus_dz, + autofocus_dz=self.autofocus_dz, save_resolution=self.save_resolution, ) @@ -341,6 +359,10 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): ) def pre_scan_routine(self, settings: HistoScanSettingsModel) -> None: + """Autofocus before starting the scan. + + :param settings: The settings for this scan as a HistoScanSettingsModel + """ self._autofocus.looping_autofocus( dz=settings.smart_stack_params.autofocus_dz, start="centre" ) @@ -348,6 +370,11 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): def new_scan_planner( self, settings: HistoScanSettingsModel, position: Mapping[str, int] ) -> ScanPlanner: + """Return a new scan planner object. + + :param settings: The settings for this scan as a HistoScanSettingsModel + :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 @@ -367,6 +394,8 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]): ) -> tuple[bool, Optional[int]]: """Perform aquisition routine. This is run at each scan location. + :param settings: The settings for this scan as a HistoScanSettingsModel + :param xyz_position: 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. """ diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 97a56e36..da23f422 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -48,6 +48,13 @@ AnyModel = Annotated[ class ActiveScanData(scan_directories.BaseScanData): + """A Model for the ScanData during an ongoing scan. + + This differs from HistoricScanData as in this model ``workflow_settings`` are the + model specified for the current ScanWorkflow. HistoricScanData loads + ``workflow_settings`` into a dictionary. + """ + workflow_settings: AnyModel """The settings for the ongoing workflow.""" @@ -432,11 +439,7 @@ class SmartScanThing(lt.Thing): stitching_settings = self.scan_data.stitching_settings if self.scan_data.stitch_automatically and stitching_settings is not None: self.logger.info("Stitching final image (may take some time)...") - self.stitch_scan( - scan_name=self.ongoing_scan.name, - correlation_resize=stitching_settings.correlation_resize, - overlap=stitching_settings.overlap, - ) + self.stitch_scan(scan_name=self.ongoing_scan.name) @lt.endpoint( "get", @@ -624,25 +627,24 @@ class SmartScanThing(lt.Thing): return FileResponse(preview_path) @lt.action - def stitch_scan( - self, - scan_name: str, - correlation_resize: Optional[float] = None, - overlap: Optional[float] = None, - ) -> None: + def stitch_scan(self, scan_name: str) -> None: """Generate a stitched image based on stage position metadata.""" - scan_data_dict = self._scan_dir_manager.get_scan_data_dict(scan_name) - if scan_data_dict is None: + scan_data = self._scan_dir_manager.get_scan_data(scan_name) + if scan_data is None: self.logger.warning( "Couldn't read scan data - it may be missing or corrupt." ) + return + if scan_data.stitching_settings is None: + # If the stitching settings are none then this type of scan cannot be + # stitiched. + return + final_stitcher = stitching.FinalStitcher( self._scan_dir_manager.img_dir_for(scan_name), logger=self.logger, - overlap=overlap, - correlation_resize=correlation_resize, stitch_tiff=self.stitch_tiff, - scan_data_dict=scan_data_dict, + stitching_settings=scan_data.stitching_settings, ) try: final_stitcher.run()