Fully subclass Rectangle basescanner for raster and snake

This commit is contained in:
jaknapper 2026-02-09 19:35:19 +00:00
parent fb7a8b4959
commit 6294cde471
3 changed files with 156 additions and 12 deletions

View file

@ -19,6 +19,7 @@
}, },
"histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow", "histo_scan_workflow": "openflexure_microscope_server.things.scan_workflows:HistoScanWorkflow",
"snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow", "snake_workflow": "openflexure_microscope_server.things.scan_workflows:SnakeWorkflow",
"raster_workflow": "openflexure_microscope_server.things.scan_workflows:RasterWorkflow",
"stage_measure": "openflexure_microscope_server.things.stage_measure:RangeofMotionThing", "stage_measure": "openflexure_microscope_server.things.stage_measure:RangeofMotionThing",
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV", "bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV" "bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"

View file

@ -433,7 +433,7 @@ class SmartSpiral(RectangleScan):
def _parse(self, planner_settings: Optional[dict] = None) -> None: def _parse(self, planner_settings: Optional[dict] = None) -> None:
super()._parse(planner_settings) super()._parse(planner_settings)
if "max_dist" not in planner_settings: if not planner_settings or "max_dist" not in planner_settings:
raise KeyError("SmartSpiral requires max_dist") raise KeyError("SmartSpiral requires max_dist")
self._max_dist = int(planner_settings["max_dist"]) self._max_dist = int(planner_settings["max_dist"])
@ -614,7 +614,9 @@ class SnakeScan(RectangleScan):
expected_keys = ["x_count", "y_count"] expected_keys = ["x_count", "y_count"]
invalid_msg = "SnakeScan requires planner_settings with keys: " invalid_msg = "SnakeScan requires planner_settings with keys: "
if not all(k in planner_settings for k in expected_keys): if not planner_settings or not all(
k in planner_settings for k in expected_keys
):
raise KeyError(invalid_msg + ",".join(expected_keys)) raise KeyError(invalid_msg + ",".join(expected_keys))
self._x_count = int(planner_settings["x_count"]) self._x_count = int(planner_settings["x_count"])
@ -654,7 +656,9 @@ class RasterScan(RectangleScan):
expected_keys = ["x_count", "y_count"] expected_keys = ["x_count", "y_count"]
invalid_msg = "SnakeScan requires planner_settings with keys: " invalid_msg = "SnakeScan requires planner_settings with keys: "
if not all(k in planner_settings for k in expected_keys): if not planner_settings or not all(
k in planner_settings for k in expected_keys
):
raise KeyError(invalid_msg + ",".join(expected_keys)) raise KeyError(invalid_msg + ",".join(expected_keys))
self._x_count = int(planner_settings["x_count"]) self._x_count = int(planner_settings["x_count"])

View file

@ -519,7 +519,7 @@ class SnakeSettingsModel(BaseModel):
save_resolution: tuple[int, int] save_resolution: tuple[int, int]
class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]): class SnakeWorkflow(ScanWorkflow[SnakeSettingsModel]):
"""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 snakes This workflow generates a list of coordinates in a rectangle, and snakes
@ -535,7 +535,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
readonly=True, readonly=True,
) )
_settings_model: type[SettingModelType] = SnakeSettingsModel _settings_model = SnakeSettingsModel
_planner_cls: type[ScanPlanner] = SnakeScan _planner_cls: type[ScanPlanner] = SnakeScan
# Thing Slots # Thing Slots
_cam: BaseCamera = lt.thing_slot() _cam: BaseCamera = lt.thing_slot()
@ -578,7 +578,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
def all_settings( def all_settings(
self, images_dir: str self, images_dir: str
) -> tuple[SettingModelType, StitchingSettings]: ) -> 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,7 +609,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
return scan_settings, stitching_settings return scan_settings, stitching_settings
def pre_scan_routine(self, settings: SettingModelType) -> None: def pre_scan_routine(self, settings: SnakeSettingsModel) -> None:
"""Autofocus before starting the scan. """Autofocus before starting the scan.
:param settings: The settings for this scan as a SnakeSettingsModel :param settings: The settings for this scan as a SnakeSettingsModel
@ -617,7 +617,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
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: SettingModelType, position: Mapping[str, int] self, settings: SnakeSettingsModel, position: Mapping[str, int]
) -> ScanPlanner: ) -> ScanPlanner:
"""Return a new scan planner object. """Return a new scan planner object.
@ -640,7 +640,7 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
) )
def acquisition_routine( def acquisition_routine(
self, settings: SettingModelType, xyz_pos: tuple[int, int, int] self, settings: SnakeSettingsModel, 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.
@ -671,16 +671,23 @@ class SnakeWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
] ]
class RasterSettingsModel(SnakeSettingsModel): class RasterSettingsModel(BaseModel):
"""The settings for a scan with the Raster Workflow. """The settings for a scan with the Raster Workflow.
This is identical to the SnakeSettings, but is subclassed for clarity. This is identical to the SnakeSettings, but is subclassed for clarity.
""" """
pass overlap: float
dx: int
dy: int
x_count: int
y_count: int
images_dir: str
autofocus_dz: int
save_resolution: tuple[int, int]
class RasterWorkflow(SnakeWorkflow[RasterSettingsModel]): class RasterWorkflow(ScanWorkflow[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
@ -699,3 +706,135 @@ class RasterWorkflow(SnakeWorkflow[RasterSettingsModel]):
_settings_model = RasterSettingsModel _settings_model = RasterSettingsModel
_planner_cls: type[ScanPlanner] = RasterScan _planner_cls: type[ScanPlanner] = 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)"),
]