Further layering of the classes for scan workflows

This commit is contained in:
Julian Stirling 2026-02-10 16:02:18 +00:00
parent 1f01717a1b
commit 15798d9b7d

View file

@ -68,6 +68,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
# CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None
_csm: Optional[CameraStageMapper] = lt.thing_slot()
_cam: BaseCamera = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
def check_before_start(self, scan_name: str) -> None:
"""Check before the scan starts. Throw an error if the scan shouldn't start.
@ -117,11 +120,42 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
def acquisition_routine(
self, settings: SettingModelType, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Overload to set the acquisition routine that happens at each scan site."""
"""Overload to set the acquisition routine that happens at each scan site.
:param settings: The settings for this scan as a HistoScanSettingsModel
: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.
"""
raise NotImplementedError(
"Each specific ScanWorkflow must implement an acquisition routine"
)
def _autofocus_and_capture(
self,
xyz_pos: tuple[int, int, int],
dz: int,
images_dir: str,
save_resolution: tuple[int, int],
) -> tuple[bool, Optional[int]]:
"""Atuofocus and then capture, this can be used as an acquisition routine.
:param dz: The dz for autofoucs.
:param images_dir: The path to the directory for saving images..
:param save_resolution: The resolution to save images at.
:return: A tuple ready to pass out of acquisition routine.
"""
self._autofocus.fast_autofocus(dz=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(images_dir, filename),
save_resolution=save_resolution,
)
return True, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
@ -129,13 +163,35 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
"Each scan workflow must implement a settings_ui method."
)
def _require_csm(self) -> CameraStageMapper:
"""Give each model the option to require CSM. Return it if present."""
if self._csm is None:
raise RuntimeError(
"CameraStageMapping not set, and is required for this workflow."
)
return self._csm
class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]):
"""A generic workflow for any scan that captures images on a rectilinear grid."""
# Thing Slots
_csm: CameraStageMapper = lt.thing_slot()
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.
"""
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.
"""
# 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.")
def _calc_displacement_from_overlap(self, overlap: float) -> tuple[int, int]:
"""Use camera stage mapping to calculate x and y displacement from given overlap.
@ -147,9 +203,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
:raises RuntimeError: If there is no camera stage mapper Thing available or if CMS isn't calibrated.
"""
csm = self._require_csm()
csm_image_res = csm.image_resolution
csm_image_res = self._csm.image_resolution
if csm_image_res is None:
raise RuntimeError("CSM not set. Scan shouldn't have progresses this far.")
@ -157,16 +211,33 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
dx_img = csm_image_res[1] * (1 - overlap)
dy_img = csm_image_res[0] * (1 - overlap)
x_move_stage = csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
x_move_stage = self._csm.convert_image_to_stage_coordinates(x=dx_img, y=0)
y_move_stage = self._csm.convert_image_to_stage_coordinates(x=0, y=dy_img)
# Assume no rotation or skew and take only the aligned axis of vector.
# Coerce to positive integer, but correct if x and y are flipped
if abs(x_move_stage["x"]) > abs(x_move_stage["y"]):
return x_move_stage["x"], y_move_stage["y"]
# If not use the other stage axes. Note "dx" will be the movement in camera y.
self.logger.info(
f"Based on an overlap of {self.overlap}, the stage will make steps of "
f"{y_move_stage['x']}, {x_move_stage['y']}"
)
return y_move_stage["x"], x_move_stage["y"]
def _get_stitching_settings_model(self) -> StitchingSettings:
"""Return a stitching settings model based on current settings."""
return StitchingSettings(
overlap=self.overlap,
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
@lt.property
def ready(self) -> bool:
"""Whether this scanworkflow is ready to start."""
return not self._csm.calibration_required
class HistoScanSettingsModel(BaseModel):
"""The settings for a scan with the HistoScanWorkflow.
@ -176,14 +247,14 @@ class HistoScanSettingsModel(BaseModel):
"""
overlap: float
max_dist: int
dx: int
dy: int
max_dist: int
skip_background: bool
smart_stack_params: SmartStackParams
class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
"""A workflow optimised for scanning Histopathology samples.
This workflow automatically plans its own path around a sample spiralling out from
@ -201,12 +272,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
)
_settings_model = HistoScanSettingsModel
_planner_cls: type[ScanPlanner] = SmartSpiral
_planner_cls = SmartSpiral
# Thing Slots
_background_detector: ChannelDeviationLUV = lt.thing_slot()
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
# Scan settings
@ -216,21 +284,9 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
This uses the settings from the ``BackgroundDetectThing``.
"""
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.
"""
max_range: int = lt.setting(default=45000)
"""The maximum distance in steps from the centre of the scan."""
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.
"""
# Stacking settings
stack_images_to_save: int = lt.setting(default=1)
@ -305,16 +361,8 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
: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],
)
stitching_settings = self._get_stitching_settings_model()
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}"
)
smart_stack_params = self.create_smart_stack_params(
images_dir=images_dir,
@ -502,7 +550,7 @@ class HistoScanWorkflow(ScanWorkflow[HistoScanSettingsModel]):
]
class RectangleSettingsModel(BaseModel):
class RegularGridSettingsModel(BaseModel):
"""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
@ -519,86 +567,28 @@ class RectangleSettingsModel(BaseModel):
save_resolution: tuple[int, int]
TypeSettings = TypeVar("TypeSettings", bound=RectangleSettingsModel)
class SnakeSettingsModel(RectangleSettingsModel):
"""Settings for Snake scan."""
pass
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
_cam: BaseCamera = lt.thing_slot()
_csm: CameraStageMapper = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
# Shared 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.
"""
class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
"""A base workflow for any workflow that uses a regular rectangular grid."""
x_count: int = lt.setting(default=3, ge=1)
"""The number of columns in the scan."""
y_count: int = lt.setting(default=2, ge=1)
"""The number of columns and rows in the scan.
Must be at least 1.
"""
"""The number of rows in the scan."""
# 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.
_settings_model = RegularGridSettingsModel
_planner_cls: type[SnakeScan] | type[RasterScan]
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[TypeSettings, StitchingSettings]:
def all_settings(
self, images_dir: str
) -> tuple[RegularGridSettingsModel, Optional[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],
)
stitching_settings = self._get_stitching_settings_model()
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,
@ -613,7 +603,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]):
return scan_settings, stitching_settings
def pre_scan_routine(self, settings: TypeSettings) -> None:
def pre_scan_routine(self, settings: RegularGridSettingsModel) -> None:
"""Perform these steps before starting the scan.
In this case, only autofocus.
@ -623,7 +613,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]):
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
def new_scan_planner(
self, settings: TypeSettings, position: Mapping[str, int]
self, settings: RegularGridSettingsModel, position: Mapping[str, int]
) -> ScanPlanner:
"""Return a new scan planner object.
@ -642,26 +632,22 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]):
)
def acquisition_routine(
self, settings: TypeSettings, xyz_pos: tuple[int, int, int]
self, settings: RegularGridSettingsModel, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Perform acquisition routine. This is run at each scan location.
"""Autofocus and capture.
:param settings: The settings for this scan as the correct setting model
:param settings: The settings for this scan as a HistoScanSettingsModel
: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.
In this method, image is always taken, so first return is True.
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),
return self._autofocus_and_capture(
xyz_pos=xyz_pos,
dz=settings.autofocus_dz,
images_dir=settings.images_dir,
save_resolution=settings.save_resolution,
)
return True, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
@ -673,7 +659,7 @@ class RectangleWorkflow(ScanWorkflow[TypeSettings], Generic[TypeSettings]):
]
class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]):
class SnakeWorkflow(RegularGridWorkflow):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and snakes
@ -688,11 +674,10 @@ class SnakeWorkflow(RectangleWorkflow[SnakeSettingsModel]):
),
readonly=True,
)
_settings_model = SnakeSettingsModel
_planner_cls = SnakeScan
class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]):
class RasterWorkflow(RegularGridWorkflow):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and always
@ -709,5 +694,4 @@ class RasterWorkflow(RectangleWorkflow[RasterSettingsModel]):
readonly=True,
)
_settings_model = RasterSettingsModel
_planner_cls = RasterScan