diff --git a/.gitignore b/.gitignore index 02c38a88..4f6925c9 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 747efe26..53ec72fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/ruff-increased-checking.toml b/ruff-increased-checking.toml index 8de47661..118cfbff 100644 --- a/ruff-increased-checking.toml +++ b/ruff-increased-checking.toml @@ -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 diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py new file mode 100644 index 00000000..75ffd0fb --- /dev/null +++ b/src/openflexure_microscope_server/scan_planners.py @@ -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)) diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index fbf116e5..5d853b69 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -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, diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/test_scan_planners.py b/tests/test_scan_planners.py new file mode 100644 index 00000000..da5ff961 --- /dev/null +++ b/tests/test_scan_planners.py @@ -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 diff --git a/tests/utilities/__init__.py b/tests/utilities/__init__.py new file mode 100644 index 00000000..275ec31c --- /dev/null +++ b/tests/utilities/__init__.py @@ -0,0 +1,3 @@ +""" +This directory contains utitlities that help with testing and debugging +""" diff --git a/tests/utilities/example_smart_spiral.pkl b/tests/utilities/example_smart_spiral.pkl new file mode 100644 index 00000000..5d5727d8 Binary files /dev/null and b/tests/utilities/example_smart_spiral.pkl differ diff --git a/tests/utilities/scan_test_helpers.py b/tests/utilities/scan_test_helpers.py new file mode 100644 index 00000000..6901361a --- /dev/null +++ b/tests/utilities/scan_test_helpers.py @@ -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()