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

Merge branch 'More-workflow-layers' into 'v3'

Browse source 
More workflow layers

Closes #628

See merge request openflexure/openflexure-microscope-server!469
This commit is contained in:
Joe Knapper 2026-02-12 13:49:51 +00:00
commit 52c592e35c
8 changed files with 688 additions and 331 deletions
Download patch file Download diff file Expand all files Collapse all files

1
ofm_config_full.json
View file

@ -19,6 +19,7 @@
},
"histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow",
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
"raster_workflow": "openflexure_microscope_server.things.scan_workflows:RasterWorkflow",
"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"

1
ofm_config_simulation.json
View file

@ -14,6 +14,7 @@
},
"histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow",
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
"raster_workflow": "openflexure_microscope_server.things.scan_workflows:RasterWorkflow",
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"
},

309
src/openflexure_microscope_server/scan_planners.py
View file

@ -1,14 +1,20 @@
"""Functionality for planning scan routes.
A scan route can be planned by a ScanPlanner class currently there
is only one type the SmartSpiral. More can be added using by
subclassing the ScanPlanner
A scan route can be planned by a ScanPlanner. There is a base class ``ScanPlanner``,
and then a child class that is still generic called RectGridPlanner that helps planning
anything where the movements are on a regtangular grid. RectGridPlanner has two usable
child classes:
* SmartSpiral - For spiralling around a samples but adjusting when background is
detected
* RegularGridPlanner - For Raster and Snake scanning.
"""
# Future annotations needed for typhinting same class in __eq__ method. Other option
# would be to import Union and use a string.
from __future__ import annotations
import enum
import logging
from copy import copy
from typing import Any, Literal, Optional, TypeAlias
@ -23,6 +29,23 @@ XYPosList: TypeAlias = list[XYPos]
XYZPosList: TypeAlias = list[XYZPos]
class DistanceMetric(enum.Enum):
"""An enum for selecting distance metrics for grids.
Grid distance metrics are:
* Chebyshev (``CHEBYSHEV``) which is the larger of the number of x or y moves
in the grid.
* Manhattan (``MANHATTAN``) which is the number of moves between the two points
following the grid. Or
* Euclidean (``EUCLIDEAN``) which is the length of the direct route.
"""
CHEBYSHEV = enum.auto()
MANHATTAN = enum.auto()
EUCLIDEAN = enum.auto()
def enforce_xy_tuple(value: XYPos) -> XYPos:
"""Check input is a tuple and is of length 2.
@ -197,7 +220,12 @@ class ScanPlanner:
return [loc.xyz_tuple for loc in self._path_history if loc.imaged]
@property
def focused_locations(self) -> XYZPosList:
def focused_locations(self) -> list[VisitedScanLocation]:
"""Property to access a copy of the focused_locations."""
return [loc for loc in self._path_history if loc.focused]
@property
def focused_locations_xyz(self) -> XYZPosList:
"""Property to access a copy of the focused_locations."""
return [loc.xyz_tuple for loc in self._path_history if loc.focused]
@ -304,7 +332,97 @@ class ScanPlanner:
return [FutureScanLocation(location) for line in grid for location in line]
class SmartSpiral(ScanPlanner):
class RectGridPlanner(ScanPlanner):
"""Base class for planners that operate on a rectangular grid."""
_dx: int = 0
_dy: int = 0
def _parse(self, planner_settings: Optional[dict] = None) -> None:
expected_keys = ["dx", "dy"]
invalid_msg = "RectGridPlanner requires planner_settings with keys: "
if not planner_settings or not all(
k in planner_settings for k in expected_keys
):
raise KeyError(invalid_msg + ",".join(expected_keys))
self._dx = int(planner_settings["dx"])
self._dy = int(planner_settings["dy"])
def _adjacent_positions(self, xy_pos: XYPos) -> XYPosList:
return [
(xy_pos[0] - self._dx, xy_pos[1]),
(xy_pos[0] + self._dx, xy_pos[1]),
(xy_pos[0], xy_pos[1] - self._dy),
(xy_pos[0], xy_pos[1] + self._dy),
]
def moves_between(
self,
starting_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation,
ending_pos: XYPos | np.ndarray | FutureScanLocation | VisitedScanLocation,
metric: DistanceMetric,
) -> float:
"""Return displacement in grid-move units as a numpy array [dx_moves, dy_moves].
:param starting_pos: the position to measure from
:param ending_pos: the position to measure to
:param metric: How the distance is calculated. See `DistanceMetric`
"""
if isinstance(starting_pos, (FutureScanLocation, VisitedScanLocation)):
starting_pos = starting_pos.xy_tuple
if isinstance(ending_pos, (FutureScanLocation, VisitedScanLocation)):
ending_pos = ending_pos.xy_tuple
move_size = np.array([self._dx, self._dy], dtype="float64")
starting_pos = np.array(starting_pos, dtype="float64")
ending_pos = np.array(ending_pos, dtype="float64")
displacement = (ending_pos - starting_pos) / move_size
if metric == DistanceMetric.CHEBYSHEV:
return float(np.max(np.abs(displacement)))
if metric == DistanceMetric.MANHATTAN:
return float(np.sum(np.abs(displacement)))
return float(np.linalg.norm(displacement))
def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos:
"""Return an (x,y) position halfway between two input positions."""
x = (xy_pos1[0] + xy_pos2[0]) // 2
y = (xy_pos1[1] + xy_pos2[1]) // 2
return (x, y)
def select_nearby_focus_site(self, next_position: XYPos) -> Optional[XYZPos]:
"""Return a focused site near the given position to estimate Z for the next move.
Looks for all previously focused locations that are within the scan
step size (self._dx, self._dy) of ``next_position``. Among these nearby focused
sites, it returns the most recently imaged one.
This is suitable for raster or snake scans, where the scan may move along a row
or column and then jump to a new row/column. If no nearby focused sites exist,
returns None.
:param next_position: The XY position where the next image will be taken.
:return: The XYZ tuple of the closest and most recent focused site, or None if
no focused locations exist.
"""
focused_locations = self.focused_locations
if not focused_locations:
return None
def sort_key(pos: VisitedScanLocation) -> float:
return self.moves_between(next_position, pos, DistanceMetric.MANHATTAN)
# Sort by the total number of dx and dy moves between sites, then by most
# recent. Using reverse=True puts the most recent, nearest at the end
nearby_focus_locations = sorted(focused_locations, key=sort_key, reverse=True)
# Pick the most recent nearby site
return nearby_focus_locations[-1].xyz_tuple
class SmartSpiral(RectGridPlanner):
"""A scan planner that spirals outward from the centre, prioritising short moves.
This planner spirals out from the centre, but prioritises short moves over rigidly
@ -321,20 +439,9 @@ class SmartSpiral(ScanPlanner):
dx and dy.
"""
# The maximum distance for the scan to run in any direction.
# Any future moves which would move beyond this distance are not appended.
_max_dist: int = 0
_dx: int = 0
_dy: 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 _is_primary_location(
self, location: FutureScanLocation | VisitedScanLocation
@ -352,21 +459,11 @@ class SmartSpiral(ScanPlanner):
]
def _parse(self, planner_settings: Optional[dict] = None) -> None:
"""Parse SmartSpiral Settings dictionary.
super()._parse(planner_settings)
* ``dx`` - the movement size in x
* ``dy`` - the movement size in y
* ``max_dist`` - The maximum distance to a location can be from the centre.
"""
expected_keys = ["max_dist", "dx", "dy"]
invalid_msg = "SmartSpiral 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))
if not planner_settings or "max_dist" not in planner_settings:
raise KeyError("SmartSpiral requires max_dist")
self._dx = int(planner_settings["dx"])
self._dy = int(planner_settings["dy"])
self._max_dist = int(planner_settings["max_dist"])
def _initial_location_list(self) -> list[FutureScanLocation]:
@ -472,21 +569,6 @@ class SmartSpiral(ScanPlanner):
# imaged points.
self._remaining_locations.append(i_loc)
def _adjacent_positions(self, xy_pos: XYPos) -> XYPosList:
"""Return 4 points +/-dx and +/-dy from the input location."""
return [
(xy_pos[0] - self._dx, xy_pos[1]),
(xy_pos[0] + self._dx, xy_pos[1]),
(xy_pos[0], xy_pos[1] - self._dy),
(xy_pos[0], xy_pos[1] + self._dy),
]
def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos:
"""Return an (x,y) position halfway between two input positions."""
x = (xy_pos1[0] + xy_pos2[0]) // 2
y = (xy_pos1[1] + xy_pos2[1]) // 2
return (x, y)
def _re_sort_remaining_locations(self, current_pos: XYPos) -> None:
"""Sort the remaining positions based on the current location."""
@ -494,8 +576,10 @@ class SmartSpiral(ScanPlanner):
def sort_key(pos: FutureScanLocation) -> tuple[bool, float, float, float]:
return (
self._is_primary_location(pos), # False sorts low
self.moves_between(current_pos, pos),
self.moves_between(self._initial_position, pos),
self.moves_between(current_pos, pos, DistanceMetric.CHEBYSHEV),
self.moves_between(
self._initial_position, pos, DistanceMetric.CHEBYSHEV
),
distance_between(current_pos, pos),
)
@ -514,110 +598,72 @@ class SmartSpiral(ScanPlanner):
Returns None if no focused locations are present
"""
# save to variable rather than search for focussed sites each time.
focused_locations = self.focused_locations
focused_locations = self.focused_locations_xyz
if not focused_locations:
return None
# must be float64 (double precision) to deal with the huge numbers involved!
# must be float64 to deal with large coordinates
current_pos = np.array(xy_pos, dtype="float64")
path_pos = np.array(focused_locations, dtype="float64")[:, :2]
focused_arr = np.array(focused_locations, dtype="float64")
# Use linalg.norm to calculate the direct distance between the points
# Note linalg.norm always uses float64
dists = np.linalg.norm((path_pos - current_pos), axis=1)
# Find focused sites within dx and dy
dx_ok = np.abs(focused_arr[:, 0] - current_pos[0]) <= self._dx
dy_ok = np.abs(focused_arr[:, 1] - current_pos[1]) <= self._dy
nearby_indices = np.where(dx_ok & dy_ok)[0]
# Get indices of all focused sites within distance_cutoff.
# Note np.where always returns a tuple of arrays, hence the trailing [0]
indices = np.where(dists <= self._distance_cutoff)[0]
# If no neighbouring sites were focused, choose the closest
if len(nearby_indices) == 0:
deltas = focused_arr[:, :2] - current_pos
dists = np.linalg.norm(deltas, axis=1)
min_dist = np.min(dists)
nearby_indices = np.where(dists == min_dist)[0]
# Handle the case that no focused positions are within this range, and
# instead use the nearest focused position. This will always return a
# height, due to the check that self._focused_locations exists.
if len(indices) == 0:
distance_cutoff = min(dists)
indices = np.where(dists <= distance_cutoff)[0]
nearby_sites = focused_arr[nearby_indices]
# Turning into an array allows slicing based on a list
focused_locations_array = np.array(focused_locations)
# Choose the lowest z
min_z = np.min(nearby_sites[:, 2])
# Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best
# if started too low, so the lowest z will perform best
candidates = focused_locations_array[indices]
min_z = np.min(candidates[:, -1])
# Among those with min z, choose the most recent
chosen_site = nearby_sites[nearby_sites[:, 2] == min_z][-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())
def moves_between(
self,
starting_pos: XYPos | np.ndarray | FutureScanLocation,
ending_pos: XYPos | np.ndarray | FutureScanLocation,
) -> float:
"""Return the larger of x moves or y moves between two xy positions.
:param starting_pos: the position to measure from
:param ending_pos: the position to measure to
"""
if isinstance(starting_pos, FutureScanLocation):
starting_pos = starting_pos.xy_tuple
if isinstance(ending_pos, FutureScanLocation):
ending_pos = ending_pos.xy_tuple
move_size = np.array([self._dx, self._dy])
starting_pos = np.array(starting_pos, dtype="float64")
ending_pos = np.array(ending_pos, dtype="float64")
displacement_in_moves = (ending_pos - starting_pos) / move_size
return np.max(np.abs(displacement_in_moves))
return (
int(chosen_site[0]),
int(chosen_site[1]),
int(chosen_site[2]),
)
class SnakeScan(ScanPlanner):
"""A scan planner that performs a snake scan, right and down from a corner.
class RegularGridPlanner(RectGridPlanner):
"""A scan planner that performs a snake or a raster scan.
Direction cannot yet be set it always scans, 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
_style: Literal["snake", "raster"]
def _parse(self, planner_settings: Optional[dict] = None) -> None:
"""Parse SnakeScan Settings dictionary.
super()._parse(planner_settings)
* ``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):
expected_keys = ["x_count", "y_count", "style"]
invalid_msg = "RegularGrid requires planner_settings with keys: "
if not planner_settings or not all(
k in planner_settings for k 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"])
style = planner_settings["style"]
if style not in ("snake", "raster"):
raise ValueError(
f"Unknown regular grid style {style}. Use snake or raster."
)
self._style = style
def _initial_location_list(self) -> list[FutureScanLocation]:
"""Set the initial list of locations for this scan planner.
@ -632,20 +678,11 @@ class SnakeScan(ScanPlanner):
y_count=self._y_count,
dx=self._dx,
dy=self._dy,
style="snake",
style=self._style,
)
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,

