New scan workflow for snake scans, move next pos into scan baseclass

Update tests that were testing wrong method, update all to new select_nearby_focus_site
This commit is contained in:
Joe Knapper 2026-02-06 14:54:31 +00:00
parent e59d5e82d8
commit f230eef2dc
4 changed files with 260 additions and 166 deletions

View file

@ -13,6 +13,7 @@
}
},
"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"
},

View file

@ -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,20 @@ 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
"""
path = []
for line in grid:
path += line
return [FutureScanLocation(location) for location in path]
class SmartSpiral(ScanPlanner):
"""A scan planner that spirals outward from the centre, prioritising short moves.
@ -385,7 +373,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 +502,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 +543,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())
@ -664,91 +636,16 @@ class SnakeScan(ScanPlanner):
style="snake",
)
# create_rectangular_scan_path provides a nested list, which is flattened here
path = []
for line in grid:
path += line
return self._grid_to_future_locations(grid)
return [FutureScanLocation(location) for location in path]
def mark_location_visited(
self, xyz_pos: XYZPos, imaged: bool = True, focused: bool = True
) -> None:
"""Mark the location as visited.
:param xyz_pos: the x_y_z position
:param imaged: true if an image was taken, false if not (due to background detect)
:param focused: true if autofocus completed successfully
"""
# Only call the base class to update the positions
super().mark_location_visited(xyz_pos, imaged, focused)
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_site = self._remaining_locations[0]
next_location = next_site.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.
Lowest position is best, as starting too high causes smart stacking to
autofocus and restart. Starting too low just requires extra movements in +z.
Nearby is defined as within 1.1 times the larger of the x and y scan offsets.
If no focused sites are within this range, use the height of the nearest
focused site.
Returns None if no focused locations are present
"""
# save to variable rather than search for focussed sites each time.
# 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
# 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 between the points
# Note linalg.norm always uses float64
dists = np.linalg.norm((path_pos - current_pos), axis=1)
# 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]
# 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]
# Turning into an array allows slicing based on a list
focused_locations_array = np.array(focused_locations)
# 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])
# Convert back into list so values are of type int instead of np.int32
return tuple(chosen_focused_site.tolist())
return focused_locations[-1]
def distance_between(

View file

@ -6,6 +6,7 @@ as well as specific workflows.
from __future__ import annotations
import os
from typing import (
Generic,
Mapping,
@ -17,7 +18,11 @@ from pydantic import BaseModel
import labthings_fastapi as lt
from openflexure_microscope_server.scan_planners import ScanPlanner, SmartSpiral
from openflexure_microscope_server.scan_planners import (
ScanPlanner,
SmartSpiral,
SnakeScan,
)
from openflexure_microscope_server.stitching import (
STITCHING_RESOLUTION,
StitchingSettings,
@ -33,6 +38,7 @@ from openflexure_microscope_server.things.background_detect import (
)
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)
@ -59,6 +65,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
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] = None
def check_before_start(self, scan_name: str) -> None:
"""Check before the scan starts. Throw an error if the scan shouldn't start.
@ -119,6 +128,42 @@ 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
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
"""
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.
@ -285,32 +330,6 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
return scan_settings, stitching_settings
def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]:
"""Use camera stage mapping to calculate x and y displacement.
:param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means
that each image should overlap its nearest neighbour by 50%.
:returns: (dx, dy) - the x and y displacements in steps
"""
csm_image_res = self._csm.image_resolution
if csm_image_res is None:
raise RuntimeError("CSM not set. Scan shouldn't have progressed this far.")
# Calculate displacements in image coordinates
dx_img = csm_image_res[1] * (1 - overlap)
dy_img = csm_image_res[0] * (1 - overlap)
x_move_stage = self._csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = self._csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
# Assume no rotation or skew and take only the aligned axis of vector.
# Coerce to positive integer, but correct if x and y are flipped
if abs(x_move_stage["x"]) > abs(x_move_stage["y"]):
return x_move_stage["x"], y_move_stage["y"]
# If not use the other stage axes. Note "dx" will be the movement in camera y.
return y_move_stage["x"], x_move_stage["y"]
def create_smart_stack_params(
self,
images_dir: str,
@ -478,3 +497,173 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
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)"),
]

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