RectangleWorkflow as a baseclass

This commit is contained in:
jaknapper 2026-02-10 14:04:01 +00:00
parent 6294cde471
commit f1845c3039
3 changed files with 59 additions and 200 deletions

View file

@ -572,6 +572,7 @@ class SmartSpiral(RectangleScan):
if not focused_locations: if not focused_locations:
return None return None
# must be float64 to deal with large coordinates
current_pos = np.array(xy_pos, dtype="float64") current_pos = np.array(xy_pos, dtype="float64")
focused_arr = np.array(focused_locations, dtype="float64") focused_arr = np.array(focused_locations, dtype="float64")
@ -580,7 +581,7 @@ class SmartSpiral(RectangleScan):
dy_ok = np.abs(focused_arr[:, 1] - current_pos[1]) <= self._dy dy_ok = np.abs(focused_arr[:, 1] - current_pos[1]) <= self._dy
nearby_indices = np.where(dx_ok & dy_ok)[0] nearby_indices = np.where(dx_ok & dy_ok)[0]
# If no neighbouring sites were focused, choose the site(s) that are closest # If no neighbouring sites were focused, choose the closest
if len(nearby_indices) == 0: if len(nearby_indices) == 0:
deltas = focused_arr[:, :2] - current_pos deltas = focused_arr[:, :2] - current_pos
dists = np.linalg.norm(deltas, axis=1) dists = np.linalg.norm(deltas, axis=1)
@ -593,10 +594,13 @@ class SmartSpiral(RectangleScan):
min_z = np.min(nearby_sites[:, 2]) min_z = np.min(nearby_sites[:, 2])
# Among those with min z, choose the most recent # Among those with min z, choose the most recent
best_sites = nearby_sites[nearby_sites[:, 2] == min_z] chosen_site = nearby_sites[nearby_sites[:, 2] == min_z][-1]
chosen_site = best_sites[-1]
return tuple(chosen_site.astype(int)) return (
int(chosen_site[0]),
int(chosen_site[1]),
int(chosen_site[2]),
)
class SnakeScan(RectangleScan): class SnakeScan(RectangleScan):

View file

