rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'v3' into 'goal_1'

Browse source 
# Conflicts:
#   webapp/src/components/tabContentComponents/slideScanContent.vue
This commit is contained in:
Antonio Anaya 2026-02-12 11:34:32 +00:00
commit 66b0693338
19 changed files with 2142 additions and 789 deletions
Download patch file Download diff file Expand all files Collapse all files

5
ofm_config_full.json
View file

@ -13,9 +13,12 @@
"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",
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
"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"

5
ofm_config_simulation.json
View file

@ -8,9 +8,12 @@
"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",
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"
},

BIN
picamera_coverage.zip
View file

Binary file not shown.

147
src/openflexure_microscope_server/scan_directories.py
View file

@ -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(

187
src/openflexure_microscope_server/scan_planners.py
View file

@ -11,7 +11,7 @@ from __future__ import annotations
import logging
from copy import copy
from typing import Any, Optional, TypeAlias
from typing import Any, Literal, Optional, TypeAlias
import numpy as np
@ -252,41 +252,15 @@ class ScanPlanner:
next_location = self._remaining_locations[0].xy_tuple
# If focussed locations exist return closest location, favouring most recent
closest_pos = self.closest_focus_site(next_location)
# Each scanner defines its own method of choosing a representative nearby site
closest_pos = self.select_nearby_focus_site(next_location)
z = None if closest_pos is None else closest_pos[2]
return next_location, z
def closest_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]:
"""Return the xyz position of the closest site where focus was achieved.
The most recently taken image is returned in the case of a tie.
:param xy_pos: The xy_position which the returned position should be closest
to.
Returns None if there if no focussed locations are present
"""
# save to variable rather than search for focussed sites each time.
focused_locations = self.focused_locations
if not focused_locations:
return None
# must be float64 (double precision) to deal with the huge numbers involved!
current_pos = np.array(xy_pos, dtype="float64")
path_pos = np.array(focused_locations, dtype="float64")[:, :2]
# Use linalg.norm to calculate the direct distance bweween the points
# Note linalg.norm always uses float64
dists = np.linalg.norm((path_pos - current_pos), axis=1)
# Get indices of all minima.
# Note np.where always returns a tuple of arrays, hence the trailing [0]
indices = np.where(dists == np.min(dists))[0]
# The last index is most recent
return focused_locations[indices[-1]]
def select_nearby_focus_site(self, next_location: XYPos) -> Optional[XYZPos]:
"""Return the focused site near xy_pos according to the tiebreak."""
raise NotImplementedError("Did you call the ScanPlanner base class?")
def mark_location_visited(
self, xyz_pos: XYZPos, imaged: bool, focused: bool
@ -316,6 +290,19 @@ class ScanPlanner:
)
)
def _grid_to_future_locations(
self,
grid: list[list[XYPos]],
) -> list[FutureScanLocation]:
"""Flatten a 2D grid of coordinates into flat list of FutureScanLocation objects.
:param grid: A 2D nested list of XY coordinates
:return: A flattened list of FutureScanLocations
"""
# Loop over each location in each line to flatten grid into single list.
return [FutureScanLocation(location) for line in grid for location in line]
class SmartSpiral(ScanPlanner):
"""A scan planner that spirals outward from the centre, prioritising short moves.
@ -385,7 +372,7 @@ class SmartSpiral(ScanPlanner):
def _initial_location_list(self) -> list[FutureScanLocation]:
"""Set the initial list of locations for this scan planner.
This is salled on initialisation.
This is called on initialisation.
For smart spiral this is just the first point
"""
@ -514,26 +501,6 @@ class SmartSpiral(ScanPlanner):
self._remaining_locations.sort(key=sort_key)
def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]:
"""Return the next location to scan and its estimated z-position.
This overrides the default behaviour of ScanPlanner to take the lowest value of
nearest neighbours as this works best for smart stack.
Note z-position may be None! This indicates that the current z position
should be used.
"""
if self.scan_complete:
raise RuntimeError("Can't get next position, scan is complete")
next_location = self._remaining_locations[0].xy_tuple
# If focused locations exist, return the neighbour with the lowest z position
closest_pos = self.select_nearby_focus_site(next_location)
z = None if closest_pos is None else closest_pos[2]
return next_location, z
def select_nearby_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]:
"""Return the xyz position of the nearby site with the lowest z position.
@ -575,7 +542,11 @@ class SmartSpiral(ScanPlanner):
# Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best
# if started too low, so the lowest z will perform best
chosen_focused_site = min(focused_locations_array[indices], key=lambda x: x[-1])
candidates = focused_locations_array[indices]
min_z = np.min(candidates[:, -1])
# Find all with the minimum z, and select the latest
chosen_focused_site = candidates[candidates[:, -1] == min_z][-1]
# Convert back into list so values are of type int instead of np.int32
return tuple(chosen_focused_site.tolist())
@ -605,6 +576,77 @@ class SmartSpiral(ScanPlanner):
return np.max(np.abs(displacement_in_moves))
class SnakeScan(ScanPlanner):
"""A scan planner that performs a snake scan, right and down from a corner.
This planner starts at the corner of the region to scan, snaking back and forth,
starting moving right and down (assuming positive dx and dy.)
"""
_dx: int = 0
_dy: int = 0
_x_count: int = 0
_y_count: int = 0
def __init__(
self, initial_position: XYPos, planner_settings: Optional[dict] = None
) -> None:
"""Set up the lists inherited from ScanPlanner, plus a distance cutoff.
Use the supplied _dx and _dy to set a distance cutoff for an image to be
considered neighbouring another
"""
super().__init__(initial_position, planner_settings)
self._distance_cutoff: float = max([self._dx, self._dy]) * 1.1
def _parse(self, planner_settings: Optional[dict] = None) -> None:
"""Parse SnakeScan Settings dictionary.
* ``dx`` - the movement size in x
* ``dy`` - the movement size in y
* ``x_count`` - The number of columns in the scan.
* ``y_count`` - The number of rows in the scan.
"""
expected_keys = ["x_count", "y_count", "dx", "dy"]
invalid_msg = "SnakeScan requires a planner_settings dictionary with keys: "
if not planner_settings:
raise ValueError(invalid_msg + ",".join(expected_keys))
if not all(keys in planner_settings for keys in expected_keys):
raise KeyError(invalid_msg + ",".join(expected_keys))
self._dx = int(planner_settings["dx"])
self._dy = int(planner_settings["dy"])
self._x_count = int(planner_settings["x_count"])
self._y_count = int(planner_settings["y_count"])
def _initial_location_list(self) -> list[FutureScanLocation]:
"""Set the initial list of locations for this scan planner.
This is called on initialisation.
For snake scan, this is the full grid, and none will be added during scanning.
"""
grid = create_rectangular_scan_path(
starting_pos=self._initial_position,
x_count=self._x_count,
y_count=self._y_count,
dx=self._dx,
dy=self._dy,
style="snake",
)
return self._grid_to_future_locations(grid)
# The noqa statement is because next_position is unused but is needed for equivalence
# with other workflows that require the next pos to select a neighbour.
def select_nearby_focus_site(self, next_position: XYPos) -> Optional[XYZPos]: # noqa: ARG002
"""For a snake scan, use the most recent focused site to predict focus."""
focused_locations = self.focused_locations
if not focused_locations:
return None
return focused_locations[-1]
def distance_between(
current_pos: XYPos | np.ndarray | FutureScanLocation,
next_pos: XYPos | np.ndarray | FutureScanLocation,
@ -620,3 +662,40 @@ def distance_between(
next_pos = np.array(next_pos, dtype="float64")
current_pos = np.array(current_pos, dtype="float64")
return float(np.linalg.norm(next_pos - current_pos))
def create_rectangular_scan_path(
starting_pos: XYPos,
x_count: int,
y_count: int,
dx: int,
dy: int,
style: Literal["snake", "raster"],
) -> list[list[XYPos]]:
"""Generate a 2D grid of (x, y) coordinates representing a rectangular scan path.
The grid is generated from starting_pos, and expanded in the
positive x and y directions using the provided step sizes. The scan order
can be either raster (left-to-right for every row) or snake (alternating
left-to-right and right-to-left per row).
:param starting_pos: Starting (x, y) position for the scan grid.
:param x_count: Number of points in the x-direction (columns).
:param y_count: Number of points in the y-direction (rows).
:param dx: Step size between points in the x-direction.
:param dy: Step size between points in the y-direction.
:param style: Scan pattern style. Either raster or snake.
:return: Nested list of (x, y) coordinates arranged by row.
"""
coords: list[list[XYPos]] = []
# Populate grid with coordinates in a regular grid
for y_index in range(y_count): # rows
row = [
(starting_pos[0] + x_index * dx, starting_pos[1] + y_index * dy)
for x_index in range(x_count)
]
if style == "snake" and y_index % 2 == 1:
row.reverse()
coords.append(row)
return coords

91
src/openflexure_microscope_server/stitching.py
View file

@ -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.

128
src/openflexure_microscope_server/things/autofocus.py
View file

@ -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)

1
src/openflexure_microscope_server/things/camera/__init__.py
View file

@ -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

671
src/openflexure_microscope_server/things/scan_workflows.py Normal file
View file

@ -0,0 +1,671 @@
"""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
import os
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,
SnakeScan,
)
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.things.stage import BaseStage
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."""
# CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None
_csm: Optional[CameraStageMapper] = lt.thing_slot()
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."
)
def _require_csm(self) -> CameraStageMapper:
"""Give each model the option to require CSM. Return it if present."""
if self._csm is None:
raise RuntimeError(
"CameraStageMapping not set, and is required for this workflow."
)
return self._csm
def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]:
"""Use camera stage mapping to calculate x and y displacement from given overlap.
:param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means
that each image should overlap its nearest neighbour by 50%.
:returns: (dx, dy) - the x and y displacements in steps
:raises RuntimeError: If there is no camera stage mapper Thing available or if CMS isn't calibrated.
"""
csm = self._require_csm()
csm_image_res = csm.image_resolution
if csm_image_res is None:
raise RuntimeError("CSM not set. Scan shouldn't have progresses this far.")
# Calculate displacements in image coordinates
dx_img = csm_image_res[1] * (1 - overlap)
dy_img = csm_image_res[0] * (1 - overlap)
x_move_stage = csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
# Assume no rotation or skew and take only the aligned axis of vector.
# Coerce to positive integer, but correct if x and y are flipped
if abs(x_move_stage["x"]) > abs(x_move_stage["y"]):
return x_move_stage["x"], y_move_stage["y"]
# If not use the other stage axes. Note "dx" will be the movement in camera y.
return y_move_stage["x"], x_move_stage["y"]
class HistoScanSettingsModel(BaseModel):
"""The settings for a scan with the HistoScanWorkflow.
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 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)"),
]
class SnakeSettingsModel(BaseModel):
"""The settings for a scan with the SnakeWorkflow.
This includes settings calculated when starting. This will be held by smart scan
during a scan and serialised to disk.
"""
overlap: float
dx: int
dy: int
x_count: int
y_count: int
images_dir: str
autofocus_dz: int
save_resolution: tuple[int, int]
class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and snakes
around them from the top left (assuming positive dx and dy).
"""
display_name: str = lt.property(default="Snake Scan", readonly=True)
ui_blurb: str = lt.property(
default=(
"This scan workflow is optimised for scanning over a rectangle. It "
"snakes down and right from the starting point, over a defined grid."
),
readonly=True,
)
_settings_model = SnakeSettingsModel
_planner_cls: type[ScanPlanner] = SnakeScan
# Thing Slots
_background_detector: ChannelDeviationLUV = lt.thing_slot()
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
# Scan settings
autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000)
"""The z distance to perform an autofocus in steps.
Must be greater than or equal to 200, and less than or equal to 2000.
"""
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x or y.
This must be between 0.1 and 0.7.
"""
x_count: int = lt.setting(default=3)
y_count: int = lt.setting(default=2)
# The noqa statement is because scan_name is unused but is needed for equivalence
# with other workflows that may want to validate the scan name.
def check_before_start(self, scan_name: str) -> None: # noqa: ARG002
"""Before starting a scan, check that camera-stage-mapping is set.
Raise error if:
- camera stage mapping is not set
"""
if self._csm.calibration_required:
raise RuntimeError("Camera Stage Mapping is not calibrated.")
@lt.property
def ready(self) -> bool:
"""Whether this scanworkflow is ready to start."""
return not self._csm.calibration_required
def all_settings(
self, images_dir: str
) -> tuple[SnakeSettingsModel, StitchingSettings]:
"""Return the workflow and stitching settings.
:param images_dir: The directory that images are to be written to.
:return: A tuple containing the settings model for this workflow and the
settings model for stitching.
"""
stitching_settings = StitchingSettings(
overlap=self.overlap,
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
dx, dy = self._calc_displacement_from_overlap(self.overlap)
self.logger.info(
f"Based on an overlap of {self.overlap}, the stage will make steps of "
f"{dx}, {dy}"
)
scan_settings = SnakeSettingsModel(
overlap=self.overlap,
dx=dx,
dy=dy,
x_count=self.x_count,
y_count=self.y_count,
images_dir=images_dir,
autofocus_dz=self.autofocus_dz,
save_resolution=self.save_resolution,
)
return scan_settings, stitching_settings
def pre_scan_routine(self, settings: SnakeSettingsModel) -> None:
"""Autofocus before starting the scan.
:param settings: The settings for this scan as a SnakeSettingsModel
"""
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
def new_scan_planner(
self, settings: SnakeSettingsModel, position: Mapping[str, int]
) -> ScanPlanner:
"""Return a new scan planner object.
:param settings: The settings for this scan as a SnakeSettingsModel
:param position: The starting position as a mapping of axes names to int.
"""
# The initial plan for the scan should be a single x,y position. All future
# moves will be planned around this point. In future, route planner could
# have multiple starting positions, each of which will be visited before the
# scan can end.
planner_settings = {
"dx": settings.dx,
"dy": settings.dy,
"x_count": settings.x_count,
"y_count": settings.y_count,
}
return self._planner_cls(
initial_position=(position["x"], position["y"]),
planner_settings=planner_settings,
)
def acquisition_routine(
self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Perform acquisition routine. This is run at each scan location.
:param settings: The settings for this scan as a SnakeSettingsModel
:param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus.
If failed to find focus, returns for the focus z-position.
"""
self._autofocus.fast_autofocus(dz=settings.autofocus_dz)
focus_height = self._stage.get_xyz_position()[2]
filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg"
self._cam.capture_and_save(
jpeg_path=os.path.join(settings.images_dir, filename),
save_resolution=settings.save_resolution,
)
imaged = True
return imaged, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
return [
property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"),
property_control_for(self, "x_count", label="Number of columns"),
property_control_for(self, "y_count", label="Number of rows"),
property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"),
]

