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

Merge branch 'further-scan-refactor' into 'v3'

Browse source 
Further scan refactor

See merge request openflexure/openflexure-microscope-server!242
This commit is contained in:
Julian Stirling 2025-04-14 16:07:37 +00:00
commit be64d4c5e0
10 changed files with 1268 additions and 348 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.gitignore vendored
View file

@ -84,3 +84,7 @@ openflexure_microscope/cobertura.xml
# web app build
/src/openflexure_microscope_server/static/
# Files created by test utilities
/tests/utilities/*.pstats
/tests/utilities/*.png

3
pyproject.toml
View file

@ -35,6 +35,7 @@ dev = [
"mypy-gitlab-code-quality",
# "pytest-gitlab-code-quality", # pytest version constraint clashes with labthings
"pytest",
"matplotlib~=3.10"
]
pi = [
"labthings-picamera2 == 0.0.2-dev0",
@ -81,6 +82,8 @@ addopts = [
norecursedirs = [".git", "build", "node_modules"]
pythonpath = ["."]
[tool.ruff.format]
# Use native line endings for all files
line-ending = "native"

5
ruff-increased-checking.toml
View file

@ -35,3 +35,8 @@ select = [
# also "D" but this even warns about docstring punctuation. Perhaps a step too
# far?
]
# Line length is set to 88 for ruff. This is what the formatter aims for
# some lines are not formatted to 88 if the formatter can't easily break
# them. Allow up to 99 before throwing a long line error
line-length = 99

375
src/openflexure_microscope_server/scan_planners.py Normal file
View file

@ -0,0 +1,375 @@
"""
This module contains functionality for planning a scan route
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
"""
from typing import TypeAlias, Optional
import logging
from copy import copy
import numpy as np
LOGGER = logging.getLogger(__name__)
XYPos: TypeAlias = tuple[int, int]
XYZPos: TypeAlias = tuple[int, int, int]
XYPosList: TypeAlias = list[XYPos]
XYZPosList: TypeAlias = list[XYZPos]
def enforce_xy_tuple(value: XYPos) -> XYPos:
"""
Used for enforcing that an input is a tuple and is the correct length
"""
if not isinstance(value, (list, tuple)):
raise ValueError("2 value tuple expected")
if not len(value) == 2:
raise ValueError("2 value tuple expected")
if isinstance(value, list):
return tuple(value)
return value
def enforce_xyz_tuple(value: XYZPos) -> XYZPos:
"""
Used for enforcing that an input is a tuple and is the correct length
"""
if not isinstance(value, (list, tuple)):
raise ValueError("3 value tuple expected")
if not len(value) == 3:
raise ValueError("3 value tuple expected")
if isinstance(value, list):
return tuple(value)
return value
class ScanPlanner:
"""
A base class for a scan planner.
This should never be used directly for a scan, it should be subclassed.
Each subclass should implement at least the methods with NotImplementedError
set:
* _parse() - to parse the planner_settings dictionary, saving values to class
variables
* _intial_location_list() - Sets the list of locations for the scan to follow
For a simple scan pattern this should be sufficent. For more complex ones that
dynanically adjust the path it is suggested to override `mark_location_visited()`
calling `super().mark_location_visited()` at the start of the method so that all
locations are adjusted.
When subclassing be sure to use enforce_xy_tuple and enforce_xyz_tuple on any user
data before running
"""
def __init__(self, intial_position: XYPos, planner_settings: Optional[dict] = None):
"""
Set up lists for the path planning, and scan history.
"""
self._initial_position = enforce_xy_tuple(intial_position)
self._parse(planner_settings)
# The remaining (x,y) locations to scan
# (This was `path` before refactoring from the long `sample_scan` code)
self._remaining_locations: XYPosList = self._intial_location_list()
# This holds a list of all (x,y,z) locations where images were taken
# this may not be equivalent to the x,y poistions ins self._path_history
# if background detect is used
# (This was not used in the `sample_scan` code)
self._imaged_locations: XYZPosList = []
# This holds a list of all (x,y,z) locations where autofocus was successful
# (This was `focused_path` before refactoring from the long `sample_scan` code)
self._focused_locations: XYZPosList = []
# This holds a list of all x,y locations visited in order since the start
# (This was `true_path` before refactoring from the long `sample_scan` code
# previously it had z set, but if we don't take an image, not z is needed and
# it slows other checks)
self._path_history: XYPosList = []
@property
def scan_complete(self) -> bool:
"""
Return True if there are no locations left to scan.
"""
return not self._remaining_locations
@property
def remaining_locations(self) -> XYPosList:
"""
Property to access a copy of the remaining_locations
"""
return copy(self._remaining_locations)
@property
def imaged_locations(self) -> XYZPosList:
"""
Property to access a copy of the imaged_locations
"""
return copy(self._imaged_locations)
@property
def focused_locations(self) -> XYZPosList:
"""
Property to access a copy of the focused_locations
"""
return copy(self._focused_locations)
@property
def path_history(self) -> XYPosList:
"""
Property to access a copy of the path_history
"""
return copy(self._path_history)
def _parse(self, planner_settings: Optional[dict] = None) -> None:
"""
Parse any settings sent to this planner and store them if needed.
"""
raise NotImplementedError("Did you call the ScanPlanner base class?")
def _intial_location_list(self) -> XYPosList:
"""
Called on initalisation. Sets the initial list of locations for this scan planner
For a simple grid scan/snake scan this would be all locations to move to.
Note for implementation that this _must_ contain (x,y) tuples, not [x, y]
lists or matching errors could occur.
"""
raise NotImplementedError("Did you call the ScanPlanner base class?")
def position_visited(self, position: XYPos) -> bool:
"""
Return True if input xy position has been visited before
"""
# Ensure tuple for correct matching!
return tuple(position) in self._path_history
def position_planned(self, position: XYPos) -> bool:
"""
Return True if input xy position is planned
"""
# Ensure tuple for correct matching!
return tuple(position) in self._remaining_locations
def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]:
"""
Return the next location to scan, and the estimated z-position
for this location.
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]
# If focussed locations exist return closest location, favouring most recent
closest_pos = self.closest_focus_site(next_location)
if closest_pos is None:
z = None
else:
z = 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
to the input xy_position, with the most recently taken image returned in
the case of a tie
Returns None if there if no focussed locations are present
"""
if not self._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(self._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 self._focused_locations[indices[-1]]
def mark_location_visited(
self, xyz_pos: XYZPos, imaged: bool, focused: bool
) -> None:
"""
Mark the location as visited
Args:
xyz_pos: the x_y_z position
imaged: true if an image was taken, false if not (due to background detect)
focused: true if autofocus completed successfully
"""
# ensure is tuple!
xyz_pos = enforce_xyz_tuple(xyz_pos)
xy_pos = xyz_pos[:2]
# Remove the expected position from the remaining locations list
# and check it's correct
expected_pos = tuple(self._remaining_locations.pop(0))
if xy_pos != expected_pos:
raise RuntimeError("Wrong scan location visited!")
# Append xy position for path_history
self._path_history.append(xy_pos)
# And full x,y,z for imaged and foucsed if appropriate
if imaged:
self._imaged_locations.append(xyz_pos)
if focused:
self._focused_locations.append(xyz_pos)
class SmartSpiral(ScanPlanner):
"""
This is a smart spiral scan that spirals out from the centre.
Each time and image is taken the four neighbouring images are added
to the list of poisitions to image (unless they are already listed or
tried). However if a location is not imaged due no sample being detected
then neibouring positions are not imaged.
The next image taken is the closes to the centre (considering the largest
of vertical or horizontal distance), ties are broken by the distance from
the current position.
"""
_max_dist: int = 0
_dx: int = 0
_dy: int = 0
def _parse(self, planner_settings: Optional[dict] = None) -> None:
"""
Parse SmartSpiral Settings. This should be a dictionary
"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))
self._dx = int(planner_settings["dx"])
self._dy = int(planner_settings["dy"])
self._max_dist = int(planner_settings["max_dist"])
def _intial_location_list(self) -> XYPosList:
"""
Called on initalisation. Sets the initial list of locations for this scan planner
For smart spiral this is just the first point
"""
return [self._initial_position]
def mark_location_visited(
self, xyz_pos: XYZPos, imaged: bool = True, focused: bool = True
) -> None:
"""
Mark the location as visited. Adjust extra positions accordingly
Args:
xyz_pos: the x_y_z position
imaged: true if an image was taken, false if not (due to background detect)
focused: true if autofocus completed successfully
"""
# First call the base class to update the positions
super().mark_location_visited(xyz_pos, imaged, focused)
xy_pos = enforce_xy_tuple(xyz_pos[:2])
if imaged:
self._add_surrounding_positions(xy_pos)
self._re_sort_remaining_locations(xy_pos)
def _add_surrounding_positions(self, xy_pos: XYPos) -> None:
"""
This adds the surrounding (4 point connectivity) positions
to the remaining locations if they are not too far away or
already planned or already visited
"""
new_positions = [
(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),
]
for new_pos in new_positions:
# Skip position if already planned or visited
if self.position_planned(new_pos) or self.position_visited(new_pos):
continue
dist = distance_between(new_pos, self._initial_position)
if dist > self._max_dist:
LOGGER.debug("Rejected moving to %s as it is out of range", new_pos)
continue
self._remaining_locations.append(new_pos)
def _re_sort_remaining_locations(self, current_pos: XYPos) -> None:
"""
Sort the remaining positions besed on the current location
"""
# Defined rather than use a lambda for readability
def sort_key(pos):
return self.moves_from_centre(pos), distance_between(current_pos, pos)
self._remaining_locations.sort(key=sort_key)
def moves_from_centre(
self,
xy_pos: XYPos,
) -> float:
"""
Return the number of moves from the centre in the x or y direction
whichever is largest
Args:
xy_pos: the position
Note this has been renamed from `steps_from_centre` as that implied
stepper motor steps not number of moves in a scan
"""
move_size = np.array([self._dx, self._dy])
starting_pos = np.array(self._initial_position, dtype="float64")
current_pos = np.array(xy_pos, dtype="float64")
displacement_in_moves = (current_pos - starting_pos) / move_size
return np.max(np.abs(displacement_in_moves))
def distance_between(
current_pos: XYPos | np.ndarray, next_pos: XYPos | np.ndarray
) -> float:
"""
Calculate the distance between the two xy positions
This was previously called `distance_to_site`
"""
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))

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

@ -1,5 +1,3 @@
# ruff: noqa: E722
import shutil
import zipfile
import threading
@ -35,7 +33,7 @@ from openflexure_microscope_server.utilities import ErrorCapturingThread
from openflexure_microscope_server.things.autofocus import AutofocusThing
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
from openflexure_microscope_server.things.background_detect import BackgroundDetectThing
from openflexure_microscope_server import scan_planners
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
@ -44,27 +42,6 @@ BackgroundDep = direct_thing_client_dependency(
)
def closest(current, focused_path):
"""Finds the index of the closest x-y position in a list from the current position,
with ties split by the later element in the list (most recently taken)
must be float64 to deal with the huge numbers involved!"""
current_pos = np.array(current[:2], dtype="float64")
path_pos = np.asarray(focused_path, dtype="float64").T[:2].T
dist_2 = np.sqrt(
np.sum((path_pos - current_pos) ** 2, axis=1, dtype="float64"), dtype="float64"
)
min_dist = np.argmin(dist_2)
mask = np.where(dist_2 == dist_2[min_dist], 1, 0)
try:
closest = np.max(np.nonzero(mask))
except:
closest = 0
return closest
def unpack_autofocus(scan_data):
"""Extract z, sharpness data from a move_and_measure call
@ -89,8 +66,12 @@ def unpack_autofocus(scan_data):
return jpeg_heights[turning[0] : turning[1]], jpeg_sizes_mb[turning[0] : turning[1]]
def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit):
# limit is the largest ratio of change in z to change in xy that's allowed
def focus_change_acceptable(prev_pos, prev_z, new_pos, new_z, fractional_limit):
"""Return true if the change in z-position is small enough.
A large change in z-position in an indication of focus failure.
fractional_limit is the largest ratio of change in z to change in xy that's allowed
"""
prev_xy = np.asarray(prev_pos, dtype="float64")
new_xy = np.asarray(new_pos, dtype="float64")
@ -104,23 +85,9 @@ def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit):
else:
movement_ratio = np.divide(focus_change, dist, dtype="float64")
if movement_ratio > limit:
return "reject"
else:
return "accept"
def steps_from_centre(current_loc, starting_loc, dx, dy):
step_size = np.array([dx, dy])
return np.max(np.abs(np.divide(np.subtract(current_loc, starting_loc), step_size)))
def distance_to_site(current_pos, next_pos):
next_pos = np.array(next_pos, dtype="float64")
current_pos = np.array(current_pos, dtype="float64")
return np.sqrt(
(next_pos[1] - current_pos[1]) ** 2 + (next_pos[0] - current_pos[0]) ** 2
)
if movement_ratio > fractional_limit:
return False
return True
class NotEnoughFreeSpaceError(IOError):
@ -128,11 +95,21 @@ class NotEnoughFreeSpaceError(IOError):
def ensure_free_disk_space(path: str, min_space: int = 500000000) -> None:
"""Raise an exception if we are running out of disk space"""
"""
Raise an exception if we are running out of disk space
Args:
path = path to a location on the disk you want to check
min_space [int] = the minimum space required in bytes
default = 500,000,000 (500MiB)
Raises:
NotEnoughFreeSpaceError if the remaining storage is below min_space
"""
du = shutil.disk_usage(path)
if du.free < min_space:
raise NotEnoughFreeSpaceError(
"There is not enough free disk space to continue."
"There is not enough free disk space to continue. "
f"(Required: {min_space}, {du})."
)
@ -202,7 +179,10 @@ class SmartScanThing(Thing):
self._background_detect: Optional[BackgroundDep] = None
self._ongoing_scan_name: Optional[str] = None
self._starting_position: Optional[Mapping[str, int]] = None
self._capture_thread: Optional[ErrorCapturingThread] = None
self._scan_images_taken: Optional[int] = None
# TODO Scan data is a dict during refactoring, should become a dataclass
self._scan_data: Optional[dict] = None
@thing_action
def sample_scan(
@ -224,7 +204,6 @@ class SmartScanThing(Thing):
background_detect Thing).
"""
started_scan = False
got_lock = self._scan_lock.acquire(timeout=0.1)
if not got_lock:
raise RuntimeError("Trying to run scan while scan is already running!")
@ -238,22 +217,24 @@ class SmartScanThing(Thing):
self._metadata_getter = metadata_getter
self._csm = csm
self._background_detect = background_detect
self._capture_thread = None
self._scan_images_taken = 0
# Don't set self._scan_data dictionary. This is done at the start of _run_scan
try:
self._check_background_is_set()
self._check_background_and_csm_set()
self._ongoing_scan_name = self._get_unique_scan_name_and_dir(scan_name)
overlap = self.overlap
# record starting position so we can return there
self._starting_position = self._stage.position
started_scan = True
self._run_scan(scan_name, overlap)
self._run_scan()
except Exception as e:
if started_scan:
# If _scan_data is set then scan started
if self._scan_data is not None:
self._return_to_starting_position()
if not isinstance(e, NotEnoughFreeSpaceError):
# Don't stich if drive is full (already logged)
self._perform_final_stitch(overlap)
self._perform_final_stitch()
# Error must be raised so UI gives correct output
raise e
finally:
@ -266,16 +247,27 @@ class SmartScanThing(Thing):
self._metadata_getter = None
self._csm = None
self._background_detect = None
self._capture_thread = None
self._ongoing_scan_name = None
self._scan_images_taken = None
self._scan_data = None
self._scan_lock.release()
@_scan_running
def _check_background_is_set(self):
"""Before starting a scan check that we've got a background set
def _check_background_and_csm_set(self):
"""Before starting a scan check that background and camera-stage-mapping are set
Raise error if it is not set but background detect is being used.
Raise error if:
- background is to be skipped but is not set
- camera stage mapping is not set
Raise warning if not using background detect that scan will go on until max steps reached
"""
if self._csm.image_resolution is None:
raise RuntimeError(
"Camera-stage mapping is not calibrated. This is required before "
"scans can be carried out."
)
if self.skip_background:
if not self._background_detect.background_distributions:
@ -367,286 +359,162 @@ class SmartScanThing(Thing):
if not os.path.exists(trial_dir):
os.makedirs(trial_dir)
# If we made the directory this is the scan name
# Save it as the most latest scan (this persists as a
# Save the scan name as the latest scan (this persists as a
# property after the scan finishes)
self._latest_scan_name = trial_unique_scan_name
# Create images directory and
os.mkdir(self.images_dir_for_scan(trial_unique_scan_name))
# Return the scan name
return trial_unique_scan_name
raise FileExistsError("Could not create a new scan folder: all names in use!")
@_scan_running
def _move_to_next_point(
self,
path: list[list[int]],
focused_path: list[list[int]],
) -> list[int]:
"""Remove the first point from the path, and move there.
self, next_point: tuple[int, int], z_estimate: Optional[int] = None
) -> tuple[int, int, int]:
"""Move to the next position (half an autofocus move below estimated z)
This will move to the next XY position in `path`, taking the `z` value
either from the current z value of the stage, or from `focused_path`.
Moves the stage to the next poistion. If no z_estimate is given then
the current stage position is used.
Returns the point we have moved to.
Returns the (x,y,z) with the chosen z_estimate
"""
loc = [path[0][0], path[0][1]]
path.remove(path[0])
if len(focused_path) > 1:
z_index = closest(loc, focused_path)
z = int(focused_path[z_index][2])
else:
z = self._stage.position["z"]
self._scan_logger.info(f"Moving to {loc}")
if z_estimate is None:
z_estimate = self._stage.position["z"]
self._scan_logger.info(f"Moving to {next_point}")
self._stage.move_absolute(
x=int(loc[0]), y=int(loc[1]), z=z - self.autofocus_dz / 2
x=next_point[0],
y=next_point[1],
z=z_estimate - self._scan_data["autofocus_dz"] / 2,
)
return loc + [z]
return (next_point[0], next_point[1], z_estimate)
@_scan_running
def _run_scan(self, scan_name, overlap):
# Define these variables so we can use them in the finally: block
# (after testing they are not None)
def _take_test_image_to_calc_displacement(self, overlap):
"""
Take a test image and use camera stage mapping to calculate x and y displacement
Return (dx, dy) - the x and y displacments in steps
"""
test_jpg = self._cam.grab_jpeg()
test_image = np.array(Image.open(test_jpg.open()))
test_image_res = list(test_image.shape[:2])
csm_image_res = [int(i) for i in self._csm.image_resolution]
if test_image_res != csm_image_res:
raise RuntimeError(
"Cannot start scan as it is set up to capture with a resolution that "
"has not been mapped.\n"
f"Scan resolution: {test_image_res}\n"
f"camera-stage-mapping resolution {csm_image_res}."
)
# get displacement matrix. note it is for (y, x) not (x, y) coordinates
csm_disp_matrix = self._csm.image_to_stage_displacement_matrix
# Calculate displacements in image coordinates
dx_img = test_image.shape[1] * (1 - overlap)
dy_img = test_image.shape[0] * (1 - overlap)
# Calculate displacements in steps as vectors using a dot product with the matrix
dx_vec = np.dot(np.array([0, dx_img]), csm_disp_matrix)
dy_vec = np.dot(np.array([dy_img, 0]), csm_disp_matrix)
# Assume no rotation or skew and take only the aligned axis of vector.
# Cooerce to positive integer
dx = int(np.abs(dx_vec[0]))
dy = int(np.abs(dy_vec[1]))
return dx, dy
@_scan_running
def _set_scan_data(self):
"""
This sets the self._scan_data dictionary. This needs to become a
dataclass.
"""
overlap = self.overlap
dx, dy = self._take_test_image_to_calc_displacement(overlap)
self._scan_logger.info(
f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}"
)
autofocus_dz = self.autofocus_dz
if autofocus_dz == 0:
self._scan_logger.info("Running scan without autofocus")
elif autofocus_dz <= 200:
self._scan_logger.warning(
f"Your autofocus range is {autofocus_dz} steps, which is too short to "
"attempt to focus. Running without autofocus"
)
autofocus_dz = 0
# Fix scan parameters in case UI is updated during scan.
self._scan_data = {
"scan_name": self._ongoing_scan_name,
"overlap": overlap,
"max_dist": self.max_range,
"dx": dx,
"dy": dy,
"autofocus_dz": autofocus_dz,
"autofocus_on": bool(autofocus_dz),
"start_time": time.strftime("%H_%M_%S-%d_%m_%Y"),
"skip_background": self.skip_background,
"stitch_automatically": self.stitch_automatically,
}
@_scan_running
def _save_scan_inputs_json(self):
# This should be a method of the scan_data dataclass
data = {
"scan_name": self._ongoing_scan_name,
"overlap": self._scan_data["overlap"],
"autofocus range": self._scan_data["autofocus_dz"],
"dx": self._scan_data["dx"],
"dy": self._scan_data["dy"],
"start time": self._scan_data["start_time"],
"skipping background": self._scan_data["skip_background"],
}
scan_inputs_fname = os.path.join(
self._ongoing_scan_images_dir, "scan_inputs.json"
)
with open(scan_inputs_fname, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
@_scan_running
def _manage_stitching_threads(self):
"""
Manage the stitching threads starting them if the are needed
and not running.
"""
if self._scan_images_taken > 3:
if not self._preview_stitch_running():
self._preview_stitch_start()
if self._scan_data["stitch_automatically"]:
if not self._correlate_running():
self._correlate_start(overlap=self._scan_data["overlap"])
@_scan_running
def _run_scan(self):
# Uset to check if finally was reached via exeption (except
# cancel by user.
scan_successful = True
capture_thread = None
try:
os.mkdir(self._ongoing_scan_images_dir)
self._scan_logger.info(f"Saving images to {self._ongoing_scan_images_dir}")
max_dist = self.max_range
if self.autofocus_dz == 0:
self._scan_logger.info("Running scan without autofocus")
elif self.autofocus_dz <= 200:
self._scan_logger.warning(
f"Your dz range is {self.autofocus_dz} steps, which is too short to attempt to focus. Running without autofocus"
)
names = []
positions = []
r = self._cam.grab_jpeg()
arr = np.array(Image.open(r.open()))
if self._csm.image_resolution is None:
raise RuntimeError(
"Camera-stage mapping is not calibrated. This is required before "
"scans can be carried out."
)
if list(arr.shape[:2]) != [int(i) for i in self._csm.image_resolution]:
self._scan_logger.error(
f"Images are, by default, {arr.shape[:2]}, but the CSM was "
f"calibrated at {self._csm.image_resolution}."
)
# Here, we calculate the x and y step size based on the desired overlap
# TODO: Consider using CSM calibration size instead
# TODO: generalise to have 2D displacements for x and y (as the
# camera and stage may not be aligned).
csm_disp_matrix = self._csm.image_to_stage_displacement_matrix
dx = int(
np.abs(
np.dot(
np.array([0, arr.shape[1] * (1 - overlap)]), csm_disp_matrix
)[0]
)
)
dy = int(
np.abs(
np.dot(
np.array([arr.shape[0] * (1 - overlap), 0]), csm_disp_matrix
)[1]
)
)
self._scan_logger.info(
f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}"
)
# construct a 2D scan path
path = [[self._stage.position["x"], self._stage.position["y"]]]
focused_path = [] # This holds a list of all points where focus succeeded
true_path = [] # This holds a list of all points visited
start_time = time.strftime("%H_%M_%S-%d_%m_%Y")
data = {
"scan_name": scan_name,
"overlap": overlap,
"autofocus range": self.autofocus_dz,
"dx": dx,
"dy": dy,
"start time": start_time,
"skipping background": self.skip_background,
}
scan_inputs_fname = os.path.join(
self._ongoing_scan_images_dir, "scan_inputs.json"
)
with open(scan_inputs_fname, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
self._set_scan_data()
self._save_scan_inputs_json()
if self._scan_images_taken != 0:
raise RuntimeError(
"_scan_images_taken should be zero before starting scanning"
)
# At the start of the loop, we simultaneously capture an image and move to the next scan point.
# We skip capturing on the first run, because we've not focused yet - and also we skip capturing if
# it looks like background.
while len(path) > 0:
loc = self._move_to_next_point(path=path, focused_path=focused_path)
msg = "_scan_images_taken should be zero before starting scanning"
raise RuntimeError(msg)
if self._scan_images_taken > 3:
if not self._preview_stitch_running():
self._preview_stitch_start()
if self.stitch_automatically:
if not self._correlate_running():
self._correlate_start(overlap=overlap)
ensure_free_disk_space(self._ongoing_scan_dir)
# Check if the image is background
if self.skip_background:
image_is_sample = self._background_detect.image_is_sample()
else:
image_is_sample = True
# if more than 92% of the image is background, treat it as background and continue
if not image_is_sample:
self._scan_logger.info(
f"Skipping {self._stage.position} as it is {round(self._background_detect.background_fraction(), 0)}% background."
)
else:
# if not, it's sample. run an autofocus and use the updated height
new_pos = [
[self._stage.position["x"] - dx, self._stage.position["y"]],
[self._stage.position["x"] + dx, self._stage.position["y"]],
[self._stage.position["x"], self._stage.position["y"] - dy],
[self._stage.position["x"], self._stage.position["y"] + dy],
]
for pos in new_pos:
if (
pos not in [sublist[:2] for sublist in true_path]
and pos not in path
):
path.append(pos)
attempts = 0
if self.autofocus_dz > 200:
while True:
jpeg_zs, jpeg_sizes = self._autofocus.looping_autofocus(
dz=self.autofocus_dz, start="base"
)
current_height = self._stage.position["z"]
time.sleep(0.2)
autofocus_success = self._autofocus.verify_focus_sharpness(
sweep_sizes=jpeg_sizes, camera=CamDep, threshold=0.92
)
self._scan_logger.info(
f"We just tested the focus! Result was {autofocus_success}"
)
if autofocus_success:
# if there have been successful autofocuses in this scan, find the closest one in x-y
# test if the change in z between them exceeds a ratio (indicating a failed autofocus)
if len(focused_path) > 0:
nearest_focused_site = focused_path[
closest(loc, focused_path)
]
result = limit_focus_change(
nearest_focused_site[0:2],
nearest_focused_site[-1],
loc[0:2],
current_height,
0.5,
)
# if there haven't been any previous autofocuses, we have to assume this one worked
else:
result = "accept"
else:
result = "reject"
# if the autofocus worked, add the current position to the list of successful locations
if result == "accept":
loc = list(self._stage.position.values())
focused_path.append(loc)
break
if attempts >= 3:
self._scan_logger.warning(
"Could not autofocus after 3 attempts."
)
break
# if the autofocus was rejected, we return to the height of the closest successful autofocus. not perfect, but better than wandering out of focus
self._scan_logger.info(
"The focus has shifted further than we expect: retrying."
)
self._stage.move_absolute(z=int(loc[2]))
attempts += 1
# Acquire the image in a thread, and continue once it's acquired (i.e. leave saving in the background)
if capture_thread: # wait for the previous capture to be saved, i.e. don't leave more than one image saving in the background
wait_start = time.time()
thread_was_alive = capture_thread.is_alive()
# If the capture thread has thrown an exception it will be raised when join is called,
# this will cause the scan to end. If we want to retry captures at a later date
# this is where we will need to catch the IOError or CaptureError from the thread.
capture_thread.join()
time.sleep(0.2)
if thread_was_alive:
wait_time = time.time() - wait_start
self._scan_logger.info(
f"Waited {wait_time:.1f}s for the previous capture to finish saving."
)
# increment capure counter as thread has completed
self._scan_images_taken += 1
acquired = Event()
name = f"image_{loc[0]}_{loc[1]}.jpg"
jpeg_path = os.path.join(self._ongoing_scan_images_dir, name)
time.sleep(0.2)
# Use ErrorCapturingThread intead of Thread. This will raise errors in the calling
# thread only when join() is called, allowing us to handle this appropriately.
capture_thread = ErrorCapturingThread(
target=self._capture_and_save,
kwargs={
"acquired": acquired,
"jpeg_path": jpeg_path,
},
)
capture_thread.start()
acquired.wait() # wait until the image is acquired
positions.append(loc[:2])
names.append(name)
# add the current position to the list of all positions visited
true_path.append(loc)
temp_path = []
for i in path:
if distance_to_site(i, true_path[0][:2]) < max_dist:
temp_path.append(i)
else:
self._scan_logger.info(
f"Rejected moving to {i} as it is out of range"
)
path = temp_path.copy()
path = sorted(
path,
key=lambda x: (
steps_from_centre(x, true_path[0][:2], dx, dy),
distance_to_site(loc[:2], x),
),
)
self.create_zip_of_scan(
logger=self._scan_logger,
scan_name=self._ongoing_scan_name,
download_zip=False,
)
# This is the main loop of the scan!
self._main_scan_loop()
except InvocationCancelledError:
scan_successful = False
@ -667,26 +535,201 @@ class SmartScanThing(Thing):
)
raise e
finally:
if capture_thread:
if self._capture_thread:
# If the capture thread had an error we capture it here
try:
capture_thread.join()
self._capture_thread.join()
except Exception as e:
# If the thread has already ended due to an exception we will
# If the scan has already ended due to an exception we will
# ignore any exceptions, but if it appeared to be successful
# we will log an error.
# we will log the error.
if scan_successful:
self._scan_logger.error(
"The scan appears to have started successfully, however the final capture raised"
f"the following error: {e}."
"The scan appears to have started successfully, however "
f"the final capture raised the following error: {e}."
"Attempting to stitch and archive images.",
exc_info=e,
)
# This is what happens if the scan completes sucessfully or the
# This is what happens if the scan completes successfully or the
# user cancels it.
self._return_to_starting_position()
self._perform_final_stitch(overlap)
self._perform_final_stitch()
@_scan_running
def _main_scan_loop(self):
planner_settings = {
"dx": self._scan_data["dx"],
"dy": self._scan_data["dy"],
"max_dist": self._scan_data["max_dist"],
}
route_planner = scan_planners.SmartSpiral(
intial_position=(self._stage.position["x"], self._stage.position["y"]),
planner_settings=planner_settings,
)
# At the start of the loop, we simultaneously capture an image and move to the next scan point.
# We skip capturing on the first run, because we've not focused yet - and also we skip capturing if
# it looks like background.
while not route_planner.scan_complete:
ensure_free_disk_space(self._ongoing_scan_dir)
self._manage_stitching_threads()
next_pos_xy, z_est = route_planner.get_next_location_and_z_estimate()
new_pos_xyz = self._move_to_next_point(next_pos_xy, z_est)
capture_image = True
# If skipping background, take an image to check if it is background
if self._scan_data["skip_background"]:
capture_image = self._background_detect.image_is_sample()
if not capture_image:
route_planner.mark_location_visited(
new_pos_xyz, imaged=False, focused=False
)
# Background fraction is actually a percentage
back_perc = round(self._background_detect.background_fraction(), 0)
msg = f"Skipping {new_pos_xyz} as it is {back_perc}% background."
self._scan_logger.info(msg)
continue
focused = False
if self._scan_data["autofocus_on"]:
closest_xyz = route_planner.closest_focus_site(new_pos_xyz[:2])
focused = self._try_autofocus(new_pos_xyz, closest_xyz)
route_planner.mark_location_visited(
new_pos_xyz, imaged=True, focused=focused
)
# wait for the previous capture to be saved, i.e. don't leave more than one image saving in the background
if self._capture_thread:
self._wait_for_capture_thread()
# increment capure counter as thread has completed
self._scan_images_taken += 1
# Add it to the incremental zip
self.create_zip_of_scan(
logger=self._scan_logger,
scan_name=self._ongoing_scan_name,
download_zip=False,
)
name = f"image_{new_pos_xyz[0]}_{new_pos_xyz[1]}.jpg"
jpeg_path = os.path.join(self._ongoing_scan_images_dir, name)
self._capture_thread, acquired = self._start_capture_thread(jpeg_path)
# wait until the image is acquired
acquired.wait()
@_scan_running
def _try_autofocus(
self,
this_xyz: tuple[int, int, int],
closest_xyz: Optional[tuple[int, int, int]],
) -> bool:
"""
Try to perform autofocus and return boolean for if successful
Args:
this_xyz is the current x,y,z position.
closest_xyz is the (x, y, z) coordinates of the closest position, this is None
if no previous images have been taken or in focus
Return True on successful autofocus.
Return False if failed after 3 tries - the position will be the initial estimate
"""
attempts = 0
max_attempts = 3
dz = self._scan_data["autofocus_dz"]
while attempts < max_attempts:
attempts += 1
_, jpeg_sizes = self._autofocus.looping_autofocus(dz=dz, start="base")
current_height = self._stage.position["z"]
time.sleep(0.2)
autofocus_sharp_enough = self._autofocus.verify_focus_sharpness(
sweep_sizes=jpeg_sizes, camera=CamDep, threshold=0.92
)
# Not sharp enough, go to start and try again
if not autofocus_sharp_enough:
z = this_xyz[2] - dz / 2 if attempts < max_attempts else this_xyz[2]
self._stage.move_absolute(z=z)
continue
# No previous positions to compare against return success
if closest_xyz is None:
return True
# Check the change in z-position is acceptable
# If the z change compared to the closest focused image exceeds
# a given fraction of the xy displacement this indicates failure
success = focus_change_acceptable(
prev_pos=closest_xyz[:2],
prev_z=closest_xyz[2],
new_pos=this_xyz[:2],
new_z=current_height,
fractional_limit=0.5,
)
# No focus change acceptable return success
if success:
return True
# Shifted to far move to start and try again
z = this_xyz[2] - dz / 2 if attempts < max_attempts else this_xyz[2]
self._stage.move_absolute(z=z)
self._scan_logger.info(
"The focus has shifted further than we expect: retrying."
)
self._scan_logger.warning("Could not autofocus after 3 attempts.")
return False
@_scan_running
def _wait_for_capture_thread(self) -> None:
"""
Wait for the capture thread to be complete.
"""
wait_start = time.time()
thread_was_alive = self._capture_thread.is_alive()
# If the capture thread has thrown an exception it will be raised
# when join is called, this will cause the scan to end. If we want
# to retry captures at a later date this is where we will need to
# catch the IOError or CaptureError from the thread.
self._capture_thread.join()
time.sleep(0.2)
if thread_was_alive:
wait_time = time.time() - wait_start
self._scan_logger.info(
f"Waited {wait_time:.1f}s for the previous capture to finish saving."
)
@_scan_running
def _start_capture_thread(
self, jpeg_path: str
) -> tuple[ErrorCapturingThread, Event]:
"""
Start the capture thread.
Args:
jpeg_path, the path to save the image once aquired
Return the thread and an event that will be set when the image is aquired
"""
acquired = Event()
time.sleep(0.2)
# Acquire the image in a thread, and continue once it's acquired
# (i.e. leave saving in the background) Use ErrorCapturingThread
# intead of Thread. This will raise errors in the calling thread
# only when join() is called, allowing us to handle this appropriately.
capture_thread = ErrorCapturingThread(
target=self._capture_and_save,
kwargs={"acquired": acquired, "jpeg_path": jpeg_path},
)
capture_thread.start()
return capture_thread, acquired
@_scan_running
def _return_to_starting_position(self):
@ -697,7 +740,7 @@ class SmartScanThing(Thing):
)
@_scan_running
def _perform_final_stitch(self, overlap):
def _perform_final_stitch(self):
"""Perform final stitch of the data"""
if self._scan_images_taken <= 3:
@ -714,12 +757,12 @@ class SmartScanThing(Thing):
self._preview_stitch_wait()
self._correlate_wait()
try:
if self.stitch_automatically:
if self._scan_data["stitch_automatically"]:
self._scan_logger.info("Stitching final image (may take some time)...")
self.stitch_scan(
logger=self._scan_logger,
scan_name=self._ongoing_scan_name,
overlap=overlap,
overlap=self._scan_data["overlap"],
)
except SubprocessError as e:
self._scan_logger.error(f"Stitching failed: {e}", exc_info=e)
@ -786,7 +829,9 @@ class SmartScanThing(Thing):
metadata
).encode("utf-8")
piexif.insert(piexif.dump(exif_dict), jpeg_path)
except:
except: # noqa: E722
# We need to capture any exception as there are many reasons metadata
# might not be added. We warn rather than log the error.
self._scan_logger.warning(f"Failed to add metadata to {jpeg_path}")
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e
@ -1063,30 +1108,45 @@ class SmartScanThing(Thing):
logger: InvocationLogger,
cmd: list[str],
) -> CompletedProcess:
"""Run a subprocess and log any output"""
"""
Run a subprocess and log any output
Raises:
ChildProcessError if exit code is not zero
"""
logger.info(f"Running command in subprocess: `{' '.join(cmd)}`")
p = Popen(cmd, stdout=PIPE, stderr=STDOUT, bufsize=1, universal_newlines=True)
os.set_blocking(p.stdout.fileno(), False)
def log_buffer(buffer):
"""A short internal function to read everything in the buffer to
a multiline string and log"""
lines = []
while line := buffer.readline():
lines.append(line)
if lines:
logger.info("".join(lines))
# Run the command piping stdout into the process for reading and
# forwarding the stdrerr to stdout
process = Popen(
cmd, stdout=PIPE, stderr=STDOUT, bufsize=1, universal_newlines=True
)
# Stop opening pipe blocking writing to it
os.set_blocking(process.stdout.fileno(), False)
logger.info(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())))
while p.poll() is None:
try:
output = p.stdout.readline()
if output != "" and output is not None:
logger.info(output)
except:
pass
for line in p.stdout:
try:
output = p.stdout.readline()
if output != "" and output is not None:
logger.info(output)
except:
pass
# Poll returns None while running, will return the error code when finnished
while process.poll() is None:
log_buffer(process.stdout)
# Once buffer is clear sleep for 0.2s before trying again.
time.sleep(0.2)
logger.info("Stitching complete")
return p
# Print everything in the buffer when program finishes
log_buffer(process.stdout)
if process.poll() == 0:
logger.info("Stitching complete")
else:
raise ChildProcessError(f"Subprocess {cmd[0]} exited with an error.")
@thing_action
def stitch_scan(
@ -1100,7 +1160,9 @@ class SmartScanThing(Thing):
Note that as this is a thing_action it needs the logger passed as
a variable if called from another thing action
"""
json_fname = "scan_inputs.json"
images_folder = self.images_dir_for_scan(scan_name=scan_name)
json_fpath = os.path.join(images_folder, json_fname)
if self.stitch_tiff:
tiff_arg = "--stitch_tiff"
@ -1109,11 +1171,24 @@ class SmartScanThing(Thing):
if overlap == 0.0:
try:
with open(os.path.join(images_folder, "scan_inputs.json")) as data_file:
with open(json_fpath, "r", encoding="utf-8") as data_file:
data_loaded = json.load(data_file)
logger.info(data_loaded)
overlap = data_loaded["overlap"]
except:
except (json.decoder.JSONDecodeError, FileNotFoundError, TypeError):
# As there is no schema or pydantic model this should handle
# The file not being there, it not being json in the file,
# or the imported data not being indexable
logger.warning(
f"Couldn't read scan data, is {json_fname} missing or corrupt? "
"Attempting stitch with overlap value of 0.1"
)
overlap = 0.1
except KeyError:
logger.warning(
"Value for overlap not found in scan data. "
"Attempting stitch with overlap value of 0.1"
)
overlap = 0.1
self.run_subprocess(
logger,

0
tests/__init__.py Normal file
View file

259
tests/test_scan_planners.py Normal file
View file

@ -0,0 +1,259 @@
import pytest
from copy import copy
from openflexure_microscope_server import scan_planners
from .utilities import scan_test_helpers
def test_enforce_xy_tuple():
bad_len_vals = [[1], [], (1,), (2, 4, 4), [1, 4, 5, 7]]
bad_type_vals = ["hi", 1, {"this": "that"}, {1, 2}]
for value in bad_len_vals + bad_type_vals:
with pytest.raises(ValueError):
scan_planners.enforce_xy_tuple(value)
assert (1, 6) == scan_planners.enforce_xy_tuple((1, 6))
assert (1, 6) == scan_planners.enforce_xy_tuple([1, 6])
def test_enforce_xyz_tuple():
bad_len_vals = [[1], [], (1,), (2, 4), [1, 4, 5, 7]]
bad_type_vals = ["hi!", 1, {"this": "that"}, {1, 2, 3}]
for value in bad_len_vals + bad_type_vals:
with pytest.raises(ValueError):
scan_planners.enforce_xyz_tuple(value)
assert (1, 6, 2) == scan_planners.enforce_xyz_tuple((1, 6, 2))
assert (1, 6, 6) == scan_planners.enforce_xyz_tuple([1, 6, 6])
def test_base_class_not_implemented():
intial_position = (100, 50)
with pytest.raises(NotImplementedError):
scan_planners.ScanPlanner(intial_position=intial_position)
def test_v_basic_smart_spiral():
intial_position = (100, 50)
planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000}
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
# Create a planner. It shouldn't be complete.
assert not planner.scan_complete
# When we start it should want to stay in the inital pos and have
# no z_estimate
xy_pos, z_pos = planner.get_next_location_and_z_estimate()
assert xy_pos == intial_position
assert z_pos is None
# Try to mark location as imaged with only xy_position
with pytest.raises(ValueError):
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_bad_smart_spiral_settings():
intial_position = (100, 50)
# Class init should raise error if no planner_settings dictionary set
with pytest.raises(ValueError):
scan_planners.SmartSpiral(intial_position=intial_position)
planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000}
keys = ["dx", "dy", "max_dist"]
# Class init should raise error if planner_settings is missing any key
for delkey in keys:
bad_planner_settings = copy(planner_settings)
del bad_planner_settings[delkey]
with pytest.raises(KeyError):
scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=bad_planner_settings
)
# Class init should raise error if planner_settings if any value can't be cast
# to int
keys = ["dx", "dy", "max_dist"]
for badkey in keys:
bad_planner_settings = copy(planner_settings)
bad_planner_settings[badkey] = "I can't be converted to an int"
with pytest.raises(ValueError):
scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=bad_planner_settings
)
def test_smart_spiral_first_few_pos():
"""
This test is VERY long, not really a "unit". It checks step-by-step
that data is added correctly for the first few postions in a scan.
This should catch basic cases of if the algorithm is updated.
"""
intial_position = (100, 50)
planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000}
# Create a planner
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
# it shouldn't start complete
assert not planner.scan_complete
# When we start it should want to stay in the inital pos and have
# no z_estimate
xy_pos1, z_pos1 = planner.get_next_location_and_z_estimate()
assert xy_pos1 == intial_position
assert z_pos1 is None
# Set a focus value
z_focus = 10
xyz_pos1 = (xy_pos1[0], xy_pos1[1], z_focus)
# if we mark this position as visited, imaged, and focused
planner.mark_location_visited(xyz_pos1, imaged=True, focused=True)
# scan is not complete
assert not planner.scan_complete
# Lists should have updated to add the position to histories
assert xyz_pos1 in planner.imaged_locations
assert xyz_pos1 in planner.focused_locations
assert xy_pos1 in planner.path_history
# remove from planned
assert xy_pos1 not in planner.remaining_locations
# Add 4 new points to planned
assert (100, 0) in planner.remaining_locations
assert (100, 100) in planner.remaining_locations
assert (150, 50) in planner.remaining_locations
assert (50, 50) in planner.remaining_locations
assert len(planner.remaining_locations) == 4
# Now the next location is updated
xy_pos2, z_pos2 = planner.get_next_location_and_z_estimate()
# move in negative x-dir first
assert xy_pos2 == (50, 50)
# Focus set from closest point
assert z_pos2 is z_focus
xyz_pos2 = (xy_pos2[0], xy_pos2[1], z_focus)
# if we mark this position as visited, imaged, and NOT focused
planner.mark_location_visited(xyz_pos2, imaged=True, focused=False)
# Check this position remove from planned
assert xy_pos2 not in planner.remaining_locations
# Check original position not re-added
assert xy_pos1 not in planner.remaining_locations
# 3 old points still planned
assert (100, 0) in planner.remaining_locations
assert (100, 100) in planner.remaining_locations
assert (150, 50) in planner.remaining_locations
# Add 3 new points to planned
assert (0, 50) in planner.remaining_locations
assert (50, 100) in planner.remaining_locations
assert (50, 0) in planner.remaining_locations
assert len(planner.remaining_locations) == 6
# Now the next location is updated
xy_pos3, z_pos3 = planner.get_next_location_and_z_estimate()
# move in negative y-dir first next
assert xy_pos3 == (50, 0)
# Focus still set as before
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
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)
# ... 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 most recent point as this was focussed
assert z_pos4 is new_z_focus
assert planner.closest_focus_site(xy_pos4) == xyz_pos3
def test_smart_spiral_stops_on_max_dist():
intial_position = (0, 0)
planner_settings = {"dx": 100, "dy": 100, "max_dist": 1000}
# Create a planner
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
while not planner.scan_complete:
xy_pos, _ = planner.get_next_location_and_z_estimate()
xyz_pos = (xy_pos[0], xy_pos[1], 0)
planner.mark_location_visited(xyz_pos, imaged=True, focused=True)
# Estimate of radius = 10 scans, pir^2 = 314 images. In reality it works out as
# 317
assert len(planner.path_history) == 317
def test_mark_wrong_location():
"""
This test is VERY long, not really a "unit". It checks step-by-step
that data is added correctly for the first few postions in a scan.
This should catch basic caseses of if the algorithm is updated.
"""
intial_position = (100, 50)
planner_settings = {"dx": 50, "dy": 50, "max_dist": 10000}
# Create a planner
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
xy_pos, _ = planner.get_next_location_and_z_estimate()
# Trying to mark the wrong location as visited raises error
wrong_xyz_pos = (xy_pos[0] + 1, xy_pos[1], 0)
with pytest.raises(RuntimeError):
planner.mark_location_visited(wrong_xyz_pos, imaged=True, focused=True)
def test_closest_focus_wth_large_numbers():
"""
The number of steps gets very large in reality runs some tests to check
that everything works well with huge numbers of steps
"""
intial_position = (0, 0)
# Set this up, but we won't use the settings
planner_settings = {"dx": 10000, "dy": 10000, "max_dist": 100000}
# Create a planner
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
# Directly overwrite the private focussed locations list for test
# For two points 1m points away it should choose the last as they are equal
planner._focused_locations = [(1000000, 0, 0), (0, 1000000, 0)]
assert planner.closest_focus_site((0, 0)) == (0, 1000000, 0)
# Try similar
planner._focused_locations = [(1234567, 0, 0), (-1234567, 0, 0)]
assert planner.closest_focus_site((0, 0)) == (-1234567, 0, 0)
# Make the first point 1 step closer
planner._focused_locations = [(1234566, 0, 0), (-1234567, 0, 0)]
assert planner.closest_focus_site((0, 0)) == (1234566, 0, 0)
def test_example_smart_spiral():
_, planner = scan_test_helpers.example_smart_spiral()
expected_planner = scan_test_helpers.get_expected_result_for_example_smart_spiral()
assert planner.path_history == expected_planner.path_history
assert planner.imaged_locations == expected_planner.imaged_locations

3
tests/utilities/__init__.py Normal file
View file

@ -0,0 +1,3 @@
"""
This directory contains utitlities that help with testing and debugging
"""

BIN
tests/utilities/example_smart_spiral.pkl Normal file
View file

Binary file not shown.

196
tests/utilities/scan_test_helpers.py Normal file
View file

@ -0,0 +1,196 @@
import os
import pickle
import numpy as np
from scipy import interpolate
from matplotlib import pyplot as plt
from matplotlib.path import Path as MatPath
from matplotlib.patches import PathPatch
from matplotlib.figure import Figure
from openflexure_microscope_server import scan_planners
THIS_DIR = os.path.dirname(os.path.realpath(__file__))
class FakeSample:
"""
A fake sample to test scan algorithms. The sample is able to return
whether a given position is sample, no image associated with the sample
"""
def __init__(self, xy_points: list[tuple[int, int]]):
"""
Create the sample from a spline interpolation around
the given points.
"""
self._sample_perimeter = interp_closed_path(xy_points, 500)
def is_sample(self, pos: tuple[int, int], im_size: tuple[int, int]) -> bool:
"""
Return whether an image at a given location with a given image size
is on the sample
This doesn't check the entire image field as this is designed to be used
where the fake sample is much larger than the image and has smooth edges
It just checks the 4 corners
"""
img_corners = [
(pos[0] + im_size[0], pos[1] + im_size[1]),
(pos[0] + im_size[0], pos[1] - im_size[1]),
(pos[0] - im_size[0], pos[1] + im_size[1]),
(pos[0] - im_size[0], pos[1] - im_size[1]),
]
return any(
self._sample_perimeter.contains_point(corner) for corner in img_corners
)
@property
def patch(self) -> PathPatch:
"""
The sample as a matplotlib patch for plotting
"""
patch = PathPatch(self._sample_perimeter)
patch.set(color=(1.0, 0.8, 1.0, 1.0))
return patch
def visualise_scan(sample: FakeSample, planner: scan_planners.ScanPlanner) -> Figure:
"""
For a given sample and scanner object return a matplotlib figure of the scan
"""
fig, ax = plt.subplots(figsize=(8, 8))
ax.add_artist(sample.patch)
xh, yh = zip(*planner._path_history)
xi, yi, _ = zip(*planner._imaged_locations)
# convert history to numpy array so can calculate quiver arrows
xh = np.array(xh)
yh = np.array(yh)
plt.quiver(
xh[:-1],
yh[:-1],
xh[1:] - xh[:-1],
yh[1:] - yh[:-1],
scale_units="xy",
angles="xy",
scale=1,
)
plt.plot(xh, yh, "r.")
plt.plot(xi, yi, "g*")
ax.axis("equal")
return fig
def interp_closed_path(xy_points: list[tuple[int, int]], n_points: int) -> MatPath:
"""
Given a lists of xy_points interpolate an n_point closed curve. This can be used
to create an arbitrary sample shape plan a scan.
Modified from:
https://stackoverflow.com/questions/33962717/interpolating-a-closed-curve-using-scipy
"""
# Use zip to seperate x and y points into tuples
x, y = zip(*xy_points)
# Append first point and convert to array
x = np.array(x + (x[0],))
y = np.array(y + (y[0],))
# fit splines to x=f(u) and y=g(u), treating both as periodic. also note that s=0
# is needed in order to force the spline fit to pass through all the input points.
spline_data, _ = interpolate.splprep([x, y], s=0, per=True)
# evaluate the spline
xi, yi = interpolate.splev(np.linspace(0, 1, n_points), spline_data)
# Convert to a matplotlib closed path
path_points = [[xp, yp] for xp, yp in (zip(xi, yi))]
return MatPath(path_points, closed=True)
def example_smart_spiral() -> tuple[FakeSample, scan_planners.ScanPlanner]:
"""
Run an example scan and return the sample scanned and the planner object
after scan is complete
"""
xy_sample_points = [
(-5000, -5000),
(-2000, 10000),
(1000, 2000),
(6000, 7000),
(9000, 2000),
]
sample = FakeSample(xy_sample_points)
img_size = (1000, 1000)
intial_position = (0, 0)
planner_settings = {"dx": 700, "dy": 700, "max_dist": 100000}
planner = scan_planners.SmartSpiral(
intial_position=intial_position, planner_settings=planner_settings
)
while not planner.scan_complete:
xy_pos, _ = planner.get_next_location_and_z_estimate()
xyz_pos = (xy_pos[0], xy_pos[1], 0)
imaged = sample.is_sample(xy_pos, img_size)
planner.mark_location_visited(xyz_pos, imaged=imaged, focused=imaged)
return sample, planner
def profile_and_save_plot_for_example_smart_spiral():
"""
Run the example scan and save a plot and the profile data
Also print the cumulative stats
This runs if you run this file directly
"""
import pstats
import cProfile
profiler = cProfile.Profile()
sample, planner = profiler.runcall(example_smart_spiral)
stats_fname = os.path.join(THIS_DIR, "scan_example_stats.pstats")
profiler.dump_stats(stats_fname)
png_fname = os.path.join(THIS_DIR, "scan_example_plot.png")
fig = visualise_scan(sample, planner)
fig.savefig(png_fname, dpi=200)
run_stats = pstats.Stats(stats_fname)
run_stats.strip_dirs()
run_stats.sort_stats("cumulative")
run_stats.print_stats("scan_planners.py")
def update_example_smart_spiral_pickle():
"""
Pickle the ScanPlanner for the example_smart_spiral(),
this is done so the history can be compared by testing to check
the algorithm is unchanged.
If the algorithm is purposefully changed then this will need to be
run to update the pickle for the test to pass.
"""
pkl_fname = os.path.join(THIS_DIR, "example_smart_spiral.pkl")
with open(pkl_fname, "wb") as pkl_file_obj:
_, planner = example_smart_spiral()
pickle.dump(planner, pkl_file_obj, pickle.HIGHEST_PROTOCOL)
def get_expected_result_for_example_smart_spiral():
"""
Return the expected ScanPlanner object for the example_smart_spiral(),
this is pickled, so that it can be committed.
"""
pkl_fname = os.path.join(THIS_DIR, "example_smart_spiral.pkl")
with open(pkl_fname, "rb") as pkl_file_obj:
planner = pickle.load(pkl_file_obj)
return planner
if __name__ == "__main__":
profile_and_save_plot_for_example_smart_spiral()