389
src/openflexure_microscope_server/things/scan_workflows.py
View file

@ -9,6 +9,7 @@ from __future__ import annotations
import os
from typing import (
Generic,
Literal,
Mapping,
Optional,
TypeVar,
@ -19,9 +20,9 @@ from pydantic import BaseModel
import labthings_fastapi as lt
from openflexure_microscope_server.scan_planners import (
RegularGridPlanner,
ScanPlanner,
SmartSpiral,
SnakeScan,
)
from openflexure_microscope_server.stitching import (
STITCHING_RESOLUTION,
@ -67,6 +68,10 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
# CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None
_csm: Optional[CameraStageMapper] = lt.thing_slot()
# Camera, stage and autofocus are all required by any scan workflow
_cam: BaseCamera = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
def check_before_start(self, scan_name: str) -> None:
"""Check before the scan starts. Throw an error if the scan shouldn't start.
@ -116,11 +121,44 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
def acquisition_routine(
self, settings: SettingModelType, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Overload to set the acquisition routine that happens at each scan site."""
"""Overload to set the acquisition routine that happens at each scan site.
:param settings: The settings for this scan, which should be a SettingModelType
:param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus.
If failed to find focus, returns for the focus z-position.
"""
raise NotImplementedError(
"Each specific ScanWorkflow must implement an acquisition routine"
)
def _autofocus_and_capture(
self,
xyz_pos: tuple[int, int, int],
dz: int,
images_dir: str,
save_resolution: tuple[int, int],
) -> tuple[bool, Optional[int]]:
"""Autofocus and then capture, this can be used as an acquisition routine.
:param dz: The dz for autofocus.
:param images_dir: The path to the directory for saving images..
:param save_resolution: The resolution to save images at.
:return: A tuple ready to pass out of acquisition routine. In this method,
image is always taken, so first return is True.
"""
self._autofocus.fast_autofocus(dz=dz)
focus_height = self._stage.get_xyz_position()[2]
filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg"
self._cam.capture_and_save(
jpeg_path=os.path.join(images_dir, filename),
save_resolution=save_resolution,
)
return True, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
@ -128,13 +166,35 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
"Each scan workflow must implement a settings_ui method."
)
def _require_csm(self) -> CameraStageMapper:
"""Give each model the option to require CSM. Return it if present."""
if self._csm is None:
raise RuntimeError(
"CameraStageMapping not set, and is required for this workflow."
)
return self._csm
class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
"""A generic workflow for any scan that captures images on a rectilinear grid."""
# Redefine _csm Thing Slot, as CSM is required for any RectGridWorkflow
_csm: CameraStageMapper = lt.thing_slot()
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x and y.
This must be between 0.1 and 0.7.
"""
autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000)
"""The z distance to perform an autofocus in steps.
Must be greater than or equal to 200, and less than or equal to 2000.
"""
# The noqa statement is because scan_name is unused but is needed for equivalence
# with other workflows that may want to validate the scan name.
def check_before_start(self, scan_name: str) -> None: # noqa: ARG002
"""Before starting a scan, check that camera-stage-mapping is set.
Raise error if:
- camera stage mapping is not set
"""
if self._csm.calibration_required:
raise RuntimeError("Camera Stage Mapping is not calibrated.")
def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]:
"""Use camera stage mapping to calculate x and y displacement from given overlap.
@ -146,26 +206,41 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
:raises RuntimeError: If there is no camera stage mapper Thing available or if CMS isn't calibrated.
"""
csm = self._require_csm()
csm_image_res = csm.image_resolution
csm_image_res = self._csm.image_resolution
if csm_image_res is None:
raise RuntimeError("CSM not set. Scan shouldn't have progresses this far.")
raise RuntimeError("CSM not set. Scan shouldn't have progressed this far.")
# Calculate displacements in image coordinates
dx_img = csm_image_res[1] * (1 - overlap)
dy_img = csm_image_res[0] * (1 - overlap)
x_move_stage = csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
x_move_stage = self._csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = self._csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
# Assume no rotation or skew and take only the aligned axis of vector.
# Coerce to positive integer, but correct if x and y are flipped
if abs(x_move_stage["x"]) > abs(x_move_stage["y"]):
return x_move_stage["x"], y_move_stage["y"]
# If not use the other stage axes. Note "dx" will be the movement in camera y.
self.logger.info(
f"Based on an overlap of {self.overlap}, the stage will make steps of "
f"{y_move_stage['x']}, {x_move_stage['y']}"
)
return y_move_stage["x"], x_move_stage["y"]
def _get_stitching_settings_model(self) -> StitchingSettings:
"""Return a stitching settings model based on current settings."""
return StitchingSettings(
overlap=self.overlap,
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
@lt.property
def ready(self) -> bool:
"""Whether this scanworkflow is ready to start."""
return not self._csm.calibration_required
class HistoScanSettingsModel(BaseModel):
"""The settings for a scan with the HistoScanWorkflow.
@ -175,14 +250,14 @@ class HistoScanSettingsModel(BaseModel):
"""
overlap: float
max_dist: int
dx: int
dy: int
max_dist: int
skip_background: bool
smart_stack_params: SmartStackParams
class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
"""A workflow optimised for scanning Histopathology samples.
This workflow automatically plans its own path around a sample spiralling out from
@ -200,12 +275,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
)
_settings_model = HistoScanSettingsModel
_planner_cls: type[ScanPlanner] = SmartSpiral
_planner_cls = SmartSpiral
# Thing Slots
_background_detector: ChannelDeviationLUV = lt.thing_slot()
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
# Scan settings
@ -215,21 +287,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
This uses the settings from the ``BackgroundDetectThing``.
"""
autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000)
"""The z distance to perform an autofocus in steps.
Must be greater than or equal to 200, and less than or equal to 2000.
"""
max_range: int = lt.setting(default=45000)
"""The maximum distance in steps from the centre of the scan."""
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x or y.
This must be between 0.1 and 0.7.
"""
# Stacking settings
stack_images_to_save: int = lt.setting(default=1)
@ -304,16 +364,8 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
:return: A tuple containing the settings model for this workflow and the
settings model for stitching.
"""
stitching_settings = StitchingSettings(
overlap=self.overlap,
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
stitching_settings = self._get_stitching_settings_model()
dx, dy = self._calc_displacement_from_overlap(self.overlap)
self.logger.info(
f"Based on an overlap of {self.overlap}, the stage will make steps of "
f"{dx}, {dy}"
)
smart_stack_params = self.create_smart_stack_params(
images_dir=images_dir,
@ -484,9 +536,6 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
"""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"
),
@ -498,11 +547,14 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
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)"),
property_control_for(
self, "skip_background", label="Detect and Skip Empty Fields "
),
]
class SnakeSettingsModel(BaseModel):
"""The settings for a scan with the SnakeWorkflow.
class RegularGridSettingsModel(BaseModel):
"""The settings for a scan with a regular grid of dx and dy for x_count, y_count steps.
This includes settings calculated when starting. This will be held by smart scan
during a scan and serialised to disk.
@ -513,12 +565,108 @@ class SnakeSettingsModel(BaseModel):
dy: int
x_count: int
y_count: int
style: Literal["snake", "raster"]
images_dir: str
autofocus_dz: int
save_resolution: tuple[int, int]
class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
"""A base workflow for any workflow that uses a regular rectangular grid."""
x_count: int = lt.setting(default=3, ge=1)
"""The number of columns in the scan."""
y_count: int = lt.setting(default=2, ge=1)
"""The number of rows in the scan."""
_settings_model = RegularGridSettingsModel
_planner_cls = RegularGridPlanner
_grid_style: Literal["snake", "raster"]
def all_settings(
self, images_dir: str
) -> tuple[RegularGridSettingsModel, Optional[StitchingSettings]]:
"""Return the workflow and stitching settings.
:param images_dir: The directory that images are to be written to.
:return: A tuple containing the settings model for this workflow and the
settings model for stitching.
"""
stitching_settings = self._get_stitching_settings_model()
dx, dy = self._calc_displacement_from_overlap(self.overlap)
scan_settings = self._settings_model(
overlap=self.overlap,
dx=dx,
dy=dy,
x_count=self.x_count,
y_count=self.y_count,
style=self._grid_style,
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: RegularGridSettingsModel) -> None:
"""Perform these steps before starting the scan.
In this case, only autofocus.
:param settings: The settings for this scan as as the relevant SettingsModel type.
"""
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
def new_scan_planner(
self, settings: RegularGridSettingsModel, 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.
"""
planner_settings = {
"dx": settings.dx,
"dy": settings.dy,
"x_count": settings.x_count,
"y_count": settings.y_count,
"style": settings.style,
}
return self._planner_cls(
initial_position=(position["x"], position["y"]),
planner_settings=planner_settings,
)
def acquisition_routine(
self, settings: RegularGridSettingsModel, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Autofocus and capture.
:param settings: The settings for this scan as a RegularGridSettingsModel
: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.
"""
return self._autofocus_and_capture(
xyz_pos=xyz_pos,
dz=settings.autofocus_dz,
images_dir=settings.images_dir,
save_resolution=settings.save_resolution,
)
@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)"),
]
class SnakeWorkflow(RegularGridWorkflow):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and snakes
@ -533,139 +681,24 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
),
readonly=True,
)
_grid_style = "snake"
_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
class RasterWorkflow(RegularGridWorkflow):
"""A workflow optimised for snaking around samples.
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.
This workflow generates a list of coordinates in a rectangle, and always
moves right across a row, then moves down a row while moving to the starting
column (assuming positive dx and dy).
"""
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x or y.
display_name: str = lt.property(default="Raster Scan", readonly=True)
ui_blurb: str = lt.property(
default=(
"This scan workflow is optimised for performing a raster scan over a rectangle. It "
"always moves down and right from the starting point, over a defined grid."
),
readonly=True,
)
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)"),
]
_grid_style = "raster"

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

