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

Merge branch 'camera_modes' into 'v3'

Browse source 
Update camera API to use defined capture modes

Closes #424, #436, #786, #405, and #593

See merge request openflexure/openflexure-microscope-server!601
This commit is contained in:
Joe Knapper 2026-06-16 13:53:04 +00:00
commit d3c4773884
23 changed files with 992 additions and 492 deletions
Download patch file Download diff file Expand all files Collapse all files

BIN
picamera_coverage.zip
View file

Binary file not shown.

102
src/openflexure_microscope_server/things/__init__.py
View file

@ -5,9 +5,13 @@ with other Things and including them in the LabThings-FastAPI config file.
"""
import os
from pathlib import PurePath
from tempfile import TemporaryDirectory
from types import TracebackType
from typing import Optional, Self
from pydantic import PrivateAttr, RootModel, model_validator
import labthings_fastapi as lt
@ -34,6 +38,8 @@ class OFMThing(lt.Thing):
self._data_dir = os.path.join(
os.path.normpath(str(app_data_dir)), os.path.normpath(self.name)
)
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
return self
def __exit__(
@ -57,3 +63,99 @@ class OFMThing(lt.Thing):
"No data directory set. Has the LabThings server been started?"
)
return self._data_dir
def create_data_path(self, path: str, absolute: bool = False) -> "RelativeDataPath":
"""Create a ``RelativeDataPath`` object with this Thing set as the saving Thing.
:param path: The relative path within the data directory of this Thing's data
dir that the data should be saved to.
:param absolute: Set to True if the current path is absolute. A relative path
will be returned. A validation error will be raised if the absolute path
is not within the data directory.
:return: A ``RelativeDataPath`` object with the saving Thing already set.
"""
if absolute:
path = os.path.relpath(path, self.data_dir)
rel_data_path = RelativeDataPath(path)
rel_data_path.set_saving_thing(self)
return rel_data_path
class RelativeDataPath(RootModel[str]):
"""A relative path that is validated, and can have a Thing assigned to it.
Use ``set_saving_thing`` or ``set_saving_thing_if_unset`` to set the Thing whose
data directory will be used for the final save.
Use the ``abs_data_path`` property to get the final path for saving.
"""
_saving_thing: Optional[OFMThing | TemporaryDirectory] = PrivateAttr(default=None)
@model_validator(mode="before")
@classmethod
def validate_relative_path(cls, value: str) -> str:
"""Validate the relative path is relative and has no parent dir references."""
p = PurePath(value)
if p.is_absolute():
raise ValueError("Absolute paths are not allowed")
if ".." in p.parts:
raise ValueError("Parent directory references are not allowed")
return os.path.normpath(value)
@property
def save_location_set(self) -> bool:
"""Return True if the saving thing is set."""
return self._saving_thing is not None
def set_saving_thing(self, thing: OFMThing | TemporaryDirectory) -> None:
"""Set the Thing that is saving the data.
The thing can also be a ``TemporaryDirectory`` object.
This will set the data directory.
"""
if self.save_location_set:
raise RuntimeError("The saving Thing for the relative path is already set")
self._saving_thing = thing
def set_saving_thing_if_unset(self, thing: OFMThing) -> None:
"""Set the Thing that is saving the data if it is not already set.
Use this in an action to set the Thing for paths set via the API.
"""
if not self.save_location_set:
self._saving_thing = thing
def save_to_tempdir(self) -> TemporaryDirectory:
"""Use a temporary directory to save raher than an ``OFMThing``.
:returns: the ``TemporaryDirectory`` object.
"""
if self.save_location_set:
raise RuntimeError("The saving Thing for the relative path is already set")
self._saving_thing = TemporaryDirectory()
return self._saving_thing
def join(self, sub_path: str) -> "RelativeDataPath":
"""Join a path to the end of this path.
:return: A new ``RelativeDataPath`` object with the path appended.
"""
new_path = RelativeDataPath(os.path.join(self.root, sub_path))
if self._saving_thing is not None:
new_path.set_saving_thing(self._saving_thing)
return new_path
@property
def abs_data_path(self) -> str:
"""The absolute data directory to save to."""
if self._saving_thing is None:
raise RuntimeError("The saving Thing for the relative path was never set")
if isinstance(self._saving_thing, TemporaryDirectory):
return os.path.join(self._saving_thing.name, self.root)
return os.path.join(self._saving_thing.data_dir, self.root)

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

@ -9,7 +9,6 @@ See repository root for licensing information.
import enum
import logging
import os
import time
from dataclasses import dataclass
from types import TracebackType
@ -737,11 +736,9 @@ 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(capture_parameters.images_dir, capture.filename),
save_resolution=capture_parameters.save_resolution,
buffer_id=capture.buffer_id,
)
path = capture_parameters.images_dir.join(capture.filename)
self._cam.save_from_memory(path=path, buffer_id=capture.buffer_id)
self._cam.clear_buffers()
return sharpest_index
@ -823,7 +820,9 @@ class AutofocusThing(lt.Thing):
camera buffer_id needed for saving.
"""
stage_location = self._stage.position
buffer_id = self._cam.capture_to_memory(buffer_max=buffer_max)
buffer_id = self._cam.capture_to_memory(
capture_mode="standard", buffer_max=buffer_max
)
return CaptureInfo(
buffer_id=buffer_id,
position=stage_location,
@ -956,14 +955,8 @@ class AutofocusThing(lt.Thing):
# 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,
)
path = capture_parameters.images_dir.join(capture.filename)
self._cam.save_from_memory(path=path, buffer_id=capture.buffer_id)
self._cam.clear_buffers()

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