@ -485,9 +485,6 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
"""A list of PropertyControl objects to create the settings in the scan tab.""" """A list of PropertyControl objects to create the settings in the scan tab."""
return [ return [
property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"), property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"),
property_control_for(
self, "skip_background", label="Detect and Skip Empty Fields "
),
property_control_for( property_control_for(
self, "stack_images_to_save", label="Images in Stack to Save" self, "stack_images_to_save", label="Images in Stack to Save"
), ),
@ -499,11 +496,14 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
property_control_for(self, "stack_dz", label="Stack dz (steps)"), property_control_for(self, "stack_dz", label="Stack dz (steps)"),
property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"), property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"),
property_control_for(self, "max_range", label="Maximum Distance (steps)"), property_control_for(self, "max_range", label="Maximum Distance (steps)"),
property_control_for(
self, "skip_background", label="Detect and Skip Empty Fields "
),
] ]
class SnakeSettingsModel(BaseModel): class RectangleSettingsModel(BaseModel):
"""The settings for a scan with the SnakeWorkflow. """The settings for a scan with a regular grid of dx and dy for x_count, y_count steps.
This includes settings calculated when starting. This will be held by smart scan This includes settings calculated when starting. This will be held by smart scan
during a scan and serialised to disk. during a scan and serialised to disk.
@ -519,49 +519,39 @@ class SnakeSettingsModel(BaseModel):
save_resolution: tuple[int, int] save_resolution: tuple[int, int]
class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]): TypeSettings = TypeVar("TypeSettings", bound=RectangleSettingsModel)
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and snakes
around them from the top left (assuming positive dx and dy).
"""
display_name: str = lt.property(default="Snake Scan", readonly=True) class SnakeSettingsModel(RectangleSettingsModel):
ui_blurb: str = lt.property( """Settings for Snake scan."""
default=(
"This scan workflow is optimised for scanning over a rectangle. It "
"snakes down and right from the starting point, over a defined grid."
),
readonly=True,
)
_settings_model = SnakeSettingsModel pass
_planner_cls: type[ScanPlanner] = SnakeScan
class RasterSettingsModel(RectangleSettingsModel):
"""Settings for Raster scan."""
pass
class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]):
"""Abstract workflow for rectangular scans."""
# Settings and planner must be set by subclass
_settings_model: type[TypeSettings]
_planner_cls: type[ScanPlanner]
# Thing Slots # Thing Slots
_cam: BaseCamera = lt.thing_slot() _cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot() _csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot() _autofocus: AutofocusThing = lt.thing_slot()
_stage: BaseStage = lt.thing_slot() _stage: BaseStage = lt.thing_slot()
# Scan settings # Shared scan settings
autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000) autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000)
"""The z distance to perform an autofocus in steps.
Must be greater than or equal to 200, and less than or equal to 2000.
"""
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7) overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x or y.
This must be between 0.1 and 0.7.
"""
x_count: int = lt.setting(default=3) x_count: int = lt.setting(default=3)
y_count: int = lt.setting(default=2) y_count: int = lt.setting(default=2)
# The noqa statement is because scan_name is unused but is needed for equivalence
# with other workflows that may want to validate the scan name.
def check_before_start(self, scan_name: str) -> None: # noqa: ARG002 def check_before_start(self, scan_name: str) -> None: # noqa: ARG002
"""Before starting a scan, check that camera-stage-mapping is set. """Before starting a scan, check that camera-stage-mapping is set.
@ -576,9 +566,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
"""Whether this scanworkflow is ready to start.""" """Whether this scanworkflow is ready to start."""
return not self._csm.calibration_required return not self._csm.calibration_required
def all_settings( def all_settings(self, images_dir: str) -> tuple[TypeSettings, StitchingSettings]:
self, images_dir: str
) -> tuple[SnakeSettingsModel, StitchingSettings]:
"""Return the workflow and stitching settings. """Return the workflow and stitching settings.
:param images_dir: The directory that images are to be written to. :param images_dir: The directory that images are to be written to.
@ -609,25 +597,21 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
return scan_settings, stitching_settings return scan_settings, stitching_settings
def pre_scan_routine(self, settings: SnakeSettingsModel) -> None: def pre_scan_routine(self, settings: TypeSettings) -> None:
"""Autofocus before starting the scan. """Perform these steps before starting the scan.
:param settings: The settings for this scan as a SnakeSettingsModel In this case, only autofocus.
""" """
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre") self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
def new_scan_planner( def new_scan_planner(
self, settings: SnakeSettingsModel, position: Mapping[str, int] self, settings: TypeSettings, position: Mapping[str, int]
) -> ScanPlanner: ) -> ScanPlanner:
"""Return a new scan planner object. """Return a new scan planner object.
:param settings: The settings for this scan as a SnakeSettingsModel :param settings: The settings for this scan as a SnakeSettingsModel
:param position: The starting position as a mapping of axes names to int. :param position: The starting position as a mapping of axes names to int.
""" """
# The initial plan for the scan should be a single x,y position. All future
# moves will be planned around this point. In future, route planner could
# have multiple starting positions, each of which will be visited before the
# scan can end.
planner_settings = { planner_settings = {
"dx": settings.dx, "dx": settings.dx,
"dy": settings.dy, "dy": settings.dy,
@ -640,13 +624,14 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
) )
def acquisition_routine( def acquisition_routine(
self, settings: SnakeSettingsModel, xyz_pos: tuple[int, int, int] self, settings: TypeSettings, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]: ) -> tuple[bool, Optional[int]]:
"""Perform acquisition routine. This is run at each scan location. """Perform acquisition routine. This is run at each scan location.
:param settings: The settings for this scan as a SnakeSettingsModel :param settings: The settings for this scan as the correct setting model
:param xyz_pos: The current position as a tuple or 3 ints. :param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus. :return: A tuple of whether an image was taken, and the z-position for focus.
In this method, image is always taken, so first return is True.
If failed to find focus, returns for the focus z-position. If failed to find focus, returns for the focus z-position.
""" """
self._autofocus.fast_autofocus(dz=settings.autofocus_dz) self._autofocus.fast_autofocus(dz=settings.autofocus_dz)
@ -657,8 +642,7 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
save_resolution=settings.save_resolution, save_resolution=settings.save_resolution,
) )
imaged = True return True, focus_height
return imaged, focus_height
@lt.property @lt.property
def settings_ui(self) -> list[PropertyControl]: def settings_ui(self) -> list[PropertyControl]:
@ -671,23 +655,26 @@ class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
] ]
class RasterSettingsModel(BaseModel): class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]):
"""The settings for a scan with the Raster Workflow. """A workflow optimised for snaking around samples.
This is identical to the SnakeSettings, but is subclassed for clarity. This workflow generates a list of coordinates in a rectangle, and snakes
around them from the top left (assuming positive dx and dy).
""" """
overlap: float display_name: str = lt.property(default="Snake Scan", readonly=True)
dx: int ui_blurb: str = lt.property(
dy: int default=(
x_count: int "This scan workflow is optimised for scanning over a rectangle. It "
y_count: int "snakes down and right from the starting point, over a defined grid."
images_dir: str ),
autofocus_dz: int readonly=True,
save_resolution: tuple[int, int] )
_settings_model = SnakeSettingsModel
_planner_cls = SnakeScan
class RasterWorkflow(ScanWorkflow[RasterSettingsModel]): class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]):
"""A workflow optimised for snaking around samples. """A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and always This workflow generates a list of coordinates in a rectangle, and always
@ -705,136 +692,4 @@ class RasterWorkflow(ScanWorkflow[RasterSettingsModel]):
) )
_settings_model = RasterSettingsModel _settings_model = RasterSettingsModel
_planner_cls: type[ScanPlanner] = RasterScan _planner_cls = RasterScan
# Thing Slots
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
# Scan settings
autofocus_dz: int = lt.setting(default=1000, ge=200, le=2000)
"""The z distance to perform an autofocus in steps.
Must be greater than or equal to 200, and less than or equal to 2000.
"""
overlap: float = lt.setting(default=0.45, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x or y.
This must be between 0.1 and 0.7.
"""
x_count: int = lt.setting(default=3)
y_count: int = lt.setting(default=2)
# The noqa statement is because scan_name is unused but is needed for equivalence
# with other workflows that may want to validate the scan name.
def check_before_start(self, scan_name: str) -> None: # noqa: ARG002
"""Before starting a scan, check that camera-stage-mapping is set.
Raise error if:
- camera stage mapping is not set
"""
if self._csm.calibration_required:
raise RuntimeError("Camera Stage Mapping is not calibrated.")
@lt.property
def ready(self) -> bool:
"""Whether this scanworkflow is ready to start."""
return not self._csm.calibration_required
def all_settings(
self, images_dir: str
) -> tuple[RasterSettingsModel, StitchingSettings]:
"""Return the workflow and stitching settings.
:param images_dir: The directory that images are to be written to.
:return: A tuple containing the settings model for this workflow and the
settings model for stitching.
"""
stitching_settings = StitchingSettings(
overlap=self.overlap,
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
dx, dy = self._calc_displacement_from_overlap(self.overlap)
self.logger.info(
f"Based on an overlap of {self.overlap}, the stage will make steps of "
f"{dx}, {dy}"
)
scan_settings = self._settings_model(
overlap=self.overlap,
dx=dx,
dy=dy,
x_count=self.x_count,
y_count=self.y_count,
images_dir=images_dir,
autofocus_dz=self.autofocus_dz,
save_resolution=self.save_resolution,
)
return scan_settings, stitching_settings
def pre_scan_routine(self, settings: RasterSettingsModel) -> None:
"""Autofocus before starting the scan.
:param settings: The settings for this scan as a SnakeSettingsModel
"""
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
def new_scan_planner(
self, settings: RasterSettingsModel, position: Mapping[str, int]
) -> ScanPlanner:
"""Return a new scan planner object.
:param settings: The settings for this scan as a SnakeSettingsModel
:param position: The starting position as a mapping of axes names to int.
"""
# The initial plan for the scan should be a single x,y position. All future
# moves will be planned around this point. In future, route planner could
# have multiple starting positions, each of which will be visited before the
# scan can end.
planner_settings = {
"dx": settings.dx,
"dy": settings.dy,
"x_count": settings.x_count,
"y_count": settings.y_count,
}
return self._planner_cls(
initial_position=(position["x"], position["y"]),
planner_settings=planner_settings,
)
def acquisition_routine(
self, settings: RasterSettingsModel, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Perform acquisition routine. This is run at each scan location.
:param settings: The settings for this scan as a RasterSettingsModel
:param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus.
If failed to find focus, returns for the focus z-position.
"""
self._autofocus.fast_autofocus(dz=settings.autofocus_dz)
focus_height = self._stage.get_xyz_position()[2]
filename = f"img_{xyz_pos[0]}_{xyz_pos[1]}_{focus_height}.jpeg"
self._cam.capture_and_save(
jpeg_path=os.path.join(settings.images_dir, filename),
save_resolution=settings.save_resolution,
)
imaged = True
return imaged, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
return [
property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"),
property_control_for(self, "x_count", label="Number of columns"),
property_control_for(self, "y_count", label="Number of rows"),
property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"),
]

View file

@ -89,7 +89,7 @@ def test_bad_smart_spiral_settings():
initial_position = (100, 50) initial_position = (100, 50)
# Class init should raise error if no planner_settings dictionary set # Class init should raise error if no planner_settings dictionary set
msg = "SmartSpiral requires a planner_settings dictionary with keys" msg = "RectangleScan requires planner_settings with keys"
with pytest.raises(ValueError, match=msg): with pytest.raises(ValueError, match=msg):
scan_planners.SmartSpiral(initial_position=initial_position) scan_planners.SmartSpiral(initial_position=initial_position)
@ -105,7 +105,7 @@ def test_bad_smart_spiral_settings():
initial_position=initial_position, planner_settings=bad_planner_settings initial_position=initial_position, planner_settings=bad_planner_settings
) )
# Class init should raise error if planner_settings if any value can't be cast # Class init should raise error if any value in planner_settings can't be cast
# to int # to int
keys = ["dx", "dy", "max_dist"] keys = ["dx", "dy", "max_dist"]
for badkey in keys: for badkey in keys: