diff --git a/src/openflexure_microscope_server/scan_planners.py b/src/openflexure_microscope_server/scan_planners.py index 7387e70d..2248a55c 100644 --- a/src/openflexure_microscope_server/scan_planners.py +++ b/src/openflexure_microscope_server/scan_planners.py @@ -5,7 +5,11 @@ is only one type the SmartSpiral. More can be added using by subclassing the ScanPlanner """ -from typing import TypeAlias, Optional +# Future annotations needed for typhinting same class in __eq__ method. Other option +# would be to import Union and use a string. +from __future__ import annotations + +from typing import TypeAlias, Optional, Any import logging from copy import copy @@ -51,6 +55,99 @@ def enforce_xyz_tuple(value: XYZPos) -> XYZPos: return value +# Disable PLW1641, this warns against hash not being set when equals is. But the +# class shouldn't be hashable. +class FutureScanLocation: # noqa PLW1641 + """Data for information on future locations to scan. + + This object is only used for internal calculation and data storage it shouldn't + be an input or output of public methods. + """ + + def __init__(self, xy_pos: XYPos, **kwargs: Any) -> None: + """Initialise FutureScanLocation with an xy position. + + :param xy_pos: The (x, y) position to scan. + :param kwargs: Any other information about the location. This will be passed + through to the VisitedScanLocation object. + """ + self._xy_pos = enforce_xy_tuple(xy_pos) + self.planner_data = copy(kwargs) + + @property + def xy_tuple(self) -> XYPos: + """The xy position tuple.""" + return self._xy_pos + + def __eq__(self, other: Any) -> bool: + """Check for equality, only checks the xy position. + + Will check against tuple or other FutureScanLocation object. + """ + if isinstance(other, FutureScanLocation): + return self._xy_pos == other.xy_tuple + if isinstance(other, tuple) and len(other) == 2: + return self._xy_pos[:2] == other + return NotImplemented + + +# Disable PLW1641, this warns against hash not being set when equals is. But the +# class shouldn't be hashable. +class VisitedScanLocation: # noqa PLW1641 + """Data for information on locations already visited during a scan. + + This object is only used for internal calculation and data storage it shouldn't + be an input or output of public methods. + """ + + def __init__( + self, xyz_pos: XYZPos, imaged: bool, focused: bool, **kwargs: Any + ) -> None: + """Initialise VisitedScanLocation with an xyz-position, and whether imaged/focused. + + :param xyz_pos: The (x, y, z) position visited. + :param imaged: True if an image was taken, False if not (due to background + detect) + :param focused: True if autofocus completed successfully + :param kwargs: Any other information about the location. This should be passed + through to from the FutureScanLocation object that requested the location + to be scanned. + """ + self._xyz_pos = enforce_xyz_tuple(xyz_pos) + self.imaged = imaged + self.focused = focused + self.planner_data = copy(kwargs) + + @property + def xyz_tuple(self) -> XYZPos: + """The xyz position tuple.""" + return self._xyz_pos + + @property + def xy_tuple(self) -> XYPos: + """The xy position tuple.""" + return self._xyz_pos[:2] + + def __eq__(self, other: Any) -> bool: + """Check for equality, only checks the xyz-position or xy-position if z isn't available. + + Will check xyz-position against 3-value tuples and other VisitedScanLocation + objects. + + Will check xy-position against 2-value tuples and FutureScanLocation objects. + """ + if isinstance(other, VisitedScanLocation): + return self._xyz_pos == other.xyz_tuple + if isinstance(other, FutureScanLocation): + return self._xyz_pos[:2] == other.xy_tuple + if isinstance(other, tuple): + if len(other) == 2: + return self._xyz_pos[:2] == other + if len(other) == 3: + return self._xyz_pos == other + return NotImplemented + + class ScanPlanner: """A base class for a scan planner. @@ -79,25 +176,10 @@ class ScanPlanner: self._initial_position = enforce_xy_tuple(initial_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._initial_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 positions 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 = [] + self._remaining_locations: list[FutureScanLocation] = ( + self._initial_location_list() + ) + self._path_history: list[VisitedScanLocation] = [] @property def scan_complete(self) -> bool: @@ -107,48 +189,57 @@ class ScanPlanner: @property def remaining_locations(self) -> XYPosList: """Property to access a copy of the remaining_locations.""" - return copy(self._remaining_locations) + return [loc.xy_tuple for loc in self._remaining_locations] @property def imaged_locations(self) -> XYZPosList: """Property to access a copy of the imaged_locations.""" - return copy(self._imaged_locations) + return [loc.xyz_tuple for loc in self._path_history if loc.imaged] @property def focused_locations(self) -> XYZPosList: """Property to access a copy of the focused_locations.""" - return copy(self._focused_locations) + return [loc.xyz_tuple for loc in self._path_history if loc.focused] @property def path_history(self) -> XYPosList: """Property to access a copy of the path_history.""" - return copy(self._path_history) + return [loc.xy_tuple for loc in 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 _initial_location_list(self) -> XYPosList: + def _initial_location_list(self) -> list[FutureScanLocation]: """Set the initial list of locations for this scan planner. This is called on initialisation. 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. + :return: A list of FutureScanLocation objects with all planned locations. """ 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.""" + def position_visited(self, position: XYPos | FutureScanLocation) -> bool: + """Return True if input scan position has been visited before.""" # Ensure tuple for correct matching! - return tuple(position) in self._path_history + return position in self._path_history - def position_planned(self, position: XYPos) -> bool: - """Return True if input xy position is planned.""" + def get_visited_location( + self, position: XYPos | XYZPos | FutureScanLocation + ) -> VisitedScanLocation: + """Return the scan location from the history that matches the input position.""" + # Ignoring type as self._path_history has type List[VisitedScanLocation], and + # VisitedScanLocation implements __eq__ for XYPos & XYZPos & FutureScanLocation + # however this is not statically detectable by MyPy + index = self._path_history.index(position) # type: ignore[arg-type] + return self._path_history[index] + + def position_planned(self, position: XYPos | FutureScanLocation) -> bool: + """Return True if input scan position position is planned.""" # Ensure tuple for correct matching! - return tuple(position) in self._remaining_locations + return position in self._remaining_locations def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]: """Return the next location to scan and its estimated z-position. @@ -159,7 +250,7 @@ class ScanPlanner: if self.scan_complete: raise RuntimeError("Can't get next position, scan is complete") - next_location = self._remaining_locations[0] + next_location = self._remaining_locations[0].xy_tuple # If focussed locations exist return closest location, favouring most recent closest_pos = self.closest_focus_site(next_location) @@ -177,12 +268,14 @@ class ScanPlanner: Returns None if there if no focussed locations are present """ - if not self._focused_locations: + # save to variable rather than search for focussed sites each time. + focused_locations = self.focused_locations + if not focused_locations: return None # must be float64 (double precision) to deal with the huge numbers involved! current_pos = np.array(xy_pos, dtype="float64") - path_pos = np.array(self._focused_locations, dtype="float64")[:, :2] + path_pos = np.array(focused_locations, dtype="float64")[:, :2] # Use linalg.norm to calculate the direct distance bweween the points # Note linalg.norm always uses float64 @@ -193,36 +286,35 @@ class ScanPlanner: indices = np.where(dists == np.min(dists))[0] # The last index is most recent - return self._focused_locations[indices[-1]] + return 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 - + :param xyz_pos: the x_y_z position + :param imaged: true if an image was taken, false if not (due to background detect) + :param focused: true if autofocus completed successfully """ # 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: + expected_pos = self._remaining_locations.pop(0) + if xyz_pos[:2] != expected_pos.xy_tuple: 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) + self._path_history.append( + VisitedScanLocation( + xyz_pos=xyz_pos, + imaged=imaged, + focused=focused, + **expected_pos.planner_data, + ) + ) class SmartSpiral(ScanPlanner): @@ -257,6 +349,21 @@ class SmartSpiral(ScanPlanner): super().__init__(initial_position, planner_settings) self._distance_cutoff: float = max([self._dx, self._dy]) * 1.1 + def _is_primary_location( + self, location: FutureScanLocation | VisitedScanLocation + ) -> bool: + """Return True if input is a primary location not a secondary (intermediate) location.""" + return location.planner_data["primary"] + + @property + def secondary_locations(self) -> XYZPosList: + """A list of all secondary (intermediate) locations.""" + return [ + loc.xyz_tuple + for loc in self._path_history + if not self._is_primary_location(loc) + ] + def _parse(self, planner_settings: Optional[dict] = None) -> None: """Parse SmartSpiral Settings dictionary. @@ -275,32 +382,33 @@ class SmartSpiral(ScanPlanner): self._dy = int(planner_settings["dy"]) self._max_dist = int(planner_settings["max_dist"]) - def _initial_location_list(self) -> XYPosList: + def _initial_location_list(self) -> list[FutureScanLocation]: """Set the initial list of locations for this scan planner. This is salled on initialisation. For smart spiral this is just the first point """ - return [self._initial_position] + return [FutureScanLocation(self._initial_position, primary=True)] 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 + """Mark the location as visited. + :param xyz_pos: the x_y_z position + :param imaged: true if an image was taken, false if not (due to background detect) + :param focused: true if autofocus completed successfully """ # 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) + if self._is_primary_location(self._path_history[-1]): + if imaged: + self._add_surrounding_positions(xy_pos) + else: + self._add_intermediate_positions(xy_pos) self._re_sort_remaining_locations(xy_pos) def _add_surrounding_positions(self, xy_pos: XYPos) -> None: @@ -312,30 +420,89 @@ class SmartSpiral(ScanPlanner): * too far away * already planned * already visited + + If the already visited position was not imaged then an intermediate location is + added. See also self._add_intermediate_positions() for adding intermediate + locations after visiting a location that was not imaged. """ - 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), - ] + new_positions = self._adjacent_positions(xy_pos) 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): + # Skip position if already planned + if self.position_planned(new_pos): + continue + if self.position_visited(new_pos): + # Get the VisitedScanLocation object if already visited + visited = self.get_visited_location(new_pos) + if visited.imaged: + # If this adjacent position was imaged successfully already then skip. + continue + # If it wasn't imaged add an intermediate location between the + # last imaged position and this one. + i_pos = self._intermediate_position(xy_pos, new_pos) + # Set primary=False surrounding images are not added once imaged. + i_loc = FutureScanLocation(i_pos, primary=False) + # Append instantly without checking max_distance as this is between + # imaged points. + self._remaining_locations.append(i_loc) 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) + new_loc = FutureScanLocation(new_pos, primary=True) + self._remaining_locations.append(new_loc) + + def _add_intermediate_positions(self, xy_pos: XYPos) -> None: + """Add intermediate points after locating a background location. + + This is called after an image is recorded that was background. Intermediate + locations are added between any adjacent locations that were successfully + imaged due to being labelled as containing sample. + + Note that in the case that an imaged location has an adjacent background image + then adding the intermediate image will be handled by + _add_surrounding_positions(). + """ + surrounding_positions = self._adjacent_positions(xy_pos) + + for surr_pos in surrounding_positions: + if self.position_visited(surr_pos): + # Get the VisitedScanLocation object if already visited + visited = self.get_visited_location(surr_pos) + if not visited.imaged: + # If it wasn't imaged then skip this position + continue + # If surrounding location was imaged add an intermediate location + # between the most recent position and this imaged position. + i_pos = self._intermediate_position(xy_pos, surr_pos) + # Set primary=False surrounding images are not added once imaged. + i_loc = FutureScanLocation(i_pos, primary=False) + # Append instantly without checking max_distance as this is between + # imaged points. + self._remaining_locations.append(i_loc) + + def _adjacent_positions(self, xy_pos: XYPos) -> XYPosList: + """Return 4 points +/-dx and +/-dy from the input location.""" + return [ + (xy_pos[0] - self._dx, xy_pos[1]), + (xy_pos[0] + self._dx, xy_pos[1]), + (xy_pos[0], xy_pos[1] - self._dy), + (xy_pos[0], xy_pos[1] + self._dy), + ] + + def _intermediate_position(self, xy_pos1: XYPos, xy_pos2: XYPos) -> XYPos: + """Return an (x,y) position halfway between two input positions.""" + x = (xy_pos1[0] + xy_pos2[0]) // 2 + y = (xy_pos1[1] + xy_pos2[1]) // 2 + return (x, y) def _re_sort_remaining_locations(self, current_pos: XYPos) -> None: """Sort the remaining positions based on the current location.""" # Defined rather than use a lambda for readability - def sort_key(pos: XYPos) -> tuple[float, float, float]: + def sort_key(pos: FutureScanLocation) -> tuple[float, float, float]: return ( self.moves_between(current_pos, pos), self.moves_between(self._initial_position, pos), @@ -356,7 +523,7 @@ class SmartSpiral(ScanPlanner): if self.scan_complete: raise RuntimeError("Can't get next position, scan is complete") - next_location = self._remaining_locations[0] + next_location = self._remaining_locations[0].xy_tuple # If focused locations exist, return the neighbour with the lowest z position closest_pos = self.select_nearby_focus_site(next_location) @@ -376,12 +543,14 @@ class SmartSpiral(ScanPlanner): Returns None if no focused locations are present """ - if not self._focused_locations: + # save to variable rather than search for focussed sites each time. + focused_locations = self.focused_locations + if not focused_locations: return None # must be float64 (double precision) to deal with the huge numbers involved! current_pos = np.array(xy_pos, dtype="float64") - path_pos = np.array(self._focused_locations, dtype="float64")[:, :2] + path_pos = np.array(focused_locations, dtype="float64")[:, :2] # Use linalg.norm to calculate the direct distance between the points # Note linalg.norm always uses float64 @@ -399,7 +568,7 @@ class SmartSpiral(ScanPlanner): indices = np.where(dists <= distance_cutoff)[0] # Turning into an array allows slicing based on a list - focused_locations_array = np.array(self._focused_locations) + focused_locations_array = np.array(focused_locations) # Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best # if started too low, so the lowest z will perform best @@ -410,8 +579,8 @@ class SmartSpiral(ScanPlanner): def moves_between( self, - starting_pos: XYPos | np.ndarray, - ending_pos: XYPos | np.ndarray, + starting_pos: XYPos | np.ndarray | FutureScanLocation, + ending_pos: XYPos | np.ndarray | FutureScanLocation, ) -> float: """Return the larger of x moves or y moves between two xy positions. @@ -419,6 +588,10 @@ class SmartSpiral(ScanPlanner): :param ending_pos: the position to measure to """ + if isinstance(starting_pos, FutureScanLocation): + starting_pos = starting_pos.xy_tuple + if isinstance(ending_pos, FutureScanLocation): + ending_pos = ending_pos.xy_tuple move_size = np.array([self._dx, self._dy]) starting_pos = np.array(starting_pos, dtype="float64") @@ -430,12 +603,17 @@ class SmartSpiral(ScanPlanner): def distance_between( - current_pos: XYPos | np.ndarray, next_pos: XYPos | np.ndarray + current_pos: XYPos | np.ndarray | FutureScanLocation, + next_pos: XYPos | np.ndarray | FutureScanLocation, ) -> float: """Calculate the distance between the two xy positions. This was previously called ``distance_to_site`` """ + if isinstance(current_pos, FutureScanLocation): + current_pos = current_pos.xy_tuple + if isinstance(next_pos, FutureScanLocation): + next_pos = next_pos.xy_tuple 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/tests/test_scan_planners.py b/tests/test_scan_planners.py index 89996945..f18d6c81 100644 --- a/tests/test_scan_planners.py +++ b/tests/test_scan_planners.py @@ -254,17 +254,27 @@ def test_closest_focus_with_large_numbers(): planner = scan_planners.SmartSpiral( initial_position=initial_position, planner_settings=planner_settings ) - # Directly overwrite the private focussed locations list for test + # Directly overwrite the private path history 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)] + + planner._path_history = [ + scan_planners.VisitedScanLocation((1000000, 0, 0), imaged=True, focused=True), + scan_planners.VisitedScanLocation((0, 1000000, 0), imaged=True, focused=True), + ] assert planner.closest_focus_site((0, 0)) == (0, 1000000, 0) # Try similar - planner._focused_locations = [(1234567, 0, 0), (-1234567, 0, 0)] + planner._path_history = [ + scan_planners.VisitedScanLocation((1234567, 0, 0), imaged=True, focused=True), + scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True), + ] assert planner.closest_focus_site((0, 0)) == (-1234567, 0, 0) # Make the first point 1 step closer - planner._focused_locations = [(1234566, 0, 0), (-1234567, 0, 0)] + planner._path_history = [ + scan_planners.VisitedScanLocation((1234566, 0, 0), imaged=True, focused=True), + scan_planners.VisitedScanLocation((-1234567, 0, 0), imaged=True, focused=True), + ] assert planner.closest_focus_site((0, 0)) == (1234566, 0, 0) @@ -292,5 +302,6 @@ def test_example_smart_spiral(): expected_planner = ( scan_test_helpers.get_expected_result_for_example_smart_spiral(sample_name) ) + assert planner.path_history == expected_planner.path_history assert planner.imaged_locations == expected_planner.imaged_locations diff --git a/tests/utilities/example_smart_spiral_core.pkl b/tests/utilities/example_smart_spiral_core.pkl index 5b7ba2de..b3b47db8 100644 Binary files a/tests/utilities/example_smart_spiral_core.pkl and b/tests/utilities/example_smart_spiral_core.pkl differ diff --git a/tests/utilities/example_smart_spiral_lobed.pkl b/tests/utilities/example_smart_spiral_lobed.pkl index a935e162..21bcf3e1 100644 Binary files a/tests/utilities/example_smart_spiral_lobed.pkl and b/tests/utilities/example_smart_spiral_lobed.pkl differ diff --git a/tests/utilities/example_smart_spiral_regular.pkl b/tests/utilities/example_smart_spiral_regular.pkl index ddf784c8..aa9b847b 100644 Binary files a/tests/utilities/example_smart_spiral_regular.pkl and b/tests/utilities/example_smart_spiral_regular.pkl differ diff --git a/tests/utilities/scan_test_helpers.py b/tests/utilities/scan_test_helpers.py index 288996ac..4d32a2fb 100644 --- a/tests/utilities/scan_test_helpers.py +++ b/tests/utilities/scan_test_helpers.py @@ -60,8 +60,9 @@ def visualise_scan(sample: FakeSample, planner: scan_planners.ScanPlanner) -> Fi """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, strict=True) - xi, yi, _ = zip(*planner._imaged_locations, strict=True) + xh, yh = zip(*planner.path_history, strict=True) + xi, yi, _zi = zip(*planner.imaged_locations, strict=True) + xs, ys, _zs = zip(*planner.secondary_locations, strict=True) # convert history to numpy array so can calculate quiver arrows xh = np.array(xh) @@ -78,6 +79,7 @@ def visualise_scan(sample: FakeSample, planner: scan_planners.ScanPlanner) -> Fi ) plt.plot(xh, yh, "r.") plt.plot(xi, yi, "g*") + plt.plot(xs, ys, "o", mfc="none", mec="blue") ax.axis("equal") return fig @@ -99,7 +101,7 @@ def interp_closed_path(xy_points: list[tuple[int, int]], n_points: int) -> MatPa # 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) + spline_data, *_unused = interpolate.splprep([x, y], s=0, per=True) # evaluate the spline xi, yi = interpolate.splev(np.linspace(0, 1, n_points), spline_data)