@ -11,22 +11,22 @@ from __future__ import annotations
import io
import json
import os
import tempfile
import time
from abc import ABC, abstractmethod
from copy import deepcopy
from datetime import datetime
from types import TracebackType
from typing import Annotated, Any, Literal, Mapping, Optional, Self, Tuple
from typing import Any, Literal, Mapping, Optional, Self
import numpy as np
import piexif
from PIL import Image
from pydantic import BaseModel, Field
from pydantic import BaseModel
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from openflexure_microscope_server.things import OFMThing
from openflexure_microscope_server.things import OFMThing, RelativeDataPath
from openflexure_microscope_server.things.background_detect import (
BackgroundDetectAlgorithm,
)
@ -34,31 +34,42 @@ from openflexure_microscope_server.ui import ActionButton, PropertyControl
from openflexure_microscope_server.utilities import coerce_thing_selector
class JPEGBlob(lt.blob.Blob):
"""A class representing a JPEG image as a LabThings FastAPI Blob."""
class ImageFormatInfo(BaseModel):
"""Basic data for image formats."""
media_type: str = "image/jpeg"
media_type: str
extension: str
supported_extensions: tuple[str, ...]
"""All supported extension (lowercase)."""
def path_matches(self, path: str) -> bool:
"""Return True if path matches one of the supported extensions."""
return path.lower().endswith(self.supported_extensions)
class PNGBlob(lt.blob.Blob):
"""A class representing a PNG image as a LabThings FastAPI Blob."""
media_type: str = "image/png"
BASE_IMAGE_FORMATS: dict[str, ImageFormatInfo] = {
"jpeg": ImageFormatInfo(
media_type="image/jpeg",
extension=".jpeg",
supported_extensions=(".jpeg", ".jpg"),
),
"png": ImageFormatInfo(
media_type="image/png",
extension=".png",
supported_extensions=(".png",),
),
}
class CaptureError(RuntimeError):
"""An error trying to capture from a CameraThing."""
PositiveInt = Annotated[int, Field(ge=1)]
NonEmptyString = Annotated[str, Field(min_length=1)]
class CaptureParams(BaseModel):
"""A class for capturing at least a single image."""
images_dir: NonEmptyString
save_resolution: tuple[PositiveInt, PositiveInt]
images_dir: RelativeDataPath
capture_mode: str
class NoImageInMemoryError(RuntimeError):
@ -71,7 +82,7 @@ class CameraMemoryBuffer:
However subclasses of BaseCamera can use this class to store other object types.
"""
_storage: dict[int, tuple[Any, Mapping[str, Any]]]
_storage: dict[int, tuple[Any, Mapping[str, Any], str]]
def __init__(self) -> None:
"""Create the buffer instance."""
@ -86,6 +97,7 @@ class CameraMemoryBuffer:
self,
image: Any,
metadata: Mapping[str, Any],
mode: str,
buffer_max: int = 1,
) -> int:
"""Add an image to the Memory buffer.
@ -104,12 +116,12 @@ class CameraMemoryBuffer:
"""
self._latest_id += 1
self._create_space(buffer_max)
self._storage[self._latest_id] = (image, metadata)
self._storage[self._latest_id] = (image, metadata, mode)
return self._latest_id
def get_image(
self, buffer_id: Optional[int] = None, remove: bool = True
) -> tuple[Any, Mapping[str, Any]]:
) -> tuple[Any, Mapping[str, Any], str]:
"""Return the image with the given id.
If no id is given the most recent image is returned. However, the
@ -169,12 +181,23 @@ class CameraMemoryBuffer:
class StreamingMode(BaseModel):
"""Description of streaming modes for the camera.
Cameras can sub class this to record camera specific information about the mode.
Cameras can sub class this to store camera specific information about the mode.
"""
description: str
class CaptureMode(BaseModel):
"""Description of still capture modes for the camera.
Cameras can sub class this to store camera specific information about the mode.
"""
description: str
save_resolution: Optional[tuple[int, int]] = None
"""The resolution to save the image. Use None to save as captured."""
class BaseCamera(OFMThing, ABC):
"""The base class for all cameras. All cameras must directly inherit from this class.
@ -191,6 +214,8 @@ class BaseCamera(OFMThing, ABC):
_memory_buffer = CameraMemoryBuffer()
supports_focus_fom: bool = False
supported_image_formats = deepcopy(BASE_IMAGE_FORMATS)
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise the base camera, this creates the background detectors.
@ -205,8 +230,10 @@ class BaseCamera(OFMThing, ABC):
# would be ideal.
self._default_background_detector = "bg_channel_deviations_luv"
self._background_detector_name: Optional[str] = None
self._framerate_monitor_running = False
if "default" and "full_resolution" not in self.streaming_modes:
required_modes = ("default", "full_resolution")
if not all(mode in self.streaming_modes for mode in required_modes):
raise KeyError(
f"Camera {type(self).__name__} doesn't define both a 'default' and a "
"'full_resolution' streaming mode."
@ -429,97 +456,11 @@ class BaseCamera(OFMThing, ABC):
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
@abstractmethod
@lt.action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 5,
) -> NDArray:
"""Acquire one image from the camera and return as an array."""
downsampled_array_factor: int = lt.property(default=2, ge=1)
"""The downsampling factor when calling capture_downsampled_array."""
@lt.action
def capture_downsampled_array(self) -> NDArray:
"""Acquire one image from the camera, downsample, and return as an array.
* The array is downsamples by the thing property `downsampled_array_factor`.
* The default capture array arguments are used.
This method provides the interface expected by the camera_stage_mapping.
"""
img = self.capture_array()
return downsample(self.downsampled_array_factor, img)
@lt.action
def capture_jpeg(
self,
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = None,
) -> JPEGBlob:
"""Acquire one image from the camera as a JPEG.
This will use the internal capture image functionally of capture_image of
the specific camera being used.
:param stream_name: A stream name supported by this camera.
:param wait: (Optional, float) Set a timeout in seconds. If None it will
use the default for the underlying camera.
"""
fname = datetime.now().strftime("%Y-%m-%d-%H%M%S.jpeg")
directory = tempfile.TemporaryDirectory()
jpeg_path = os.path.join(directory.name, fname)
img = self.capture_image(stream_name, wait)
capture_metadata = self._capture_metadata()
self._save_capture(
path=jpeg_path,
image=img,
metadata=capture_metadata,
)
return JPEGBlob.from_temporary_directory(directory, fname)
@lt.action
def capture_png(
self,
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = None,
) -> PNGBlob:
"""Acquire one image from the camera as a PNG.
This will use the internal capture image functionally of capture_image of
the specific camera being used.
:param stream_name: A stream name supported by this camera.
:param wait: (Optional, float) Set a timeout in seconds. If None it will
use the default for the underlying camera.
"""
fname = datetime.now().strftime("%Y-%m-%d-%H%M%S.jpeg")
directory = tempfile.TemporaryDirectory()
png_path = os.path.join(directory.name, fname)
img = self.capture_image(stream_name, wait)
capture_metadata = self._capture_metadata()
self._save_capture(
path=png_path,
image=img,
metadata=capture_metadata,
)
return PNGBlob.from_temporary_directory(directory, fname)
@lt.action
def grab_jpeg(
self,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
) -> lt.blob.Blob:
"""Acquire one image from the preview stream and return as blob of JPEG data.
Note: in rare cases the JPEG stream may be broken. This can cause an OS error
@ -527,7 +468,7 @@ class BaseCamera(OFMThing, ABC):
complete, this error will not be raised until the data is accessed. Consider
using ``grab_jpeg_as_array`` instead.
This differs from ``capture_jpeg`` in that it does not pause the MJPEG
This differs from ``capture`` in that it does not pause the MJPEG
preview stream. Instead, we simply return the next frame from that
stream (either "main" for the preview stream, or "lores" for the low
resolution preview). No metadata is returned.
@ -536,7 +477,9 @@ class BaseCamera(OFMThing, ABC):
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
frame = self._thing_server_interface.call_async_task(stream.grab_frame)
return JPEGBlob.from_bytes(frame)
blob = lt.blob.Blob.from_bytes(frame)
blob.media_type = BASE_IMAGE_FORMATS["jpeg"].media_type
return blob
@lt.action
def grab_as_array(
@ -549,7 +492,7 @@ class BaseCamera(OFMThing, ABC):
this method over directly grabbing the frame and converting to a numpy array
via PIL.
This differs from ``capture_array`` in that it does not pause the MJPEG
This differs from ``capture_as_array`` in that it does not pause the MJPEG
preview stream.
"""
stream = (
@ -575,32 +518,118 @@ class BaseCamera(OFMThing, ABC):
)
return self._thing_server_interface.call_async_task(stream.next_frame_size)
@lt.property
def capture_modes(self) -> Mapping[str, CaptureMode]:
"""Modes the camera can use for capturing."""
return {
"quick": CaptureMode(
description="Capture without altering the stream settings.",
),
"standard": CaptureMode(description="The standard capture mode."),
}
def _validate_capture_mode(self, capture_mode: str) -> str:
"""Check input capture mode exists, always returns a valid mode.
:param capture_mode: The capture mode to check. If this isn't valid a warning
will be logged.
:return: The input capture mode if it is supported, or "standard".
"""
if capture_mode not in self.capture_modes:
name = type(self).__name__
self.logger.warning(
f"{name} has no capture mode {capture_mode}. Using the 'standard' "
"capture mode instead."
)
capture_mode = "standard"
return capture_mode
@abstractmethod
def capture_image(
@lt.action
def capture_as_array(
self,
stream_name: Literal["main", "lores", "full"],
wait: Optional[float] = None,
) -> Image.Image:
"""Capture a PIL image from stream stream_name with timeout wait."""
capture_mode: str = "standard",
raw: bool = False,
) -> NDArray:
"""Acquire one image from the camera and return as an array."""
downsampled_array_factor: int = lt.property(default=2, ge=1)
"""The downsampling factor when calling capture_downsampled_array."""
@lt.action
def capture_and_save(
def capture_downsampled_array(self) -> NDArray:
"""Acquire one image from the camera, downsample, and return as an array.
* The array is downsampled by the thing property `downsampled_array_factor`.
* The default capture array arguments are used.
This method provides the interface expected by the camera_stage_mapping.
"""
img = self.capture_as_array(capture_mode="quick")
return downsample(self.downsampled_array_factor, img)
@lt.action
def capture(
self,
jpeg_path: str,
save_resolution: Optional[Tuple[int, int]] = None,
capture_mode: str = "standard",
image_format: str = "jpeg",
retain_image: bool = True,
) -> lt.blob.Blob:
"""Acquire one image from the camera.
This will use the internal capture image functionally of _capture_image of
the specific camera being used.
:param capture_mode: The mode to use, must be one of ``capture_modes``.
:param image_format: The image format to use, must be one of
``supported_image_formats``
:param retain_image: (Default True) True to save image to the microscope,
False to only save temporarily for transfer.
:returns: A LabThings Blob that with access to the captured file.
"""
format_info = self.supported_image_formats[image_format]
fname = datetime.now().strftime("%Y-%m-%d-%H%M%S") + format_info.extension
path = RelativeDataPath(fname)
tmpdir = None
if retain_image:
path.set_saving_thing(self)
else:
tmpdir = path.save_to_tempdir()
self.capture_and_save_to_path(path, capture_mode)
if tmpdir is None:
blob = lt.blob.Blob.from_file(path.abs_data_path)
else:
blob = lt.blob.Blob.from_temporary_directory(tmpdir, fname)
blob.media_type = format_info.media_type
return blob
def capture_and_save_to_path(
self,
path: RelativeDataPath,
capture_mode: str = "standard",
) -> None:
"""Capture an image and save it to disk.
:param jpeg_path: The path to save the file to
: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, capture_metadata = self._robust_image_capture()
This is not an action as it exposes a direct path for saving
self._save_capture(jpeg_path, image, capture_metadata, save_resolution)
:param path: The path to save the file to, this should be a ``RelativeDataPath``
object. If the saving Thing is not set for the path, the camera's data
directory will be used.
:param capture_mode: (Optional) The name of the capture mode as defined by the
camera.
"""
buffer_id = self.capture_to_memory(capture_mode=capture_mode)
self.save_from_memory(path=path, buffer_id=buffer_id)
@lt.action
def capture_to_memory(self, buffer_max: int = 1) -> int:
def capture_to_memory(
self, capture_mode: str = "standard", buffer_max: int = 1, max_attempts: int = 5
) -> int:
"""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
@ -608,66 +637,102 @@ class BaseCamera(OFMThing, ABC):
:param buffer_max: The maximum number of images that should be in the buffer
once this images is added. Default is 1.
:param max_attempts: The maximum number of times to attempt the capture.
:returns: the buffer id of the image captured
"""
image, metadata = self._robust_image_capture()
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
success = False
for capture_attempts in range(max_attempts):
try:
ofm_metadata = self._collect_ofm_metadata()
image = self._capture_image(capture_mode=capture_mode)
success = True
break
except TimeoutError:
self.logger.warning(
f"Attempt {capture_attempts + 1} to capture image timed out. "
"Do you have enough RAM?"
)
if not success:
raise CaptureError(
f"An error occurred while capturing after {max_attempts} attempts"
)
return self._memory_buffer.add_image(
image, ofm_metadata, capture_mode, buffer_max=buffer_max
)
@lt.action
def save_from_memory(
self,
jpeg_path: str,
save_resolution: Optional[Tuple[int, int]] = None,
path: RelativeDataPath,
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 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.
Note this is not exposed as an action as it allows arbitrary paths on disk to
be written to.
:param path: The path to save the file to
: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, mode = self._memory_buffer.get_image(buffer_id)
self._save_capture(
path=jpeg_path,
image=image,
metadata=metadata,
save_resolution=save_resolution,
)
mode_info = self.capture_modes[mode]
save_resolution = mode_info.save_resolution
path.set_saving_thing_if_unset(self)
resolved_path = path.abs_data_path
if save_resolution is not None and image.size != save_resolution:
image = image.resize(save_resolution, Image.Resampling.BOX)
try:
save_kwargs: dict[str, Any] = {}
if BASE_IMAGE_FORMATS["jpeg"].path_matches(resolved_path):
# 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
save_kwargs = {"quality": 95, "subsampling": 0}
if (
BASE_IMAGE_FORMATS["png"].path_matches(resolved_path)
and image.mode == "RGBX"
):
image = image.convert("RGB")
image.save(resolved_path, **save_kwargs)
try:
self._add_metadata_to_capture(resolved_path, dict(metadata))
except Exception:
# We need to capture any exception as there are many reasons metadata
# might not be added. We warn rather than log the error.
self.logger.exception(f"Failed to add metadata to {resolved_path}")
except Exception as e:
raise IOError(f"An error occurred while saving {resolved_path}") from e
@abstractmethod
def _capture_image(self, capture_mode: str = "standard") -> Image.Image:
"""Capture a PIL image from the camera.
This unlike the ``grab_*`` methods this may pause the stream or temporarily
switch streaming mode to capture the image if required by the mode.
"""
@lt.action
def clear_buffers(self) -> None:
"""Clear all images in memory."""
self._memory_buffer.clear()
def _robust_image_capture(self) -> Tuple[Image.Image, Mapping[str, Any]]:
"""Capture an image in memory and return it with metadata.
def _collect_ofm_metadata(self) -> dict:
"""Return the metadata for a capture.
This robust capturing method attempts to capture the image five times
each time with a 5 second timeout set.
:raises CaptureError: if the capture fails for any reason
:returns: tuple with PIL Image, and dictionary of metadata.
This is information from the thing states, the time, and make/model names.
"""
for capture_attempts in range(5):
try:
capture_metadata = self._capture_metadata()
image = self.capture_image(stream_name="main", wait=5)
return image, capture_metadata
except TimeoutError:
self.logger.warning(
f"Attempt {capture_attempts + 1} to capture image timed out. Do you have enough RAM?"
)
raise CaptureError("An error occurred while capturing after 5 attempts")
def _capture_metadata(self) -> dict:
"""Return the metadata for a capture, from the thing states, time and known names."""
metadata = self._thing_server_interface.get_thing_states()
current_time = datetime.now()
return {
@ -678,7 +743,7 @@ class BaseCamera(OFMThing, ABC):
"things_states": metadata,
}
def _add_metadata_to_capture(self, jpeg_path: str, capture_metadata: dict) -> None:
def _add_metadata_to_capture(self, path: str, capture_metadata: dict) -> None:
"""Add the EXIF metadata for a JPEG image.
This adds:
@ -687,7 +752,7 @@ class BaseCamera(OFMThing, ABC):
- Camera Make and Model
"""
# Load existing EXIF
exif_dict = piexif.load(jpeg_path)
exif_dict = piexif.load(path)
user_metadata = capture_metadata["things_states"]
capture_time = capture_metadata["capture_time"]
@ -721,45 +786,7 @@ class BaseCamera(OFMThing, ABC):
exif_dict["0th"][piexif.ImageIFD.Model] = capture_metadata["model"]
# Write the updated EXIF back to the file
piexif.insert(piexif.dump(exif_dict), jpeg_path)
def _save_capture(
self,
path: str,
image: Image.Image,
metadata: Mapping[str, Any],
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Save the captured image and metadata to disk.
A warning is logged if metadata cannot be added.
:raises IOError: if the file cannot be saved
nothing is returned on success
"""
if save_resolution is not None and image.size != save_resolution:
image = image.resize(save_resolution, Image.Resampling.BOX)
try:
save_kwargs: dict[str, Any] = {}
if path.endswith("jpg"):
# 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
save_kwargs = {"quality": 95, "subsampling": 0}
image.save(path, **save_kwargs)
try:
self._add_metadata_to_capture(path, dict(metadata))
except Exception:
# We need to capture any exception as there are many reasons metadata
# might not be added. We warn rather than log the error.
self.logger.exception(f"Failed to add metadata to {path}")
except Exception as e:
raise IOError(f"An error occurred while saving {path}") from e
piexif.insert(piexif.dump(exif_dict), path)
settling_time: float = lt.setting(default=0.2, ge=0)
"""The settling time when calling the ``settle()`` method."""

46
src/openflexure_microscope_server/things/camera/opencv.py
View file

@ -11,7 +11,7 @@ from __future__ import annotations
import logging
from threading import Thread
from types import TracebackType
from typing import Literal, Optional, Self
from typing import Optional, Self
import cv2
from PIL import Image
@ -136,41 +136,33 @@ class OpenCVCamera(BaseCamera):
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
self.capture_array()
self.capture_as_array()
@lt.action
def capture_array(
def capture_as_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "full",
wait: Optional[float] = None,
capture_mode: str = "standard",
raw: bool = False,
) -> NDArray:
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
It's likely to be highly inefficient - raw and/or uncompressed captures using
binary image formats will be added in due course.
"""
if wait is not None:
LOGGER.warning("OpenCV camera has no wait option. Use None.")
LOGGER.warning(f"OpenCV camera doesn't respect {stream_name=}")
"""Acquire one image from the camera and return as an array."""
if raw is True:
raise NotImplementedError(
"OpenCV camera camera doesn't support raw capture."
)
# Warn if the capture mode is incorrect, but don't read the coerced value as
# this camera only supports one mode.
self._validate_capture_mode(capture_mode)
ret, frame = self.cap.read()
if not ret:
raise RuntimeError("Failed to capture frame from camera.")
return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
def capture_image(
self,
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = None,
) -> Image.Image:
"""Acquire one image from the camera and return as a PIL image.
This function will produce a JPEG image.
"""
if wait is not None:
LOGGER.warning("OpenCV camera has no wait option. Use None.")
LOGGER.warning(f"OpenCV camera doesn't respect {stream_name=}")
return Image.fromarray(self.capture_array())
def _capture_image(self, capture_mode: str = "standard") -> Image.Image:
"""Acquire one image from the camera and return as a PIL image."""
# Warn if the capture mode is incorrect, but don't read the coerced value as
# this camera only supports one mode.
self._validate_capture_mode(capture_mode)
return Image.fromarray(self.capture_as_array())
@lt.property
def manual_camera_settings(self) -> list[PropertyControl]:

242
src/openflexure_microscope_server/things/camera/picamera.py
View file

@ -54,7 +54,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from . import BaseCamera, StreamingMode
from . import BaseCamera, CaptureMode, StreamingMode
from . import picamera_recalibrate_utils as recalibrate_utils
from . import picamera_tuning_file_utils as tf_utils
@ -112,6 +112,20 @@ class PiCamera2StreamingMode(StreamingMode):
return {"output_size": self.sensor_mode_resolution, "bit_depth": self.bit_depth}
class PiCamera2CaptureMode(CaptureMode):
"""Capture mode configuration for the PiCamera2."""
streaming_mode: Optional[str]
"""The streaming mode the camera should be in for capture.
Use None to use the active mode.
"""
stream_name: Literal["lores", "main"]
"""The Picamera stream name to capture."""
timeout: float = 5.0
"""The timeout. A number above 10 risks hardlocking the Pi."""
class StreamingPiCamera2(BaseCamera, ABC):
"""A Thing that provides and interface to the Raspberry Pi Camera.
@ -382,6 +396,7 @@ class StreamingPiCamera2(BaseCamera, ABC):
* On closing of the context manager the stream will restart.
"""
already_streaming = self.stream_active
streaming_mode = self.streaming_mode
with self._picamera_lock:
if pause_stream and already_streaming:
self._stop_streaming(stop_web_stream=False)
@ -389,7 +404,7 @@ class StreamingPiCamera2(BaseCamera, ABC):
yield self._picamera
finally:
if pause_stream and already_streaming:
self._start_streaming()
self._start_streaming(streaming_mode)
def __exit__(
self,
@ -409,6 +424,35 @@ class StreamingPiCamera2(BaseCamera, ABC):
def streaming_modes(self) -> Mapping[str, PiCamera2StreamingMode]:
"""Modes the camera can stream in."""
@abstractmethod
@lt.property
def capture_modes(self) -> Mapping[str, PiCamera2CaptureMode]:
"""Modes the camera can use for capturing."""
def _create_picam_config_from_mode_info(
self,
picam: Picamera2,
controls: dict[str, Any],
mode_info: PiCamera2StreamingMode,
) -> dict[str, Any]:
"""Create the dictionary to pass to ``Picamera2.configure``.
:param controls: The controls for the camera. Note that running
``_get_persistent_controls()`` should be done before getting the picamera
lock to ensure that the current settings are read from the camera.
"""
if mode_info.scaler_crop is not None:
controls["ScalerCrop"] = mode_info.scaler_crop
stream_config = picam.create_video_configuration(
main={"size": mode_info.main_resolution},
lores={"size": mode_info.lores_resolution, "format": "YUV420"},
sensor=mode_info.sensor_mode_dict,
controls=controls,
)
stream_config["buffer_count"] = mode_info.buffer_count
return stream_config
def _start_streaming(self, mode: str = "default") -> None:
"""Start the MJPEG stream. This is where persistent controls are sent to camera.
@ -425,24 +469,21 @@ class StreamingPiCamera2(BaseCamera, ABC):
raise ValueError(f"Unknown mode {mode}")
self.streaming_mode = mode
mode_info = self.streaming_modes[mode]
# This must be before getting the picamera hardware lock.
controls = self._get_persistent_controls()
if mode_info.scaler_crop is not None:
controls["ScalerCrop"] = mode_info.scaler_crop
mode_info = self.streaming_modes[mode]
with self._streaming_picamera() as picam:
try:
if picam.started:
picam.stop()
picam.stop_encoder() # make sure there are no other encoders going
stream_config = picam.create_video_configuration(
main={"size": mode_info.main_resolution},
lores={"size": mode_info.lores_resolution, "format": "YUV420"},
sensor=mode_info.sensor_mode_dict,
stream_config = self._create_picam_config_from_mode_info(
picam=picam,
controls=controls,
mode_info=mode_info,
)
stream_config["buffer_count"] = mode_info.buffer_count
picam.configure(stream_config)
LOGGER.info("Starting picamera MJPEG stream...")
stream_name = "lores" if mode_info.use_lores_as_preview else "main"
@ -487,65 +528,63 @@ class StreamingPiCamera2(BaseCamera, ABC):
cam.capture_metadata()
@contextmanager
def _switch_to_single_capture_mode(self) -> Iterator[Picamera2]:
"""Get the picamera lock, pause stream and switch into still capture config.
def _ensure_mode_for_capture(
self, capture_mode_info: PiCamera2CaptureMode
) -> Iterator[Picamera2]:
"""Ensure in correct mode for capture.
Restarts stream when complete.
If the camera is already in the correct mode, the stream isn't paused and
this is the same as using ``self._streaming_picamera()``.
Otherwise, pause stream, and switch mode. Mode is reset and stream
restarts after the context manager closes.
"""
mode_info = self.streaming_modes["full_resolution"]
with self._streaming_picamera(pause_stream=True) as cam:
LOGGER.debug("Reconfiguring camera for full resolution capture")
cam.configure(
cam.create_still_configuration(sensor=mode_info.sensor_mode_dict)
)
cam.start()
time.sleep(self._sensor_info.short_pause)
yield cam
required_streaming_mode = capture_mode_info.streaming_mode
if (
required_streaming_mode is None
or required_streaming_mode == self.streaming_mode
):
with self._streaming_picamera() as cam:
yield cam
else:
streaming_mode_info = self.streaming_modes[required_streaming_mode]
def capture_image(
self,
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = 0.9,
) -> Image.Image:
# This must be before getting the picamera hardware lock.
controls = self._get_persistent_controls()
with self._streaming_picamera(pause_stream=True) as cam:
LOGGER.debug("Reconfiguring camera for full resolution capture")
stream_config = self._create_picam_config_from_mode_info(
picam=cam,
controls=controls,
mode_info=streaming_mode_info,
)
cam.configure(stream_config)
cam.start()
time.sleep(self._sensor_info.short_pause)
yield cam
def _capture_image(self, capture_mode: str = "standard") -> Image.Image:
"""Acquire one image from the camera and return it as a PIL Image.
If the ``stream_name`` parameter is ``main`` or ``lores``, it will be captured
from the main preview stream, or the low-res preview stream, respectively. This
means the camera won't be reconfigured, and the stream will not pause (though
it may miss one frame).
If ``full`` resolution is requested, we will briefly pause the MJPEG stream and
reconfigure the camera to capture a full resolution image. This will capture an
image at the full resolution of the current sensor mode. If the current sensor
mode bins or crops the image, this may not be the native resolution of the
camera sensor.
:param stream_name: (Optional) The PiCamera2 stream to use, should be one of
["main", "lores", "full"]. Default = "main". Note that "raw" images cannot
be captured as PIL images. Use capture_array
:param wait: (Optional, float) Set a timeout in seconds. Default = 0.9s,
lower than the 1s timeout for the camera. This ensures that our code times
out and returns before the camera times out. If None is set the default
value of 0.9 will be used to prevent the possibility of the camera locking.
:param capture_mode: The capture mode to use. See the description field of each
mode in ``capture_modes`` for more detail.
:raises TimeoutError: if this time is exceeded during capture.
"""
if wait is None:
wait = 0.9
if stream_name in ["main", "lores", "raw"]:
with self._streaming_picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
elif stream_name == "full":
with self._switch_to_single_capture_mode() as cam:
return cam.capture_image(name="main", wait=wait)
else:
raise ValueError(f'Unknown stream name "{stream_name}"')
capture_mode = self._validate_capture_mode(capture_mode)
capture_mode_info = self.capture_modes[capture_mode]
with self._ensure_mode_for_capture(capture_mode_info) as cam:
return cam.capture_image(
capture_mode_info.stream_name, wait=capture_mode_info.timeout
)
@lt.action
def capture_array(
def capture_as_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 0.9,
capture_mode: str = "standard",
raw: bool = False,
) -> NDArray:
"""Acquire one image from the camera and return as an array.
@ -553,26 +592,26 @@ class StreamingPiCamera2(BaseCamera, ABC):
It's likely to be highly inefficient - raw and/or uncompressed captures using
binary image formats will be added in due course.
:param stream_name: (Optional) The PiCamera2 stream to use, should be one of
["main", "lores", "raw", "full"]. Default = "main"
:param wait: (Optional, float) Set a timeout in seconds. Default = 0.9s,
lower than the 1s timeout for the camera. This ensures that our code times
out and returns before the camera times out. If None is set the default
value of 0.9 will be used to prevent the possibility of the camera locking.
:param capture_mode: (Optional) The name of the capture mode as defined by the
camera.
:param raw: Whether to capture RAW data. Capturing RAW data may ignore some
of the camera mode settings.
:raises TimeoutError: if this time is exceeded during capture.
"""
if stream_name == "raw":
# Raw cannot used capture_image.
if wait is None:
wait = 0.9
with self._switch_to_single_capture_mode() as cam:
return cam.capture_array(name="raw", wait=wait)
if raw:
# Raw cannot use _capture_image.
capture_mode = self._validate_capture_mode(capture_mode)
capture_mode_info = self.capture_modes[capture_mode]
with self._ensure_mode_for_capture(capture_mode_info) as cam:
return cam.capture_array(name="raw", wait=capture_mode_info.timeout)
# Note that internally the PiCamera creates a PIL image and then converts to
# numpy with ``np.array(Image.open(io.BytesIO(self.make_buffer(name))))``.
# As such we use capture_image to get an Image from the picamera and return
# As such we use _capture_image to get an Image from the picamera and return
# as array
return np.array(self.capture_image(stream_name, wait))
return np.array(self._capture_image(capture_mode))
@lt.property
def camera_configuration(self) -> Mapping:
@ -946,6 +985,31 @@ class PiCameraV2(StreamingPiCamera2):
),
}
@lt.property
def capture_modes(self) -> Mapping[str, PiCamera2CaptureMode]:
"""Modes the camera can use for capturing."""
return {
"standard": PiCamera2CaptureMode(
description=(
"The standard mode, 2MP image created from downsampling an 8MP "
"capture."
),
save_resolution=(1640, 1232),
streaming_mode="full_resolution",
stream_name="main",
),
"full": PiCamera2CaptureMode(
description="A full resolution 8MP capture.",
streaming_mode="full_resolution",
stream_name="main",
),
"quick": PiCamera2CaptureMode(
description="Capture without altering the stream settings.",
streaming_mode=None,
stream_name="main",
),
}
class PiCameraHQ(StreamingPiCamera2):
"""A Thing that provides and interface to the Raspberry Pi Camera HQ."""
@ -971,9 +1035,10 @@ class PiCameraHQ(StreamingPiCamera2):
),
"full_resolution": PiCamera2StreamingMode(
description=(
"Streaming the camera in 12MP full resolution. The preview stream "
"sent to the UI will be the low resolution (lores) stream. "
"This allows better image capture at expense of preview quality."
"Streaming the camera with an 8MP area in full-resolution area "
"from the centre of the 12MP sensor.. The preview stream sent to "
"the UI will be the low resolution (lores) stream. This allows "
"better image capture at expense of preview quality."
),
main_resolution=(2800, 2800),
lores_resolution=(350, 350),
@ -983,3 +1048,30 @@ class PiCameraHQ(StreamingPiCamera2):
scaler_crop=(628, 120, 2800, 2800),
),
}
@lt.property
def capture_modes(self) -> Mapping[str, PiCamera2CaptureMode]:
"""Modes the camera can use for capturing."""
return {
"standard": PiCamera2CaptureMode(
description=(
"The standard mode, 2MP image created from downsampling an 8MP "
"capture from the centre of the 12MP sensor."
),
save_resolution=(1400, 1400),
streaming_mode="full_resolution",
stream_name="main",
),
"full": PiCamera2CaptureMode(
description=(
"An 8MP full resolution capture from the centre of the 12MP sensor."
),
streaming_mode="full_resolution",
stream_name="main",
),
"quick": PiCamera2CaptureMode(
description="Capture without altering the stream settings.",
streaming_mode=None,
stream_name="main",
),
}

