diff --git a/ofm_config_full.json b/ofm_config_full.json index 6d4686ec..50e0fa7c 100644 --- a/ofm_config_full.json +++ b/ofm_config_full.json @@ -13,9 +13,11 @@ "smart_scan": { "class": "openflexure_microscope_server.things.smart_scan:SmartScanThing", "kwargs": { - "scans_folder": "/var/openflexure/scans/" + "scans_folder": "/var/openflexure/scans/", + "default_workflow": "histo_scan_workflow" } }, + "histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow", "stage_measure": "openflexure_microscope_server.things.stage_measure:RangeofMotionThing", "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/ofm_config_simulation.json b/ofm_config_simulation.json index ef886fb4..367088de 100644 --- a/ofm_config_simulation.json +++ b/ofm_config_simulation.json @@ -8,9 +8,11 @@ "smart_scan": { "class": "openflexure_microscope_server.things.smart_scan:SmartScanThing", "kwargs": { - "scans_folder": "./openflexure/scans/" + "scans_folder": "./openflexure/scans/", + "default_workflow": "histo_scan_workflow" } }, + "histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow", "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/picamera_coverage.zip b/picamera_coverage.zip index f22a204a..0e85922e 100644 Binary files a/picamera_coverage.zip and b/picamera_coverage.zip differ diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 6770a9cd..a5ddb6fe 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -8,7 +8,7 @@ import shutil import threading import zipfile from datetime import datetime, timedelta -from typing import Any, Mapping, Optional +from typing import Any, Mapping, Optional, Self from pydantic import ( BaseModel, @@ -16,8 +16,10 @@ from pydantic import ( ValidationError, field_serializer, field_validator, + model_validator, ) +from openflexure_microscope_server.stitching import StitchingSettings from openflexure_microscope_server.utilities import make_name_safe, requires_lock LOGGER = logging.getLogger(__name__) @@ -29,6 +31,7 @@ STITCH_REGEX = re.compile(r"stitched\.jpe?g$") IMAGE_REGEX = re.compile(r"-?[0-9]+_-?[0-9]+\.jpe?g$") SCAN_DATA_FILENAME = "scan_data.json" +SCAN_DATA_SCHEMA_VERSION = 2 class NotEnoughFreeSpaceError(IOError): @@ -47,8 +50,20 @@ class ScanInfo(BaseModel): dzi: Optional[str] -class ScanData(BaseModel): - """Data about a scan to be saved to a JSON file in the directory. +class BaseScanData(BaseModel): + """Data about a scan not including workflow specific data. + + For including workflow specific data see also: + + * ActiveScanData which subclasses this including the BaseModel used by the + ScanWorkflow + * HistoricScanData which has the workflow specific data loaded as a dictionary. + + Separating historic and active data allows workflows to use any BaseModel for its + settings, but for the data to be reloaded even if that model has updated or is not + available. Historic scan data loaded from disk is used for stitching and for + creating a ScanInfo object for communicating with the UI. These uses are clearly + typed by this model. This serialises into a human readable format where possible with @@ -60,42 +75,20 @@ class ScanData(BaseModel): model_config = ConfigDict(extra="forbid") + schema_version: int = SCAN_DATA_SCHEMA_VERSION + scan_name: str """The name of the scan i.e. scan_0001""" starting_position: Mapping[str, int] """The starting position in dictionary format.""" - overlap: float - """The overlap between adjacent images as a fraction of the image size.""" - - max_dist: int - """The maximum distance the scan could move (in steps) from the starting position.""" - - dx: int - """The number of steps between adjacent images in x.""" - - dy: int - """The number of steps between adjacent images in y.""" - - autofocus_dz: int - """The z range used for autofocus (in steps).""" - - autofocus_on: bool - """Whether autofocus is on.""" - start_time: datetime """The time the scan started.""" - skip_background: bool - """Whether automatic background detection is on, skipping locations with no sample.""" - stitch_automatically: bool """Whether the scan is set to automatically stitch when complete.""" - correlation_resize: float - """The resize factor applied to images when the stitching program is correlating.""" - save_resolution: tuple[int, int] """The resolution that scan images are saved at.""" @@ -114,13 +107,24 @@ class ScanData(BaseModel): This should be set with ``set_final_data()`` to ensure duration is set. """ - def set_final_data(self, result: str) -> None: - """Set the final data for the scan, scan duration is automatically calculated. + stitching_settings: Optional[StitchingSettings] + """The data needed to stitch a scan. - :param result: A string describing the result. - """ - self.duration = datetime.now() - self.start_time - self.scan_result = result + Set to None for types of scan that cannot be stitched. + """ + + workflow: str + """The class name of the workflow Thing.""" + + @model_validator(mode="after") + def validate_schema_version(self) -> Self: + """Validate the schema version is as the current one.""" + if self.schema_version != SCAN_DATA_SCHEMA_VERSION: + raise ValueError( + f"Unsupported schema version {self.schema_version}, " + f"expected {SCAN_DATA_SCHEMA_VERSION}" + ) + return self @field_validator("start_time", mode="before") @classmethod @@ -171,6 +175,59 @@ class ScanData(BaseModel): return "Unknown" if value is None else value +class HistoricScanData(BaseScanData): + """A Model for the scan data 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.""" + + @model_validator(mode="before") + @classmethod + def coerce_legacy(cls, data: dict) -> dict: + """Coerce any scan data from before version 2 into the version 2 format.""" + # Before the current version no schema_version was set + if "schema_version" in data: + return data + + if "correlation_resize" and "overlap" in data: + correlation_resize = data.pop("correlation_resize") + # Note we don't pop overlap, as it is a setting for the legacy workflow as well + # as a stitching setting. + # 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"] = StitchingSettings( + correlation_resize=correlation_resize, + overlap=overlap, + ) + else: + data["stitching_settings"] = None + + # Add any legacy workflow settings that are found + legacy_keys = [ + "overlap", + "max_dist", + "dx", + "dy", + "autofocus_dz", + "autofocus_on", + "skip_background", + ] + workflow_settings = {} + for key in legacy_keys: + if key in data: + workflow_settings[key] = data.pop(key) + + data["workflow"] = "Legacy" + data["workflow_settings"] = workflow_settings + return data + + class ScanDirectoryManager: """A class for managing interactions with scan directories.""" @@ -260,13 +317,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 @@ -482,7 +535,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 @@ -498,18 +551,18 @@ class ScanDirectory: except (json.decoder.JSONDecodeError, IOError): return None - def get_scan_data(self) -> Optional[ScanData]: - """Return the scan data from the json file as a ScanData model. + def get_scan_data(self) -> Optional[HistoricScanData]: + """Return the scan data from the json file as a HistoricScanData model. - :return: The data as a ScanData model or None if it couldn't be loaded or + :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 try: - return ScanData(**data_dict) + return HistoricScanData(**data_dict) except ValidationError: LOGGER.warning(f"Could not validate scan data for {self.name}.") return None @@ -560,7 +613,7 @@ class ScanDirectory: files.append(os.path.relpath(full_path, self.dir_path)) return files - def save_scan_data(self, scan_data: ScanData) -> None: + def save_scan_data(self, scan_data: BaseScanData) -> None: """Save the scan data for this scan to disk.""" if self.scan_data_path is None: raise FileNotFoundError( diff --git a/src/openflexure_microscope_server/stitching.py b/src/openflexure_microscope_server/stitching.py index e505971f..9a69f7c1 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,22 +221,18 @@ 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 ScanData for this scan as a dictionary. This is used - to read/calculate overlap and correlation_resize if they are not provided. """ + if not isinstance(stitching_settings, StitchingSettings): + raise StitcherValidationError( + "Final stitcher requires settings to be set as a StitchingSettings " + "model" + ) 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 ) @@ -240,57 +246,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/autofocus.py b/src/openflexure_microscope_server/things/autofocus.py index b1165888..266d4bd1 100644 --- a/src/openflexure_microscope_server/things/autofocus.py +++ b/src/openflexure_microscope_server/things/autofocus.py @@ -33,8 +33,8 @@ class NotStreamingError(RuntimeError): """No images captured from stream. The camera is almost certainly not streaming.""" -class StackParams(BaseModel): - """A class for holding for stack parameters, and returning computed ones.""" +class SmartStackParams(BaseModel): + """A class for holding for smart stack parameters, and returning computed ones.""" stack_dz: int images_to_save: int @@ -80,7 +80,8 @@ class StackParams(BaseModel): ) if min_images_to_test > MAX_TEST_IMAGE_COUNT: raise ValueError( - f"Testing with more than {MAX_TEST_IMAGE_COUNT} images is likely to focus on the cover slip, or strike the sample." + f"Testing with more than {MAX_TEST_IMAGE_COUNT} images is likely to " + "focus on the cover slip, or strike the sample." ) if min_images_to_test % 2 == 0 or min_images_to_test <= 0: raise ValueError( @@ -97,7 +98,7 @@ class StackParams(BaseModel): return images_to_save @model_validator(mode="after") - def check_image_limits(self) -> "StackParams": + def check_image_limits(self) -> "SmartStackParams": """Ensure the number of images to save isn't more than the minimum tested.""" if self.images_to_save > self.min_images_to_test: raise ValueError("Can't save more images than the minimum number tested.") @@ -145,7 +146,7 @@ class StackParams(BaseModel): @dataclass class CaptureInfo: - """The information from a capture in a z_stack.""" + """The information from a capture in a smart_z_stack.""" buffer_id: int position: Mapping[str, int] @@ -452,107 +453,10 @@ class AutofocusThing(lt.Thing): "Looping autofocus couldn't converge on a focus location." ) - stack_images_to_save: int = lt.setting(default=1) - """The number of images to save in a stack. - - Defaults to 1 unless you need to see either side of focus - """ - - stack_min_images_to_test: int = lt.setting(default=9) - """The minimum number of images to capture in a stack. - - This many images are captures and tested for focus, if the focus is not central - enough more images may be captured. After new images are captured the number sets - the number of images used for checking if focus is central. - - Defaults to 9 which balances reliability and speed. - """ - - stack_dz: int = lt.setting(default=50) - """Distance in steps between images in a z-stack. - - Suggested values: - - * 50 for 60-100x - * 100 for 40x - * 200 for 20x - """ - - @lt.action - def create_stack_params( - self, - images_dir: str, - autofocus_dz: int, - save_resolution: tuple[int, int], - ) -> StackParams: - """Set up the parameters used for all stacks in a scan. - - :param images_dir: the folder to save all images - :param autofocus_dz: the range to autofocus over if a stack fails - :param save_resolution: The resolution to save the captures to disk with - - :returns: A StackParams object with the required parameters. - """ - # Coerce min_images_to_test parameter - min_images_to_test = self.stack_min_images_to_test - if min_images_to_test < MIN_TEST_IMAGE_COUNT: - self.logger.warning( - f"Cannot test only {min_images_to_test} image(s) as this will fail. " - "Setting min images to test to lowest possible value of" - f"{MIN_TEST_IMAGE_COUNT}." - ) - min_images_to_test = MIN_TEST_IMAGE_COUNT - elif min_images_to_test > MAX_TEST_IMAGE_COUNT: - self.logger.warning( - f"Testing {min_images_to_test} images will cause defocus. " - "Setting min images to test to highest possible value of " - f"{MAX_TEST_IMAGE_COUNT}." - ) - min_images_to_test = MAX_TEST_IMAGE_COUNT - elif min_images_to_test % 2 == 0: - min_images_to_test += 1 - self.logger.warning( - "Minimum number of images to test should be odd, setting to " - f"{min_images_to_test}." - ) - # Set the Thing property to the coerced value - self.stack_min_images_to_test = min_images_to_test - - # Coerce the images to save parameter to be positive, odd, and less than - # min_images_to_save - images_to_save = self.stack_images_to_save - if images_to_save <= 0: - self.logger.warning( - "At least 1 images must be saved. Setting images to save to 1." - ) - images_to_save = 1 - elif images_to_save > min_images_to_test: - self.logger.warning( - f"Cannot save {images_to_save} images as this above the minimum " - f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}." - ) - images_to_save = min_images_to_test - elif images_to_save % 2 == 0: - images_to_save += 1 - self.logger.warning( - f"Images to save should be odd, setting to {images_to_save}." - ) - # Set the Thing property to the coerced value - self.stack_images_to_save = images_to_save - - return StackParams( - stack_dz=self.stack_dz, - images_to_save=self.stack_images_to_save, - min_images_to_test=self.stack_min_images_to_test, - autofocus_dz=autofocus_dz, - images_dir=images_dir, - save_resolution=save_resolution, - ) - @lt.action def run_smart_stack( self, - stack_parameters: StackParams, + stack_parameters: SmartStackParams, save_on_failure: bool = False, check_turning_points: bool = True, ) -> tuple[bool, int]: @@ -564,7 +468,7 @@ class AutofocusThing(lt.Thing): The sharpest image, and optionally images around the sharpest, will be saved to the images_dir with their coordinates in the filename. - :param stack_parameters: A StackParams object containing the required + :param stack_parameters: A SmartStackParams object containing the required parameters to run a stack. :param save_on_failure: Whether to save an image even if no focus was found. :param check_turning_points: Whether to check the number of turning points in @@ -579,7 +483,7 @@ class AutofocusThing(lt.Thing): attempt = 0 while True: attempt += 1 - success, captures, sharpest_id = self.z_stack( + success, captures, sharpest_id = self.smart_z_stack( stack_parameters=stack_parameters, check_turning_points=check_turning_points, ) @@ -630,7 +534,7 @@ class AutofocusThing(lt.Thing): self, sharpest_id: int, captures: list[CaptureInfo], - stack_parameters: StackParams, + stack_parameters: SmartStackParams, ) -> int: """Save the required captures to disk. @@ -640,7 +544,7 @@ class AutofocusThing(lt.Thing): :param sharpest_id: the buffer id index of the sharpest image :param captures: a list of captures, including file name, image data and metadata - :param stack_parameters: a StackParams object holding stack parameters + :param stack_parameters: a SmartStackParams object holding stack parameters """ sharpest_index = _get_capture_index_by_id(captures, sharpest_id) slice_to_save = stack_parameters.slice_to_save(sharpest_index) @@ -655,19 +559,23 @@ class AutofocusThing(lt.Thing): self._cam.clear_buffers() return sharpest_index - def z_stack( + def smart_z_stack( self, - stack_parameters: StackParams, + stack_parameters: SmartStackParams, check_turning_points: bool, ) -> tuple[bool, list[CaptureInfo], int]: """Capture a series of images checking that sharpest image central. + This is part of run_smart_stack. This is the actual z_stackng stacking method + called by the action run_smart_stack. The action also handles resetting, + autofocussing, and retrying. + The images are separated in z offset by stack_parameters.stack_dz, as they are captured the last stack_parameters.min_images_to_test images are checked to see if the sharpest image is central enough in the stack. If it is the stack completes. - :param stack_parameters: a StackParams object holding stack parameters + :param stack_parameters: a SmartStackParams object holding stack parameters :param check_turning_points: Whether to check the number of turning points in the sharpnesses of the images in the stack is exactly 1. (May fail with thick samples) diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py index d8c37011..cb599bdc 100644 --- a/src/openflexure_microscope_server/things/camera/__init__.py +++ b/src/openflexure_microscope_server/things/camera/__init__.py @@ -596,6 +596,7 @@ class BaseCamera(lt.Thing): """Validate and set background_detector_name.""" if name not in self._all_background_detectors: self.logger.warning(f"{name} is not a valid background detector name.") + return self._background_detector_name = name @property diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py new file mode 100644 index 00000000..dc6a30a0 --- /dev/null +++ b/src/openflexure_microscope_server/things/scan_workflows.py @@ -0,0 +1,480 @@ +"""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 ( + Generic, + Mapping, + Optional, + TypeVar, +) + +from pydantic import BaseModel + +import labthings_fastapi as lt + +from openflexure_microscope_server.scan_planners import ScanPlanner, SmartSpiral +from openflexure_microscope_server.stitching import ( + STITCHING_RESOLUTION, + StitchingSettings, +) +from openflexure_microscope_server.things.autofocus import ( + MAX_TEST_IMAGE_COUNT, + MIN_TEST_IMAGE_COUNT, + AutofocusThing, + SmartStackParams, +) +from openflexure_microscope_server.things.background_detect import ( + ChannelDeviationLUV, +) +from openflexure_microscope_server.things.camera import BaseCamera +from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper +from openflexure_microscope_server.ui import PropertyControl, property_control_for + +SettingModelType = TypeVar("SettingModelType", bound=BaseModel) + + +class ScanWorkflow(Generic[SettingModelType], lt.Thing): + """A base class for all Scanworkflows. + + Scan workflows set the behaviour of a scan, including the background detection, + scan planning, acquisition routine. + """ + + display_name: str = lt.property(default="Base Workflow", readonly=True) + ui_blurb: str = lt.property( + default="If you see this message, something is wrong.", readonly=True + ) + + _settings_model: type[SettingModelType] + + # All workflows must have a set class for scan planning + _planner_cls: type[ScanPlanner] + + # All workflows set a save resolution + save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232)) + """A tuple of the image resolution to capture.""" + + def check_before_start(self, scan_name: str) -> None: + """Check before the scan starts. Throw an error if the scan shouldn't start. + + The scan_name is passed to this function to enable workflows to validate the + scan name if needed. + """ + raise NotImplementedError( + "Each specific ScanWorkflow must implement a check_before_start." + ) + + @lt.property + def ready(self) -> bool: + """Whether this scanworkflow is ready to start.""" + raise NotImplementedError( + "Each specific ScanWorkflow must implement a ready property." + ) + + def all_settings( + self, images_dir: str + ) -> 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 StitchingSettings object or None + is returned if it is not possible to stitch the scan. + """ + raise NotImplementedError( + "Each specific ScanWorkflow must implement a `all_settings`. method." + ) + + 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." + ) + + 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." + ) + + 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.""" + raise NotImplementedError( + "Each specific ScanWorkflow must implement an acquisition routine" + ) + + @lt.property + def settings_ui(self) -> list[PropertyControl]: + """A list of PropertyControl objects to create the settings in the scan tab.""" + raise NotImplementedError( + "Each scan workflow must implement a settings_ui method." + ) + + +class HistoScanSettingsModel(BaseModel): + """The settings for a scan with the HistoScanWorkflow. + + This includes settings calculated when starting. This will be held by smart scan + during a scan and serialised to disk. + """ + + overlap: float + max_dist: int + dx: int + dy: int + skip_background: bool + smart_stack_params: SmartStackParams + + +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. + """ + + display_name: str = lt.property(default="Histo Scan", readonly=True) + ui_blurb: str = lt.property( + default=( + "This scan workflow is optimised for scanning H&E stained biopsies. It " + "spirals out from the starting location, scanning only where it detects " + "sample. It also works well for many other flat, well-featured samples." + ), + readonly=True, + ) + + _settings_model = HistoScanSettingsModel + _planner_cls: type[ScanPlanner] = 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 + + skip_background: bool = lt.setting(default=True) + """Whether to detect and skip empty fields of view. + + 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) + """The number of images to save in a stack. + + Defaults to 1 unless you need to see either side of focus + """ + + stack_min_images_to_test: int = lt.setting(default=9) + """The minimum number of images to capture in a stack. + + This many images are captures and tested for focus, if the focus is not central + enough more images may be captured. After new images are captured, this value sets + the number of images used for checking if focus is achieved. + + Defaults to 9 which balances reliability and speed. + """ + + stack_dz: int = lt.setting(default=50) + """Distance in steps between images in a z-stack. + + Suggested values: + + * 50 for 60-100x + * 100 for 40x + * 200 for 20x + """ + + # 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 background and camera-stage-mapping are set. + + Raise error if: + - background is to be skipped but is not set + - camera stage mapping is not set + + Raise warning if not using background detect that scan will go on until max steps reached + """ + if self._csm.calibration_required: + raise RuntimeError("Camera Stage Mapping is not calibrated.") + + if self.skip_background: + if not self._background_detector.ready: + raise RuntimeError( + "Background is not set: you need to calibrate background detection." + ) + else: + self.logger.warning( + "This scan will run in a spiral from the starting point " + f"until you cancel it, or until it has moved by {self.max_range} steps " + "in every direction. Make sure you watch it run to stop it leaving " + "the area of interest, or (worse) leading the microscope's range " + "of motion." + ) + + @lt.property + def ready(self) -> bool: + """Whether this scanworkflow is ready to start.""" + if self._csm.calibration_required: + return False + if not self.skip_background: + return True + return self._background_detector.ready + + def all_settings( + self, images_dir: str + ) -> 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], + ) + + 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, + autofocus_dz=self.autofocus_dz, + save_resolution=self.save_resolution, + ) + + scan_settings = HistoScanSettingsModel( + overlap=self.overlap, + max_dist=self.max_range, + dx=dx, + dy=dy, + skip_background=self.skip_background, + smart_stack_params=smart_stack_params, + ) + + 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, + autofocus_dz: int, + save_resolution: tuple[int, int], + ) -> SmartStackParams: + """Set up the parameters used for all stacks in a scan. + + :param images_dir: the folder to save all images + :param autofocus_dz: the range to autofocus over if a stack fails + :param save_resolution: The resolution to save the captures to disk with + + :returns: A StackSmartParams object with the required parameters. + """ + # Coerce min_images_to_test parameter + min_images_to_test = self.stack_min_images_to_test + if min_images_to_test < MIN_TEST_IMAGE_COUNT: + self.logger.warning( + f"Cannot test only {min_images_to_test} image(s) as this will fail. " + "Setting min images to test to lowest possible value of" + f"{MIN_TEST_IMAGE_COUNT}." + ) + min_images_to_test = MIN_TEST_IMAGE_COUNT + elif min_images_to_test > MAX_TEST_IMAGE_COUNT: + self.logger.warning( + f"Testing {min_images_to_test} images will cause defocus. " + "Setting min images to test to highest possible value of " + f"{MAX_TEST_IMAGE_COUNT}." + ) + min_images_to_test = MAX_TEST_IMAGE_COUNT + elif min_images_to_test % 2 == 0: + min_images_to_test += 1 + self.logger.warning( + "Minimum number of images to test should be odd, setting to " + f"{min_images_to_test}." + ) + # Set the Thing property to the coerced value + self.stack_min_images_to_test = min_images_to_test + + # Coerce the images to save parameter to be positive, odd, and less than + # min_images_to_save + images_to_save = self.stack_images_to_save + if images_to_save <= 0: + self.logger.warning( + "At least 1 images must be saved. Setting images to save to 1." + ) + images_to_save = 1 + elif images_to_save > min_images_to_test: + self.logger.warning( + f"Cannot save {images_to_save} images as this above the minimum " + f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}." + ) + images_to_save = min_images_to_test + elif images_to_save % 2 == 0: + images_to_save += 1 + self.logger.warning( + f"Images to save should be odd, setting to {images_to_save}." + ) + # Set the Thing property to the coerced value + self.stack_images_to_save = images_to_save + + return SmartStackParams( + stack_dz=self.stack_dz, + images_to_save=self.stack_images_to_save, + min_images_to_test=self.stack_min_images_to_test, + autofocus_dz=autofocus_dz, + images_dir=images_dir, + save_resolution=save_resolution, + ) + + 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" + ) + + 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 + # 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 acquisition_routine( + self, settings: HistoScanSettingsModel, 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 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. + """ + if settings.skip_background: + # If skipping background, take an image to check if current field of view + # is background + image_array = self._cam.grab_as_array(stream_name="lores") + capture_image, bg_message = self._background_detector.image_is_sample( + image_array + ) + del image_array + + if not capture_image: + # Return early if the image is background. + msg = f"Skipping {xyz_pos} as it is {bg_message}." + self.logger.info(msg) + return False, None + + save_on_failure = not settings.skip_background + + focus_height: Optional[int] + focused, focus_height = self._autofocus.run_smart_stack( + stack_parameters=settings.smart_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 + + if not imaged: + msg = f"Stack failed at {xyz_pos}. Treating as background." + self.logger.info(msg) + + # 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: + focus_height = None + + 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, "skip_background", label="Detect and Skip Empty Fields " + ), + property_control_for( + self, "stack_images_to_save", label="Images in Stack to Save" + ), + property_control_for( + self, + "stack_min_images_to_test", + label="Minimum number of images to test for focus", + ), + property_control_for(self, "stack_dz", label="Stack dz (steps)"), + property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), + property_control_for(self, "max_range", label="Maximum Distance (steps)"), + ] diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index ebbe37f2..3f8b36e0 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -1,8 +1,8 @@ """The core sample scanning functionality for the OpenFlexure Microscope. -SmartScan provides sample scanning functionality including automatic background -detection (via the ``CameraThing``) and automatic path planning via -`scan_planners`. It manages the directories of past scans via `scan_directories`. +SmartScan provides sample scanning functionality. This functionality can be customised +by different ``ScanWorkflow`` Things which control the path planning and acquisition +routines. It manages the directories of past scans via `scan_directories`. It also controls external processes for live stitching composite images, and the creation of the final stitched images. """ @@ -12,35 +12,64 @@ import threading import time from datetime import datetime from subprocess import SubprocessError +from types import TracebackType from typing import ( + Annotated, Any, Callable, Concatenate, Mapping, Optional, ParamSpec, + Self, TypeVar, ) -import numpy as np from fastapi import HTTPException from fastapi.responses import FileResponse -from pydantic import BaseModel +from pydantic import BaseModel, PlainSerializer import labthings_fastapi as lt -from openflexure_microscope_server import scan_directories, scan_planners, stitching +from openflexure_microscope_server import scan_directories, stitching +from openflexure_microscope_server.utilities import coerce_thing_selector # Things -from .autofocus import AutofocusThing, StackParams from .camera import BaseCamera -from .camera_stage_mapping import CameraStageMapper, CSMUncalibratedError +from .scan_workflows import ScanWorkflow from .stage import BaseStage T = TypeVar("T") P = ParamSpec("P") +# This allows ActiveScanData to hold arbitrary workflow settings models during a scan. +AnyModel = Annotated[ + BaseModel, + PlainSerializer(lambda value: value.model_dump(), return_type=dict), +] + + +class ActiveScanData(scan_directories.BaseScanData): + """A model for the scan data 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.""" + + def set_final_data(self, result: str) -> None: + """Set the final data for the scan, scan duration is automatically calculated. + + :param result: A string describing the result. + """ + self.duration = datetime.now() - self.start_time + self.scan_result = result + + class ScanListInfo(BaseModel): """The information to be sent to the Scan List tab.""" @@ -99,13 +128,15 @@ class SmartScanThing(lt.Thing): past scans. """ - _autofocus: AutofocusThing = lt.thing_slot() _cam: BaseCamera = lt.thing_slot() - _csm: CameraStageMapper = lt.thing_slot() _stage: BaseStage = lt.thing_slot() + _all_workflows: Mapping[str, ScanWorkflow] = lt.thing_slot() def __init__( - self, thing_server_interface: lt.ThingServerInterface, scans_folder: str + self, + thing_server_interface: lt.ThingServerInterface, + scans_folder: str, + default_workflow: str, ) -> None: """Initialise a SmartScanThing saving to and loading from the input directory. @@ -116,19 +147,54 @@ class SmartScanThing(lt.Thing): super().__init__(thing_server_interface) self._scan_dir_manager = scan_directories.ScanDirectoryManager(scans_folder) self._scan_lock = threading.Lock() + self._default_workflow = default_workflow + self._workflow_name = default_workflow - # Variables set by the scan - _stack_params: Optional[StackParams] = None + def __enter__(self) -> Self: + """Open hardware connection when the Thing context manager is opened.""" + valid_name = coerce_thing_selector( + thing_mapping=self._all_workflows, + selected=self.workflow_name, + default=self._default_workflow, + ) + if valid_name is None: + raise RuntimeError( + "Could not set Scan Workflow. A Scan Workflow must be present in your " + "configuration." + ) + self._workflow_name = valid_name + return self + + def __exit__( + self, + _exc_type: type[BaseException], + _exc_value: Optional[BaseException], + _traceback: Optional[TracebackType], + ) -> None: + """Clean up after context manager is closed. + + In this case it doesn't need to do anything. + """ + + # Note that the default detector name is set at init. This is over written if + # setting is loaded from disk. + @lt.setting + def workflow_name(self) -> str: + """The name of the scan workflow selector.""" + return self._workflow_name + + @workflow_name.setter + def _set_workflow_name(self, name: str) -> None: + """Validate and set workflow_name.""" + if name not in self._all_workflows: + self.logger.warning(f"'{name}' is not a valid scan workflow name.") + return + self._workflow_name = name @property - def stack_params(self) -> StackParams: - """The parameters for z-stacking during the onging scan. - - Only read this property is a scan is ongoing or it will raise an error. - """ - if self._stack_params is None: - raise RuntimeError("Cannot get stack parameters as they are not set.") - return self._stack_params + def _workflow(self) -> ScanWorkflow: + """The active scan workflow object.""" + return self._all_workflows[self.workflow_name] _ongoing_scan: Optional[scan_directories.ScanDirectory] = None @@ -142,11 +208,11 @@ class SmartScanThing(lt.Thing): raise ScanNotRunningError("Cannot get ongoing scan if scan is not running.") return self._ongoing_scan - _scan_data: Optional[scan_directories.ScanData] = None + _scan_data: Optional[ActiveScanData] = None @property - def scan_data(self) -> scan_directories.ScanData: - """The ScanData object jolding information about the of the ongoing scan. + def scan_data(self) -> ActiveScanData: + """The ActiveScanData object holding information about the of the ongoing scan. Only read this property is a scan is ongoing or it will raise an error. """ @@ -156,18 +222,6 @@ class SmartScanThing(lt.Thing): _preview_stitcher: Optional[stitching.PreviewStitcher] = None - @property - def preview_stitcher(self) -> stitching.PreviewStitcher: - """The PreviewStitcher object for stitching live previews. - - Only read this property is a scan is ongoing or it will raise an error. - """ - if self._preview_stitcher is None: - raise ScanNotRunningError( - "No preview stitcher agailable as scan is not running." - ) - return self._preview_stitcher - _latest_scan_name: Optional[str] = None @lt.property @@ -177,26 +231,28 @@ class SmartScanThing(lt.Thing): @lt.action def sample_scan(self, scan_name: str = "") -> None: - """Move the stage to cover an area, taking images that can be tiled together. + """Move the stage to cover an area, taking images. - The stage will move in a pattern that grows outwards from the starting point, - stopping once it is surrounded by "background" (as detected by the - camera Thing) or reaches the "max_range" measured in steps. + The way the stage moves depends on the selected workflow. + If images overlap for a scan workflow then the images can be stitched together + into a larger composite image. """ got_lock = self._scan_lock.acquire(timeout=0.1) if not got_lock: raise RuntimeError("Trying to run scan while scan is already running!") - # `scan_data` should already be None. This is added as a precaution as - # the presence of `scan_data` is used during error handling to - # determine whether the scan started. - self._scan_data = None try: - self._check_background_and_csm_set() + # `scan_data` should already be None. This is added as a precaution as + # the presence of `scan_data` is used during error handling to + # determine whether the scan started. + self._scan_data = None + # probably make workflow a context manager with a lock? + workflow = self._workflow + + workflow.check_before_start(scan_name) self._ongoing_scan = self._scan_dir_manager.new_scan_dir(scan_name) self._latest_scan_name = self.ongoing_scan.name - self._autofocus.looping_autofocus(dz=self.autofocus_dz, start="centre") - self._run_scan() + self._run_scan(workflow) except Exception as e: # If _scan_data is set then scan started if self._scan_data is not None: @@ -216,40 +272,10 @@ class SmartScanThing(lt.Thing): self._scan_lock.release() # Ensure any PreviewStitcher created cannot be reused. self._preview_stitcher = None - self._stack_params = None # Remove any scan folders containing zero images. self.purge_empty_scans() - @_scan_running - def _check_background_and_csm_set(self) -> None: - """Before starting a scan, check that background and camera-stage-mapping are set. - - Raise error if: - - background is to be skipped but is not set - - camera stage mapping is not set - - Raise warning if not using background detect that scan will go on until max steps reached - """ - self._csm.assert_calibration() - - if self.skip_background: - if ( - self._cam.background_detector is None - or not self._cam.background_detector.ready - ): - raise RuntimeError( - "Background is not set: you need to calibrate background detection." - ) - else: - self.logger.warning( - "This scan will run in a spiral from the starting point " - f"until you cancel it, or until it has moved by {self.max_range} steps " - "in every direction. Make sure you watch it run to stop it leaving " - "the area of interest, or (worse) leading the microscope's range " - "of motion." - ) - @_scan_running def _move_to_next_point( self, next_point: tuple[int, int], z_estimate: Optional[int] = None @@ -275,90 +301,33 @@ class SmartScanThing(lt.Thing): return (next_point[0], next_point[1], z_estimate) @_scan_running - def _calc_displacement_from_test_image(self, overlap: float) -> tuple[int, int]: - """Take a test image and 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 - """ - if ( - self._csm.image_resolution is None - or self._csm.image_to_stage_displacement_matrix is None - ): - raise CSMUncalibratedError("Camera stage mapping is not calibrated") - test_image = self._cam.grab_as_array() - - test_image_res = list(test_image.shape) - - csm_image_res = [int(i) for i in self._csm.image_resolution] - - # If current stream width is different to csm calibration width, - # perform the conversion here - res_ratio = csm_image_res[0] / test_image_res[0] - - # get displacement matrix. note it is for (y, x) not (x, y) coordinates - csm_disp_matrix = np.array(self._csm.image_to_stage_displacement_matrix) - csm_disp_matrix *= res_ratio - - # Calculate displacements in image coordinates - dx_img = test_image.shape[1] * (1 - overlap) - dy_img = test_image.shape[0] * (1 - overlap) - - # Calculate displacements in steps as vectors using a dot product with the matrix - dx_vec = np.dot(np.array([0, dx_img]), csm_disp_matrix) - dy_vec = np.dot(np.array([dy_img, 0]), csm_disp_matrix) - - # Assume no rotation or skew and take only the aligned axis of vector. - # Coerce to positive integer - dx = int(np.abs(dx_vec[0])) - dy = int(np.abs(dy_vec[1])) - - return dx, dy - - @_scan_running - def _collect_scan_data(self) -> scan_directories.ScanData: + def _collect_scan_data(self, workflow: ScanWorkflow) -> ActiveScanData: """Collect and return the data for this scan so it cannot be changed mid-scan.""" # Record starting position so it can be returned to at end of scan. starting_position = self._stage.position - overlap = self.overlap - dx, dy = self._calc_displacement_from_test_image(overlap) - correlation_resize = stitching.STITCHING_RESOLUTION[0] / self.save_resolution[0] - self.logger.debug( - f"Resizing images when correlating by a factor of {correlation_resize}" + images_dir = self.ongoing_scan.images_dir + # Type narrowing + if images_dir is None: + raise RuntimeError("Couldn't run scan, images directory was not created.") + + workflow_settings, stitching_settings = workflow.all_settings( + images_dir=images_dir ) - self.logger.info( - f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}" - ) - - 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 + # If stitching settings is None then this workflow doesn't support stitching. + auto_stitch = self.stitch_automatically and stitching_settings is not None # Fix scan parameters in case UI is updated during scan. - return scan_directories.ScanData( + return ActiveScanData( scan_name=self.ongoing_scan.name, starting_position=starting_position, - overlap=overlap, - max_dist=self.max_range, - dx=dx, - dy=dy, - autofocus_dz=autofocus_dz, - autofocus_on=bool(autofocus_dz), start_time=datetime.now(), - skip_background=self.skip_background, - stitch_automatically=self.stitch_automatically, - correlation_resize=correlation_resize, - save_resolution=self.save_resolution, + stitch_automatically=auto_stitch, + save_resolution=workflow.save_resolution, + workflow=type(workflow).__name__, + workflow_settings=workflow_settings, + stitching_settings=stitching_settings, ) @_scan_running @@ -374,13 +343,16 @@ 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._preview_stitcher 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: - self.preview_stitcher.start() + if self.scan_data.image_count > 3 and not self._preview_stitcher.running: + self._preview_stitcher.start() @_scan_running - def _run_scan(self) -> None: + def _run_scan(self, workflow: ScanWorkflow) -> None: """Prepare and run the main scan, and perform final actions on completion. The result (or exception) from the main scan loop determines whether the @@ -389,26 +361,29 @@ class SmartScanThing(lt.Thing): """ try: self._cam.start_streaming(main_resolution=(3280, 2464)) - self._scan_data = self._collect_scan_data() + self._scan_data = self._collect_scan_data(workflow) + + workflow.pre_scan_routine(self._scan_data.workflow_settings) self.ongoing_scan.save_scan_data(self._scan_data) images_dir = self.ongoing_scan.images_dir + # Type narrowing if images_dir is None: raise RuntimeError( "Couldn't run scan, images directory was not created." ) - self._stack_params = self._autofocus.create_stack_params( - images_dir=images_dir, - autofocus_dz=self.autofocus_dz, - save_resolution=self.scan_data.save_resolution, - ) - self._preview_stitcher = stitching.PreviewStitcher( - images_dir, - overlap=self.scan_data.overlap, - correlation_resize=self.scan_data.correlation_resize, - ) + + # If stitching settings are None then this type of scan can't be stitched + if self.scan_data.stitching_settings is not None: + # Settings exist, so create preview stitcher + stitching_settings = self.scan_data.stitching_settings + self._preview_stitcher = stitching.PreviewStitcher( + images_dir, + overlap=stitching_settings.overlap, + correlation_resize=stitching_settings.correlation_resize, + ) # This is the main loop of the scan! - self._main_scan_loop() + self._main_scan_loop(workflow) self._save_final_scan_data(scan_result="success") except lt.exceptions.InvocationCancelledError: @@ -437,24 +412,15 @@ class SmartScanThing(lt.Thing): self._perform_final_stitch() @_scan_running - def _main_scan_loop(self) -> None: + def _main_scan_loop(self, workflow: ScanWorkflow) -> None: """Run the main loop of the scan. 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, @@ -472,43 +438,31 @@ 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.acquisition_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: + focused = True + 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 @@ -532,18 +486,15 @@ class SmartScanThing(lt.Thing): self.logger.info("Waiting for background processes to finish...") - # Actually check the sticher exists rather than using self.preview_sticher as + # Check the sticher exists rather than using self.preview_sticher as # this method can be called during exception handling. if self._preview_stitcher is not None: self._preview_stitcher.wait() - if self.scan_data.stitch_automatically: + 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=self.scan_data.correlation_resize, - overlap=self.scan_data.overlap, - ) + self.stitch_scan(scan_name=self.ongoing_scan.name) @lt.endpoint( "get", @@ -568,26 +519,9 @@ class SmartScanThing(lt.Thing): raise HTTPException(404, "File not found") return FileResponse(preview_path) - save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232)) - """A tuple of the image resolution to capture.""" - - max_range: int = lt.setting(default=45000) - """The maximum distance in steps from the centre of the scan.""" - stitch_tiff: bool = lt.setting(default=False) """Whether or not to also produce a pyramidal tiff at the end of a scan.""" - skip_background: bool = lt.setting(default=True) - """Whether to detect and skip empty fields of view. - - This uses the settings from the ``BackgroundDetectThing``.""" - - autofocus_dz: int = lt.setting(default=1000) - """The z distance to perform an autofocus in steps.""" - - overlap: float = lt.setting(default=0.45) - """The fraction (0-1) that adjacent images should overlap in x or y.""" - stitch_automatically: bool = lt.setting(default=True) """Whether to run a final stitch at the end of a successful scan.""" @@ -748,25 +682,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() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..a7d77bfe --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,63 @@ +"""Fixtures for all test suites.""" + +import logging +import re +from collections.abc import Iterable +from contextlib import contextmanager +from typing import Optional + +import pytest + + +@pytest.fixture +def check_side_effect(caplog): + """Supply a context manager for checking code either logs, raises an error, or doesn't.""" + + @contextmanager + def _checker( + side_effect: Optional[type | int | Iterable[int]], + match: Optional[str | Iterable[str]] = None, + ) -> None: + """Check the code within this context has the expected side effect. + + :param side_effect: The side effect of the code within this context. An int + should be a logging level number, or a list of level numbers if multiple + logs are expected. + :param match: Optionally a match regex for raises or the logger, can be a list + if side effect is a list of levels. + """ + # If the side effect is an exception + if isinstance(side_effect, type) and issubclass(side_effect, BaseException): + with pytest.raises(side_effect, match=match): + yield + return + + # If the side effect is None or logging + caplog.clear() + + with caplog.at_level(logging.INFO): + yield + + if side_effect is None: + # No side effect + assert caplog.records == [] + elif isinstance(side_effect, Iterable): + # Multiple logs + if match is not None: + assert isinstance(match, Iterable) + assert len(match) == len(side_effect) + assert len(caplog.records) == len(side_effect) + for i, level in enumerate(side_effect): + record = caplog.records[i] + assert record.levelno == level + if match is not None: + assert re.match(match[i], record.message) is not None + else: + # Single log + assert len(caplog.records) == 1 + record = caplog.records[0] + assert record.levelno == side_effect + if match is not None: + assert re.match(match, record.message) is not None + + return _checker diff --git a/tests/unit_tests/test_scan_data.py b/tests/unit_tests/test_scan_data.py index b14ba061..cc29935b 100644 --- a/tests/unit_tests/test_scan_data.py +++ b/tests/unit_tests/test_scan_data.py @@ -5,7 +5,11 @@ from copy import copy from datetime import datetime, timedelta from math import floor -from openflexure_microscope_server.scan_directories import ScanData +from pydantic import BaseModel + +from openflexure_microscope_server.scan_directories import HistoricScanData +from openflexure_microscope_server.stitching import StitchingSettings +from openflexure_microscope_server.things.smart_scan import ActiveScanData MOCK_START_TIME = datetime( year=2024, @@ -28,8 +32,8 @@ MOCK_END_TIME = datetime( ) -def _fake_scan_data(**kwargs) -> ScanData: - """Make fake scan data, the start time is now. Final properties are not added. +def _fake_legacy_scan_data(**kwargs) -> HistoricScanData: + """Make fake legacy scan data. :param **kwargs: Key word arguments can be used to override other values. """ @@ -50,12 +54,73 @@ def _fake_scan_data(**kwargs) -> ScanData: } for key, value in kwargs.items(): data_dict[key] = value - return ScanData(**data_dict) + return HistoricScanData(**data_dict) + + +class MockWorkflowSettingModel(BaseModel): + """A mock model to check that ActiveScanData can hold arbitrary models.""" + + setting_1: int + setting_2: int + setting_3: str + + +def fake_active_scan_data(): + """Fake scan data for and active scan. + + The start time is now. Final properties are not added. + """ + return ActiveScanData( + schema_version=2, + scan_name="fake_scan_0001", + starting_position={"x": 123, "y": 456, "z": 789}, + start_time=copy(MOCK_START_TIME), + stitch_automatically=True, + save_resolution=(1000, 1000), + stitching_settings=StitchingSettings(correlation_resize=0.25, overlap=0.1), + workflow="MockWorkflow", + workflow_settings=MockWorkflowSettingModel( + setting_1=1, + setting_2=2, + setting_3="three", + ), + ) + + +def assert_active_and_historic_data_equivalent(active_data, historic_data): + """Raise and error if active and historic scan data is not equivalent.""" + assert isinstance(active_data, ActiveScanData) + assert isinstance(historic_data, HistoricScanData) + # For the round trip to be equal we must remove microseconds from the start + # time as they are not saved + active_data.start_time = active_data.start_time.replace(microsecond=0) + for key in active_data.model_fields: + if key == "workflow_settings": + # For workflow_settings check the base model serialises to the historic + # data. + active_wf_setting_dict = active_data.workflow_settings.model_dump() + assert historic_data.workflow_settings == active_wf_setting_dict + continue + + assert getattr(active_data, key) == getattr(historic_data, key) + + +def test_legacy_data_validates(): + """Check that legacy scan data validates.""" + scan_data = _fake_legacy_scan_data() + assert isinstance(scan_data, HistoricScanData) + assert scan_data.image_count == 0 + assert scan_data.duration is None + assert scan_data.scan_result is None + # Most importantly legacy stitching data should now be in the StitchingSettings + # model + assert scan_data.stitching_settings.correlation_resize == 0.25 + assert scan_data.stitching_settings.overlap == 0.1 def test_set_final_data(): - """Check that adding final data to a ScanData object works as expected.""" - scan_data = _fake_scan_data() + """Check that adding final data to a ActiveScanData object works as expected.""" + scan_data = fake_active_scan_data() assert scan_data.image_count == 0 assert scan_data.duration is None @@ -75,8 +140,8 @@ def test_set_final_data(): def test_custom_serialisation(): - """Check that the custom serialisation in ScanData works as expected.""" - scan_data = _fake_scan_data() + """Check that the custom serialisation in ActiveScanData works as expected.""" + scan_data = fake_active_scan_data() # Serialise to string then load directly as json scan_data_dict = json.loads(scan_data.model_dump_json()) assert scan_data_dict["start_time"] == "2024-12-25_11:00:00" @@ -95,26 +160,23 @@ def test_custom_serialisation(): def test_round_trip_not_finalised(): - """Check that ScanData without final data can be serialised and deserialised.""" - scan_data = _fake_scan_data() + """Check that ActiveScanData without final data can be serialised and deserialised.""" + scan_data = fake_active_scan_data() scan_data_dict = json.loads(scan_data.model_dump_json()) - scan_data_reloaded = ScanData(**scan_data_dict) + scan_data_reloaded = HistoricScanData(**scan_data_dict) - # For the round trip to be equal we must remove microseconds from the start - # time as they are not saved - scan_data.start_time = scan_data.start_time.replace(microsecond=0) - assert scan_data == scan_data_reloaded + assert_active_and_historic_data_equivalent(scan_data, scan_data_reloaded) def test_round_trip_finalised(): - """Check that finalised ScanData can be serialised and deserialised.""" - scan_data = _fake_scan_data() + """Check that finalised HistoricScanData can be serialised and deserialised.""" + scan_data = fake_active_scan_data() # Finalise the data. scan_data.image_count += 123 scan_data.set_final_data(result="Success") scan_data_dict = json.loads(scan_data.model_dump_json()) - scan_data_reloaded = ScanData(**scan_data_dict) + scan_data_reloaded = HistoricScanData(**scan_data_dict) # For the round trip to be equal we must remove microseconds from the start # time and duration as they are not saved @@ -122,4 +184,4 @@ def test_round_trip_finalised(): scan_data.start_time = scan_data.start_time.replace(microsecond=0) scan_data.duration = timedelta(seconds=floor(scan_data.duration.total_seconds())) - assert scan_data == scan_data_reloaded + assert_active_and_historic_data_equivalent(scan_data, scan_data_reloaded) diff --git a/tests/unit_tests/test_scan_directories.py b/tests/unit_tests/test_scan_directories.py index ff7af3eb..603587a3 100644 --- a/tests/unit_tests/test_scan_directories.py +++ b/tests/unit_tests/test_scan_directories.py @@ -21,7 +21,10 @@ from openflexure_microscope_server.scan_directories import ( get_files_in_zip, ) -from .test_scan_data import _fake_scan_data +from .test_scan_data import ( + assert_active_and_historic_data_equivalent, + fake_active_scan_data, +) from .utilities import assert_unique_of_length # Use our own dir in the root temp dir not a dynamically generated one so we @@ -173,7 +176,7 @@ def test_scan_sequence_and_listing(caplog): scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) # Create some scan data and mark it as successful to get an end date. - scan_data = _fake_scan_data() + scan_data = fake_active_scan_data() scan_data.set_final_data(result="Success") # Make 4 scans scan_dir = scan_dir_manager.new_scan_dir("fake_scan") @@ -362,26 +365,33 @@ def test_get_scan_data_path(): assert scan_dir_manager.get_scan_data_path(scan_name) is None -def test_get_scan_data_dict(): - """Check that the dictionary for the scan data is returned, or None if doesn't exist.""" +def test_get_scan_data(): + """Check that the scan data is returned, or None if doesn't exist.""" _clear_scan_dir() scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) scan_dir = scan_dir_manager.new_scan_dir("fake_scan") scan_name = scan_dir.name # Doesn't yet exist - assert scan_dir_manager.get_scan_data_dict(scan_name) is None + assert scan_dir_manager.get_scan_data(scan_name) is None - fake_data = {"foo": 1, "bar": "foobar"} + fake_active_data = fake_active_scan_data() with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file: - json.dump(fake_data, json_file) + json.dump(fake_active_data.model_dump(), json_file) # Should now be able to load this fake data from disk - assert scan_dir_manager.get_scan_data_dict(scan_name) == fake_data + fake_historic_data = scan_dir_manager.get_scan_data(scan_name) + assert_active_and_historic_data_equivalent(fake_active_data, fake_historic_data) # Check None is returned if the data cannot be read. with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file: json_file.write("this is not json") - assert scan_dir_manager.get_scan_data_dict(scan_name) is None + assert scan_dir_manager.get_scan_data(scan_name) is None + + # Check None is returned if the data cannot or is json but cannot be serialised to + # the data model + with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file: + json_file.write(json.dumps({"foo": "bar"})) + assert scan_dir_manager.get_scan_data(scan_name) is None def test_empty_scan_info(): @@ -442,29 +452,6 @@ def test_zipping_scan_data(): assert not file.endswith(".dzi") -def test_saving_and_loading_scan_data(): - """Test that scan data is saved and loaded as expected.""" - _clear_scan_dir() - scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) - scan_dir = scan_dir_manager.new_scan_dir("fake_scan") - scan_name = scan_dir.name - - # Should start without a scan data file. - assert not os.path.isfile(scan_dir.scan_data_path) - # Create - scan_data_obj = _fake_scan_data() - scan_dir.save_scan_data(scan_data_obj) - # File should now exist - assert os.path.isfile(scan_dir.scan_data_path) - - # Dump the scan json to a string an reload it - # Note that more detailed checking of the dumping and loading of ScanData is in - # tests/test_scan_data.py - scan_data_dict = json.loads(scan_data_obj.model_dump_json()) - # What is loaded from file should be the same as from dumping and loading. - assert scan_dir_manager.get_scan_data_dict(scan_name) == scan_data_dict - - def test_saving_scan_data_error(): """Test that saving scan data if there is no images directory raises FileNotFoundError.""" _clear_scan_dir() @@ -475,7 +462,7 @@ def test_saving_scan_data_error(): shutil.rmtree(scan_dir.images_dir) # Should raise FileNotFoundError. with pytest.raises(FileNotFoundError): - scan_dir.save_scan_data(_fake_scan_data()) + scan_dir.save_scan_data(fake_active_scan_data()) def test_all_files(): diff --git a/tests/unit_tests/test_scan_workflows.py b/tests/unit_tests/test_scan_workflows.py new file mode 100644 index 00000000..6406d5ab --- /dev/null +++ b/tests/unit_tests/test_scan_workflows.py @@ -0,0 +1,352 @@ +"""Tests for ScanWorkflow things.""" + +import itertools +import logging + +import pytest +from pydantic import BaseModel + +import labthings_fastapi as lt +from labthings_fastapi.testing import create_thing_without_server + +from openflexure_microscope_server.scan_planners import SmartSpiral +from openflexure_microscope_server.stitching import StitchingSettings +from openflexure_microscope_server.things.autofocus import SmartStackParams +from openflexure_microscope_server.things.camera_stage_mapping import csm_img_to_stage +from openflexure_microscope_server.things.scan_workflows import ( + HistoScanSettingsModel, + HistoScanWorkflow, + ScanWorkflow, +) +from openflexure_microscope_server.ui import PropertyControl + + +def test_partial_base_classes(): + """Create a partial class and check it raises the correct errors.""" + + class MinimalSetings(BaseModel): + """Some minimal settings for a workflow that doesn't work.""" + + foo: str = "bar" + + class BadWorkflow(ScanWorkflow[MinimalSetings]): + """Can initialise. Other properties and methods error.""" + + display_name: str = lt.property(default="Bad Workflow", readonly=True) + + bad_workflow = create_thing_without_server(BadWorkflow) + + settings = MinimalSetings() + with pytest.raises(NotImplementedError): + bad_workflow.check_before_start(settings) + + with pytest.raises(NotImplementedError): + bad_workflow.ready + + with pytest.raises(NotImplementedError): + bad_workflow.all_settings("Fake Dir") + + with pytest.raises(NotImplementedError): + bad_workflow.pre_scan_routine(settings) + + with pytest.raises(NotImplementedError): + bad_workflow.new_scan_planner(settings, position={"x": 0, "y": 0, "z": 0}) + + with pytest.raises(NotImplementedError): + bad_workflow.acquisition_routine(settings, xyz_pos=[0, 0, 0]) + + with pytest.raises(NotImplementedError): + bad_workflow.settings_ui + + +@pytest.fixture +def histo_workflow(): + """Return a HistoScanWorkflow thing with slots mocked.""" + return create_thing_without_server(HistoScanWorkflow, mock_all_slots=True) + + +# Use itertools to iterate over every true/false permutation +@pytest.mark.parametrize( + ("csm_calibrated", "skip_background", "bg_ready"), + itertools.product([True, False], repeat=3), +) +def test_histo_workflow_ready( + histo_workflow, csm_calibrated, skip_background, bg_ready, caplog +): + """Check ready property and the pre-run check work for each permutation.""" + histo_workflow._csm.calibration_required = not csm_calibrated + histo_workflow.skip_background = skip_background + histo_workflow._background_detector.ready = bg_ready + + if not csm_calibrated: + # For all permutations not ready if CSM isn't calibrated. + assert not histo_workflow.ready + with pytest.raises( + RuntimeError, match="Camera Stage Mapping is not calibrated." + ): + histo_workflow.check_before_start("scan_name") + elif not skip_background: + # CSM is ready and background is not skipped: Always ready, but warns on check + assert histo_workflow.ready + with caplog.at_level(logging.WARNING): + histo_workflow.check_before_start("scan_name") + assert len(caplog.records) == 1 + elif not bg_ready: + # Skipping background, but detector not ready. Raises error + assert not histo_workflow.ready + with pytest.raises(RuntimeError, match="Background is not set"): + histo_workflow.check_before_start("scan_name") + else: + # Finally CSM calibrated, skipping background, and detector ready: This is + # ready and should not warn. + assert histo_workflow.ready + with caplog.at_level(logging.WARNING): + histo_workflow.check_before_start("scan_name") + assert len(caplog.records) == 0 + + +def test_histo_workflow_settings_generation(histo_workflow, mocker): + """Check the settings models generate as expected.""" + mocker.patch.object( + histo_workflow, "_calc_displacement_from_overlap", return_value=(123, 456) + ) + workflow_settings, stitching_settings = histo_workflow.all_settings("/this/img_dir") + ## Check type + assert isinstance(workflow_settings, HistoScanSettingsModel) + assert isinstance(stitching_settings, StitchingSettings) + assert isinstance(workflow_settings.smart_stack_params, SmartStackParams) + + # Check stitching defaults + assert stitching_settings.correlation_resize == 0.5 + assert stitching_settings.overlap == 0.45 + # Check some workflow defaults + assert workflow_settings.overlap == 0.45 + assert workflow_settings.max_dist == 45000 + assert workflow_settings.skip_background + assert workflow_settings.smart_stack_params.stack_dz == 50 + assert workflow_settings.smart_stack_params.images_to_save == 1 + assert workflow_settings.smart_stack_params.min_images_to_test == 9 + # Check values from calculating overlap are as expected (from above mock) + assert workflow_settings.dx == 123 + assert workflow_settings.dy == 456 + # And that the input image dir is passed to stack the stack parameter for saving + assert workflow_settings.smart_stack_params.images_dir == "/this/img_dir" + + +# A CSM that is "normal" changing from camera maxtrix coordinates (y,x) to normal +# (x, y) coordinates +CSM_NORMAL = [ + [0.02, 0.25], + [0.25, 0.01], +] +# A CSM that is "normal" changing from camera maxtrix coordinates (y,x) to normal +# (x, y) coordinates, but the output y sign is flipped. +CSM_FLIP_Y = [ + [0.02, -0.25], + [0.25, 0.01], +] +# A CSM that is for a rotated compared to normal camera changing from camera maxtrix +# thus dx (the x motor) controls y and the y motor controls x +CSM_ROTATED = [ + [0.25, 0.02], + [0.01, 0.25], +] + + +@pytest.mark.parametrize( + ("csm_matrix", "overlap", "expected_steps"), + [ + (CSM_NORMAL, 0.5, (100, 75)), + (CSM_NORMAL, 0.25, (150, 112)), + (CSM_FLIP_Y, 0.5, (100, -75)), + (CSM_FLIP_Y, 0.25, (150, -112)), + (CSM_ROTATED, 0.5, (75, 100)), + (CSM_ROTATED, 0.25, (112, 150)), + ], +) +def test_histo_calculate_overlap(csm_matrix, overlap, expected_steps, histo_workflow): + """Check the dx and dy values are as expected for given overlap and matrix. + + Note that the matrices on the dominant axes all have 0.25 magnitude. And the + mock camera is set to be 800px in x and 600px in y. For 0 overlap the movement + should be dx of 200 (due to the 0.25 factor). + """ + + def apply_csm(x: float, y: float, **_kwargs: float) -> dict[str, int]: + """Convert image coordinates to stage coordinates.""" + return csm_img_to_stage(csm_matrix, x=x, y=y) + + histo_workflow._csm.convert_image_to_stage_coordinates.side_effect = apply_csm + + # First check this error if CSM isn't calibrated: + histo_workflow._csm.calibration_required = True + histo_workflow._csm.image_resolution = None + with pytest.raises(RuntimeError, match="CSM not set"): + histo_workflow._calc_displacement_from_overlap(0.25) + + histo_workflow._csm.calibration_required = False + # the img resolution is in matrix coords, so (y, x) not (x, y) + histo_workflow._csm.image_resolution = (600, 800) + dx, dy = histo_workflow._calc_displacement_from_overlap(overlap) + assert (dx, dy) == expected_steps + + +def test_histo_pre_scan_routine(histo_workflow, mocker): + """Check the pre-scan routine does autofocusses.""" + # Rather than create a whole Setting class, just create a mock with the value + # we need set + mock_settings = mocker.Mock() + mock_settings.smart_stack_params.autofocus_dz = 1234 + # Run the function + histo_workflow.pre_scan_routine(mock_settings) + # Check the autofocus was run using the mocked slot. + assert histo_workflow._autofocus.looping_autofocus.call_count == 1 + call_kwargs = histo_workflow._autofocus.looping_autofocus.call_args.kwargs + assert call_kwargs["dz"] == 1234 + assert call_kwargs["start"] == "centre" + + +def test_histo_new_scan_planner(histo_workflow, mocker): + """Check the pre-scan routine does autofocusses.""" + # Rather than create a whole Setting class, just create a mock with the 3 values + # we need set + mock_settings = mocker.Mock() + mock_settings.dx = 666 + mock_settings.dy = 999 + mock_settings.max_dist = 54321 + + planner = histo_workflow.new_scan_planner( + mock_settings, position={"x": 1, "y": 2, "z": 3} + ) + assert isinstance(planner, SmartSpiral) + # Check the settings were read + assert planner._dx == 666 + assert planner._dy == 999 + assert planner._max_dist == 54321 + + +def test_histo_acquisition_with_background_detected(histo_workflow, mocker, caplog): + """If background is detected it should return without a stack.""" + # Mocking for settings as above + mock_settings = mocker.Mock() + mock_settings.skip_background = True + + histo_workflow._background_detector.image_is_sample.return_value = ( + False, + "totally empty", + ) + + with caplog.at_level(logging.INFO): + imaged, focus_height = histo_workflow.acquisition_routine( + mock_settings, xyz_pos=(1, 2, 3) + ) + + # Check returns + assert imaged is False + assert focus_height is None # None meaning not focussed + # Check there is a log explaining what happened + assert len(caplog.records) == 1 + assert caplog.records[0].message == "Skipping (1, 2, 3) as it is totally empty." + assert caplog.records[0].levelname == "INFO" + + # And that smart stack never ran + assert histo_workflow._autofocus.run_smart_stack.call_count == 0 + + +def test_histo_acquisition_not_skipping_background(histo_workflow, mocker, caplog): + """Check when not skipping background an image is always saved.""" + # Mocking for settings as above + mock_settings = mocker.Mock() + mock_settings.skip_background = False + + # Set up background detector to say it is empty, this shouldn't stop stacking. + histo_workflow._background_detector.image_is_sample.return_value = ( + False, + "totally empty", + ) + + with caplog.at_level(logging.INFO): + # First set smart stack to report failure + histo_workflow._autofocus.run_smart_stack.return_value = (False, 123) + imaged, focus_height = histo_workflow.acquisition_routine( + mock_settings, xyz_pos=(1, 2, 3) + ) + # Check return + assert imaged is True # Always images if not skipping background + assert focus_height is None # None meaning didn't focus + + # And again reporting a successful stack + histo_workflow._autofocus.run_smart_stack.return_value = (True, 123) + imaged, focus_height = histo_workflow.acquisition_routine( + mock_settings, xyz_pos=(1, 2, 3) + ) + # Check return + assert imaged is True # Always images if not skipping background + assert focus_height == 123 + + # Should never warn + assert len(caplog.records) == 0 + + # And that smart stack never ran + assert histo_workflow._autofocus.run_smart_stack.call_count == 2 + + +def test_histo_acquisition_on_sample(histo_workflow, mocker, caplog): + """Check when skipping background, but background not detected.""" + # Mocking for settings as above + mock_settings = mocker.Mock() + mock_settings.skip_background = True + + # Set up background detector to say there is sample. + histo_workflow._background_detector.image_is_sample.return_value = (True, None) + + with caplog.at_level(logging.INFO): + # First set smart stack to report failure + histo_workflow._autofocus.run_smart_stack.return_value = (False, 123) + imaged, focus_height = histo_workflow.acquisition_routine( + mock_settings, xyz_pos=(1, 2, 3) + ) + # Check return: + # Don't save failed stack when skipping background, assume it is actually + # background + assert imaged is False + assert focus_height is None # None meaning didn't focus + + # And again reporting a successful stack + histo_workflow._autofocus.run_smart_stack.return_value = (True, 123) + imaged, focus_height = histo_workflow.acquisition_routine( + mock_settings, xyz_pos=(1, 2, 3) + ) + # Check return + assert imaged is True # Always images if not skipping background + assert focus_height == 123 + + # Should never warn + assert len(caplog.records) == 1 + expected_message = "Stack failed at (1, 2, 3). Treating as background." + assert caplog.records[0].message == expected_message + assert caplog.records[0].levelname == "INFO" + + # And that smart stack never ran + assert histo_workflow._autofocus.run_smart_stack.call_count == 2 + + +def test_histo_workflow_settings_ui(histo_workflow): + """Check that the workflow specifies the expected controls.""" + ui = histo_workflow.settings_ui + + assert len(ui) == 7 + for element in ui: + assert isinstance(element, PropertyControl) + + names = [el.property_name for el in ui] + expected_names = [ + "overlap", + "skip_background", + "stack_images_to_save", + "stack_min_images_to_test", + "stack_dz", + "autofocus_dz", + "max_range", + ] + assert names == expected_names diff --git a/tests/unit_tests/test_smart_scan.py b/tests/unit_tests/test_smart_scan.py index 4a59e47e..a41d96d5 100644 --- a/tests/unit_tests/test_smart_scan.py +++ b/tests/unit_tests/test_smart_scan.py @@ -17,20 +17,24 @@ import logging import os import shutil import tempfile +from collections.abc import Iterable +from dataclasses import dataclass from datetime import datetime from typing import Callable, Optional +from unittest import mock import pytest from fastapi import HTTPException +from pydantic import BaseModel from labthings_fastapi.exceptions import InvocationCancelledError from labthings_fastapi.testing import create_thing_without_server -from openflexure_microscope_server.scan_directories import ( - NotEnoughFreeSpaceError, - ScanData, -) +from openflexure_microscope_server.scan_directories import NotEnoughFreeSpaceError +from openflexure_microscope_server.stitching import StitchingSettings +from openflexure_microscope_server.things.scan_workflows import ScanWorkflow from openflexure_microscope_server.things.smart_scan import ( + ActiveScanData, ScanNotRunningError, SmartScanThing, ) @@ -50,10 +54,33 @@ def _clear_scan_dir() -> None: def smart_scan_thing(): """Return a smart scan thing as a fixture.""" return create_thing_without_server( - SmartScanThing, scans_folder=SCAN_DIR, mock_all_slots=True + SmartScanThing, + scans_folder=SCAN_DIR, + default_workflow="mock-_all_workflows", + mock_all_slots=True, ) +def custom_smart_scan_thing(default_workflow, all_workflows): + """Set up a custom smart scan thing with workflows adjusted. + + This allows setting a default workflow and to adjust all workflows from simple + single item mock from `mock_all_slots`. + """ + smart_scan_thing = create_thing_without_server( + SmartScanThing, + scans_folder=SCAN_DIR, + default_workflow=default_workflow, + mock_all_slots=True, + ) + # Pop the existing mock workflow and add specified ones (if any) + smart_scan_thing._all_workflows.pop("mock-_all_workflows") + for key, thing in all_workflows.items(): + smart_scan_thing._all_workflows[key] = thing + + return smart_scan_thing + + def test_initial_properties(smart_scan_thing): """Check the initial values of properties. @@ -64,17 +91,139 @@ def test_initial_properties(smart_scan_thing): assert smart_scan_thing.latest_scan_name is None +@dataclass +class WorkflowSelectorTestCase: + """The information from a capture in a smart_z_stack.""" + + workflows: dict[str, mock.MagicMock] + default_wf: str + expected_wf: Optional[str] + loaded_wf: Optional[str] = None + side_effect: Optional[type | int | Iterable[int]] = None + """The side effect of entering the Thing (None, Logger level number, Error type)""" + match: Optional[str | Iterable[str]] = None + + +SELECTOR_CASES = [ + WorkflowSelectorTestCase( + workflows={}, + default_wf="foo", + expected_wf=None, + side_effect=RuntimeError, + match="Could not set Scan Workflow", + ), + WorkflowSelectorTestCase( + workflows={ + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + }, + default_wf="foo", + expected_wf="foo", + side_effect=None, + ), + WorkflowSelectorTestCase( + workflows={ + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + }, + default_wf="bar", + expected_wf="bar", + side_effect=None, + ), + WorkflowSelectorTestCase( + workflows={ + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + }, + default_wf="wrong", + expected_wf="foo", + side_effect=logging.WARNING, + match="Could not select default key 'wrong'", + ), + WorkflowSelectorTestCase( + workflows={ + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + }, + default_wf="wrong", + loaded_wf="bar", + expected_wf="bar", + side_effect=None, + ), + WorkflowSelectorTestCase( + workflows={ + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + }, + default_wf="wrong", + loaded_wf="wrong", + expected_wf="foo", + side_effect=[logging.WARNING, logging.WARNING], + match=[ + "Could not select 'wrong' from Thing mapping", + "Could not select default key 'wrong'", + ], + ), +] + + +@pytest.mark.parametrize("case", SELECTOR_CASES) +def test_workflow_set_on_enter(case, check_side_effect): + """Check workflow is set on enter.""" + smart_scan_thing = custom_smart_scan_thing(case.default_wf, case.workflows) + with check_side_effect(case.side_effect, match=case.match): + # Load in "loaded" as the sever would + smart_scan_thing._workflow_name = case.loaded_wf + with smart_scan_thing: + assert smart_scan_thing._workflow_name == case.expected_wf + + +def test_setting_workflows(caplog): + """Check that setting workflow works, or warns if incorrect.""" + workflows = { + "foo": mock.MagicMock(spec=ScanWorkflow), + "bar": mock.MagicMock(spec=ScanWorkflow), + } + + smart_scan_thing = custom_smart_scan_thing("foo", workflows) + with caplog.at_level(logging.WARNING), smart_scan_thing: + assert smart_scan_thing._workflow_name == "foo" + assert smart_scan_thing._workflow is workflows["foo"] + # Can't set None, warns doesn't change + smart_scan_thing.workflow_name = None + assert len(caplog.records) == 1 + assert smart_scan_thing._workflow_name == "foo" + assert smart_scan_thing._workflow is workflows["foo"] + # Can't set a different name + smart_scan_thing.workflow_name = "wrong" + assert len(caplog.records) == 2 # another log + assert smart_scan_thing._workflow_name == "foo" + assert smart_scan_thing._workflow is workflows["foo"] + + # can set a valid name + smart_scan_thing.workflow_name = "bar" + assert len(caplog.records) == 2 # No extra logs + assert smart_scan_thing._workflow_name == "bar" + assert smart_scan_thing._workflow is workflows["bar"] + + def test_inaccessible_scan_methods(smart_scan_thing): """Test that method with @_scan_running decorator is inaccessible. The @_scan_running decorator makes these functions inaccessible unless - a scan is running. + a scan is running. Also test properties that raise same error. """ with pytest.raises(ScanNotRunningError): smart_scan_thing._run_scan() with pytest.raises(ScanNotRunningError): smart_scan_thing._manage_stitching_threads() + # Properties + with pytest.raises(ScanNotRunningError): + smart_scan_thing.scan_data + with pytest.raises(ScanNotRunningError): + smart_scan_thing.ongoing_scan + def test_private_delete_scan(smart_scan_thing, caplog): """Test the private _delete_scan method deletes directories or warns if it can't.""" @@ -223,23 +372,30 @@ MOCK_SCAN_DIR = "scans/test_name_0001/images/" MOCK_START_POS = {"x": 123, "y": 456, "z": 789} +class MockWorkflowSettingModel(BaseModel): + """A mock model to check that ActiveScanData can hold arbitrary models.""" + + foo: str = "bar" + bar: str = "foo" + dx: int = 123 + dy: int = 456 + + def _expected_scan_data(): - """Return the expected ScanData object for a SmartScan with default properties.""" + """Return the expected ActiveScanData object for a SmartScan with default properties.""" expected_dict = { "scan_name": MOCK_SCAN_NAME, "starting_position": MOCK_START_POS, - "overlap": 0.45, - "max_dist": 45000, - "dx": 100, - "dy": 100, - "autofocus_dz": 1000, - "autofocus_on": True, - "skip_background": True, - "stitch_automatically": True, - "correlation_resize": 0.5, "save_resolution": (1640, 1232), + "stitch_automatically": True, + "stitching_settings": { + "overlap": 0.45, + "correlation_resize": 0.5, + }, + "workflow": "Mock", + "workflow_settings": MockWorkflowSettingModel(), } - return ScanData(start_time=datetime.now(), **expected_dict) + return ActiveScanData(start_time=datetime.now(), **expected_dict) @pytest.fixture @@ -247,14 +403,14 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker): """Return a scan thing that is mocked so that _collect_scan_data will run.""" # Set the lock so it thinks the scan is running with smart_scan_thing._scan_lock: - mocker.patch.object( - smart_scan_thing, - "_calc_displacement_from_test_image", - return_value=[100, 100], - ) - smart_scan_thing._stage.position = MOCK_START_POS + smart_scan_thing._workflow.all_settings.return_value = ( + MockWorkflowSettingModel(), + StitchingSettings(correlation_resize=0.5, overlap=0.45), + ) + smart_scan_thing._workflow.save_resolution = (1640, 1232) + mock_ongoing_scan = mocker.Mock() mock_ongoing_scan.name = MOCK_SCAN_NAME mock_ongoing_scan.images_dir = MOCK_SCAN_DIR @@ -264,10 +420,10 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker): def test_collect_scan_data(scan_thing_mocked_for_scan_data): - """Run _collect_scan_data, and check the ScanData object has the expected values.""" + """Run _collect_scan_data, and check the ActiveScanData object has the expected values.""" scan_thing = scan_thing_mocked_for_scan_data - data = scan_thing._collect_scan_data() + data = scan_thing._collect_scan_data(scan_thing._workflow) expected_data = _expected_scan_data() time_diff = expected_data.start_time - data.start_time assert abs(time_diff.total_seconds()) < 1 @@ -277,17 +433,17 @@ def test_collect_scan_data(scan_thing_mocked_for_scan_data): def test_save_final_scan_data(scan_thing_mocked_for_scan_data): - """Run _save_final_scan_data, check save is called with final results in ScanData.""" + """Run _save_final_scan_data, check save is called with final results in ActiveScanData.""" scan_thing = scan_thing_mocked_for_scan_data - scan_thing._scan_data = scan_thing._collect_scan_data() + scan_thing._scan_data = scan_thing._collect_scan_data(scan_thing._workflow) scan_thing._scan_data.image_count = 44 scan_thing._save_final_scan_data("Mocked!") # _ongoing_scan is a mock so we can check that save_scan data was called and get # the value scan_thing._ongoing_scan.save_scan_data.assert_called() final_data = scan_thing._ongoing_scan.save_scan_data.call_args[0][0] - assert isinstance(final_data, ScanData) + assert isinstance(final_data, ActiveScanData) assert final_data.scan_result == "Mocked!" assert final_data.image_count == 44 assert final_data.duration.total_seconds() < 1 @@ -322,10 +478,10 @@ def check_run_scan(scan_thing, caplog, expected_exception=None): """ if expected_exception is None: with caplog.at_level(logging.WARNING): - scan_thing._scan_data = scan_thing._run_scan() + scan_thing._scan_data = scan_thing._run_scan(scan_thing._workflow) else: with pytest.raises(expected_exception), caplog.at_level(logging.WARNING): - scan_thing._scan_data = scan_thing._run_scan() + scan_thing._scan_data = scan_thing._run_scan(scan_thing._workflow) # The preview stitcher object should still exist. And images dir should be set. assert scan_thing._preview_stitcher.images_dir == MOCK_SCAN_DIR @@ -341,7 +497,7 @@ def check_run_scan(scan_thing, caplog, expected_exception=None): def test_run_scan(scan_thing_mocked_for_run_scan, caplog): - """Run _save_final_scan_data, check save is called with final results in ScanData.""" + """Run _save_final_scan_data, check save is called with final results in ActiveScanData.""" result, logs, calls = check_run_scan(scan_thing_mocked_for_run_scan, caplog) assert result == "success" diff --git a/tests/unit_tests/test_stack.py b/tests/unit_tests/test_stack.py index 6683c719..7cc4fe7e 100644 --- a/tests/unit_tests/test_stack.py +++ b/tests/unit_tests/test_stack.py @@ -19,12 +19,13 @@ from openflexure_microscope_server.things.autofocus import ( AutofocusThing, CaptureInfo, NotAPeakError, - StackParams, + SmartStackParams, _count_turning_points, _get_capture_by_id, _get_capture_index_by_id, _get_peak_turning_point, ) +from openflexure_microscope_server.things.scan_workflows import HistoScanWorkflow RANDOM_GENERATOR = np.random.default_rng() @@ -64,7 +65,7 @@ def test_stack_params_validation(save_ims, extra_ims): # Coerce min_images_to_test as the max extra ims depends on save_ims so is hard # to do automatically in hypothesis. This clamps the number between 3 and 9. min_images_to_test = max(min(save_ims + extra_ims, 9), 3) - StackParams( + SmartStackParams( stack_dz=50, images_to_save=save_ims, min_images_to_test=min_images_to_test, @@ -91,7 +92,7 @@ def test_stack_params_not_enough_test_images(save_ims, extra_ims): "Can't save more images than the minimum number tested)" ) with pytest.raises(ValueError, match=match): - StackParams( + SmartStackParams( stack_dz=50, images_to_save=save_ims, min_images_to_test=save_ims + extra_ims, @@ -116,7 +117,7 @@ def test_stack_params_negative_images_to_save(save_ims, extra_ims): "Images to save must be positive and odd)" ) with pytest.raises(ValueError, match=match): - StackParams( + SmartStackParams( stack_dz=50, images_to_save=save_ims, min_images_to_test=save_ims + extra_ims, @@ -142,7 +143,7 @@ def test_even_min_images_to_test(save_ims, extra_ims): "Minimum number of images to test should be positive and odd)" ) with pytest.raises(ValueError, match=match): - StackParams( + SmartStackParams( stack_dz=50, images_to_save=save_ims, min_images_to_test=save_ims + extra_ims, @@ -166,7 +167,7 @@ def test_even_images_to_save(save_ims, extra_ims): "Images to save must be positive and odd)" ) with pytest.raises(ValueError, match=match): - StackParams( + SmartStackParams( stack_dz=50, images_to_save=save_ims, min_images_to_test=save_ims + extra_ims, @@ -177,11 +178,11 @@ def test_even_images_to_save(save_ims, extra_ims): def test_computed_stack_params(): - """Test StackParams computed properties are as expected. + """Test SmartStackParams computed properties are as expected. Not using hypothesis or we will just copy in the same formulas. """ - stack_parameters = StackParams( + stack_parameters = SmartStackParams( stack_dz=50, images_to_save=5, min_images_to_test=9, @@ -267,20 +268,28 @@ def autofocus_thing(): return create_thing_without_server(AutofocusThing, mock_all_slots=True) -def test_create_stack(autofocus_thing, caplog): +@pytest.fixture +def histo_scan_workflow(): + """Return an autofocus thing connected to a server.""" + return create_thing_without_server(HistoScanWorkflow, mock_all_slots=True) + + +def test_create_stack(histo_scan_workflow, caplog): """Run create stack with default values and check there is no coercion or logging.""" - initial_min_images_to_test = autofocus_thing.stack_min_images_to_test - initial_images_to_save = autofocus_thing.stack_images_to_save + initial_min_images_to_test = histo_scan_workflow.stack_min_images_to_test + initial_images_to_save = histo_scan_workflow.stack_images_to_save with caplog.at_level(logging.INFO): - stack_params = autofocus_thing.create_stack_params( + stack_params = histo_scan_workflow.create_smart_stack_params( autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) ) assert len(caplog.records) == 0 - assert autofocus_thing.stack_min_images_to_test == initial_min_images_to_test - assert autofocus_thing.stack_images_to_save == initial_images_to_save - assert stack_params.min_images_to_test == autofocus_thing.stack_min_images_to_test - assert stack_params.images_to_save == autofocus_thing.stack_images_to_save + assert histo_scan_workflow.stack_min_images_to_test == initial_min_images_to_test + assert histo_scan_workflow.stack_images_to_save == initial_images_to_save + assert ( + stack_params.min_images_to_test == histo_scan_workflow.stack_min_images_to_test + ) + assert stack_params.images_to_save == histo_scan_workflow.stack_images_to_save @pytest.mark.parametrize( @@ -293,13 +302,13 @@ def test_create_stack(autofocus_thing, caplog): ], ) def test_coercing_stack_test_ims( - initial_test_ims, coerced_test_ims, expected_log_start, autofocus_thing, caplog + initial_test_ims, coerced_test_ims, expected_log_start, histo_scan_workflow, caplog ): """Run create stack with images to test set to values requiring coercion, and check result.""" - autofocus_thing.stack_min_images_to_test = initial_test_ims + histo_scan_workflow.stack_min_images_to_test = initial_test_ims with caplog.at_level(logging.WARNING): - stack_params = autofocus_thing.create_stack_params( + stack_params = histo_scan_workflow.create_smart_stack_params( autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) ) @@ -308,7 +317,9 @@ def test_coercing_stack_test_ims( # Check the value is coerced in the stack_params assert stack_params.min_images_to_test == coerced_test_ims # Check that the setting in the Thing was updated to the coerced value - assert stack_params.min_images_to_test == autofocus_thing.stack_min_images_to_test + assert ( + stack_params.min_images_to_test == histo_scan_workflow.stack_min_images_to_test + ) @pytest.mark.parametrize( @@ -321,13 +332,13 @@ def test_coercing_stack_test_ims( ], ) def test_coercing_stack_save_ims( - initial_save_ims, coerced_save_ims, expected_log_start, autofocus_thing, caplog + initial_save_ims, coerced_save_ims, expected_log_start, histo_scan_workflow, caplog ): """Run create stack with images to save set to values requiring coercion, and check result.""" - autofocus_thing.stack_images_to_save = initial_save_ims + histo_scan_workflow.stack_images_to_save = initial_save_ims with caplog.at_level(logging.WARNING): - stack_params = autofocus_thing.create_stack_params( + stack_params = histo_scan_workflow.create_smart_stack_params( autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) ) @@ -336,13 +347,13 @@ def test_coercing_stack_save_ims( # Check the value is coerced in the stack_params assert stack_params.images_to_save == coerced_save_ims # Check that the setting in the Thing was updated to the coerced value - assert stack_params.images_to_save == autofocus_thing.stack_images_to_save + assert stack_params.images_to_save == histo_scan_workflow.stack_images_to_save @pytest.mark.parametrize("pass_on", [1, 2, 3, 4]) -def test_run_smart_stack(pass_on, autofocus_thing, mocker): +def test_run_smart_stack(pass_on, histo_scan_workflow, autofocus_thing, mocker): """Test Running smart stack with the stack passing on different attempts.""" - stack_params = autofocus_thing.create_stack_params( + stack_params = histo_scan_workflow.create_smart_stack_params( autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) ) assert stack_params.max_attempts == 3 @@ -364,8 +375,8 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker): failed_return = (False, fake_captures, "pick_me") return_list = [failed_return] * (pass_on - 1) + [successful_return] - # Mock z_stack and looping_autofocus - autofocus_thing.z_stack = mocker.Mock(side_effect=return_list) + # Mock smart_z_stack and looping_autofocus + autofocus_thing.smart_z_stack = mocker.Mock(side_effect=return_list) autofocus_thing.looping_autofocus = mocker.Mock() # Run it @@ -380,9 +391,10 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker): # Final z is the one from the id returned by the stack "pick_me" assert final_z == 555 - # z_stack should run up until the time it passes. Running no more than max_attempts + # smart_z_stack should run up until the time it passes. Running no more than + # max_attempts n_stacks = min(pass_on, stack_params.max_attempts) - assert autofocus_thing.z_stack.call_count == n_stacks + assert autofocus_thing.smart_z_stack.call_count == n_stacks # Move absolute should be 1 less time that the number of times z_stack_run assert autofocus_thing._stage.move_absolute.call_count == n_stacks - 1 # As should looping autofocus @@ -396,14 +408,16 @@ def test_run_smart_stack(pass_on, autofocus_thing, mocker): assert autofocus_thing._cam.save_from_memory.call_count == (1 if success else 0) -def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, mocker): - """Set up a z_stack, run it, and return the result. +def setup_and_run_smart_z_stack( + check_returns, check_turning_points, histo_scan_workflow, autofocus_thing, mocker +): + """Set up a smart_z_stack, run it, and return the result. :param check_returns: The return values from check_stack_result. Note that if this is a list, it will be set as a side effect (and should be a list of tuples of results). If it a tuple (or anything else), it is set as a return value. """ - stack_params = autofocus_thing.create_stack_params( + stack_params = histo_scan_workflow.create_smart_stack_params( autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232) ) stack_params.settling_time = 0 # Don't settle or tests take forever. @@ -413,58 +427,70 @@ def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns) else: autofocus_thing.check_stack_result = mocker.Mock(return_value=check_returns) - return autofocus_thing.z_stack( + return autofocus_thing.smart_z_stack( stack_parameters=stack_params, check_turning_points=check_turning_points, ) -def test_z_stack_turning_toggle_passed(autofocus_thing, mocker): +def test_z_stack_turning_toggle_passed(histo_scan_workflow, autofocus_thing, mocker): """Check that the toggling of turning points is passed to the check.""" check_returns = ("success", "mock_id") for check_turning in [True, False]: - setup_and_run_z_stack(check_returns, check_turning, autofocus_thing, mocker) + setup_and_run_smart_z_stack( + check_returns, check_turning, histo_scan_workflow, autofocus_thing, mocker + ) check_kwargs = autofocus_thing.check_stack_result.call_args.kwargs assert check_kwargs["check_turning_points"] == check_turning -def test_z_stack_returns_on_success_and_restart(autofocus_thing, mocker): +def test_z_stack_returns_on_success_and_restart( + histo_scan_workflow, autofocus_thing, mocker +): """Check that if the check returns success or restart then the stack exits with correct return value.""" for result in ["success", "restart"]: check_returns = (result, "mock_id") - ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) + ret = setup_and_run_smart_z_stack( + check_returns, True, histo_scan_workflow, autofocus_thing, mocker + ) assert autofocus_thing.check_stack_result.call_count == 1 # Check the number of images taken is exactly the call count. ims_taken = autofocus_thing.capture_stack_image.call_count - assert ims_taken == autofocus_thing.stack_min_images_to_test + assert ims_taken == histo_scan_workflow.stack_min_images_to_test # And the result is as expected. assert ret[0] == (result == "success") -def test_z_stack_exits_if_focus_never_found(autofocus_thing, mocker): +def test_z_stack_exits_if_focus_never_found( + histo_scan_workflow, autofocus_thing, mocker +): """Check that if the check returns continue the stack exits eventually with a failure.""" check_returns = ("continue", "mock_id") - ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) + ret = setup_and_run_smart_z_stack( + check_returns, True, histo_scan_workflow, autofocus_thing, mocker + ) assert autofocus_thing.check_stack_result.call_count == EXTRA_STACK_CAPTURES + 1 # Check the number of images taken is the maximum possible, set by the min images to # test and the number of extra images that can be taken ims_taken = autofocus_thing.capture_stack_image.call_count - max_ims = autofocus_thing.stack_min_images_to_test + EXTRA_STACK_CAPTURES + max_ims = histo_scan_workflow.stack_min_images_to_test + EXTRA_STACK_CAPTURES assert ims_taken == max_ims # And the result is as expected. assert not ret[0] -def test_z_stack_return(autofocus_thing, mocker): +def test_z_stack_return(histo_scan_workflow, autofocus_thing, mocker): """Check z-stack returns as expected for more complex cases the fixed results above.""" for i in range(2, EXTRA_STACK_CAPTURES): check_returns = [ ("restart" if j == i - 1 else "continue", f"id_{j}") for j in range(i) ] - ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) + ret = setup_and_run_smart_z_stack( + check_returns, True, histo_scan_workflow, autofocus_thing, mocker + ) # Calculate images taken - images_taken = autofocus_thing.stack_min_images_to_test + i - 1 + images_taken = histo_scan_workflow.stack_min_images_to_test + i - 1 assert autofocus_thing.capture_stack_image.call_count == images_taken # Check it reports a failure assert not ret[0] @@ -473,7 +499,9 @@ def test_z_stack_return(autofocus_thing, mocker): check_returns = [ ("success" if j == i - 1 else "continue", f"id_{j}") for j in range(i) ] - ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker) + ret = setup_and_run_smart_z_stack( + check_returns, True, histo_scan_workflow, autofocus_thing, mocker + ) # Calculate images taken assert autofocus_thing.capture_stack_image.call_count == images_taken # Check it reports a success diff --git a/tests/unit_tests/test_stitching.py b/tests/unit_tests/test_stitching.py index 12bcfbfb..349d975c 100644 --- a/tests/unit_tests/test_stitching.py +++ b/tests/unit_tests/test_stitching.py @@ -11,15 +11,16 @@ import time from copy import copy import pytest +from pydantic import BaseModel import labthings_fastapi as lt from openflexure_microscope_server.stitching import ( - STITCHING_RESOLUTION, BaseStitcher, FinalStitcher, PreviewStitcher, StitcherValidationError, + StitchingSettings, ) from ..shared_utils.lt_test_utils import LabThingsTestEnv @@ -88,87 +89,41 @@ FINAL_EXPECTED_COMMAND = [ FAKE_DIR, ] - -def test_final_stitcher_command_defaults(caplog): - """Check the FinalStitcher stitches with expected default values. - - It should warn when default values are used as they are a fallback. - """ - n_logs = 0 - # Test with no dictionary data and with irrelevant dictionary data. - with caplog.at_level(logging.WARNING): - for data_dict in [None, {"irrelevant": "data"}]: - stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER, scan_data_dict=data_dict) - # Should log for overlap being None and correlation_resize being None - n_logs += 2 - assert len(caplog.records) == n_logs - assert stitcher.command == FINAL_EXPECTED_COMMAND +DEFAULT_SETTINGS = StitchingSettings(correlation_resize=0.5, overlap=0.1) -def test_final_stitcher_command_tiff(caplog): +def test_final_stitcher_command_tiff(): """Check that the tiff can be requested.""" # Modify defaults expected_command = copy(FINAL_EXPECTED_COMMAND) expected_command[4] = "--stitch_tiff" - stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER, stitch_tiff=True) - # Should log for overlap being None and correlation_resize being None - assert len(caplog.records) == 2 + stitcher = FinalStitcher( + FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS, stitch_tiff=True + ) assert stitcher.command == expected_command -def test_final_stitcher_command_set_val_directly(): - """Check that values are set as expected when directly input.""" - # Modify defaults - expected_command = copy(FINAL_EXPECTED_COMMAND) - expected_command[8] = "0.36" - expected_command[10] = "0.25" - # Test with no data dictionary, irrelevant data, and also the wrong data - # When wrong data is submitted, it should take the directly input data. - dict_vals = [ - None, - {"irrelevant": "data"}, - {"overlap": 0.2, "save_resolution": [5, 5]}, - ] - for data_dict in dict_vals: - stitcher = FinalStitcher( - FAKE_DIR, - logger=LOGGER, - overlap=0.4, - correlation_resize=0.25, - scan_data_dict=data_dict, - ) - assert stitcher.command == expected_command - - -def test_final_stitcher_command_set_with_dict(): +def test_final_stitcher_command_with_settings(): """Check that values are set as expected when set from a ScanData dictionary.""" # Modify defaults expected_command = copy(FINAL_EXPECTED_COMMAND) expected_command[8] = "0.36" expected_command[10] = "0.25" - # Check same thing works with a dictionary, resize is calculated from the saved image - # resolution. Make 4x bigger than STITCHING_RESOLUTION to get 0.25 - resolution = [dim * 4 for dim in STITCHING_RESOLUTION] - # Check legacy key as well as current one: - for resolution_key in ["save_resolution", "capture resolution"]: - stitcher = FinalStitcher( - FAKE_DIR, - logger=LOGGER, - scan_data_dict={"overlap": 0.4, resolution_key: resolution}, - ) - assert stitcher.command == expected_command + + stitcher = FinalStitcher( + FAKE_DIR, + logger=LOGGER, + stitching_settings=StitchingSettings(correlation_resize=0.25, overlap=0.4), + ) + assert stitcher.command == expected_command def _validation_error_tester(scan_path, **kwargs): - """Check each type of stitcher throws a validation error for the given init args.""" - # If scan_data_dict is in the kwargs only test the scan_data_dict - if "scan_data_dict" not in kwargs: - with pytest.raises(StitcherValidationError): - BaseStitcher(scan_path, **kwargs).command - with pytest.raises(StitcherValidationError): - PreviewStitcher(scan_path, **kwargs).command + """Check stitcher throws a validation error for the given init args.""" with pytest.raises(StitcherValidationError): - FinalStitcher(scan_path, logger=LOGGER, **kwargs).command + BaseStitcher(scan_path, **kwargs).command + with pytest.raises(StitcherValidationError): + PreviewStitcher(scan_path, **kwargs).command def test_validation_error(): @@ -176,10 +131,23 @@ def test_validation_error(): The stitcher should throw a validation error each attempt. """ + # Tests for preview (and base) stitcher _validation_error_tester("/dir;rm -rf /;", overlap=".2", correlation_resize=".25") _validation_error_tester(FAKE_DIR, overlap=".2", correlation_resize=".25;rm -rf /;") _validation_error_tester(FAKE_DIR, overlap=".2;rm -rf /;", correlation_resize=".25") - _validation_error_tester(FAKE_DIR, scan_data_dict={"overlap": ".2;rm -rf /;"}) + + class EvilModel(BaseModel): + overlap: str + correlation_resize: str + + with pytest.raises(StitcherValidationError): + FinalStitcher( + FAKE_DIR, + logger=LOGGER, + stitching_settings=EvilModel( + overlap=".2;rm -rf /;", correlation_resize=".25" + ), + ) def test_extra_arg_validation(): @@ -188,7 +156,9 @@ def test_extra_arg_validation(): Currently extra args do not come from user input. But this makes checks more future-proof. """ - stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER) + stitcher = FinalStitcher( + FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS + ) stitcher._extra_args = ["&&rm -rf /&&"] with pytest.raises(StitcherValidationError): stitcher.command @@ -233,7 +203,9 @@ class StitchingTestThing(lt.Thing): @lt.action def run_final(self): """Run the final stitcher.""" - stitcher = FinalStitcher(FAKE_DIR, logger=self.logger) + stitcher = FinalStitcher( + FAKE_DIR, logger=self.logger, stitching_settings=DEFAULT_SETTINGS + ) # Send in the argument HANG to mock-stitch and it just hang for 10s stitcher._extra_args = ["HANG"] stitcher.run() @@ -277,9 +249,8 @@ def test_final_stitching_command(caplog, mocker): mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd) with caplog.at_level(logging.INFO): - # Input values to prevent logging stitcher = FinalStitcher( - FAKE_DIR, logger=LOGGER, overlap=0.1, correlation_resize=0.5 + FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS ) # For the final stitcher it will always complete before returning. stitcher.run() @@ -315,16 +286,17 @@ def test_final_stitching_command_cancelled(stitching_test_env, mocker): assert re.match(r"^Invocation [0-9a-f-]+ was cancelled", logs[-1]["message"]) -def test_final_stitching_command_error(caplog, mocker): +def test_final_stitching_command_error(mocker): """Check that ChildProcessError is raised if the final stitch errors.""" mock_cmd = f"python {MOCK_STITCHER}" mocker.patch("openflexure_microscope_server.stitching.STITCHING_CMD", mock_cmd) - with caplog.at_level(logging.INFO): - stitcher = FinalStitcher(FAKE_DIR, logger=LOGGER) - # Send in the argument ERROR to mock-stitch and it will raise an error rather - # than echo. - stitcher._extra_args = ["ERROR"] - with pytest.raises(ChildProcessError): - stitcher.run() + stitcher = FinalStitcher( + FAKE_DIR, logger=LOGGER, stitching_settings=DEFAULT_SETTINGS + ) + # Send in the argument ERROR to mock-stitch and it will raise an error rather + # than echo. + stitcher._extra_args = ["ERROR"] + with pytest.raises(ChildProcessError): + stitcher.run() diff --git a/webapp/src/components/tabContentComponents/slideScanContent.vue b/webapp/src/components/tabContentComponents/slideScanContent.vue index 0fdf6e1f..eeef6d86 100644 --- a/webapp/src/components/tabContentComponents/slideScanContent.vue +++ b/webapp/src/components/tabContentComponents/slideScanContent.vue @@ -1,51 +1,32 @@