@ -208,6 +208,16 @@ class SmartScanThing(lt.Thing):
raise ScanNotRunningError("Cannot get ongoing scan if scan is not running.")
return self._ongoing_scan
@lt.property
def all_workflow_names(self) -> list[str]:
"""Return a list of all available Scan Workflows."""
return list(self._all_workflows.keys())
@lt.property
def workflow_display_names(self) -> dict[str, str]:
"""Return a list of the display names of all available Scan Workflows."""
return {name: wf.display_name for name, wf in self._all_workflows.items()}
_scan_data: Optional[ActiveScanData] = None
@property

276
tests/unit_tests/test_scan_planners.py
View file

@ -89,8 +89,8 @@ def test_bad_smart_spiral_settings():
initial_position = (100, 50)
# Class init should raise error if no planner_settings dictionary set
msg = "SmartSpiral requires a planner_settings dictionary with keys"
with pytest.raises(ValueError, match=msg):
msg = "RectGridPlanner requires planner_settings with keys"
with pytest.raises(KeyError, match=msg):
scan_planners.SmartSpiral(initial_position=initial_position)
planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000}
@ -105,7 +105,7 @@ def test_bad_smart_spiral_settings():
initial_position=initial_position, planner_settings=bad_planner_settings
)
# Class init should raise error if planner_settings if any value can't be cast
# Class init should raise error if any value in planner_settings can't be cast
# to int
keys = ["dx", "dy", "max_dist"]
for badkey in keys:
@ -317,12 +317,18 @@ def test_example_smart_spiral():
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."""
def test_snake_planner_basic_grid():
"""Check that snake scan planner 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_settings = {
"dx": 100,
"dy": 100,
"x_count": 1,
"y_count": 1,
"style": "snake",
}
planner = scan_planners.SnakeScan(
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
@ -352,11 +358,17 @@ def test_snake_scan_basic_grid():
def test_snake_scan_basic_length():
"""SnakeScan should generate the correct number of locations."""
"""Snake scan planner should generate the correct number of locations."""
initial_position = (100, 50)
planner_settings = {"dx": 100, "dy": 100, "x_count": 3, "y_count": 4}
planner_settings = {
"dx": 100,
"dy": 100,
"x_count": 3,
"y_count": 4,
"style": "snake",
}
planner = scan_planners.SnakeScan(
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
@ -369,9 +381,15 @@ def test_snake_scan_basic_length():
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_settings = {
"dx": 10,
"dy": 10,
"x_count": 4,
"y_count": 3,
"style": "snake",
}
planner = scan_planners.SnakeScan(
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
@ -399,9 +417,9 @@ def test_snake_scan_ordering():
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_settings = {"dx": 5, "dy": 5, "x_count": 4, "y_count": 1, "style": "snake"}
planner = scan_planners.SnakeScan(
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
@ -412,9 +430,9 @@ def test_snake_scan_single_row():
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_settings = {"dx": 5, "dy": 5, "x_count": 1, "y_count": 4, "style": "snake"}
planner = scan_planners.SnakeScan(
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
@ -425,3 +443,229 @@ def test_snake_scan_single_column():
(0, 10),
(0, 15),
]
def test_snake_scan_z_propagation():
"""Test that snake planner selects correct previous focus height."""
initial_position = (0, 0)
planner_settings = {
"dx": 50,
"dy": 50,
"x_count": 5,
"y_count": 5,
"style": "snake",
}
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
expected_z = 1
while not planner.scan_complete:
xy_pos, z_est = planner.get_next_location_and_z_estimate()
print(xy_pos, z_est)
# First position should have no estimate
if xy_pos == (0, 0):
assert z_est is None
else:
# Should estimate from previous focused point
assert z_est == expected_z - 1
xyz_pos = (xy_pos[0], xy_pos[1], expected_z)
planner.mark_location_visited(
xyz_pos,
imaged=True,
focused=True,
)
expected_z += 1
def test_raster_planner_basic_grid():
"""Check that raster scan planner generates a single point for a 1x1 scan."""
initial_position = (100, 50)
planner_settings = {
"dx": 100,
"dy": 100,
"x_count": 1,
"y_count": 1,
"style": "raster",
}
planner = scan_planners.RegularGridPlanner(
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_raster_scan_basic_length():
"""Raster scan planner should generate the correct number of locations."""
initial_position = (100, 50)
planner_settings = {
"dx": 100,
"dy": 100,
"x_count": 3,
"y_count": 4,
"style": "raster",
}
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
coords = planner.remaining_locations
assert len(coords) == 3 * 4
def test_raster_scan_ordering():
"""Test that raster scan returns a path in the right order."""
initial_position = (0, 0)
planner_settings = {
"dx": 10,
"dy": 10,
"x_count": 4,
"y_count": 3,
"style": "raster",
}
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
coords = planner.remaining_locations
expected = [
(0, 0),
(10, 0),
(20, 0),
(30, 0),
(0, 10),
(10, 10),
(20, 10),
(30, 10),
(0, 20),
(10, 20),
(20, 20),
(30, 20),
]
assert coords == expected
def test_raster_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, "style": "raster"}
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
assert planner.remaining_locations == [(0, 0), (5, 0), (10, 0), (15, 0)]
def test_raster_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, "style": "raster"}
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings=planner_settings,
)
assert planner.remaining_locations == [
(0, 0),
(0, 5),
(0, 10),
(0, 15),
]
def test_raster_z_propagation():
"""Test that snake planner selects correct previous focus height.
Constructs a 5x5 grid in a raster pattern, and test that for each movement,
the chosen next z position is either
- None, for the first point
- the start of the previous row, for the first point in a row
- the previous site otherwise
"""
initial_position = (0, 0)
x_count = 5
y_count = 5
dx = 50
dy = 50
planner = scan_planners.RegularGridPlanner(
initial_position=initial_position,
planner_settings={
"dx": dx,
"dy": dy,
"x_count": x_count,
"y_count": y_count,
"style": "raster",
},
)
visited_positions = []
current_z = 1
while not planner.scan_complete:
visited_count = len(visited_positions)
xy_pos, z_est = planner.get_next_location_and_z_estimate()
print(visited_count)
if visited_count == 0:
assert z_est is None
else:
# check whether this site is at the start of a new row
if visited_count % x_count == 0:
# if so, get the z position from the start of the previous row
expected_z = visited_positions[-x_count][2]
else:
expected_z = visited_positions[-1][2]
assert z_est == expected_z
xyz_pos = (xy_pos[0], xy_pos[1], current_z)
planner.mark_location_visited(
xyz_pos,
imaged=True,
focused=True,
)
visited_positions.append(xyz_pos)
current_z += 1

2
tests/unit_tests/test_scan_workflows.py
View file

@ -342,11 +342,11 @@ def test_histo_workflow_settings_ui(histo_workflow):
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",
"skip_background",
]
assert names == expected_names

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

@ -2,6 +2,20 @@
<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" v-observe-visibility="visibilityChanged" class="uk-padding-small">
<!-- Workflow Selection Dropdown -->
<div class="uk-margin">
<label class="uk-form-label">Workflow</label>
<select
class="uk-select uk-form-small"
:value="workflowName"
@change="setWorkflow($event.target.value)"
>
<option v-for="(label, name) in workflowOptions" :key="name" :value="name">
{{ label }}
</option>
</select>
</div>
<h4 v-if="workflowDisplayName" class="workflow-name">
{{ workflowDisplayName }}
</h4>
@ -145,6 +159,7 @@ export default {
workflowSettings: [],
workflowDisplayName: undefined,
workflowBlurb: undefined,
workflowOptions: [],
};
},
@ -159,6 +174,7 @@ export default {
async created() {
this.readSettings();
this.workflowOptions = await this.readThingProperty("smart_scan", "workflow_display_names");
},
methods: {
@ -216,6 +232,21 @@ export default {
setTimeout(this.pollScan, 1000); // keep rescheduling until it's stopped
}
},
async setWorkflow(name) {
try {
this.workflowName = name;
await this.writeThingProperty("smart_scan", "workflow_name", name);
// refresh UI
await this.readSettings();
} catch (err) {
this.modalError(err);
// revert if server rejected
this.workflowName = await this.readThingProperty("smart_scan", "workflow_name");
}
},
async downloadZipFile(response) {
const scan_name = response.input.scan_name;
const filename = `${scan_name}_images.zip`;