diff --git a/picamera_coverage.zip b/picamera_coverage.zip
index ef76d9f9..3dd24bdf 100644
Binary files a/picamera_coverage.zip and b/picamera_coverage.zip differ
diff --git a/src/openflexure_microscope_server/things/__init__.py b/src/openflexure_microscope_server/things/__init__.py
index 255e0379..b5855e6e 100644
--- a/src/openflexure_microscope_server/things/__init__.py
+++ b/src/openflexure_microscope_server/things/__init__.py
@@ -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)
diff --git a/src/openflexure_microscope_server/things/autofocus.py b/src/openflexure_microscope_server/things/autofocus.py
index 43cc3e2b..bff91e81 100644
--- a/src/openflexure_microscope_server/things/autofocus.py
+++ b/src/openflexure_microscope_server/things/autofocus.py
@@ -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()
diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py
index b2579654..fa32d6d6 100644
--- a/src/openflexure_microscope_server/things/camera/__init__.py
+++ b/src/openflexure_microscope_server/things/camera/__init__.py
@@ -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."""
diff --git a/src/openflexure_microscope_server/things/camera/opencv.py b/src/openflexure_microscope_server/things/camera/opencv.py
index 00135bad..4f31ea8a 100644
--- a/src/openflexure_microscope_server/things/camera/opencv.py
+++ b/src/openflexure_microscope_server/things/camera/opencv.py
@@ -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]:
diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py
index 35a20710..c769947d 100644
--- a/src/openflexure_microscope_server/things/camera/picamera.py
+++ b/src/openflexure_microscope_server/things/camera/picamera.py
@@ -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",
+ ),
+ }
diff --git a/src/openflexure_microscope_server/things/camera/simulation.py b/src/openflexure_microscope_server/things/camera/simulation.py
index 9b09e14e..e5fe90c1 100644
--- a/src/openflexure_microscope_server/things/camera/simulation.py
+++ b/src/openflexure_microscope_server/things/camera/simulation.py
@@ -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
diff --git a/src/openflexure_microscope_server/things/scan_workflows.py b/src/openflexure_microscope_server/things/scan_workflows.py
index fe4ee898..40febefd 100644
--- a/src/openflexure_microscope_server/things/scan_workflows.py
+++ b/src/openflexure_microscope_server/things/scan_workflows.py
@@ -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:
diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py
index 84463e17..328a76cb 100644
--- a/src/openflexure_microscope_server/things/smart_scan.py
+++ b/src/openflexure_microscope_server/things/smart_scan.py
@@ -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
diff --git a/src/openflexure_microscope_server/things/stage_measure.py b/src/openflexure_microscope_server/things/stage_measure.py
index 0f633ee1..c6dc122b 100644
--- a/src/openflexure_microscope_server/things/stage_measure.py
+++ b/src/openflexure_microscope_server/things/stage_measure.py
@@ -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:
diff --git a/tests/hardware_specific_tests/picamera2/test_acquisition.py b/tests/hardware_specific_tests/picamera2/test_acquisition.py
index 7ac49e3a..ed864e1b 100644
--- a/tests/hardware_specific_tests/picamera2/test_acquisition.py
+++ b/tests/hardware_specific_tests/picamera2/test_acquisition.py
@@ -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))
diff --git a/tests/hardware_specific_tests/picamera2/test_exposure_time_drift.py b/tests/hardware_specific_tests/picamera2/test_exposure_time_drift.py
index 9dff64e4..1710309e 100644
--- a/tests/hardware_specific_tests/picamera2/test_exposure_time_drift.py
+++ b/tests/hardware_specific_tests/picamera2/test_exposure_time_drift.py
@@ -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
diff --git a/tests/hardware_specific_tests/picamera2/test_streaming_mode.py b/tests/hardware_specific_tests/picamera2/test_streaming_mode.py
index b1184f80..a327a271 100644
--- a/tests/hardware_specific_tests/picamera2/test_streaming_mode.py
+++ b/tests/hardware_specific_tests/picamera2/test_streaming_mode.py
@@ -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.
diff --git a/tests/integration_tests/test_actions.py b/tests/integration_tests/test_actions.py
index 5de893a0..d16df380 100644
--- a/tests/integration_tests/test_actions.py
+++ b/tests/integration_tests/test_actions.py
@@ -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)
diff --git a/tests/unit_tests/test_base_camera.py b/tests/unit_tests/test_base_camera.py
index aba5b4ea..6efc72e3 100644
--- a/tests/unit_tests/test_base_camera.py
+++ b/tests/unit_tests/test_base_camera.py
@@ -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
diff --git a/tests/unit_tests/test_camera_buffer.py b/tests/unit_tests/test_camera_buffer.py
index 1ea6620d..fece4900 100644
--- a/tests/unit_tests/test_camera_buffer.py
+++ b/tests/unit_tests/test_camera_buffer.py
@@ -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"
diff --git a/tests/unit_tests/test_metadata.py b/tests/unit_tests/test_metadata.py
index c7ec6b6f..4431d009 100644
--- a/tests/unit_tests/test_metadata.py
+++ b/tests/unit_tests/test_metadata.py
@@ -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())
diff --git a/tests/unit_tests/test_ofm_thing.py b/tests/unit_tests/test_ofm_thing.py
new file mode 100644
index 00000000..593c49d5
--- /dev/null
+++ b/tests/unit_tests/test_ofm_thing.py
@@ -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
diff --git a/tests/unit_tests/test_scan_workflows.py b/tests/unit_tests/test_scan_workflows.py
index 0bb7b13c..a4ba6e0c 100644
--- a/tests/unit_tests/test_scan_workflows.py
+++ b/tests/unit_tests/test_scan_workflows.py
@@ -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
diff --git a/tests/unit_tests/test_simulated_camera.py b/tests/unit_tests/test_simulated_camera.py
index 99f486be..2c7d4a81 100644
--- a/tests/unit_tests/test_simulated_camera.py
+++ b/tests/unit_tests/test_simulated_camera.py
@@ -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):
diff --git a/tests/unit_tests/test_smart_scan.py b/tests/unit_tests/test_smart_scan.py
index a56773b6..d8fe3038 100644
--- a/tests/unit_tests/test_smart_scan.py
+++ b/tests/unit_tests/test_smart_scan.py
@@ -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
diff --git a/tests/unit_tests/test_stack.py b/tests/unit_tests/test_stack.py
index ced295e6..40b48e87 100644
--- a/tests/unit_tests/test_stack.py
+++ b/tests/unit_tests/test_stack.py
@@ -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")
diff --git a/webapp/src/components/tabContentComponents/controlComponents/paneControl.vue b/webapp/src/components/tabContentComponents/controlComponents/paneControl.vue
index b9f0a000..b3a1b5fe 100644
--- a/webapp/src/components/tabContentComponents/controlComponents/paneControl.vue
+++ b/webapp/src/components/tabContentComponents/controlComponents/paneControl.vue
@@ -7,20 +7,9 @@
-
-
-