rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'fast-stack' into 'v3'

Browse source 
A stack that tests whether the sharpest image is towards the middle

See merge request openflexure/openflexure-microscope-server!271
This commit is contained in:
Julian Stirling 2025-06-27 20:02:36 +00:00
commit 3e0264bcbc
10 changed files with 1111 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -94,3 +94,6 @@ openflexure/
# js version file
/webapp/src/version.js
# Hypothesis
.hypothesis

3
pyproject.toml
View file

@ -37,7 +37,8 @@ dev = [
"mypy-gitlab-code-quality",
# "pytest-gitlab-code-quality", # pytest version constraint clashes with labthings
"pytest",
"matplotlib~=3.10"
"matplotlib~=3.10",
"hypothesis",
]
pi = [
"picamera2~=0.3.27",

63
src/openflexure_microscope_server/scan_planners.py
View file

@ -20,6 +20,12 @@ XYPosList: TypeAlias = list[XYPos]
XYZPosList: TypeAlias = list[XYZPos]
# 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.
# This wis based of a 4:3 aspect ratio. So x moves are 1.33 times larger than y
NEIGHBOUR_CUTOFF = 1.4
def enforce_xy_tuple(value: XYPos) -> XYPos:
"""
Used for enforcing that an input is a tuple and is the correct length
@ -345,6 +351,63 @@ class SmartSpiral(ScanPlanner):
self._remaining_locations.sort(key=sort_key)
def get_next_location_and_z_estimate(self) -> tuple[XYPos, Optional[int]]:
"""
Return the next location to scan, and the estimated z-position
for this location. This overrides the default behaviour of ScanPlanner
to take the lowest value of nearest neighbours as this works best for smart stack
Note z-position may be None! This indicates that the current z position
should be used.
"""
if self.scan_complete:
raise RuntimeError("Can't get next position, scan is complete")
next_location = self._remaining_locations[0]
# If focused locations exist, return the neighbour with the lowest z position
closest_pos = self.select_nearby_focus_site(next_location)
if closest_pos is None:
z = None
else:
z = closest_pos[2]
return next_location, z
def select_nearby_focus_site(self, xy_pos: XYPos) -> Optional[XYZPos]:
"""
Return the xyz position of the nearby site with the lowest z position.
Lowest position is best, as starting too high causes smart stacking to
autofocus and restart. Starting too low just requires extra movements in +z.
Nearby is defined as within NEIGHBOUR_CUTOFF times the distance to the closest neighbour.
Returns None if there if no focused locations are present
"""
if not self._focused_locations:
return None
# must be float64 (double precision) to deal with the huge numbers involved!
current_pos = np.array(xy_pos, dtype="float64")
path_pos = np.array(self._focused_locations, dtype="float64")[:, :2]
# Use linalg.norm to calculate the direct distance between the points
# Note linalg.norm always uses float64
dists = np.linalg.norm((path_pos - current_pos), axis=1)
# Get indices of all focused sites within NEIGHBOUR_CUTOFF the minimum distance.
# Note np.where always returns a tuple of arrays, hence the trailing [0]
indices = np.where(dists <= NEIGHBOUR_CUTOFF * np.min(dists))[0]
# Turning into an array allows slicing based on a list
focused_locations_array = np.array(self._focused_locations)
# Choose the lowest (smallest z) of the neighbouring sites. Smart stack works best
# if started too low, so the lowest z will perform best
chosen_focused_site = min(focused_locations_array[indices], key=lambda x: x[-1])
# Convert back into list so values are of type int instead of np.int32
return tuple(chosen_focused_site.tolist())
def moves_between(
self,
starting_pos: XYPos | np.ndarray,

486
src/openflexure_microscope_server/things/autofocus.py
View file

@ -9,10 +9,9 @@ See repository root for licensing information.
from contextlib import contextmanager
import logging
import time
from typing import Annotated, Mapping, Optional, Sequence
from typing import Annotated, Mapping, Optional, Sequence, Literal
import os
import shutil
import glob
from dataclasses import dataclass
from fastapi import Depends
import numpy as np
@ -21,17 +20,165 @@ from pydantic import BaseModel
from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.types.numpy import NDArray
from labthings_fastapi.dependencies.invocation import InvocationLogger
from .camera import RawCameraDependency as Camera
from .camera import CameraDependency as WrappedCamera
from .stage import StageDependency as Stage
STACK_OVERSHOOT = 200
SETTLING_TIME = 0.3
class StackParams:
"""A class for holding for scan parameters
All arguments are keyword only
:param stack_dz: The number of motor steps between images
:param images_to_save: The number of images to save to disk
:param min_images_to_test: The minimum number of images in the stack before, the
stack is evaluated for focus. As more images are captured evaluation of the focus
is always evaluated with the same number of images. i.e. if min_images_to_test=9,
then 9 images are captured, if the stack is not well focused, a 10th image is
captured and images 2 to 10 are evaluated for focus
:param autofocus_dz: The number of steps in a full autofocus (when required)
:param images_dir: The directory to save images to disk
:param save_resolution: The resolution to save the captures to disk with
"""
def __init__(
self,
*,
stack_dz: int,
images_to_save: int,
min_images_to_test: 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 save must be positive and odd")
self.stack_dz = stack_dz
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
settling_time: float = 0.3
"""Time (in seconds) between moving and capturing an image"""
backlash_correction: int = 250
"""
Distance (in steps) to overshoot a move and then undo, to account for backlash
"""
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
"""
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
"""
max_attempts: int = 3
"""Maximum number of times to attempt fast stack"""
@property
def stack_z_range(self) -> int:
"""The range of the z stack, in steps
Note that this is the range of the minimum number of images captured,
which is also the range of the images stored in memory that can be
saved."""
return self.stack_dz * (self.min_images_to_test - 1)
@property
def steps_undershoot(self) -> int:
"""
The distance to deliberately undershoot the estimated optimal starting point
"""
# Starting too low by "steps_undershoot" makes smart stacking faster.
# Starting a stack too high requires it to move to the start,
# autofocus and then re-stack. Starting slightly too low only
# requires extra +z movements and captures.
return self.stack_dz * self.img_undershoot
@property
def max_images_to_test(self) -> int:
"""The maximum number of images that will be captured and tested in a stack
This is 15 images more then the minimum number that are captured.
"""
return self.min_images_to_test + 15
def slice_to_save(self, sharpest_index):
"""Return the slice of images to save given the index of the sharpest image"""
images_each_side = (self.images_to_save - 1) // 2
return slice(
max(sharpest_index - images_each_side, 0),
sharpest_index + images_each_side + 1,
)
@dataclass
class CaptureInfo:
"""
The information from a capture in a z_stack
"""
buffer_id: int
position: dict[str, int]
sharpness: int
@property
def filename(self) -> str:
"""The filename for this image generated from the position"""
return f"{self.position['x']}_{self.position['y']}_{self.position['z']}.jpeg"
def _get_capture_by_id(captures: list[CaptureInfo], buffer_id: int) -> CaptureInfo:
"""Return the capture from a list of CaptureInfo objects with the matching id.
:param captures: A list of capture objects
:param buffer_id: The buffer id of the image to return
:returns: the CaptureInfo object of the capture with matching id
:raises: ValueError if buffer_id does not match the buffer_id of any captures
"""
return captures[_get_capture_index_by_id(captures, buffer_id)]
def _get_capture_index_by_id(captures: list[CaptureInfo], buffer_id: int) -> int:
"""Return the index of the capture with the matching id from a list of CaptureInfo
objects
:param captures: A list of capture objects
:param buffer_id: The buffer id of the image to return
:returns: the list index of the capture with matching id
:raises: ValueError if buffer_id does not match the buffer_id of any captures
"""
ids = [capture.buffer_id for capture in captures]
if buffer_id not in ids:
raise ValueError(f"No capture has a buffer id of {buffer_id}")
return ids.index(buffer_id)
class SharpnessDataArrays(BaseModel):
@ -117,9 +264,7 @@ class JPEGSharpnessMonitor:
stop = len(jpeg_times)
logging.debug("changing stop to %s", (stop))
jpeg_times = jpeg_times[start:stop]
jpeg_zs: np.ndarray = np.interp(
jpeg_times, stage_times, stage_zs
) # np.ndarray[float]
jpeg_zs: np.ndarray = np.interp(jpeg_times, stage_times, stage_zs)
return jpeg_times, jpeg_zs, jpeg_sizes[start:stop]
def sharpest_z_on_move(self, index: int) -> int:
@ -252,14 +397,24 @@ class AutofocusThing(Thing):
return heights.tolist(), sizes.tolist()
@thing_property
def stack_images_to_capture(self) -> int:
def stack_images_to_save(self) -> int:
"""The number of images to capture and save in a stack
Defaults to 1 unless you need to see either side of focus"""
return self.thing_settings.get("stack_images_to_capture", 1)
return self.thing_settings.get("stack_images_to_save", 1)
@stack_images_to_capture.setter
def stack_images_to_capture(self, value: int) -> None:
self.thing_settings["stack_images_to_capture"] = value
@stack_images_to_save.setter
def stack_images_to_save(self, value: int) -> None:
self.thing_settings["stack_images_to_save"] = value
@thing_property
def stack_min_images_to_test(self) -> int:
"""The number of images to test for successful focusing in a stack
Defaults to 9, which balances reliability and speed"""
return self.thing_settings.get("stack_min_images_to_test", 9)
@stack_min_images_to_test.setter
def stack_min_images_to_test(self, value: int) -> None:
self.thing_settings["stack_min_images_to_test"] = value
@thing_property
def stack_dz(self) -> int:
@ -274,85 +429,260 @@ class AutofocusThing(Thing):
self.thing_settings["stack_dz"] = value
@thing_action
def run_z_stack(
def run_smart_stack(
self,
cam: WrappedCamera,
stage: Stage,
logger: InvocationLogger,
metadata_getter: GetThingStates,
sharpness_monitor: SharpnessMonitorDep,
images_dir: str,
stack_dir: str,
capture_resolution: tuple[int, int],
autofocus_dz: int,
save_resolution: tuple[int, int],
) -> tuple[bool, int]:
"""Run a smart stack, which captures images offset in z, testing
whether the sharpest image is towards the centre of the stack.
The sharpest image, and optionally images around the sharpest,
will be saved using their coordinates to images_dir
:param cam: Camera Dependency supplied by LabThings dependency injection
:param stage: Stage Dependency supplied by LabThings dependency injection
:param sharpness_monitor: Sharpness Monitor Dependency (for focus detection)
supplied by LabThings dependency injection
: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 the images should be saved at, the
images will be resampled if this doesn't match the camera's capture resolution
:returns: A tuple containing:
- A boolean, True if stack was successfully
- The z position of the sharpest image
"""
# Set the variables to prevent changes from the GUI or other windows
stack_parameters = StackParams(
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,
)
trys = 0
# Loop until a stack is successful
while trys < stack_parameters.max_attempts:
success, captures, sharpest_id = self.z_stack(
stack_parameters=stack_parameters,
cam=cam,
stage=stage,
)
if success:
break
# The z position of the first images in the previous attempt.
initial_z_pos = captures[0].position["z"]
# If a stack is not successful, move to the start and autofocus
self.reset_stack(
initial_z_pos,
stack_parameters.autofocus_dz,
stage,
sharpness_monitor,
)
# Save stack_parameters.image_to_save images centred on the sharpest capture.
# If the smart_stack failed the exact number of images saved may not be
# stack_parameters.image_to_save
self.save_stack(
sharpest_id=sharpest_id,
captures=captures,
stack_parameters=stack_parameters,
cam=cam,
)
# Return whether or not the smart stack was successful, and the z position of
# the sharpest image, for path planning and tracking
return success, _get_capture_by_id(captures, sharpest_id).position["z"]
def reset_stack(
self,
initial_z_pos: list[int],
autofocus_dz: int,
stage: Stage,
sharpness_monitor: SharpnessMonitorDep,
) -> None:
"""Run a z stack, saving all images to stack_dir and copying the
central image to stack_dir"""
stack_dz = self.stack_dz
images_to_capture = self.stack_images_to_capture
"""Return to the initial height of the current stack, and run
a looping autofocus.
stack_z_range = stack_dz * (images_to_capture - 1)
if stack_z_range > 0:
# Perform backlash corrected move. See issue #420
stage.move_relative(z=-(STACK_OVERSHOOT + stack_z_range / 2))
stage.move_relative(z=STACK_OVERSHOOT)
time.sleep(0.3)
Arguments:
initial_z_pos: The initial z positions of previous captures
autofocus_dz: the range in steps to autofocus
variables stage and sharpness_monitor are Thing dependencies passed through from
the calling action
"""
stage.move_absolute(z=initial_z_pos)
self.looping_autofocus(
stage=stage,
sharpness_monitor=sharpness_monitor,
dz=autofocus_dz,
)
for capture_count in range(images_to_capture):
time.sleep(SETTLING_TIME)
def save_stack(
self,
sharpest_id: int,
captures: list[list],
stack_parameters: StackParams,
cam: WrappedCamera,
) -> int:
"""Save the required captures to disk. Will save the sharpest image,
and any images either side of focus.
jpeg_path = os.path.join(stack_dir, f"{capture_count}.jpeg")
start_time = time.time()
cam.capture_to_memory(logger=logger, metadata_getter=metadata_getter)
captured_time = time.time()
Arguments:
sharpest_id: the buffer id index of the sharpest image
captures: a list of captures, including file name, image data and metadata
stack_parameters: a StackParams object holding stack parameters
variables logger and capture are Thing dependencies passed through from the
calling action
"""
if capture_count + 1 < images_to_capture:
stage.move_relative(z=stack_dz)
moved_time = time.time()
sharpest_index = _get_capture_index_by_id(captures, sharpest_id)
slice_to_save = stack_parameters.slice_to_save(sharpest_index)
# Loop through the range, saving each capture to disk
for capture in captures[slice_to_save]:
cam.save_from_memory(
jpeg_path=jpeg_path,
logger=logger,
save_resolution=capture_resolution,
jpeg_path=os.path.join(stack_parameters.images_dir, capture.filename),
save_resolution=stack_parameters.save_resolution,
buffer_id=capture.buffer_id,
)
saved_time = time.time()
time_remaining = SETTLING_TIME - (saved_time - start_time)
if time_remaining > 0:
time.sleep(time_remaining)
logger.info(f"Settled for an extra {round(time_remaining, 3)} seconds")
logger.debug(f"Capturing took {round(captured_time - start_time, 2)} s")
logger.debug(f"Saving took {round(saved_time - moved_time, 2)} s")
logger.debug(
f"Effective settling time was {round(time.time() - moved_time, 2)} s"
cam.clear_buffers()
return sharpest_index
def z_stack(
self,
stack_parameters: StackParams,
cam: WrappedCamera,
stage: Stage,
) -> tuple[bool, list[CaptureInfo], Optional[int]]:
"""Capture a series of images offset by stack_parameters.stack_dz, and test whether
the sharpest image is towards the centre of the stack.
:param stack_parameters: a StackParams object holding stack parameters
:param cam: Camera Dependency to be passed through from the calling action
:param stage: Stage Dependency to be passed through from the calling action
:returns: A tuple of
- the stack result (True for successful stack, False for failed stack),
- a list of CaptureInfo objects,
- the buffer_id of the shapest image (or None if the stack failed)
"""
# Move down by the height of the z stack, plus an overshoot
# Better to start too low and take too many images than too high and need to refocus
stage.move_relative(
z=-(
stack_parameters.steps_undershoot
+ stack_parameters.backlash_correction
+ stack_parameters.stack_z_range / 2
)
)
stage.move_relative(z=stack_parameters.backlash_correction)
captures = []
# Always check for focus using the the last `min_images_to_test` in the
# stack so check is fair
ims_to_check = slice(-stack_parameters.min_images_to_test, None)
# If the sharpest image isn't found within the maximum number of images
# end the loop and return "restart"
while len(captures) < stack_parameters.max_images_to_test:
time.sleep(stack_parameters.settling_time)
# Append a new image to the stack
captures.append(
self.capture_stack_image(
cam,
stage,
buffer_max=stack_parameters.min_images_to_test,
)
)
self.copy_sharpest_image_from_stack(images_dir, stack_dir, logger)
# If the number of images is enough to test, test them
if len(captures) >= stack_parameters.min_images_to_test:
result, capture_id = self.check_stack_result(captures[ims_to_check])
def copy_central_image_from_stack(
if result == "success":
return True, captures, capture_id
if result == "restart":
return False, captures, None
# If reached here the result was "continue"
stage.move_relative(z=stack_parameters.stack_dz)
return False, captures, None
def capture_stack_image(
self,
images_dir: str,
stack_dir: str,
):
"""Gets a list of images in a folder (stack_dir), sorts them, and copies the central image
to images dir."""
image_list = glob.glob(os.path.join(stack_dir, "*"))
image_list.sort()
central_index = (len(image_list) - 1) // 2
central_image = image_list[central_index]
xy_location = os.path.basename(stack_dir)
cam: WrappedCamera,
stage: Stage,
buffer_max: int,
) -> CaptureInfo:
"""Capture another image and return the capture information.
shutil.copy(central_image, os.path.join(images_dir, f"{xy_location}.jpeg"))
The capture is stored by the camera Thing, and can be saved by ID.
def copy_sharpest_image_from_stack(
self,
images_dir: str,
stack_dir: str,
logger: InvocationLogger,
) -> None:
"""Gets a list of images in a folder (stack_dir), sorts them by filesize, and copies the sharpest
image to images dir."""
image_list = glob.glob(os.path.join(stack_dir, "*"))
image_list = sorted(image_list, key=os.path.getsize)
sharpest_image = image_list[-1]
xy_location = os.path.basename(stack_dir)
:param cam: Camera Dependency to be passed through from the calling action
:param stage: Stage Dependency to be passed through from the calling action
:buffer_max: The maximum number of images to tell the camera to keep in memory
for saving once the stack is complete
logger.info(sharpest_image)
shutil.copy(sharpest_image, os.path.join(images_dir, f"{xy_location}.jpeg"))
:return: A CaptureInfo object containing the capture information including its
camera buffer_id needed for saving.
"""
stage_location = stage.position
buffer_id = cam.capture_to_memory(buffer_max=buffer_max)
return CaptureInfo(
buffer_id=buffer_id,
position=stage_location,
sharpness=cam.grab_jpeg_size(stream_name="lores"),
)
def check_stack_result(
self, captures: list[CaptureInfo]
) -> tuple[Literal["success", "continue", "restart"], int]:
"""Test a list of captures, to decide whether the sharpest image from a
stack is centrally enough in the stack
:param captures: a list of the capture objects to for testing if the
sharpeness has converged in the centre
:return: A tuple with two values:
- result - which is one of three literal values:
'success' if the sharpest image is towards the centre
'continue' if the sharpest image is in the final two images of the list
'restart' if the sharpest image is in the first two images of the list
- capture_id - the buffer id of the sharpest image
"""
sharpest_index = np.argmax([capture.sharpness for capture in captures])
# The buffer id of the sharpest image
capture_id = captures[sharpest_index].buffer_id
sharpness_length = len(captures)
# If only testing one image, then by definition the sharpest is central
if sharpness_length == 1:
return "success", capture_id
# If testing three images, test if the centre is the sharpest
if sharpness_length == 3:
if sharpest_index == 1:
return "success", capture_id
if sharpest_index == 0:
return "restart", capture_id
return "continue", capture_id
# For larger stacks, test if the best image is not within two of the edge of the stack
# ie for a stack of 7 images, best image must be between 3rd and and 5th
exclusion_range = 2
if sharpest_index < exclusion_range:
return "restart", capture_id
if sharpest_index >= sharpness_length - exclusion_range:
return "continue", capture_id
return "success", capture_id

189
src/openflexure_microscope_server/things/camera/__init__.py
View file

@ -7,7 +7,7 @@ See repository root for licensing information.
"""
from __future__ import annotations
from typing import Literal, Optional, Tuple
from typing import Literal, Optional, Tuple, Any
import json
from pydantic import RootModel
@ -50,13 +50,116 @@ class NoImageInMemoryError(RuntimeError):
"""An error called if no image in in memory when an method is called to use that image"""
class CameraMemoryBuffer:
"""
A class that holds images in memory. The images are by default PIL images.
However subclasses of BaseCamera can use this class to store other object types
"""
_storage: dict[int, tuple[Any, Optional[dict]]]
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 = {}
# A simple id system where each capture id is just the number of captures since
# the server starts
self._latest_id: int = 0
def add_image(
self, image: Any, metadata: Optional[dict] = None, buffer_max: int = 1
) -> int:
"""
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._create_space(buffer_max)
self._storage[self._latest_id] = (image, metadata)
return self._latest_id
def get_image(
self, buffer_id: Optional[int] = None, remove: bool = True
) -> tuple[Any, Optional[dict]]:
"""
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
if buffer_id is None:
# Get the latest image and metadata tuple from storage
try:
image_tuple = list(self._storage.values())[-1]
except IndexError as e:
raise NoImageInMemoryError("No image in memory to retrieve.") from e
# Clear the storage so images don't get retrieved out of order
self._storage.clear()
return image_tuple
try:
if remove:
return self._storage.pop(buffer_id)
return self._storage[buffer_id]
except KeyError as e:
raise NoImageInMemoryError(
"No image with matching id in memory to retrieve."
) from e
def clear(self):
"""
Clear all images from memory
"""
self._storage.clear()
def _create_space(self, buffer_max: int) -> None:
"""
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 buffer_max <= 1:
self._storage.clear()
return
# Number to remove to get the storage down to 1 less than the buffer length
to_remove = len(self._storage) - (buffer_max - 1)
# If if there is space. Nothing to do, just return
if to_remove < 1:
return
keys_to_remove = list(self._storage.keys())[:to_remove]
for key in keys_to_remove:
del self._storage[key]
class BaseCamera(Thing):
"""The base class for all cameras. All cameras must directly inherit from this class"""
_memory_image: Optional[Image] = None
_memory_metadata: Optional[dict] = None
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
def __enter__(self) -> None:
raise NotImplementedError("CameraThings must define their own __enter__ method")
@ -64,17 +167,6 @@ class BaseCamera(Thing):
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
raise NotImplementedError("CameraThings must define their own __exit__ method")
@thing_property
def image_in_memory(self) -> bool:
"""True if an image is in memory ready to be saved"""
return self._memory_image is not None and self._memory_metadata is not None
@thing_action
def clear_image_memory(self) -> None:
"""Clear any image in memory"""
self._memory_image = None
self._memory_metadata = None
@thing_action
def start_streaming(
self, main_resolution: tuple[int, int], buffer_count: int
@ -133,6 +225,18 @@ class BaseCamera(Thing):
frame = portal.call(stream.grab_frame)
return JPEGBlob.from_bytes(frame)
@thing_action
def grab_jpeg_size(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size"""
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
return portal.call(stream.next_frame_size)
@thing_action
def capture_image(
self,
@ -154,9 +258,13 @@ class BaseCamera(Thing):
) -> None:
"""Capture an image and save it to disk
save_resolution can be set to resize the image before saving. By default this is None
meaning that the image is saved at original resoltion.
: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 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(
metadata_getter,
@ -176,17 +284,25 @@ class BaseCamera(Thing):
self,
logger: InvocationLogger,
metadata_getter: GetThingStates,
buffer_max: int = 1,
) -> None:
"""
Capture an image to memory. This can be saved later with `save_from_memory`
Note that only one image is held in memory so this will overwrite any image
in memory.
: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
"""
self._memory_image, self._memory_metadata = self._robust_image_capture(
metadata_getter,
logger=logger,
)
image, metadata = self._robust_image_capture(metadata_getter, logger)
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
@thing_action
def save_from_memory(
@ -194,21 +310,33 @@ class BaseCamera(Thing):
jpeg_path: str,
logger: InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None,
buffer_id: Optional[int] = None,
) -> None:
"""
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`
"""
if not self.image_in_memory:
raise NoImageInMemoryError("No image in memory to save.")
image, metadata = self._memory_buffer.get_image(buffer_id)
self._save_capture(
jpeg_path=jpeg_path,
image=self._memory_image,
metadata=self._memory_metadata,
image=image,
metadata=metadata,
logger=logger,
save_resolution=save_resolution,
)
self.clear_image_memory()
@thing_action
def clear_buffers(self) -> None:
"""Clear all images in memory"""
self._memory_buffer.clear()
def _robust_image_capture(
self,
@ -248,10 +376,13 @@ class BaseCamera(Thing):
if save_resolution is not None and image.size != save_resolution:
image = image.resize(save_resolution, Image.BOX)
try:
# Per PIL documentation, (https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg)
# there are two factors when saving a JPEG. subsampling affects the colour, quality the pixels
# Per PIL documentation,
# (https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg)
# there are two factors when saving a JPEG. Subsampling affects the colour,
# quality affects the pixels.
# subsampling = 0 disables subsampling of colour
# quality = 95 is the maximum recommended - above this, JPEG compression is disabled, file size increases and quality is barely or not affected
# quality = 95 is the maximum recommended - above this, JPEG compression is
# disabled, file size increases and quality is barely or not affected
image.save(jpeg_path, quality=95, subsampling=0)
try:
# Load EXIF metadata from image so it can be added to.

53
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -70,8 +70,6 @@ def ensure_free_disk_space(path: str, min_space: int = 500000000) -> None:
class ScanInfo(BaseModel):
"""Summary information about a scan folder"""
"""Summary information about a scan folder"""
name: str
created: datetime
modified: datetime
@ -423,7 +421,7 @@ class SmartScanThing(Thing):
"""
overlap = self.overlap
dx, dy = self._calc_displacement_from_test_image(overlap)
stitch_resize = STITCHING_RESOLUTION[0] / self.capture_resolution[0]
stitch_resize = STITCHING_RESOLUTION[0] / self.save_resolution[0]
self._scan_logger.debug(
f"Resizing images when stitching by a factor of {stitch_resize}"
@ -456,7 +454,7 @@ class SmartScanThing(Thing):
"skip_background": self.skip_background,
"stitch_automatically": self.stitch_automatically,
"stitch_resize": stitch_resize,
"capture_resolution": self.capture_resolution,
"save_resolution": self.save_resolution,
}
@_scan_running
@ -475,7 +473,7 @@ class SmartScanThing(Thing):
"dy": self._scan_data["dy"],
"start time": self._scan_data["start_time"],
"skipping background": self._scan_data["skip_background"],
"capture resolution": self._scan_data["capture_resolution"],
"capture resolution": self._scan_data["save_resolution"],
}
scan_inputs_fname = os.path.join(
@ -655,35 +653,18 @@ class SmartScanThing(Thing):
self._scan_logger.info(msg)
continue
focused = False
if self._scan_data["autofocus_on"]:
self._autofocus.looping_autofocus(
dz=self._scan_data["autofocus_dz"], start="centre"
)
current_pos_xyz = (
new_pos_xyz[0],
new_pos_xyz[1],
self._stage.position["z"],
)
# Without a test for autofocus success, assume it worked
focused = True
focused, focused_height = self._autofocus.run_smart_stack(
images_dir=self._ongoing_scan_images_dir,
autofocus_dz=self._scan_data["autofocus_dz"],
save_resolution=self._scan_data["save_resolution"],
)
current_pos_xyz = (new_pos_xyz[0], new_pos_xyz[1], focused_height)
route_planner.mark_location_visited(
current_pos_xyz, imaged=True, focused=focused
)
site_folder = os.path.join(
self._ongoing_scan_images_dir,
"stacks",
f"{new_pos_xyz[0]}_{new_pos_xyz[1]}",
)
os.makedirs(site_folder, exist_ok=True)
self._autofocus.run_z_stack(
images_dir=self._ongoing_scan_images_dir,
stack_dir=site_folder,
capture_resolution=self._scan_data["capture_resolution"],
)
# increment capure counter as thread has completed
self._scan_images_taken += 1
# Add it to the incremental zip
@ -752,14 +733,14 @@ class SmartScanThing(Thing):
return FileResponse(path)
@thing_property
def capture_resolution(self) -> tuple[int, int]:
def save_resolution(self) -> tuple[int, int]:
"""A tuple of the image resolution to capture. Should be in a
4:3 aspect ratio"""
return self.thing_settings.get("capture_resolution", ((1640, 1232)))
return self.thing_settings.get("save_resolution", ((1640, 1232)))
@capture_resolution.setter
def capture_resolution(self, value: tuple[int, int]) -> None:
self.thing_settings["capture_resolution"] = value
@save_resolution.setter
def save_resolution(self, value: tuple[int, int]) -> None:
self.thing_settings["save_resolution"] = value
@thing_property
def max_range(self) -> int:
@ -1126,8 +1107,8 @@ class SmartScanThing(Thing):
try:
with open(json_fpath, "r", encoding="utf-8") as data_file:
data_loaded = json.load(data_file)
capture_resolution = data_loaded["capture resolution"]
stitch_resize = STITCHING_RESOLUTION[0] / capture_resolution[0]
save_resolution = data_loaded["capture resolution"]
stitch_resize = STITCHING_RESOLUTION[0] / save_resolution[0]
except (json.decoder.JSONDecodeError, FileNotFoundError, TypeError):
# As there is no schema or pydantic model this should handle
# the file not being there, it not being json in the file,

210
tests/test_camera_buffer.py Normal file
View file

@ -0,0 +1,210 @@
"""
Tests for the built in buffer for the camera.
"""
from random import randint
import numpy as np
from PIL import Image
import pytest
from openflexure_microscope_server.things.camera import (
CameraMemoryBuffer,
NoImageInMemoryError,
)
RANDOM_GENERATOR = np.random.default_rng()
def random_image():
"""Create a random image"""
imarray = RANDOM_GENERATOR.integers(
low=0, high=255, size=(100, 100, 3), dtype="uint8"
)
return Image.fromarray(imarray)
def random_metadata():
"""Create a misc dictionary to pretend to be metadata"""
# Not very metadata like, but we are just checking that the same dict it
# is returned
return {"a": randint(1, 100), "b": randint(1, 100)}
def test_add_and_get_image():
"""Check images can be captured and retrieved"""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image)
returned_image, _ = mem_buf.get_image(buffer_id)
# It is the same image
assert misc_image is returned_image
# It is now removed from memory
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id)
def test_add_and_get_image_twice():
"""Check images can be retrieved twice if remove flag set false"""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image)
returned_image, _ = mem_buf.get_image(buffer_id, remove=False)
# It is the same image
assert misc_image is returned_image
# It is still in memory
returned_image, _ = mem_buf.get_image(buffer_id)
assert misc_image is returned_image
# It is now removed from memory
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id)
def test_get_without_id():
"""Check images can be captured and retrieved without ID"""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
mem_buf.add_image(misc_image)
returned_image, _ = mem_buf.get_image()
# It is the same image
assert misc_image is returned_image
# It is now removed from memory
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image()
def test_get_two_images():
"""Check two images can be retrieved."""
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, buffer_max=2)
buffer_id2 = mem_buf.add_image(misc_image2, buffer_max=2)
returned_image1, _ = mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
# It they the same images
assert misc_image1 is returned_image1
assert misc_image2 is returned_image2
# They are removed from memory
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(returned_image2)
def test_get_two_images_without_setting_buffer_size():
"""Check two images can't be retrieved if the buffer size isn't set"""
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1)
buffer_id2 = mem_buf.add_image(misc_image2)
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
# Image 2 the expected image
assert misc_image2 is returned_image2
def test_buffer_size_changing():
"""Check buffer size resets back to 1 when if not set"""
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
misc_image3 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, buffer_max=3)
buffer_id2 = mem_buf.add_image(misc_image2, buffer_max=3)
# Third capture doen't set buffer size, so it will be reset
buffer_id3 = mem_buf.add_image(misc_image3)
# As buffer size was reset, images 1 and 2 are deleted
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id2)
returned_image3, _ = mem_buf.get_image(buffer_id3)
# Image 3 the expected image
assert misc_image3 is returned_image3
def test_capture_two_images_get_without_id():
"""Check that all images are deleted when getting without id"""
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
mem_buf.add_image(misc_image1, buffer_max=2)
mem_buf.add_image(misc_image2, buffer_max=2)
returned_image, _ = mem_buf.get_image()
# When buffer_id is not specified, the most recent image (image2) is expected to
# be retrieved
assert returned_image is misc_image2
# Check all images were wiped from memory, but trying get_image without an id
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image()
def test_buffer_size_respected():
"""Capture 10 images with a buffer size of 5. Check only last 5 exist"""
mem_buf = CameraMemoryBuffer()
images = []
buffer_ids = []
for i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, buffer_max=5)
images.append(image)
buffer_ids.append(buffer_id)
for i, (image, buffer_id) in enumerate(zip(images, buffer_ids)):
if i < 5:
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id)
else:
returned_image, _ = mem_buf.get_image(buffer_id)
assert image is returned_image
def test_clear_buffer():
"""Capture 10 images clear the buffer and check they are gone"""
mem_buf = CameraMemoryBuffer()
images = []
buffer_ids = []
for i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, buffer_max=10)
images.append(image)
buffer_ids.append(buffer_id)
# Clear the buffer
mem_buf.clear()
# They are now gone
for i, (image, buffer_id) in enumerate(zip(images, buffer_ids)):
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id)
def test_get_metadata_too():
"""Capture 10 images with metadata and check metadata is returned as expected"""
mem_buf = CameraMemoryBuffer()
images = []
metadatas = []
buffer_ids = []
for i in range(10):
image = random_image()
metadata = random_metadata()
buffer_id = mem_buf.add_image(image, metadata, buffer_max=10)
images.append(image)
metadatas.append(metadata)
buffer_ids.append(buffer_id)
# Preallocate zipped data, to avoid long confusing lines
zipped = zip(images, metadatas, buffer_ids)
# Check both image and metdata
for i, (image, metadata, buffer_id) in enumerate(zipped):
returned_image, returned_metadata = mem_buf.get_image(buffer_id)
assert image is returned_image
assert metadata is returned_metadata

4
tests/test_scan_planners.py
View file

@ -183,8 +183,8 @@ def test_smart_spiral_first_few_pos():
xy_pos4, z_pos4 = planner.get_next_location_and_z_estimate()
# ...(100, 0)...
assert xy_pos4 == (100, 0)
# ... and it should get its focus from the most recent point as this was focussed
assert z_pos4 is new_z_focus
# ... and it should get its focus from the lowest neighbouring point
assert z_pos4 is z_focus
assert planner.closest_focus_site(xy_pos4) == xyz_pos3

237
tests/test_stack.py Normal file
View file

@ -0,0 +1,237 @@
"""
Tests for the smart/fast stacking.
Currently these tests don't test the Thing itself, just surrounding functionality
"""
from typing import Optional
from random import randint
import pytest
import numpy as np
from hypothesis import given, strategies as st
from openflexure_microscope_server.things.autofocus import (
StackParams,
CaptureInfo,
_get_capture_by_id,
_get_capture_index_by_id,
)
from openflexure_microscope_server.things.smart_scan import IMAGE_REGEX
RANDOM_GENERATOR = np.random.default_rng()
def odd_integers(min_value=0, max_value=1000):
"""A hypothesis strategy for odd integers"""
min_base = (min_value) // 2
max_base = (max_value - 1) // 2
# Ensure the range allows at least one odd number
if min_base > max_base:
return st.nothing()
return st.integers(min_value=min_base, max_value=max_base).map(lambda x: 2 * x + 1)
def even_integers(min_value=0, max_value=1000):
"""A hypothesis strategy for even integers"""
min_base = (min_value + 1) // 2
max_base = (max_value) // 2
# Ensure the range allows at least one even number
if min_base > max_base:
return st.nothing()
return st.integers(min_value=min_base, max_value=max_base).map(lambda x: 2 * x)
@given(
save_ims=odd_integers(min_value=0, max_value=10),
extra_ims=even_integers(min_value=0, max_value=10),
)
def test_stack_params_validation(save_ims, extra_ims):
"""Tests specifically the validation on the image numbers
save_ims is the number to save (must be odd and positive)
extra_ims is how many more images there are in min_images_to_test than
images_to_save. (even so that, min_images_to_test is odd and larger than
images_to_save
"""
StackParams(
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),
)
@given(
save_ims=odd_integers(min_value=0, max_value=10),
extra_ims=even_integers(min_value=-10, max_value=-1),
)
def test_stack_params_not_enough_test_images(save_ims, extra_ims):
"""Set the extra_ims negative so that min_images_to_test is smaller
than images_to_save.
For arguments see test_stack_params_validation
"""
with pytest.raises(ValueError):
StackParams(
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),
)
@given(
save_ims=odd_integers(min_value=-10, max_value=-1),
extra_ims=even_integers(min_value=0, max_value=10),
)
def test_stack_params_negative_images_to_save(save_ims, extra_ims):
"""save_ims is negative so images_to_save is negative, failing validation
For arguments see test_stack_params_validation
"""
with pytest.raises(ValueError):
StackParams(
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),
)
@given(
save_ims=odd_integers(min_value=0, max_value=10),
extra_ims=odd_integers(min_value=0, max_value=10),
)
def test_even_min_images_to_test(save_ims, extra_ims):
"""extra_ims is odd so min_images_to_test is even, failing validation
For arguments see test_stack_params_validation
"""
with pytest.raises(ValueError):
StackParams(
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),
)
@given(
save_ims=even_integers(min_value=0, max_value=10),
extra_ims=odd_integers(min_value=0, max_value=10),
)
def test_even_images_to_save(save_ims, extra_ims):
"""save_ims is even so images_to_save is even, failing validation
For arguments see test_stack_params_validation
"""
with pytest.raises(ValueError):
StackParams(
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),
)
def test_computed_stack_params():
"""
Test StackParams computed properties are as expected
Not using hypothesis or we will just copy in the same formulas.
"""
stack_parameters = StackParams(
stack_dz=50,
images_to_save=5,
min_images_to_test=9,
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
)
assert stack_parameters.stack_z_range == 8 * 50
assert stack_parameters.steps_undershoot == stack_parameters.img_undershoot * 50
assert stack_parameters.max_images_to_test == 9 + 15
sharpnesses = [50, 77, 234, 324, 390, 496, 569, 454, 333, 222, 178, 70]
max_ind = np.argmax(sharpnesses)
slice_to_save = stack_parameters.slice_to_save(max_ind)
# Check the slice corresponds to the index for the 5 images centred on the sharpest
assert sharpnesses[slice_to_save] == [390, 496, 569, 454, 333]
# For a failed smart stack the slice may be truncated. Check it truncates correctly
sharpnesses = [496, 569, 454, 333, 222, 178, 70, 69, 66, 50, 45, 40]
max_ind = np.argmax(sharpnesses)
slice_to_save = stack_parameters.slice_to_save(max_ind)
# Check the slice corresponds to the index for the first 4 images
assert sharpnesses[slice_to_save] == [496, 569, 454, 333]
# And again for sharpest at the end
sharpnesses = [13, 21, 26, 31, 39, 49, 50, 77, 234, 324, 390, 496, 569]
max_ind = np.argmax(sharpnesses)
slice_to_save = stack_parameters.slice_to_save(max_ind)
# Check the slice corresponds to the index for the final 3 images
assert sharpnesses[slice_to_save] == [390, 496, 569]
def random_capture(set_id: Optional[int] = None):
"""
Create a capture with random values
:param set_id: Optional, use to set a fixed id rather than a random one
"""
buffer_id = set_id if set_id is not None else randint(0, 1000)
return CaptureInfo(
buffer_id=buffer_id,
position={
"x": randint(-100000, 100000),
"y": randint(-100000, 100000),
"z": randint(-100000, 100000),
},
sharpness=randint(0, 100000),
)
def test_capture_filename_matches_regex():
"""
For 100 random captures check the image always matches the regex
"""
for _ in range(100):
assert IMAGE_REGEX.search(random_capture().filename)
@given(st.integers(min_value=0, max_value=5000))
def test_retrieval_of_captures(start):
"""
For 20 random captures, check each can be retrieved correctly by id
"""
captures = [random_capture(start + i) for i in range(20)]
for i, capture in enumerate(captures):
buffer_id = capture.buffer_id
assert _get_capture_index_by_id(captures, buffer_id) == i
assert _get_capture_by_id(captures, buffer_id) is capture
# Check errors are raised when supplying ids that aren't in the list
with pytest.raises(ValueError):
_get_capture_index_by_id(captures, start - 1)
with pytest.raises(ValueError):
_get_capture_index_by_id(captures, start + 21)
with pytest.raises(ValueError):
_get_capture_by_id(captures, start - 1)
with pytest.raises(ValueError):
_get_capture_by_id(captures, start + 21)

11
webapp/src/components/tabContentComponents/slideScanContent.vue
View file

@ -34,8 +34,15 @@
<div class="uk-margin">
<propertyControl
thing-name="autofocus"
property-name="stack_images_to_capture"
label="Images in Stack to Capture"
property-name="stack_images_to_save"
label="Images in Stack to Save"
/>
</div>
<div class="uk-margin">
<propertyControl
thing-name="autofocus"
property-name="stack_min_images_to_test"
label="Minimum number of images to test for focus"
/>
</div>
<div class="uk-margin">