37
src/openflexure_microscope_server/things/camera/simulation.py
View file

@ -14,7 +14,7 @@ import re
import time
from threading import Thread
from types import TracebackType
from typing import Literal, Optional, Self, overload
from typing import Optional, Self, overload
import numpy as np
from PIL import Image, ImageFilter
@ -479,10 +479,10 @@ class SimulatedCamera(BaseCamera):
"""
@lt.action
def capture_array(
def capture_as_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "full",
wait: Optional[float] = None,
capture_mode: str = "standard",
raw: bool = False,
) -> NDArray:
"""Acquire one image from the camera and return as an array.
@ -490,29 +490,26 @@ class SimulatedCamera(BaseCamera):
It's likely to be highly inefficient - raw and/or uncompressed captures using
binary image formats will be added in due course.
:param stream_name: Currently ignored, this argument exists to ensure consistent API across camera Things.
:param wait: Currently ignored, this argument exists to ensure consistent API across camera Things.
:param capture_mode: (Optional) The name of the capture mode as defined by the
camera.
:param raw: Raw Capture is not implemented for the simulation microscope.
Setting this to True will result in an error.
"""
if wait is not None:
LOGGER.warning("Simulation camera has no wait option. Use None.")
LOGGER.warning(f"Simulation camera camera doesn't respect {stream_name=}")
if raw is True:
raise NotImplementedError("Simulation camera doesn't support raw capture.")
# Warn if the capture mode is incorrect, but don't read the coerced value as
# this camera only supports one mode.
self._validate_capture_mode(capture_mode)
return np.array(self.generate_frame())
def capture_image(
self,
stream_name: Literal["main", "lores", "full"],
wait: Optional[float] = None,
) -> Image.Image:
def _capture_image(self, capture_mode: str = "standard") -> Image.Image:
"""Capture to a PIL image. This is not exposed as a ThingAction.
It is used for capture to memory.
:param stream_name: Currently ignored, this argument exists to ensure consistent API across camera Things.
:param wait: Currently ignored, this argument exists to ensure consistent API across camera Things.
"""
if wait is not None:
LOGGER.warning("Simulation camera has no wait option. Use None.")
LOGGER.warning(f"Simulation camera camera doesn't respect {stream_name=}")
# Warn if the capture mode is incorrect, but don't read the coerced value as
# this camera only supports one mode.
self._validate_capture_mode(capture_mode)
return self.generate_frame()
@lt.action

86
src/openflexure_microscope_server/things/scan_workflows.py
View file

@ -4,7 +4,6 @@ This module contains the base ``ScanWorkflow`` class that all workflows should s
as well as specific workflows.
"""
import os
from typing import (
Generic,
Literal,
@ -27,6 +26,7 @@ from openflexure_microscope_server.stitching import (
TARGET_STITCHING_DIMENSION,
StitchingSettings,
)
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.autofocus import (
MAX_TEST_IMAGE_COUNT,
MIN_TEST_IMAGE_COUNT,
@ -55,6 +55,10 @@ from openflexure_microscope_server.ui import (
SettingModelType = TypeVar("SettingModelType", bound=BaseModel)
class WorkflowStartError(lt.exceptions.InvocationError):
"""The scan workflow cannot start, as the requested configuration is invalid."""
class ScanWorkflow(Generic[SettingModelType], lt.Thing):
"""A base class for all Scanworkflows.
@ -75,7 +79,7 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
_planner_cls: type[ScanPlanner]
# All workflows set a save resolution
save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232))
capture_mode: str = lt.setting(default="standard")
"""A tuple of the image resolution to capture."""
# CSM may not be set, and isn't required for a workflow. Allow for it to exist or be None
@ -85,15 +89,19 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
_stage: BaseStage = lt.thing_slot()
_autofocus: AutofocusThing = lt.thing_slot()
def check_before_start(self, scan_name: str) -> None:
# 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
"""Check before the scan starts. Throw an error if the scan shouldn't start.
The scan_name is passed to this function to enable workflows to validate the
scan name if needed.
"""
raise NotImplementedError(
"Each specific ScanWorkflow must implement a check_before_start."
)
if self.capture_mode not in self._cam.capture_modes:
cam_name = type(self._cam).__name__
raise WorkflowStartError(
f"{cam_name} has no capure mode {self.capture_mode}"
)
@lt.property
def ready(self) -> bool:
@ -103,19 +111,27 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
)
def all_settings(
self, images_dir: str
) -> tuple[SettingModelType, Optional[StitchingSettings]]:
self, images_dir: RelativeDataPath
) -> tuple[SettingModelType, Optional[StitchingSettings], tuple[int, int]]:
"""Return the scan settings and the stitching settings.
- The specific settings for this scan workflow are returned as a Base Model of
the type set when defining the class.
- Stitiching settings are returned either as a StitchingSettings object or None
is returned if it is not possible to stitch the scan.
- The save resolution as determined by a test image.
"""
raise NotImplementedError(
"Each specific ScanWorkflow must implement a `all_settings` method."
)
def _get_save_resolution(self) -> tuple[int, int]:
"""Return the save resolution as determined by a test image."""
# Capture an example image.
image = self._cam._capture_image(capture_mode=self.capture_mode)
# Check size to create a unit faction for downsampling.
return image.size
def pre_scan_routine(self, settings: SettingModelType) -> None:
"""Overload to set the routine that happens before each scan."""
raise NotImplementedError(
@ -148,14 +164,14 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
self,
xyz_pos: tuple[int, int, int],
dz: int,
images_dir: str,
save_resolution: tuple[int, int],
images_dir: RelativeDataPath,
capture_mode: str,
) -> tuple[bool, Optional[int]]:
"""Autofocus and then capture, this can be used as an acquisition routine.
:param dz: The dz for autofocus.
:param images_dir: The path to the directory for saving images..
:param save_resolution: The resolution to save images at.
:param images_dir: The path to the directory for saving images.
:param capture_mode: The name of the camera capture mode.
:return: A tuple ready to pass out of acquisition routine. In this method,
image is always taken, so first return is True.
@ -164,9 +180,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
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,
self._cam.capture_and_save_to_path(
path=images_dir.join(filename),
capture_mode=capture_mode,
)
return True, focus_height
@ -217,16 +233,15 @@ class RectGridWorkflow(
must be above this. 3000 is a sensible limit for 20x objectives.
"""
# 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
def check_before_start(self, scan_name: str) -> None:
"""Before starting a scan, check that camera-stage-mapping is set.
Raise error if:
- camera stage mapping is not set
"""
super().check_before_start(scan_name)
if self._csm.calibration_required:
raise RuntimeError("Camera Stage Mapping is not calibrated.")
raise WorkflowStartError("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.
@ -261,11 +276,11 @@ class RectGridWorkflow(
)
return y_move_stage["x"], x_move_stage["y"]
def _get_stitching_settings_model(self) -> StitchingSettings:
def _get_stitching_settings_model(
self, save_resolution: tuple[int, int]
) -> StitchingSettings:
"""Return a stitching settings model based on current settings."""
# Use the save resolution and target stitch resolution to choose a unit fraction,
# which makes correlating faster
width, height = self.save_resolution
width, height = save_resolution
# Target area in pixels
target_area = TARGET_STITCHING_DIMENSION**2
# Find N so that (width/N) * (height/N) ~ target_area
@ -285,17 +300,18 @@ class RectGridWorkflow(
return self._settings_model(**base_kwargs)
def all_settings(
self, images_dir: str
) -> tuple[RectGridSettingModelType, Optional[StitchingSettings]]:
self, images_dir: RelativeDataPath
) -> tuple[RectGridSettingModelType, Optional[StitchingSettings], tuple[int, int]]:
"""Return scan settings and the 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.
:return: A tuple containing the settings model for this workflow, the
settings model for stitching, and the save resolution.
"""
save_resolution = self._get_save_resolution()
# Developer Note: When subclassing RectGridWorkflow rather than override
# this method first consider overriding _build_scan_settings
stitching_settings = self._get_stitching_settings_model()
stitching_settings = self._get_stitching_settings_model(save_resolution)
dx, dy = self._calc_displacement_from_overlap(self.overlap)
base_kwargs = {
@ -303,14 +319,14 @@ class RectGridWorkflow(
"dx": dx,
"dy": dy,
"capture_params": CaptureParams(
images_dir=images_dir, save_resolution=self.save_resolution
images_dir=images_dir, capture_mode=self.capture_mode
),
"autofocus_params": AutofocusParams(dz=self.autofocus_dz),
}
scan_settings = self._build_scan_settings(base_kwargs)
return scan_settings, stitching_settings
return scan_settings, stitching_settings, save_resolution
@lt.property
def ready(self) -> bool:
@ -510,20 +526,22 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel], SmartStackMixi
# 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 background and camera-stage-mapping are set.
"""Before starting a scan, check that background and CSM are set.
Raise error if:
- background is to be skipped but is not set
- camera stage mapping is not set
Raise warning if not using background detect that scan will go on until max steps reached
Raise warning if not using background detect that scan will go on until max
steps reached.
"""
super().check_before_start(scan_name)
if self._csm.calibration_required:
raise RuntimeError("Camera Stage Mapping is not calibrated.")
raise WorkflowStartError("Camera Stage Mapping is not calibrated.")
if self.skip_background:
if not self._background_detector.ready:
raise RuntimeError(
raise WorkflowStartError(
"Background is not set: you need to calibrate background detection."
)
else:

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

@ -304,8 +304,8 @@ class SmartScanThing(OFMThing):
# Ensure any PreviewStitcher created cannot be reused.
self._preview_stitcher = None
# Remove any scan folders containing zero images.
self.purge_empty_scans()
# Remove any scan folders containing zero images.
self.purge_empty_scans()
@_scan_running
def _move_to_next_point(
@ -343,8 +343,8 @@ class SmartScanThing(OFMThing):
if images_dir is None:
raise RuntimeError("Couldn't run scan, images directory was not created.")
workflow_settings, stitching_settings = workflow.all_settings(
images_dir=images_dir
workflow_settings, stitching_settings, save_resolution = workflow.all_settings(
images_dir=self.create_data_path(images_dir, absolute=True)
)
# If stitching settings is None then this workflow doesn't support stitching.
@ -356,7 +356,7 @@ class SmartScanThing(OFMThing):
starting_position=starting_position,
start_time=datetime.now(),
stitch_automatically=auto_stitch,
save_resolution=workflow.save_resolution,
save_resolution=save_resolution,
workflow=type(workflow).__name__,
workflow_settings=workflow_settings,
stitching_settings=stitching_settings,
@ -392,6 +392,9 @@ class SmartScanThing(OFMThing):
starting x,y,z position.
"""
try:
# Change into streaming mode instantly so mode is correct when collecting
# Scan Data
self._cam.change_streaming_mode(mode="full_resolution")
self._scan_data = self._collect_scan_data(workflow)
images_dir = self.ongoing_scan.images_dir
# Type narrowing
@ -401,7 +404,6 @@ class SmartScanThing(OFMThing):
)
self.ongoing_scan.save_scan_data(self._scan_data)
self._cam.change_streaming_mode(mode="full_resolution")
workflow.pre_scan_routine(self._scan_data.workflow_settings)
# If stitching settings are None then this type of scan can't be stitched

3
src/openflexure_microscope_server/things/stage_measure.py
View file

@ -197,9 +197,6 @@ class RangeofMotionThing(OFMThing):
"time": total_time,
}
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
datafile_path = os.path.join(self.data_dir, f"rom_data_{timestamp}.json")
with open(datafile_path, "w", encoding="utf-8") as f:

50
tests/hardware_specific_tests/picamera2/test_acquisition.py
View file

@ -7,11 +7,8 @@ import numpy as np
from PIL import Image
def test_jpeg_and_array(picamera_client):
"""Check that a jpeg grabbed from the stream is the same size as other captures.
Compare it to an array capture and a jpeg capture.
"""
def test_quick_capture_size(picamera_client):
"""Check that a jpeg grabbed from the stream is the same size as a quick capture."""
# Grab a jpeg from the stream
blob = picamera_client.grab_jpeg()
mjpeg_frame = Image.open(blob.open())
@ -20,13 +17,17 @@ def test_jpeg_and_array(picamera_client):
assert mjpeg_frame.format == "JPEG"
# Capture a jpeg
blob = picamera_client.capture_jpeg(stream_name="main")
blob = picamera_client.capture(
capture_mode="quick",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.format == "JPEG"
# Capture an array
arrlist = picamera_client.capture_array(stream_name="main")
arrlist = picamera_client.capture_as_array(capture_mode="quick")
array_main = np.array(arrlist)
# Verify image sizes are the same
@ -34,6 +35,41 @@ def test_jpeg_and_array(picamera_client):
assert array_main.shape[1::-1] == jpeg_capture.size
def test_format(picamera_client):
"""Check capture format is as requested."""
# Capture a jpeg
blob = picamera_client.capture(
capture_mode="quick",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.format == "JPEG"
blob = picamera_client.capture(
capture_mode="quick",
image_format="png",
retain_image=True,
)
png_capture = Image.open(blob.open())
png_capture.verify()
assert png_capture.format == "PNG"
def test_standard_capture_size(picamera_client):
"""Check standard capture mode captures at expected size."""
# Capture a jpeg
blob = picamera_client.capture(
capture_mode="standard",
image_format="jpeg",
retain_image=True,
)
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.size == (1640, 1232)
def test_record_framerate(picamera_client):
"""Check that framerate monitoring creates a valid JSON log with good data."""
log_file = Path(picamera_client.record_framerate(duration=1.0))

8
tests/hardware_specific_tests/picamera2/test_exposure_time_drift.py
View file

@ -39,7 +39,7 @@ def _test_exposure_time_drift(desired_time: int) -> None:
assert abs(pre_capture_et - desired_time) < EXPOSURE_TOL
for i in range(10):
client.capture_jpeg(stream_name="full")
client.capture(capture_mode="full")
if i == 0:
# Exposure can update on first capture, due to frame rate restrictions
first_et = client.exposure_time
@ -56,7 +56,7 @@ def _test_exposure_time_drift(desired_time: int) -> None:
time.sleep(0.5)
# Check before and after capture
assert client.exposure_time == frame_et
client.capture_jpeg(stream_name="full")
client.capture(capture_mode="full")
assert client.exposure_time == frame_et
print("Exposure time didn't change!!")
print(f"End of test for exposure target {desired_time}")
@ -82,7 +82,7 @@ def test_exposure_time_on_start_and_stop_stream():
# Take a couple of images to make sure that the exposure is adjusted to
# a hardware compatible value.
for _i in range(2):
client.capture_jpeg(stream_name="full")
client.capture(capture_mode="full")
# Save this time.
set_time = client.exposure_time
assert abs(set_time - desired_time) < EXPOSURE_TOL
@ -112,7 +112,7 @@ def _load_camera_and_return_exposure(tmpdir: str) -> int:
# Take a couple of images to make sure that the exposure is adjusted to
# a hardware compatible value.
for _i in range(2):
client.capture_jpeg(stream_name="full")
client.capture(capture_mode="full")
# Save this time.
return client.exposure_time

2
tests/hardware_specific_tests/picamera2/test_streaming_mode.py
View file

@ -14,7 +14,7 @@ def test_streaming_mode():
with camera_test_client() as client:
for mode, res in ["default", (820, 616)], ["full_resolution", (3280, 2464)]:
client.change_streaming_mode(mode=mode)
arr = np.array(client.capture_array(stream_name="main"))
arr = np.array(client.capture_as_array(capture_mode="quick"))
# Check that the array dimensions match the requested image size.
# Note: Numpy array shape is (y,x), but the sensor is set with (x,y)
# hence the need to compare index 0 with index 1.

29
tests/integration_tests/test_actions.py
View file

@ -12,6 +12,7 @@ it has been moved as it artificaially inflated coverage.
"""
import json
import logging
import numpy as np
import piexif
@ -40,11 +41,26 @@ def test_grab_jpeg(simulation_test_env):
assert image.size == (820, 616)
@pytest.mark.parametrize("fmt", ["jpeg", "png"])
def test_capture_and_metadata(simulation_test_env, fmt):
"""Check that the position is encoded into the image metadata."""
@pytest.mark.parametrize(
"image_format",
[
"jpeg",
pytest.param(
"png", marks=pytest.mark.xfail(reason="piexif does not support PNG EXIF")
),
],
)
def test_capture_and_metadata(simulation_test_env, image_format, caplog):
"""Capture an image and check a attributes.
- Check that the position is encoded into the image metadata
- Check the dimensions
- Check the format
"""
camera = simulation_test_env.get_thing_client("camera")
blob = getattr(camera, f"capture_{fmt}")()
with caplog.at_level(logging.WARNING):
blob = camera.capture(image_format=image_format)
assert len(caplog.messages) == 0
image = Image.open(blob.open())
exif_dict = piexif.load(image.info["exif"])
encoded_metadata = exif_dict["Exif"][piexif.ExifIFD.UserComment]
@ -52,6 +68,7 @@ def test_capture_and_metadata(simulation_test_env, fmt):
assert "position" in metadata["stage"]
assert metadata["stage"]["position"] == {"x": 0, "y": 0, "z": 0}
assert image.size == (820, 616)
assert image.format == image_format.upper()
def test_stage(simulation_test_env):
@ -77,10 +94,10 @@ def test_stage(simulation_test_env):
assert start["z"] == pos["z"]
def test_capture_array(simulation_test_env):
def test_capture_as_array(simulation_test_env):
"""Capture array from simulation and check the size is as expected."""
camera = simulation_test_env.get_thing_client("camera")
array = np.asarray(camera.capture_array())
array = np.asarray(camera.capture_as_array())
assert array.shape == (616, 820, 3)

76
tests/unit_tests/test_base_camera.py
View file

@ -5,10 +5,16 @@ test_simulated_camera.py and for testing the consistency of camera APIs see
test_cameras.py.
"""
import os
from dataclasses import dataclass, field
from typing import Optional
import numpy as np
import pytest
from PIL import Image
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.camera import CaptureMode
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
from openflexure_microscope_server.things.stage.dummy import DummyStage
@ -61,3 +67,73 @@ def test_handle_broken_frame(test_env):
for _i in range(15):
array = camera.grab_as_array()
assert isinstance(array, np.ndarray)
@dataclass
class MemorySaveTestCase:
"""Inputs and expected outputs for testing ``save_from_memory``.
The default save kwargs assume a jpeg.
"""
filename: str = "foobar.jpeg"
save_resolution: Optional[tuple[int, int]] = None
resize_needed: bool = False
convert_needed: bool = False
save_kwargs: dict[str, int] = field(
default_factory=lambda: {"quality": 95, "subsampling": 0}
)
SAVE_TEST_CASES = [
# Default test case is a jpeg, check it works with all extensions.
MemorySaveTestCase("foobar.jpeg"),
MemorySaveTestCase("foobar.jpg"),
MemorySaveTestCase("foobar.JPEG"),
MemorySaveTestCase("foobar.JPG"),
MemorySaveTestCase("foobar.png.jpeg"),
MemorySaveTestCase("foobar.png", save_kwargs={}, convert_needed=True),
MemorySaveTestCase("foobar.PNG", save_kwargs={}, convert_needed=True),
MemorySaveTestCase("foobar.jpeg.png", save_kwargs={}, convert_needed=True),
MemorySaveTestCase(save_resolution=None, resize_needed=False),
MemorySaveTestCase(save_resolution=(1000, 1200), resize_needed=False),
MemorySaveTestCase(save_resolution=(2000, 2400), resize_needed=True),
]
@pytest.mark.parametrize("test_case", SAVE_TEST_CASES)
def test_save_from_memory(test_case, test_env, mocker):
"""Check the correct image is retrieved and saved with correct settings."""
camera = test_env.get_thing_by_type(SimulatedCamera)
camera._memory_buffer = mocker.Mock()
camera._add_metadata_to_capture = mocker.Mock()
mode = CaptureMode(description="foo", save_resolution=test_case.save_resolution)
capture_modes_mock = mocker.PropertyMock(return_value={"standard": mode})
mocker.patch.object(type(camera), "capture_modes", capture_modes_mock)
mock_image = mocker.Mock()
# Make resize and convert return itself so we can track further calls of the Image
# object after a resize
mock_image.resize.return_value = mock_image
mock_image.convert.return_value = mock_image
mock_image.size = (1000, 1200)
mock_image.mode = "RGBX"
camera._memory_buffer.get_image.return_value = (
mock_image,
{"meta": "data"},
"standard",
)
camera._data_dir = os.path.normpath("/fake/data/dir")
camera.save_from_memory(RelativeDataPath(test_case.filename), 33)
assert camera._memory_buffer.get_image.call_count == 1
assert camera._memory_buffer.get_image.call_args.args == (33,)
assert camera._add_metadata_to_capture.call_count == 1
assert mock_image.resize.call_count == (1 if test_case.resize_needed else 0)
assert mock_image.convert.call_count == (1 if test_case.convert_needed else 0)
assert mock_image.save.call_count == 1
assert mock_image.save.call_args.kwargs == test_case.save_kwargs

87
tests/unit_tests/test_camera_buffer.py
View file

@ -33,8 +33,8 @@ 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, random_metadata())
returned_image, _ = mem_buf.get_image(buffer_id)
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
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
@ -46,12 +46,12 @@ 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, random_metadata())
returned_image, _ = mem_buf.get_image(buffer_id, remove=False)
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
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)
returned_image, _, _ = mem_buf.get_image(buffer_id)
assert misc_image is returned_image
# It is now removed from memory
with pytest.raises(NoImageInMemoryError):
@ -62,8 +62,8 @@ 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, random_metadata())
returned_image, _ = mem_buf.get_image()
mem_buf.add_image(misc_image, random_metadata(), "standard")
returned_image, _, _ = mem_buf.get_image()
# It is the same image
assert misc_image is returned_image
# It is now removed from memory
@ -76,11 +76,21 @@ def test_get_two_images():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2)
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2)
returned_image1, _ = mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
# It they the same images
buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=2
)
buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=2
)
buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=2
)
buffer_id2 = mem_buf.add_image(
misc_image2, random_metadata(), "standard", buffer_max=2
)
returned_image1, _, _ = mem_buf.get_image(buffer_id1)
returned_image2, _, _ = mem_buf.get_image(buffer_id2)
# Assert they are the same images
assert misc_image1 is returned_image1
assert misc_image2 is returned_image2
# They are removed from memory
@ -95,11 +105,11 @@ def test_get_two_images_without_setting_buffer_size():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata())
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata())
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), "standard")
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), "standard")
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
returned_image2, _, _ = mem_buf.get_image(buffer_id2)
# Image 2 the expected image
assert misc_image2 is returned_image2
@ -110,17 +120,21 @@ def test_buffer_size_changing():
misc_image1 = random_image()
misc_image2 = random_image()
misc_image3 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=3)
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=3)
buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=3
)
buffer_id2 = mem_buf.add_image(
misc_image2, random_metadata(), "standard", buffer_max=3
)
# Third capture doesn't set buffer size, so it will be reset
buffer_id3 = mem_buf.add_image(misc_image3, random_metadata())
buffer_id3 = mem_buf.add_image(misc_image3, random_metadata(), "standard")
# 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
returned_image3, _, _ = mem_buf.get_image(buffer_id3)
# Image 3 is the expected image
assert misc_image3 is returned_image3
@ -129,9 +143,9 @@ def test_capture_two_images_get_without_id():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2)
mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2)
returned_image, _ = mem_buf.get_image()
mem_buf.add_image(misc_image1, random_metadata(), "standard", buffer_max=2)
mem_buf.add_image(misc_image2, random_metadata(), "standard", 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
@ -148,7 +162,9 @@ def test_buffer_size_respected():
buffer_ids = []
for _i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=5)
buffer_id = mem_buf.add_image(
image, random_metadata(), "standard", buffer_max=5
)
images.append(image)
buffer_ids.append(buffer_id)
@ -157,7 +173,7 @@ def test_buffer_size_respected():
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id)
else:
returned_image, _ = mem_buf.get_image(buffer_id)
returned_image, _, _ = mem_buf.get_image(buffer_id)
assert image is returned_image
@ -169,7 +185,9 @@ def test_clear_buffer():
buffer_ids = []
for _i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=10)
buffer_id = mem_buf.add_image(
image, random_metadata(), "standard", buffer_max=10
)
images.append(image)
buffer_ids.append(buffer_id)
@ -192,7 +210,7 @@ def test_get_metadata_too():
for _i in range(10):
image = random_image()
metadata = random_metadata()
buffer_id = mem_buf.add_image(image, metadata, buffer_max=10)
buffer_id = mem_buf.add_image(image, metadata, "standard", buffer_max=10)
images.append(image)
metadatas.append(metadata)
buffer_ids.append(buffer_id)
@ -202,6 +220,19 @@ def test_get_metadata_too():
# Check both image and metadata
for image, metadata, buffer_id in zipped:
returned_image, returned_metadata = mem_buf.get_image(buffer_id)
returned_image, returned_metadata, _ = mem_buf.get_image(buffer_id)
assert image is returned_image
assert metadata is returned_metadata
def test_mode_is_returned():
"""Check that the correct mode name is returned with the image from the buffer."""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
_, _, mode = mem_buf.get_image(buffer_id)
assert mode == "standard"
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "foobar")
_, _, mode = mem_buf.get_image(buffer_id)
assert mode == "foobar"