409
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -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()

63
tests/conftest.py Normal file
View file

@ -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

100
tests/unit_tests/test_scan_data.py
View file

@ -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)

53
tests/unit_tests/test_scan_directories.py
View file

@ -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():

127
tests/unit_tests/test_scan_planners.py
View file

@ -143,6 +143,8 @@ def test_smart_spiral_first_few_pos():
# if we mark this position as visited, imaged, and focused
planner.mark_location_visited(xyz_pos1, imaged=True, focused=True)
# current visited path is [[100, 50, 10]]
# scan is not complete
assert not planner.scan_complete
@ -172,6 +174,8 @@ def test_smart_spiral_first_few_pos():
# if we mark this position as visited, imaged, and NOT focused
planner.mark_location_visited(xyz_pos2, imaged=True, focused=False)
# current visited path is [[100, 50, 10], [50, 50, 10]]
# Check this position remove from planned
assert xy_pos2 not in planner.remaining_locations
# Check original position not re-added
@ -195,19 +199,22 @@ def test_smart_spiral_first_few_pos():
assert z_pos3 is z_focus
# Check that the closest focus site to pos3 is pos 1 as
# pos 2 is not focussed
assert planner.closest_focus_site(xy_pos3) == xyz_pos1
assert planner.select_nearby_focus_site(xy_pos3) == xyz_pos1
new_z_focus = 20
xyz_pos3 = (xy_pos3[0], xy_pos3[1], new_z_focus)
# Finally check that if this is focused...
planner.mark_location_visited(xyz_pos3, imaged=True, focused=True)
# current visited path is [[100, 50, 10], [50, 50, 10], [50, 0, 20]]
# ... then the new 4th point ...
xy_pos4, z_pos4 = planner.get_next_location_and_z_estimate()
# ...(100, 0)...
assert xy_pos4 == (100, 0)
# ... and it should get its focus from the lowest neighbouring point
# lowest neighbour to [100, 0] is [100, 50, 10]
assert z_pos4 is z_focus
assert planner.closest_focus_site(xy_pos4) == xyz_pos3
assert planner.select_nearby_focus_site(xy_pos4) == xyz_pos1
def test_smart_spiral_stops_on_max_dist():
@ -265,20 +272,20 @@ def test_closest_focus_with_large_numbers():
scan_planners.VisitedScanLocation((1000000, 0, 0), imaged=True, focused=True),
scan_planners.VisitedScanLocation((0, 1000000, 0), imaged=True, focused=True),
]
assert planner.closest_focus_site((0, 0)) == (0, 1000000, 0)
assert planner.select_nearby_focus_site((0, 0)) == (0, 1000000, 0)
# Try similar
planner._path_history = [
scan_planners.VisitedScanLocation((1234567, 0, 0), imaged=True, focused=True),
scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True),
]
assert planner.closest_focus_site((0, 0)) == (-1234567, 0, 0)
assert planner.select_nearby_focus_site((0, 0)) == (-1234567, 0, 0)
# Make the first point 1 step closer
planner._path_history = [
scan_planners.VisitedScanLocation((1234566, 0, 0), imaged=True, focused=True),
scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True),
]
assert planner.closest_focus_site((0, 0)) == (1234566, 0, 0)
assert planner.select_nearby_focus_site((0, 0)) == (1234566, 0, 0)
def test_example_smart_spiral():
@ -308,3 +315,113 @@ def test_example_smart_spiral():
assert planner.path_history == expected_planner.path_history
assert planner.imaged_locations == expected_planner.imaged_locations
def test_snake_scan_basic_grid():
"""Check that SnakeScan generates a single point for a 1x1 scan."""
initial_position = (100, 50)
planner_settings = {"dx": 100, "dy": 100, "x_count": 1, "y_count": 1}
planner = scan_planners.SnakeScan(
initial_position=initial_position,
planner_settings=planner_settings,
)
assert not planner.scan_complete
# When we start it should want to stay in the initial pos and have
# no z_estimate
xy_pos, z_pos = planner.get_next_location_and_z_estimate()
assert xy_pos == initial_position
assert z_pos is None
# Try to mark location as imaged with only xy_position
with pytest.raises(ValueError, match="3 value tuple expected"):
planner.mark_location_visited(xy_pos, imaged=False, focused=False)
# scan still not complete
assert not planner.scan_complete
# if we mark this position as visited but not imaged
planner.mark_location_visited(
(xy_pos[0], xy_pos[1], 10), imaged=False, focused=False
)
# scan is now complete
assert planner.scan_complete
# if scan is complete, asking for the next location returns an error
with pytest.raises(RuntimeError):
planner.get_next_location_and_z_estimate()
def test_snake_scan_basic_length():
"""SnakeScan should generate the correct number of locations."""
initial_position = (100, 50)
planner_settings = {"dx": 100, "dy": 100, "x_count": 3, "y_count": 4}
planner = scan_planners.SnakeScan(
initial_position=initial_position,
planner_settings=planner_settings,
)
coords = planner.remaining_locations
assert len(coords) == 3 * 4
def test_snake_scan_ordering():
"""Test that snake scan returns a path in the right order."""
initial_position = (0, 0)
planner_settings = {"dx": 10, "dy": 10, "x_count": 4, "y_count": 3}
planner = scan_planners.SnakeScan(
initial_position=initial_position,
planner_settings=planner_settings,
)
coords = planner.remaining_locations
expected = [
(0, 0),
(10, 0),
(20, 0),
(30, 0),
(30, 10),
(20, 10),
(10, 10),
(0, 10),
(0, 20),
(10, 20),
(20, 20),
(30, 20),
]
assert coords == expected
def test_snake_scan_single_row():
"""Test edge case of a single row scan."""
initial_position = (0, 0)
planner_settings = {"dx": 5, "dy": 5, "x_count": 4, "y_count": 1}
planner = scan_planners.SnakeScan(
initial_position=initial_position,
planner_settings=planner_settings,
)
assert planner.remaining_locations == [(0, 0), (5, 0), (10, 0), (15, 0)]
def test_snake_scan_single_column():
"""Test edge case of a single column scan."""
initial_position = (0, 0)
planner_settings = {"dx": 5, "dy": 5, "x_count": 1, "y_count": 4}
planner = scan_planners.SnakeScan(
initial_position=initial_position,
planner_settings=planner_settings,
)
assert planner.remaining_locations == [
(0, 0),
(0, 5),
(0, 10),
(0, 15),
]

