Docstrings and stop using Pydantic for ScanParameters

This commit is contained in:
Julian Stirling 2025-06-26 19:46:25 +01:00
parent 43d5f2fdfe
commit 2aea6a2569
3 changed files with 113 additions and 61 deletions

View file

@ -21,7 +21,8 @@ XYZPosList: TypeAlias = list[XYZPos]
# how many times the minimum distance between images to include as a "nearby" image # how many times the minimum distance between images to include as a "nearby" image
# default 1.4 includes images offset in x or y, but not diagonally # default 1.4 includes images offset in x or y, but not diagonally.
# This wis based of a 4:3 aspect ratio. So x moves are 1.33 times larger than y
NEIGHBOUR_CUTOFF = 1.4 NEIGHBOUR_CUTOFF = 1.4

View file

@ -15,7 +15,7 @@ from dataclasses import dataclass
from fastapi import Depends from fastapi import Depends
import numpy as np import numpy as np
from pydantic import BaseModel, computed_field from pydantic import BaseModel
from labthings_fastapi.thing import Thing from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
@ -29,8 +29,10 @@ from .stage import StageDependency as Stage
# TODO Capture reolution should be save resolution for consistency # TODO Capture reolution should be save resolution for consistency
class StackParams(BaseModel): class StackParams:
"""Pydantic model for scan parameters """A class for holding for scan parameters
All arguments are keyword only
:param stack_dz: The number of motor steps between images :param stack_dz: The number of motor steps between images
:param images_to_save: The number of images to save to disk :param images_to_save: The number of images to save to disk
@ -42,41 +44,58 @@ class StackParams(BaseModel):
:param autofocus_dz: The number of steps in a full autofocus (when required) :param autofocus_dz: The number of steps in a full autofocus (when required)
:param images_dir: The directory to save images to disk :param images_dir: The directory to save images to disk
:param save_resolution: The resolution to save the captures to disk with :param save_resolution: The resolution to save the captures to disk with
""" """
# These parameters must be set on initialisation def __init__(
stack_dz: int self,
images_to_save: int *,
min_images_to_test: int stack_dz: int,
autofocus_dz: int images_to_save: int,
images_dir: str min_images_to_test: int,
save_resolution: tuple[int, int] autofocus_dz: int,
images_dir: str,
save_resolution: tuple[int, int],
) -> None:
if min_images_to_test < images_to_save:
raise ValueError("Can't save more images than the minimum number tested")
if min_images_to_test % 2 == 0 or min_images_to_test <= 0:
raise ValueError(
"Minimum number of images to test should be positive and odd"
)
if images_to_save % 2 == 0 or images_to_save <= 0:
raise ValueError("Images to svae must be positive and odd")
# The following parameters have values that apply well to a standard scan self.stack_dz = stack_dz
# and are not altered by default when running a scan self.images_to_save = images_to_save
self.min_images_to_test = min_images_to_test
self.autofocus_dz = autofocus_dz
self.images_dir = images_dir
self.save_resolution = save_resolution
# Using docstrings under variables as this is how pdoc would expect
# attributed to be documented
# time (in seconds) between moving and capturing an image
settling_time: float = 0.3 settling_time: float = 0.3
"""time (in seconds) between moving and capturing an image"""
# distance (in steps) to overshoot a move and then undo, to account for backlash
backlash_correction: int = 250 backlash_correction: int = 250
"""
distance (in steps) to overshoot a move and then undo, to account for backlash
"""
# how many images can be appended to the stack after the predicted peak to test
# for focus before assuming the focus was passed, and restarting the stack
stack_height_limit: int = 15 stack_height_limit: int = 15
"""
how many images can be appended to the stack after the predicted peak to test
for focus before assuming the focus was passed, and restarting the stack
"""
# how far below (in factors of stack_dz) the estimated optimal starting position to
# begin the stack. Better to start slightly too low and require many images, rather
# than too high and needing to autofocus and restart the stack
img_undershoot: int = 5 img_undershoot: int = 5
"""
how far below (in factors of stack_dz) the estimated optimal starting position to
begin the stack. Better to start slightly too low and require many images, rather
than too high and needing to autofocus and restart the stack
"""
# Per pydantic docs, even with the @property applied before @computed_field,
# mypy may throw a Decorated property not supported error (mypy issue #1362)
# To avoid this error message, add # type: ignore[prop-decorator] to the
# @computed_field line.
@computed_field
@property @property
def stack_z_range(self) -> int: def stack_z_range(self) -> int:
"""The range of the z stack, in steps """The range of the z stack, in steps
@ -86,7 +105,6 @@ class StackParams(BaseModel):
saved.""" saved."""
return self.stack_dz * (self.min_images_to_test - 1) return self.stack_dz * (self.min_images_to_test - 1)
@computed_field
@property @property
def steps_undershoot(self) -> int: def steps_undershoot(self) -> int:
""" """
@ -99,7 +117,6 @@ class StackParams(BaseModel):
# requires extra +z movements and captures. # requires extra +z movements and captures.
return self.stack_dz * self.img_undershoot return self.stack_dz * self.img_undershoot
@computed_field
@property @property
def max_images_to_test(self) -> int: def max_images_to_test(self) -> int:
"""The maximum number of images that will be captured and tested in a stack """The maximum number of images that will be captured and tested in a stack
@ -436,9 +453,6 @@ class AutofocusThing(Thing):
save_resolution=save_resolution, save_resolution=save_resolution,
) )
# Ensure the stack settings are appropriate
self.validate_stack_inputs(stack_parameters)
success = False success = False
# Loop until a stack is successful # Loop until a stack is successful
while not success: # TODO add a maximum number? while not success: # TODO add a maximum number?
@ -508,7 +522,7 @@ class AutofocusThing(Thing):
Arguments: Arguments:
sharpest_id: the buffer id index of the sharpest image sharpest_id: the buffer id index of the sharpest image
captures: a list of captures, including file name, image data and metadata captures: a list of captures, including file name, image data and metadata
stack_parameters: a StackParams Pydantic model with stack settings stack_parameters: a StackParams object holding stack parameters
variables logger and capture are Thing dependencies passed through from the variables logger and capture are Thing dependencies passed through from the
calling action calling action
""" """
@ -542,7 +556,7 @@ class AutofocusThing(Thing):
Arguments: Arguments:
images_dir: a string of the path to write all images images_dir: a string of the path to write all images
stack_parameters: a StackParams Pydantic model with stack settings stack_parameters: a StackParams object holding stack parameters
variables stage to metadata_getter are Thing dependencies passed through from the calling action variables stage to metadata_getter are Thing dependencies passed through from the calling action
""" """
# Move down by the height of the z stack, plus an overshoot # Move down by the height of the z stack, plus an overshoot
@ -606,25 +620,6 @@ class AutofocusThing(Thing):
sharpness=cam.grab_jpeg_size(stream_name="lores"), sharpness=cam.grab_jpeg_size(stream_name="lores"),
) )
def validate_stack_inputs(self, stack_parameters) -> None:
"""Check the stack settings are appropriate, and raise an error if not
Arguments:
min_images_to_test: number of images in the stack to test for focus
images_to_save: number of images to be captured around the focused image
"""
if stack_parameters.min_images_to_test < stack_parameters.images_to_save:
raise RuntimeError(
"Can't capture more images than are tested. Please increase number to test, or decrease number to capture"
)
if stack_parameters.min_images_to_test % 2 == 0:
raise RuntimeError("Images to test should be odd")
if (
stack_parameters.min_images_to_test <= 0
or stack_parameters.images_to_save <= 0
):
raise RuntimeError("Stack parameters need to be at least 1")
def check_stack_result( def check_stack_result(
self, captures: list[CaptureInfo] self, captures: list[CaptureInfo]
) -> tuple[Literal["success", "continue", "restart"], int]: ) -> tuple[Literal["success", "continue", "restart"], int]:

View file

@ -60,12 +60,31 @@ class CameraMemoryBuffer:
_storage: dict[int, tuple[Any, Optional[dict]]] _storage: dict[int, tuple[Any, Optional[dict]]]
def __init__(self): def __init__(self):
# This dictionary is the main store for data. Dictionaries are ordered since
# Python 3.6, so the order in the dictionary is the capture order
self._storage = {} self._storage = {}
# A simple id system where each capture id is just the number of captures since
# the server starts
self._latest_id: int = 0 self._latest_id: int = 0
def add_image( def add_image(
self, image: Any, metadata: Optional[dict], buffer_max: int = 1 self, image: Any, metadata: Optional[dict] = None, buffer_max: int = 1
) -> None: ) -> None:
"""
Add an image to the Memory buffer
This will add an image to the memory buffer. By default the buffer will
be cleared. To allow saving multiple images the buffer_max must be set
every time an image is added.
:param image: The image to add. A PIL image is recommended, but cameras
can choose to use other formats
:param metadata: Optional, a dictionary of the image metadata.
:param buffer_max: The maximum number of images that should be in the buffer
once this images is added. Default is 1.
:return buffer_id: The id in the buffer for this image
"""
self._latest_id += 1 self._latest_id += 1
self._create_space(buffer_max) self._create_space(buffer_max)
self._storage[self._latest_id] = (image, metadata) self._storage[self._latest_id] = (image, metadata)
@ -75,14 +94,22 @@ class CameraMemoryBuffer:
self, buffer_id: Optional[int] = None, remove: bool = True self, buffer_id: Optional[int] = None, remove: bool = True
) -> tuple[Any, Optional[dict]]: ) -> tuple[Any, Optional[dict]]:
""" """
If the buffer ID is not set, always clear whole buffer Return the image with the given id.
If no id is given the most recent image is returned. However, the
buffer is also cleared, otherwise it would be possible to accidentally
retrieve images out of order.
:param buffer_id: The buffer id of the image to retrieve
:param remove: True (default) to remove this image from the buffer, False
to leave the image in the buffer.
""" """
# No id given # No id given
if buffer_id is None: if buffer_id is None:
# Get the latest image and metadata tuple from storage # Get the latest image and metadata tuple from storage
try: try:
image_tuple = list(self._storage.value())[-1] image_tuple = list(self._storage.values())[-1]
except IndexError as e: except IndexError as e:
raise NoImageInMemoryError("No image in memory to save.") from e raise NoImageInMemoryError("No image in memory to save.") from e
# Clear the storage so images don't get retrieved out of order # Clear the storage so images don't get retrieved out of order
@ -94,11 +121,17 @@ class CameraMemoryBuffer:
return self._storage[buffer_id] return self._storage[buffer_id]
def clear(self): def clear(self):
"""
Clear all images from memory
"""
self._storage.clear() self._storage.clear()
def _create_space(self, buffer_max: int) -> None: def _create_space(self, buffer_max: int) -> None:
""" """
Create space to add an image. Create space to add an image.
:param buffer_max: The maximum number of images that should be in the buffer
once another images is added.
""" """
# If only one image to be stored just clear the storage and return # If only one image to be stored just clear the storage and return
if buffer_max <= 1: if buffer_max <= 1:
@ -220,9 +253,13 @@ class BaseCamera(Thing):
) -> None: ) -> None:
"""Capture an image and save it to disk """Capture an image and save it to disk
:param jpeg_path: The path to save the file to
save_resolution can be set to resize the image before saving. By default this :param logger: This should be injected automatically by Labthings FastAPI
is None meaning that the image is saved at original resoltion. when calling the action
:param metadata_getter: This should be injected automatically by Labthings
FastAPI when calling the action
:param save_resolution: can be set to resize the image before saving. By
default this is None meaning that the image is saved at original resolution.
""" """
image, metadata = self._robust_image_capture( image, metadata = self._robust_image_capture(
metadata_getter, metadata_getter,
@ -239,7 +276,10 @@ class BaseCamera(Thing):
@thing_action @thing_action
def capture_to_memory( def capture_to_memory(
self, logger: InvocationLogger, metadata_getter: GetThingStates, buffer_max: int self,
logger: InvocationLogger,
metadata_getter: GetThingStates,
buffer_max: int = 1,
) -> None: ) -> None:
""" """
Capture an image to memory. This can be saved later with `save_from_memory` Capture an image to memory. This can be saved later with `save_from_memory`
@ -247,7 +287,14 @@ class BaseCamera(Thing):
Note that only one image is held in memory so this will overwrite any image Note that only one image is held in memory so this will overwrite any image
in memory. in memory.
Return the buffer id of the image captured :param logger: This should be injected automatically by Labthings FastAPI
when calling the action
:param metadata_getter: This should be injected automatically by Labthings
FastAPI when calling the action
:param buffer_max: The maximum number of images that should be in the buffer
once this images is added. Default is 1.
:return: the buffer id of the image captured
""" """
image, metadata = self._robust_image_capture(metadata_getter, logger) image, metadata = self._robust_image_capture(metadata_getter, logger)
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max) return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
@ -262,6 +309,14 @@ class BaseCamera(Thing):
) -> None: ) -> None:
""" """
Save an image that has been captured to memory. Save an image that has been captured to memory.
:param jpeg_path: The path to save the file to
:param logger: This should be injected automatically by Labthings FastAPI
when calling the action
:param save_resolution: can be set to resize the image before saving. By
default this is None meaning that the image is saved at original resolution.
:param buffer_id: The buffer id of the image to save, this was returned by
`capture_to_memory`
""" """
image, metadata = self._memory_buffer.get_image(buffer_id) image, metadata = self._memory_buffer.get_image(buffer_id)
@ -275,6 +330,7 @@ class BaseCamera(Thing):
@thing_action @thing_action
def clear_buffers(self) -> None: def clear_buffers(self) -> None:
"""Clear all images in memory"""
self._memory_buffer.clear() self._memory_buffer.clear()
def _robust_image_capture( def _robust_image_capture(