4
tests/unit_tests/test_metadata.py
View file

@ -169,9 +169,9 @@ def test_picamera_metadata_written_to_exif(mock_picam_thing, temp_jpeg, mocker):
mock_interface.get_thing_states.return_value = camera.thing_state
camera._thing_server_interface = mock_interface
capture_metadata = camera._capture_metadata()
ofm_metadata = camera._collect_ofm_metadata()
camera._add_metadata_to_capture(str(temp_jpeg), capture_metadata)
camera._add_metadata_to_capture(str(temp_jpeg), ofm_metadata)
exif_dict = piexif.load(str(temp_jpeg))
user_comment = json.loads(exif_dict["Exif"][piexif.ExifIFD.UserComment].decode())

128
tests/unit_tests/test_ofm_thing.py Normal file
View file

@ -0,0 +1,128 @@
"""Tests for the for thing defined in the base __init__ of the things dir."""
import os
from tempfile import TemporaryDirectory
import pytest
from pydantic import ValidationError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things import OFMThing, RelativeDataPath
@pytest.mark.parametrize(
("path", "valid"),
[
("foo.png", True),
("foo/bar/", True),
("./foo/bar", True),
("/foo/bar", False),
("../foo/bar", False),
("foo/../bar", False),
],
)
def test_rel_data_path_validation(path, valid):
"""Check validation for a number of cases."""
if valid:
path_obj = RelativeDataPath(path)
assert path_obj.root == os.path.normpath(path)
else:
with pytest.raises(ValidationError):
RelativeDataPath(path)
def test_setting_saving_things():
"""Check that the saving thing can be set."""
thing1 = create_thing_without_server(OFMThing)
thing1._data_dir = os.path.join("data", "thing1")
thing2 = create_thing_without_server(OFMThing)
thing2._data_dir = os.path.join("data", "thing2")
path = RelativeDataPath("foo.png")
assert not path.save_location_set
with pytest.raises(
RuntimeError, match="The saving Thing for the relative path was never set"
):
path.abs_data_path
# Set the saving thing and check the response is as expected
path.set_saving_thing(thing1)
assert path.save_location_set
assert path.abs_data_path == os.path.join("data", "thing1", "foo.png")
assert path._saving_thing is thing1
# Check there is an error is trying to overwrite it with a new thing...
with pytest.raises(
RuntimeError, match="The saving Thing for the relative path is already set"
):
path.set_saving_thing(thing2)
# or a temporary dir
with pytest.raises(
RuntimeError, match="The saving Thing for the relative path is already set"
):
path.save_to_tempdir()
# Check no error when using set_saving_thing_if_unset
path.set_saving_thing_if_unset(thing2)
assert path.save_location_set
assert path.abs_data_path == os.path.join("data", "thing1", "foo.png")
assert path._saving_thing is thing1
# Check set_saving_thing_if_unset works on a new path.
path2 = RelativeDataPath("foo2.png")
assert not path2.save_location_set
path2.set_saving_thing_if_unset(thing2)
assert path2.save_location_set
assert path2.abs_data_path == os.path.join("data", "thing2", "foo2.png")
assert path2._saving_thing is thing2
def test_saving_thing_propagates_on_join():
"""Check when joining to a path the saving thing propagates to the new path."""
thing = create_thing_without_server(OFMThing)
thing._data_dir = os.path.join("data", "thing")
path = RelativeDataPath("foo")
# No thing set
assert not path.save_location_set
# After joining no thing set
path2 = path.join("bar")
assert not path2.save_location_set
# Set thing and check joined path is as expected
path2.set_saving_thing(thing)
assert path2.save_location_set
assert path2.abs_data_path == os.path.join("data", "thing", "foo", "bar")
# Do another join. Set thing remains
path3 = path2.join("file.png")
assert path3.save_location_set
assert path3.abs_data_path == os.path.join(
"data", "thing", "foo", "bar", "file.png"
)
def test_temp_dir_for_saving_thing():
"""Check that the saving thing can actually be a temporary directory."""
path = RelativeDataPath("foo.png")
tmpdir = path.save_to_tempdir()
assert isinstance(tmpdir, TemporaryDirectory)
assert path.save_location_set
assert path.abs_data_path == os.path.join(tmpdir.name, "foo.png")
def test_create_data_path_from_thing():
"""Check the create_data_path method of ``OFMThing``.
It should create a RelativeDataPath with itself set as the saving thing.
"""
thing = create_thing_without_server(OFMThing)
thing._data_dir = os.path.join("data", "thing")
path = thing.create_data_path("file.png")
assert path.save_location_set
assert path.abs_data_path == os.path.join("data", "thing", "file.png")
assert path._saving_thing is thing

