Initial commit, splitting params into focus, capture, stack

This commit is contained in:
Joe Knapper 2026-02-18 17:55:23 +00:00 committed by Julian Stirling
parent de08bb1740
commit e750129de2
4 changed files with 150 additions and 76 deletions

View file

@ -33,15 +33,25 @@ class NotStreamingError(RuntimeError):
"""No images captured from stream. The camera is almost certainly not streaming."""
class SmartStackParams(BaseModel):
"""A class for holding for smart stack parameters, and returning computed ones."""
class AutofocusParams(BaseModel):
"""A class for running autofocus routines."""
dz: int
sharpness_method: str = "jpeg"
class CaptureParams(BaseModel):
"""A class for capturing at least a single image."""
images_dir: str
save_resolution: tuple[int, int]
class StackParams(BaseModel):
"""A class for holding stack parameters, and returning computed ones."""
stack_dz: int
images_to_save: int
min_images_to_test: int
autofocus_dz: int
images_dir: str
save_resolution: tuple[int, int]
# Using docstrings under variables as this is how pdoc would expect
# attributed to be documented
@ -49,6 +59,12 @@ class SmartStackParams(BaseModel):
settling_time: float = 0.3
"""Time (in seconds) between moving and capturing an image"""
class SmartStackParams(StackParams):
"""A class for holding smart stack parameters, and returning computed ones."""
min_images_to_test: int
backlash_correction: int = 250
"""
Distance (in steps) to overshoot a move and then undo, to account for backlash
@ -457,6 +473,8 @@ class AutofocusThing(lt.Thing):
def run_smart_stack(
self,
stack_parameters: SmartStackParams,
capture_parameters: CaptureParams,
autofocus_parameters: AutofocusParams,
save_on_failure: bool = False,
check_turning_points: bool = True,
) -> tuple[bool, int]:
@ -498,7 +516,7 @@ class AutofocusThing(lt.Thing):
initial_z_pos = captures[0].position["z"]
# If a stack is not successful, move to the start and autofocus
try:
self.reset_stack(initial_z_pos, stack_parameters.autofocus_dz)
self.reset_stack(initial_z_pos, autofocus_parameters.dz)
except NoFocusFoundError:
break
@ -510,6 +528,7 @@ class AutofocusThing(lt.Thing):
sharpest_id=sharpest_id,
captures=captures,
stack_parameters=stack_parameters,
capture_parameters=capture_parameters,
)
return success, _get_capture_by_id(captures, sharpest_id).position["z"]
@ -535,6 +554,7 @@ class AutofocusThing(lt.Thing):
sharpest_id: int,
captures: list[CaptureInfo],
stack_parameters: SmartStackParams,
capture_parameters: CaptureParams,
) -> int:
"""Save the required captures to disk.
@ -552,8 +572,8 @@ class AutofocusThing(lt.Thing):
# Loop through the range, saving each capture to disk
for capture in captures[slice_to_save]:
self._cam.save_from_memory(
jpeg_path=os.path.join(stack_parameters.images_dir, capture.filename),
save_resolution=stack_parameters.save_resolution,
jpeg_path=os.path.join(capture_parameters.images_dir, capture.filename),
save_resolution=capture_parameters.save_resolution,
buffer_id=capture.buffer_id,
)
self._cam.clear_buffers()
@ -712,6 +732,56 @@ class AutofocusThing(lt.Thing):
return "success", capture_id
@lt.action
def run_basic_stack(
self,
stack_parameters: StackParams,
capture_parameters: CaptureParams,
) -> tuple[int, list[int]]:
"""Capture a simple z-stack with no focus testing or fitting.
This performs a fixed stack of images spaced by `stack_dz`,
saving all images captured. No sharpness testing, restart
logic, or autofocus is performed.
:param stack_parameters: SmartStackParams defining stack spacing,
image count, directory, and save resolution.
:returns:
- Final z position
- List of z positions captured
"""
captures: list[CaptureInfo] = []
z_positions: list[int] = []
# Capture images_to_save images
for _ in range(stack_parameters.images_to_save):
time.sleep(stack_parameters.settling_time)
capture = self.capture_stack_image(
buffer_max=stack_parameters.images_to_save
)
captures.append(capture)
z_positions.append(capture.position["z"])
self._stage.move_relative(z=stack_parameters.stack_dz)
# Save all captures
for capture in captures:
self._cam.save_from_memory(
jpeg_path=os.path.join(
capture_parameters.images_dir,
capture.filename,
),
save_resolution=capture_parameters.save_resolution,
buffer_id=capture.buffer_id,
)
self._cam.clear_buffers()
final_z = self._stage.position["z"]
return final_z, z_positions
class NotAPeakError(lt.exceptions.InvocationError):
"""The data to fit isn't a peak."""