352
tests/unit_tests/test_scan_workflows.py Normal file
View file

@ -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

218
tests/unit_tests/test_smart_scan.py
View file

@ -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"

120
tests/unit_tests/test_stack.py
View file

@ -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

126
tests/unit_tests/test_stitching.py
View file

@ -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()

128
webapp/src/components/tabContentComponents/slideScanContent.vue
View file

@ -1,51 +1,32 @@
<template>
<div uk-grid class="uk-height-1-1 uk-margin-remove uk-padding-remove">
<div class="control-component uk-padding-small">
<div v-show="!scanning" class="uk-padding-small">
<div v-show="!scanning" v-observe-visibility="visibilityChanged" class="uk-padding-small">
<h4 v-if="workflowDisplayName" class="workflow-name">
{{ workflowDisplayName }}
</h4>
<p class="workflow-blurb">{{ workflowBlurb }}</p>
<ul uk-accordion="multiple: true">
<li>
<a class="uk-accordion-title" href="#">Configure</a>
<a class="uk-accordion-title" href="#">Scan Settings</a>
<div class="uk-accordion-content">
<div
v-for="(setting, index) in workflowSettings"
:key="'detector_setting' + index"
class="uk-margin"
>
<server-specified-property-control :property-data="setting" />
</div>
</div>
</li>
<li class="uk-open">
<a class="uk-accordion-title" href="#">Stitching Settings</a>
<div class="uk-accordion-content">
<div class="uk-margin">
<propertyControl
thing-name="smart_scan"
property-name="max_range"
label="Maximum Distance (steps)"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="smart_scan"
property-name="autofocus_dz"
label="Autofocus Range (steps)"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="autofocus"
property-name="stack_dz"
label="Stack dz (steps)"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="autofocus"
property-name="stack_images_to_save"
label="Images in Stack to Save"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="autofocus"
property-name="stack_min_images_to_test"
label="Minimum number of images to test for focus"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="smart_scan"
property-name="overlap"
label="Image Overlap (0-1)"
property-name="stitch_automatically"
label="Automatically Stitch Images Together"
/>
</div>
<div class="uk-margin">
@ -57,25 +38,6 @@
</div>
</div>
</li>
<li class="uk-open">
<a class="uk-accordion-title" href="#">Scan Settings</a>
<div class="uk-accordion-content">
<div class="uk-margin">
<propertyControl
thing-name="smart_scan"
property-name="skip_background"
label="Detect and Skip Empty Fields"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="smart_scan"
property-name="stitch_automatically"
label="Automatically Stitch Images Together"
/>
</div>
</div>
</li>
</ul>
<label class="uk-form-label" for="form-stacked-text">Sample ID</label>
<div class="uk-form-controls">
@ -149,6 +111,7 @@
<script>
import streamDisplay from "./streamContent.vue";
import propertyControl from "../labThingsComponents/propertyControl.vue";
import ServerSpecifiedPropertyControl from "../labThingsComponents/serverSpecifiedPropertyControl.vue";
import actionLogDisplay from "../labThingsComponents/actionLogDisplay.vue";
import actionProgressBar from "../labThingsComponents/actionProgressBar.vue";
import MiniStreamDisplay from "../genericComponents/miniStreamDisplay.vue";
@ -179,6 +142,10 @@ export default {
log: [],
lastStitchedImage: null,
scan_name: "",
workflowName: undefined,
workflowSettings: [],
workflowDisplayName: undefined,
workflowBlurb: undefined,
};
},
@ -191,12 +158,49 @@ export default {
},
},
async created() {
this.readSettings();
},
methods: {
visibilityChanged(isVisible) {
if (isVisible) {
this.readSettings();
}
},
async readSettings() {
this.workflowName = await this.readThingProperty("smart_scan", "workflow_name");
if (this.workflowName) {
this.ready = await this.readThingProperty(this.workflowName, "ready");
this.workflowSettings = await this.readThingProperty(this.workflowName, "settings_ui");
console.log(this.workflowSettings);
this.workflowDisplayName = await this.readThingProperty(this.workflowName, "display_name");
this.workflowBlurb = await this.readThingProperty(this.workflowName, "ui_blurb");
}
},
onScanError: function (error) {
this.scanRunning = false;
this.modalError(error);
},
correlateCurrentScan() {},
/**
* Transition the UI into "scanning" mode and begin polling for scan updates.
*
* IMPORTANT:
* This method does NOT start a scan on the server.
*
* The <ActionButton> component is responsible for:
* - initiating the scan action on the backend when the user clicks the button
* - detecting and resuming an already-running scan when the page loads
*
* As a result, this method may be invoked in two cases:
* 1. Immediately after the user clicks "Start Smart Scan"
* 2. Automatically on page load if <ActionButton> detects an ongoing scan
*
* This function only:
* - updates local UI state to reflect that scanning is in progress
* - clears any previous preview image
* - starts the polling loop that fetches scan progress and preview images
*/
startScanning() {
this.lastStitchedImage = null;
this.scanning = true;
@ -235,4 +239,10 @@ export default {
.control-component {
width: 33%;
}
.workflow-name {
margin-bottom: 0.5rem;
}
.workflow-blurb {
color: #a2a2a2;
}
</style>