37
tests/unit_tests/test_scan_workflows.py
View file

@ -4,6 +4,7 @@ import itertools
import logging
import pytest
from PIL import Image
from pydantic import BaseModel
import labthings_fastapi as lt
@ -11,6 +12,7 @@ from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_planners import SmartSpiral
from openflexure_microscope_server.stitching import StitchingSettings
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.autofocus import SmartStackParams
from openflexure_microscope_server.things.camera_stage_mapping import csm_img_to_stage
from openflexure_microscope_server.things.scan_workflows import (
@ -44,8 +46,6 @@ def test_partial_base_classes():
bad_workflow = create_thing_without_server(BadWorkflow)
settings = MinimalSettings()
with pytest.raises(NotImplementedError):
bad_workflow.check_before_start(settings)
with pytest.raises(NotImplementedError):
bad_workflow.ready
@ -69,7 +69,17 @@ def test_partial_base_classes():
@pytest.fixture
def histo_workflow():
"""Return a HistoScanWorkflow thing with slots mocked."""
return create_thing_without_server(HistoScanWorkflow, mock_all_slots=True)
workflow = create_thing_without_server(HistoScanWorkflow, mock_all_slots=True)
workflow._cam.capture_modes = {"standard": "Mock"}
workflow._cam._capture_image.return_value = Image.new("RGB", (1111, 1222))
return workflow
def test_histo_workflow_save_resolution(histo_workflow):
"""Check that the camera is used to get the save resolution."""
width, height = histo_workflow._get_save_resolution()
assert width == 1111
assert height == 1222
# Use itertools to iterate over every true/false permutation
@ -117,7 +127,12 @@ def test_histo_workflow_settings_generation(histo_workflow, mocker):
mocker.patch.object(
histo_workflow, "_calc_displacement_from_overlap", return_value=(123, 456)
)
workflow_settings, stitching_settings = histo_workflow.all_settings("/this/img_dir")
img_dir = RelativeDataPath("this/img_dir")
workflow_settings, stitching_settings, save_res = histo_workflow.all_settings(
img_dir
)
assert save_res == (1111, 1222)
## Check type
assert isinstance(workflow_settings, HistoScanSettingsModel)
assert isinstance(stitching_settings, StitchingSettings)
@ -137,7 +152,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.capture_params.images_dir == "/this/img_dir"
assert workflow_settings.capture_params.images_dir.root == "this/img_dir"
def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker):
@ -148,8 +163,9 @@ def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker
# Different when False
histo_workflow.equal_distances = False
workflow_settings, _stitching_settings = histo_workflow.all_settings(
"/this/img_dir"
img_dir = RelativeDataPath("this/img_dir")
workflow_settings, _stitching_settings, _save_res = histo_workflow.all_settings(
img_dir
)
assert workflow_settings.dx == 123
@ -157,8 +173,8 @@ def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker
# Same when set True
histo_workflow.equal_distances = True
workflow_settings, _stitching_settings = histo_workflow.all_settings(
"/this/img_dir"
workflow_settings, _stitching_settings, _save_res = histo_workflow.all_settings(
img_dir
)
assert workflow_settings.dx == 123
@ -455,9 +471,8 @@ def test_correlation_resize(histo_workflow, save_res, expected_resize):
target area, taking the square root, and rounding to the nearest integer N. Then
correlation_resize = 1 / N.
"""
histo_workflow.save_resolution = save_res
histo_workflow.overlap = 0.1
settings = histo_workflow._get_stitching_settings_model()
settings = histo_workflow._get_stitching_settings_model(save_res)
assert isinstance(settings, StitchingSettings)
assert settings.correlation_resize == expected_resize

8
tests/unit_tests/test_simulated_camera.py
View file

@ -159,12 +159,12 @@ def test_infinite_sample(camera, stage):
camera.noise_level = 0
assert not camera.repeating
cached_canvas = camera.canvas
array_not_repeating = camera.capture_array()
array_not_repeating = camera.capture_as_array()
camera.repeating = True
time.sleep(0.2) # Ensure frame regenerates
# Canvas shouldn't regenerate
assert camera.canvas is cached_canvas
array_repeating = camera.capture_array()
array_repeating = camera.capture_as_array()
# Images are identical whether or not repeating
assert np.array_equal(array_not_repeating, array_repeating)
@ -174,13 +174,13 @@ def test_infinite_sample(camera, stage):
camera.repeating = False
time.sleep(0.2) # Ensure frame regenerates
# If not repeating the array is just background
assert np.all(camera.capture_array() == simulation.BG_COLOR)
assert np.all(camera.capture_as_array() == simulation.BG_COLOR)
# Turn on repeating
camera.repeating = True
time.sleep(0.2) # Ensure frame regenerates
# Sample is now infinite, so not all background
assert not np.all(camera.capture_array() == simulation.BG_COLOR)
assert not np.all(camera.capture_as_array() == simulation.BG_COLOR)
def test_simulation_cam_calibration(camera):

3
tests/unit_tests/test_smart_scan.py
View file

@ -413,13 +413,14 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker):
smart_scan_thing._workflow.all_settings.return_value = (
MockWorkflowSettingModel(),
StitchingSettings(correlation_resize=0.5, overlap=0.45),
(1640, 1232),
)
smart_scan_thing._workflow.save_resolution = (1640, 1232)
mock_ongoing_scan = mocker.Mock()
mock_ongoing_scan.name = MOCK_SCAN_NAME
mock_ongoing_scan.images_dir = MOCK_SCAN_DIR
smart_scan_thing._ongoing_scan = mock_ongoing_scan
smart_scan_thing._data_dir = "scans"
yield smart_scan_thing

45
tests/unit_tests/test_stack.py
View file

@ -8,10 +8,13 @@ import numpy as np
import pytest
from hypothesis import given
from hypothesis import strategies as st
from PIL import Image
from pydantic import ValidationError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import IMAGE_REGEX
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.autofocus import (
EXTRA_STACK_CAPTURES,
AutofocusThing,
@ -262,6 +265,10 @@ def histo_scan_workflow():
workflow._csm.calibration_required = False
workflow._csm.convert_image_to_stage_coordinates = lambda x, y: {"x": x, "y": y}
# And set up camera
workflow._cam.capture_modes = {"standard": "Mock"}
workflow._cam._capture_image.return_value = Image.new("RGB", (1000, 1000))
return workflow
@ -341,7 +348,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."""
scan_settings, _ = histo_scan_workflow.all_settings(images_dir="dummy")
scan_settings, _, _ = histo_scan_workflow.all_settings(RelativeDataPath("dummy"))
assert scan_settings.smart_stack_params.max_attempts == 3
# Set up returns from z-stack
@ -878,35 +885,15 @@ def test_invalid_stack_settling_raises():
@pytest.mark.parametrize(
("bad_path", "match_err"),
("bad_path", "error_type"),
[
("", "String should have at least 1 character"),
(None, "Input should be a valid string"),
(67, "Input should be a valid string"),
(None, TypeError),
(67, TypeError),
("../dangerous", ValidationError),
("/usr/bin/bash", ValidationError),
],
)
def test_invalid_capture_dir_raises(bad_path, match_err):
def test_invalid_capture_dir_raises(bad_path, error_type):
"""Test basic stack raises expected error for bad image dir paths."""
with pytest.raises(ValueError, match=match_err):
CaptureParams(images_dir=bad_path, save_resolution=(20, 20))
@pytest.mark.parametrize(
("bad_res", "match_err"),
[
((-100, 50), "Input should be greater than or equal to 1"),
((20, 0), "Input should be greater than or equal to 1"),
("", "Input should be a valid tuple"),
(None, "Input should be a valid tuple"),
(67, "Input should be a valid tuple"),
(
["path"],
"Input should be a valid integer, unable to parse string as an integer",
),
((20, 20, 20), "Tuple should have at most 2 items"),
],
)
def test_invalid_capture_res_raises(bad_res, match_err):
"""Test basic stack raises expected error for invalid save resolutions."""
with pytest.raises(ValueError, match=match_err):
CaptureParams(images_dir="dummy", save_resolution=bad_res)
with pytest.raises(error_type):
CaptureParams(images_dir=bad_path, capture_mode="foo")

17
webapp/src/components/tabContentComponents/controlComponents/paneControl.vue
View file

@ -7,20 +7,9 @@
<div class="uk-margin">
<action-button
thing="camera"
action="capture_jpeg"
:submit-data="{ stream_name: 'main' }"
:submit-label="'Low Resolution'"
@response="handleCaptureResponse"
@error="modalError"
/>
</div>
<div class="uk-margin">
<action-button
thing="camera"
action="capture_jpeg"
:submit-data="{ stream_name: 'full' }"
submit-label="Full Resolution"
action="capture"
:submit-data="{ capture_mode: 'standard' }"
submit-label="Capture"
:submit-on-event="'globalCaptureEvent'"
@response="handleCaptureResponse"
@error="modalError"