View file

@ -31,7 +31,9 @@ from openflexure_microscope_server.stitching import (
from openflexure_microscope_server.things.autofocus import (
MAX_TEST_IMAGE_COUNT,
MIN_TEST_IMAGE_COUNT,
AutofocusParams,
AutofocusThing,
CaptureParams,
SmartStackParams,
)
from openflexure_microscope_server.things.background_detect import (
@ -101,7 +103,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
is returned if it is not possible to stitch the scan.
"""
raise NotImplementedError(
"Each specific ScanWorkflow must implement a `all_settings`. method."
"Each specific ScanWorkflow must implement a `all_settings` method."
)
def pre_scan_routine(self, settings: SettingModelType) -> None:
@ -239,6 +241,36 @@ class RectGridWorkflow(ScanWorkflow[SettingModelType], Generic[SettingModelType]
correlation_resize=STITCHING_RESOLUTION[0] / self.save_resolution[0],
)
def all_settings(
self, images_dir: str
) -> tuple[SettingModelType, Optional[StitchingSettings]]:
"""Return the scan settings and the stitching settings.
- `images_dir` is used to create the CaptureParams stored in the settings model.
- The returned SettingModelType now contains param objects:
* capture_params: CaptureParams
* autofocus_params: AutofocusParams
* stack_params: StackParams (if relevant)
"""
stitching_settings = self._get_stitching_settings_model()
dx, dy = self._calc_displacement_from_overlap(self.overlap)
capture_params = CaptureParams(
images_dir=images_dir, save_resolution=self.save_resolution
)
autofocus_params = AutofocusParams(dz=self.autofocus_dz)
scan_settings = self._settings_model(
overlap=self.overlap,
dx=dx,
dy=dy,
capture_params=capture_params,
autofocus_params=autofocus_params,
)
return scan_settings, stitching_settings
@lt.property
def ready(self) -> bool:
"""Whether this scanworkflow is ready to start."""
@ -257,6 +289,8 @@ class HistoScanSettingsModel(BaseModel):
dy: int
max_dist: int
skip_background: bool
capture_params: CaptureParams
autofocus_params: AutofocusParams
smart_stack_params: SmartStackParams
@ -361,7 +395,7 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
def all_settings(
self, images_dir: str
) -> tuple[HistoScanSettingsModel, StitchingSettings]:
"""Return the workflow and stitching settings.
"""Return the workflow settings 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
@ -370,11 +404,12 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
stitching_settings = self._get_stitching_settings_model()
dx, dy = self._calc_displacement_from_overlap(self.overlap)
smart_stack_params = self.create_smart_stack_params(
capture_params = CaptureParams(
images_dir=images_dir,
autofocus_dz=self.autofocus_dz,
save_resolution=self.save_resolution,
)
autofocus_params = AutofocusParams(dz=self.autofocus_dz)
smart_stack_params = self.create_smart_stack_params()
scan_settings = HistoScanSettingsModel(
overlap=self.overlap,
@ -382,6 +417,8 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
dx=dx,
dy=dy,
skip_background=self.skip_background,
capture_params=capture_params,
autofocus_params=autofocus_params,
smart_stack_params=smart_stack_params,
)
@ -389,15 +426,8 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
def create_smart_stack_params(
self,
images_dir: str,
autofocus_dz: int,
save_resolution: tuple[int, int],
) -> SmartStackParams:
"""Set up the parameters used for all stacks in a scan.
:param images_dir: the folder to save all images
:param autofocus_dz: the range to autofocus over if a stack fails
:param save_resolution: The resolution to save the captures to disk with
"""Set up the parameters used for all smart stacks in a scan.
:returns: A StackSmartParams object with the required parameters.
"""
@ -452,9 +482,6 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
stack_dz=self.stack_dz,
images_to_save=self.stack_images_to_save,
min_images_to_test=self.stack_min_images_to_test,
autofocus_dz=autofocus_dz,
images_dir=images_dir,
save_resolution=save_resolution,
)
def pre_scan_routine(self, settings: HistoScanSettingsModel) -> None:
@ -463,7 +490,7 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
:param settings: The settings for this scan as a HistoScanSettingsModel
"""
self._autofocus.looping_autofocus(
dz=settings.smart_stack_params.autofocus_dz, start="centre"
dz=settings.autofocus_params.dz, start="centre"
)
def new_scan_planner(
@ -518,6 +545,8 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
focus_height: Optional[int]
focused, focus_height = self._autofocus.run_smart_stack(
stack_parameters=settings.smart_stack_params,
capture_parameters=settings.capture_params,
autofocus_parameters=settings.autofocus_params,
save_on_failure=save_on_failure,
)
# An image was captured if we are focussed or we are not skipping background.
@ -575,9 +604,8 @@ class RegularGridSettingsModel(BaseModel):
x_count: int
y_count: int
style: Literal["snake", "raster"]
images_dir: str
autofocus_dz: int
save_resolution: tuple[int, int]
capture_params: CaptureParams
autofocus_params: AutofocusParams
class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
@ -604,6 +632,13 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
stitching_settings = self._get_stitching_settings_model()
dx, dy = self._calc_displacement_from_overlap(self.overlap)
capture_params = CaptureParams(
images_dir=images_dir,
save_resolution=self.save_resolution,
)
autofocus_params = AutofocusParams(dz=self.autofocus_dz)
scan_settings = self._settings_model(
overlap=self.overlap,
dx=dx,
@ -611,9 +646,8 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
x_count=self.x_count,
y_count=self.y_count,
style=self._grid_style,
images_dir=images_dir,
autofocus_dz=self.autofocus_dz,
save_resolution=self.save_resolution,
capture_params=capture_params,
autofocus_params=autofocus_params,
)
return scan_settings, stitching_settings
@ -625,7 +659,9 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
:param settings: The settings for this scan as as the relevant SettingsModel type.
"""
self._autofocus.looping_autofocus(dz=settings.autofocus_dz, start="centre")
self._autofocus.looping_autofocus(
dz=settings.autofocus_params.dz, start="centre"
)
def new_scan_planner(
self, settings: RegularGridSettingsModel, position: Mapping[str, int]
@ -659,9 +695,9 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
"""
return self._autofocus_and_capture(
xyz_pos=xyz_pos,
dz=settings.autofocus_dz,
images_dir=settings.images_dir,
save_resolution=settings.save_resolution,
dz=settings.autofocus_params.dz,
images_dir=settings.capture_params.images_dir,
save_resolution=settings.capture_params.save_resolution,
)
@lt.property

View file

@ -130,7 +130,7 @@ def test_histo_workflow_settings_generation(histo_workflow, mocker):
assert workflow_settings.dx == 123
assert workflow_settings.dy == 456
# And that the input image dir is passed to stack the stack parameter for saving
assert workflow_settings.smart_stack_params.images_dir == "/this/img_dir"
assert workflow_settings.capture_params.images_dir == "/this/img_dir"
# A CSM that is "normal" changing from camera maxtrix coordinates (y,x) to normal
@ -196,7 +196,7 @@ def test_histo_pre_scan_routine(histo_workflow, mocker):
# Rather than create a whole Setting class, just create a mock with the value
# we need set
mock_settings = mocker.Mock()
mock_settings.smart_stack_params.autofocus_dz = 1234
mock_settings.autofocus_params.dz = 1234
# Run the function
histo_workflow.pre_scan_routine(mock_settings)
# Check the autofocus was run using the mocked slot.

View file

@ -66,12 +66,7 @@ def test_stack_params_validation(save_ims, extra_ims):
# to do automatically in hypothesis. This clamps the number between 3 and 9.
min_images_to_test = max(min(save_ims + extra_ims, 9), 3)
SmartStackParams(
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=min_images_to_test,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
stack_dz=50, images_to_save=save_ims, min_images_to_test=min_images_to_test
)
@ -96,9 +91,6 @@ def test_stack_params_not_enough_test_images(save_ims, extra_ims):
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
)
@ -121,9 +113,6 @@ def test_stack_params_negative_images_to_save(save_ims, extra_ims):
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
)
@ -147,9 +136,6 @@ def test_even_min_images_to_test(save_ims, extra_ims):
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
)
@ -171,9 +157,6 @@ def test_even_images_to_save(save_ims, extra_ims):
stack_dz=50,
images_to_save=save_ims,
min_images_to_test=save_ims + extra_ims,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
)
@ -183,12 +166,7 @@ def test_computed_stack_params():
Not using hypothesis or we will just copy in the same formulas.
"""
stack_parameters = SmartStackParams(
stack_dz=50,
images_to_save=5,
min_images_to_test=9,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
stack_dz=50, images_to_save=5, min_images_to_test=9
)
assert stack_parameters.stack_z_range == 8 * 50
@ -279,9 +257,7 @@ def test_create_stack(histo_scan_workflow, caplog):
initial_min_images_to_test = histo_scan_workflow.stack_min_images_to_test
initial_images_to_save = histo_scan_workflow.stack_images_to_save
with caplog.at_level(logging.INFO):
stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
)
stack_params = histo_scan_workflow.create_smart_stack_params()
assert len(caplog.records) == 0
assert histo_scan_workflow.stack_min_images_to_test == initial_min_images_to_test
@ -308,9 +284,7 @@ def test_coercing_stack_test_ims(
histo_scan_workflow.stack_min_images_to_test = initial_test_ims
with caplog.at_level(logging.WARNING):
stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
)
stack_params = histo_scan_workflow.create_smart_stack_params()
assert len(caplog.records) == 1
assert str(caplog.records[0].msg).startswith(expected_log_start)
@ -338,9 +312,7 @@ def test_coercing_stack_save_ims(
histo_scan_workflow.stack_images_to_save = initial_save_ims
with caplog.at_level(logging.WARNING):
stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
)
stack_params = histo_scan_workflow.create_smart_stack_params()
assert len(caplog.records) == 1
assert str(caplog.records[0].msg).startswith(expected_log_start)
@ -353,9 +325,7 @@ def test_coercing_stack_save_ims(
@pytest.mark.parametrize("pass_on", [1, 2, 3, 4])
def test_run_smart_stack(pass_on, histo_scan_workflow, autofocus_thing, mocker):
"""Test Running smart stack with the stack passing on different attempts."""
stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
)
stack_params = histo_scan_workflow.create_smart_stack_params()
assert stack_params.max_attempts == 3
# Set up returns from z-stack
@ -417,9 +387,7 @@ def setup_and_run_smart_z_stack(
is a list, it will be set as a side effect (and should be a list of tuples of
results). If it a tuple (or anything else), it is set as a return value.
"""
stack_params = histo_scan_workflow.create_smart_stack_params(
autofocus_dz=2000, images_dir="/this/is/fake", save_resolution=(1640, 1232)
)
stack_params = histo_scan_workflow.create_smart_stack_params()
stack_params.settling_time = 0 # Don't settle or tests take forever.
autofocus_thing.capture_stack_image = mocker.Mock()