Merge branch 'snake-scan' into 'v3'
SnakeScan as a scan planner See merge request openflexure/openflexure-microscope-server!464
This commit is contained in:
Julian Stirling
commit
9ab5a461c0
5 changed files with 475 additions and 86 deletions
|
|
@ -18,6 +18,7 @@
|
|||
}
|
||||
},
|
||||
"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"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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] = 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.
|
||||
|
||||
|
|
@ -119,6 +128,44 @@ 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
|
||||
|
||||
: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.
|
||||
|
|
@ -285,32 +332,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 +499,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)"),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue