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

Merge branch 'streaming-modes' into 'v3'

Browse source 
Streaming modes

Closes #707

See merge request openflexure/openflexure-microscope-server!600
This commit is contained in:
Julian Stirling 2026-05-27 15:57:19 +00:00
commit 60694f3532
15 changed files with 252 additions and 214 deletions
Download patch file Download diff file Expand all files Collapse all files

7
ofm_config_full.json
View file

@ -1,11 +1,6 @@
{
"things": {
"camera": {
"class": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2",
"kwargs": {
"camera_board": "picamera_v2"
}
},
"camera": "openflexure_microscope_server.things.camera.picamera:PiCameraV2",
"stage": "openflexure_microscope_server.things.stage.sangaboard:SangaboardThing",
"autofocus": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"gallery": "openflexure_microscope_server.things.gallery.GalleryThing",

BIN
picamera_coverage.zip
View file

Binary file not shown.

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

@ -13,6 +13,7 @@ import json
import os
import tempfile
import time
from abc import ABC, abstractmethod
from datetime import datetime
from types import TracebackType
from typing import Annotated, Any, Literal, Mapping, Optional, Self, Tuple
@ -165,7 +166,16 @@ class CameraMemoryBuffer:
del self._storage[key]
class BaseCamera(OFMThing):
class StreamingMode(BaseModel):
"""Description of streaming modes for the camera.
Cameras can sub class this to record camera specific information about the mode.
"""
description: str
class BaseCamera(OFMThing, ABC):
"""The base class for all cameras. All cameras must directly inherit from this class.
The connection to the camera hardware should be added to the ``__enter__`` method not
@ -196,6 +206,12 @@ class BaseCamera(OFMThing):
self._default_background_detector = "bg_channel_deviations_luv"
self._background_detector_name: Optional[str] = None
if "default" and "full_resolution" not in self.streaming_modes:
raise KeyError(
f"Camera {type(self).__name__} doesn't define both a 'default' and a "
"'full_resolution' streaming mode."
)
def __enter__(self) -> Self:
"""Open hardware connection when the Thing context manager is opened."""
super().__enter__()
@ -236,21 +252,51 @@ class BaseCamera(OFMThing):
"""
return False
@lt.property
def streaming_modes(self) -> Mapping[str, StreamingMode]:
"""Modes the camera can stream in."""
return {
"default": StreamingMode(
description=(
"The standard mode that balances capture resolution and stream "
"size."
)
),
"full_resolution": StreamingMode(
description=(
"Streaming the camera in full resolution. For this camera, this is "
"identical to the default mode."
)
),
}
streaming_mode: str = lt.property(default="default", readonly=True)
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int] = (800, 800), buffer_count: int = 1
) -> None:
"""Start (or stop and restart) the camera.
def change_streaming_mode(self, mode: str = "default") -> None:
"""Change the mode the camera is streaming in.
:param main_resolution: the resolution to use for the main stream.
:param buffer_count: number of images in the stream buffer.
:param mode: Must be a key from ``streaming_modes``
Note that the default values for both parameters should be set appropriately
for the specific camera when defining a new Camera Thing.
When creating a subclass. Subclass the method ``_start_streaming`` rather than
this method.
"""
raise NotImplementedError(
"CameraThings must define their own start_streaming method"
)
if not self.stream_active:
raise RuntimeError(
"Cannot change streaming mode before streaming is started."
)
if mode not in self.streaming_modes:
self.logger.warning(
f"Streaming mode {mode}, is not known. Expecting a mode from "
f"{self.streaming_modes.keys()}. Streaming in default mode."
)
mode = "default"
if mode != self.streaming_mode:
self._start_streaming(mode=mode)
@abstractmethod
def _start_streaming(self, mode: str = "default") -> None:
"""Start (or stop and restart) the camera."""
def kill_mjpeg_streams(self) -> None:
"""Kill the streams now as the server is shutting down.
@ -373,20 +419,17 @@ class BaseCamera(OFMThing):
return datafile_path
@abstractmethod
@lt.property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active."""
raise NotImplementedError(
"CameraThings must define their own stream_active method"
)
@abstractmethod
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
raise NotImplementedError(
"CameraThings must define their own discard_frames method"
)
@abstractmethod
@lt.action
def capture_array(
self,
@ -394,9 +437,6 @@ class BaseCamera(OFMThing):
wait: Optional[float] = 5,
) -> NDArray:
"""Acquire one image from the camera and return as an array."""
raise NotImplementedError(
"CameraThings must define their own capture_array method"
)
downsampled_array_factor: int = lt.property(default=2, ge=1)
"""The downsampling factor when calling capture_downsampled_array."""
@ -535,15 +575,13 @@ class BaseCamera(OFMThing):
)
return self._thing_server_interface.call_async_task(stream.next_frame_size)
@abstractmethod
def capture_image(
self,
stream_name: Literal["main", "lores", "full"],
wait: Optional[float] = None,
) -> Image.Image:
"""Capture a PIL image from stream stream_name with timeout wait."""
raise NotImplementedError(
"CameraThings must define their own capture_image method"
)
@lt.action
def capture_and_save(

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

@ -53,7 +53,7 @@ class OpenCVCamera(BaseCamera):
self.logger.warning(f"{self.camera_name} not found.")
self._camera_name = next(iter(self.cameras))
self._start_stream()
self._start_streaming()
return self
def __exit__(
@ -66,7 +66,7 @@ class OpenCVCamera(BaseCamera):
Before releasing the camera the capture thread is closed.
"""
self._stop_stream()
self._stop_streaming()
super().__exit__(exc_type, exc_value, traceback)
_camera_name = ""
@ -89,12 +89,15 @@ class OpenCVCamera(BaseCamera):
self._camera_name = value
if value not in self.cameras:
raise ValueError(f"{value} is not a valid camera name.")
self._start_stream()
self._start_streaming()
def _start_stream(self) -> None:
def _start_streaming(self, mode: str = "default") -> None:
"""Start the camera stream or restart if running."""
if mode not in self.streaming_modes:
raise ValueError(f"Unknown mode {mode}")
self.streaming_mode = mode
if self.stream_active:
self._stop_stream()
self._stop_streaming()
self.cap = cv2.VideoCapture(
self.cameras[self.camera_name], opencv_utils.BACKEND
)
@ -102,7 +105,7 @@ class OpenCVCamera(BaseCamera):
self._capture_thread = Thread(target=self._capture_frames)
self._capture_thread.start()
def _stop_stream(self) -> None:
def _stop_streaming(self) -> None:
"""Stop the camera stream."""
if self.stream_active:
self._capture_enabled = False

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

@ -22,12 +22,12 @@ import logging
import os
import tempfile
import time
from abc import ABC, abstractmethod
from contextlib import contextmanager
from threading import RLock
from types import TracebackType
from typing import (
TYPE_CHECKING,
Annotated,
Any,
Iterator,
Literal,
@ -41,7 +41,6 @@ from picamera2 import Picamera2
from picamera2.encoders import MJPEGEncoder
from picamera2.outputs import Output
from PIL import Image
from pydantic import BaseModel, BeforeValidator
import labthings_fastapi as lt
from labthings_fastapi.exceptions import ServerNotRunningError
@ -55,7 +54,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from . import BaseCamera
from . import BaseCamera, StreamingMode
from . import picamera_recalibrate_utils as recalibrate_utils
from . import picamera_tuning_file_utils as tf_utils
@ -64,11 +63,6 @@ if TYPE_CHECKING:
LOGGER = logging.getLogger(__name__)
SUPPORTED_CAMS_SENSOR_INFO = {
"picamera_v2": recalibrate_utils.IMX219_SENSOR_INFO,
"picamera_hq": recalibrate_utils.IMX477_SENSOR_INFO,
}
class PicameraModelError(RuntimeError):
"""There is a problem Picamera sensor model set by the configuration."""
@ -101,38 +95,28 @@ class PicameraStreamOutput(Output):
self.stream.add_frame(frame)
class SensorMode(BaseModel):
"""A Pydantic model holding all the information about a specific sensor mode.
class PiCamera2StreamingMode(StreamingMode):
"""Streaming mode configuration for the PiCamera2."""
This data is as reported by the PiCamera2 module.
"""
unpacked: str
main_resolution: tuple[int, int]
lores_resolution: tuple[int, int]
sensor_mode_resolution: tuple[int, int]
bit_depth: int
size: tuple[int, int]
fps: float
crop_limits: tuple[int, int, int, int]
exposure_limits: tuple[Optional[int], Optional[int], Optional[int]]
format: Annotated[str, BeforeValidator(repr)]
use_lores_as_preview: bool
buffer_count: int = 6
scaler_crop: Optional[tuple[int, int, int, int]] = None
@property
def sensor_mode_dict(self) -> dict[str, Any]:
"""The dictionary expected by PiCamera2 for setting sensor mode."""
return {"output_size": self.sensor_mode_resolution, "bit_depth": self.bit_depth}
class SensorModeSelector(BaseModel):
"""A Pydantic model holding the two values needed to select a PiCamera Sensor mode.
These values are the output size and the bit depth.
This is a Pydantic model so that it can sent by FastAPI
"""
output_size: tuple[int, int]
bit_depth: int
class StreamingPiCamera2(BaseCamera):
class StreamingPiCamera2(BaseCamera, ABC):
"""A Thing that provides and interface to the Raspberry Pi Camera.
Currently the Thing only supports the PiCamera v2 board. This needs
generalisation.
This is an abstract base class for all picamera models. Use a subclass specific
to the camera model.
"""
_focus_fom: int
@ -140,31 +124,34 @@ class StreamingPiCamera2(BaseCamera):
tuning: dict = lt.setting(default_factory=dict, readonly=True)
"""The Raspberry PiCamera Tuning File JSON."""
# Subclasses should defined both of these.
_camera_board: str
_sensor_info: recalibrate_utils.SensorInfo
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
camera_num: int = 0,
camera_board: str = "picamera_v2",
) -> None:
"""Initialise the camera with the given camera number.
This makes no connection to the camera (except to get the default tuning file).
:param thing_server_interface: The interface between this Thing and the server.
:param camera_num: The number of the camera. This should generally be left as 0
as most Raspberry Pi boards only support 1 camera.
:param camera_board: The camera board used. Supported options are "picamera_v2"
and "picamera_hq".
"""
super().__init__(thing_server_interface)
self._setting_save_in_progress = False
self._camera_num = camera_num
self._camera_board = camera_board
if camera_board not in SUPPORTED_CAMS_SENSOR_INFO:
raise PicameraModelError(
f"The camera_board {camera_board} is not supported. Supported boards "
f"are {SUPPORTED_CAMS_SENSOR_INFO.keys()}."
# Check the subclass defined the _camera_board and _sensor_info
if not hasattr(self, "_camera_board") or not hasattr(self, "_sensor_info"):
raise AttributeError(
f"{type(self).__name__} must define both both _camera_board and "
"_sensor_info attributes"
)
self._sensor_info = SUPPORTED_CAMS_SENSOR_INFO[camera_board]
self._picamera_lock = RLock()
self._picamera = None
@ -188,9 +175,6 @@ class StreamingPiCamera2(BaseCamera):
# trigger a ServerNotRunningError
self._colour_gains = tf_utils.get_colour_gains_from_lst(self.tuning)
stream_resolution: tuple[int, int] = lt.property(default=(820, 616))
"""Resolution to use for the MJPEG stream."""
mjpeg_bitrate: Optional[int] = lt.property(default=100000000)
"""Bitrate for MJPEG stream (None for default)."""
@ -302,41 +286,6 @@ class StreamingPiCamera2(BaseCamera):
"Sharpness": 1,
}
_sensor_modes: Optional[list[SensorMode]] = None
@lt.property
def sensor_modes(self) -> list[SensorMode]:
"""All the available modes the current sensor supports."""
if not self._sensor_modes:
with self._streaming_picamera() as cam:
self._sensor_modes = cam.sensor_modes
return self._sensor_modes
_sensor_mode: Optional[dict] = None
@lt.property
def sensor_mode(self) -> Optional[SensorModeSelector]:
"""The intended sensor mode of the camera."""
if self._sensor_mode is None:
return None
return SensorModeSelector(**self._sensor_mode)
@sensor_mode.setter
def _set_sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None:
"""Change the sensor mode used."""
if new_mode is None:
self._sensor_mode = None
elif isinstance(new_mode, SensorModeSelector):
self._sensor_mode = new_mode.model_dump()
elif isinstance(new_mode, dict):
self._sensor_mode = new_mode
# By pausing the stream on when accessing, streaming_picamera
# self._sensor_mode will be read and set when the stream restarts
# after the context manager closes.
with self._streaming_picamera(pause_stream=True):
pass
@lt.property
def sensor_resolution(self) -> Optional[tuple[int, int]]:
"""The native resolution of the camera's sensor."""
@ -406,13 +355,11 @@ class StreamingPiCamera2(BaseCamera):
"""Start streaming when the Thing context manager is opened.
This opens the picamera connection, initialises the camera, sets the
sensor_modes property, and then starts the streams.
property, and then starts the streams.
"""
super().__enter__()
self._initialise_picamera(check_sensor_model=True)
# Sensor modes is a cached property read it once after initialising the camera
_modes = self.sensor_modes
self.start_streaming()
self._start_streaming()
return self
@property
@ -437,12 +384,12 @@ class StreamingPiCamera2(BaseCamera):
already_streaming = self.stream_active
with self._picamera_lock:
if pause_stream and already_streaming:
self.stop_streaming(stop_web_stream=False)
self._stop_streaming(stop_web_stream=False)
try:
yield self._picamera
finally:
if pause_stream and already_streaming:
self.start_streaming()
self._start_streaming()
def __exit__(
self,
@ -451,59 +398,54 @@ class StreamingPiCamera2(BaseCamera):
traceback: Optional[TracebackType],
) -> None:
"""Close the picamera connection when the Thing context manager is closed."""
self.stop_streaming()
self._stop_streaming()
with self._streaming_picamera() as cam:
cam.close()
del self._picamera
super().__exit__(exc_type, exc_value, traceback)
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 6
) -> None:
@abstractmethod
@lt.property
def streaming_modes(self) -> Mapping[str, PiCamera2StreamingMode]:
"""Modes the camera can stream in."""
def _start_streaming(self, mode: str = "default") -> None:
"""Start the MJPEG stream. This is where persistent controls are sent to camera.
Sets the camera resolutions based on input parameters, and sets the low-res
resolution to (320, 240). Note: (320, 240) is a standard from the Pi Camera
manual.
Sets the camera stream resoltuons based on input mode
Create two streams:
* ``lores_mjpeg_stream`` for autofocus at low-res resolution
* ``mjpeg_stream`` for preview. This is the ``main_resolution`` if this is less
than (1280, 960), or the low-res resolution if above. This allows for
high resolution capture without streaming high resolution video.
main_resolution: the resolution for the main configuration. Defaults to
(820, 616), 1/4 sensor size.
buffer_count: the number of frames to hold in the buffer. Higher uses more memory,
lower may cause dropped frames. Value must be between 1 and 8, Defaults to 6.
* ``lores_mjpeg_stream`` for autofocus at ``lores_resolution``
* ``mjpeg_stream`` for preview. This is at the ``main_resolution`` unless
``use_lores_as_preview`` is True, in which case it is a copy of
``lores_mjpeg_stream``
"""
if mode not in self.streaming_modes:
raise ValueError(f"Unknown mode {mode}")
self.streaming_mode = mode
mode_info = self.streaming_modes[mode]
controls = self._get_persistent_controls()
# Buffer count can't be negative, zero, or too high.
if buffer_count < 1 or buffer_count > 8:
# 8 is slightly arbitrary. 6 is the PiCamera default for video
# and the documentation only says that setting values higher gives
# diminishing returns, and that the true maximum is hardware dependent
raise ValueError(
f"Can't set a buffer count of {buffer_count}. "
"Buffer count must be an integer from 1-8"
)
if mode_info.scaler_crop is not None:
controls["ScalerCrop"] = mode_info.scaler_crop
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": main_resolution},
lores={"size": (320, 240), "format": "YUV420"},
sensor=self._sensor_mode,
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"] = buffer_count
stream_config["buffer_count"] = mode_info.buffer_count
picam.configure(stream_config)
LOGGER.info("Starting picamera MJPEG stream...")
stream_name = "lores" if main_resolution[0] > 1280 else "main"
stream_name = "lores" if mode_info.use_lores_as_preview else "main"
picam.start_recording(
MJPEGEncoder(self.mjpeg_bitrate),
PicameraStreamOutput(self.mjpeg_stream),
@ -515,16 +457,12 @@ class StreamingPiCamera2(BaseCamera):
name="lores",
)
except Exception as e:
LOGGER.exception("Error while starting preview: {e}")
LOGGER.exception(e)
LOGGER.error(f"Error while starting preview: {e}.")
else:
self.stream_active = True
LOGGER.debug(
"Started MJPEG stream at %s on port %s", self.stream_resolution, 1
)
LOGGER.debug("Started MJPEG stream in %s mode.", mode)
@lt.action
def stop_streaming(self, stop_web_stream: bool = True) -> None:
def _stop_streaming(self, stop_web_stream: bool = True) -> None:
"""Stop the MJPEG stream."""
with self._streaming_picamera() as picam:
try:
@ -549,14 +487,17 @@ class StreamingPiCamera2(BaseCamera):
cam.capture_metadata()
@contextmanager
def _switch_to_still_capture_mode(self) -> Iterator[Picamera2]:
def _switch_to_single_capture_mode(self) -> Iterator[Picamera2]:
"""Get the picamera lock, pause stream and switch into still capture config.
Restarts stream when complete.
"""
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=self._sensor_mode))
cam.configure(
cam.create_still_configuration(sensor=mode_info.sensor_mode_dict)
)
cam.start()
time.sleep(self._sensor_info.short_pause)
yield cam
@ -595,7 +536,7 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
elif stream_name == "full":
with self._switch_to_still_capture_mode() as cam:
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}"')
@ -625,7 +566,7 @@ class StreamingPiCamera2(BaseCamera):
# Raw cannot used capture_image.
if wait is None:
wait = 0.9
with self._switch_to_still_capture_mode() as cam:
with self._switch_to_single_capture_mode() as cam:
return cam.capture_array(name="raw", wait=wait)
# 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))))``.
@ -968,3 +909,77 @@ class StreamingPiCamera2(BaseCamera):
"gamma_correction": self.gamma_correction,
}
return state
class PiCameraV2(StreamingPiCamera2):
"""A Thing that provides and interface to the Raspberry Pi Camera V2."""
_camera_board = "picamera_v2"
_sensor_info = recalibrate_utils.IMX219_SENSOR_INFO
@lt.property
def streaming_modes(self) -> Mapping[str, PiCamera2StreamingMode]:
"""Modes the camera can stream in."""
return {
"default": PiCamera2StreamingMode(
description=(
"The standard mode that balances capture resolution and stream "
"size."
),
main_resolution=(820, 616),
lores_resolution=(320, 240),
sensor_mode_resolution=(3280, 2464),
bit_depth=10,
use_lores_as_preview=False,
),
"full_resolution": PiCamera2StreamingMode(
description=(
"Streaming the camera in 8MP 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."
),
main_resolution=(3280, 2464),
lores_resolution=(320, 240),
sensor_mode_resolution=(3280, 2464),
bit_depth=10,
use_lores_as_preview=True,
),
}
class PiCameraHQ(StreamingPiCamera2):
"""A Thing that provides and interface to the Raspberry Pi Camera HQ."""
_camera_board = "picamera_hq"
_sensor_info = recalibrate_utils.IMX477_SENSOR_INFO
@lt.property
def streaming_modes(self) -> Mapping[str, PiCamera2StreamingMode]:
"""Modes the camera can stream in."""
return {
"default": PiCamera2StreamingMode(
description=(
"The standard mode that balances capture resolution and stream "
"size."
),
main_resolution=(700, 700),
lores_resolution=(350, 350),
sensor_mode_resolution=(4056, 3040),
bit_depth=12,
use_lores_as_preview=False,
scaler_crop=(628, 120, 2800, 2800),
),
"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."
),
main_resolution=(2800, 2800),
lores_resolution=(350, 350),
sensor_mode_resolution=(4056, 3040),
bit_depth=12,
use_lores_as_preview=True,
scaler_crop=(628, 120, 2800, 2800),
),
}

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

@ -413,7 +413,7 @@ class SimulatedCamera(BaseCamera):
"""Start the capture thread when the Thing context manager is opened."""
super().__enter__()
self.generate_canvas()
self.start_streaming()
self._start_streaming()
return self
def __exit__(
@ -428,13 +428,10 @@ class SimulatedCamera(BaseCamera):
self._capture_thread.join()
super().__exit__(exc_type, exc_value, traceback)
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 1
) -> None:
def _start_streaming(self, mode: str = "default") -> None:
"""Start the live stream.
The start_streaming method is used a camera ``Thing`` to begin streaming
The _start_streaming method is used a camera ``Thing`` to begin streaming
images or to adjust the stream resolution if streaming is already active.
The simulation camera does not currently support the resolution argument.
@ -442,13 +439,11 @@ class SimulatedCamera(BaseCamera):
If called while already streaming, the warning will be emitted and no other
action will be taken.
:param main_resolution: Currently ignored, this argument exists to ensure consistent API across camera Things.
:param buffer_count: Currently ignored, this argument exists to ensure consistent API across camera Things.
:param mode: The name of the streaming mode to use.
"""
LOGGER.warning(
f"Simulation camera doesn't respect {main_resolution=} or {buffer_count=} "
"arguments."
)
if mode not in self.streaming_modes:
raise ValueError(f"Unknown mode {mode}")
self.streaming_mode = mode
if not self.stream_active:
self._capture_enabled = True
self._capture_thread = Thread(target=self._capture_frames)

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

@ -401,7 +401,7 @@ class SmartScanThing(OFMThing):
)
self.ongoing_scan.save_scan_data(self._scan_data)
self._cam.start_streaming(main_resolution=(3280, 2464))
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
@ -436,7 +436,7 @@ class SmartScanThing(OFMThing):
# if it ended due to an exception.
# Start streaming in the default resolution again as soon as possible
self._cam.start_streaming()
self._cam.change_streaming_mode(mode="default")
# This is what happens if the scan completes successfully or the
# user cancels it.

4
tests/conftest.py
View file

@ -107,9 +107,9 @@ def mock_picam_thing(mocker):
},
)
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
from openflexure_microscope_server.things.camera.picamera import PiCameraV2
return create_thing_without_server(StreamingPiCamera2)
return create_thing_without_server(PiCameraV2)
@pytest.fixture

4
tests/hardware_specific_tests/picamera2/cam_test_utils.py
View file

@ -5,7 +5,7 @@ from contextlib import contextmanager
from typing import Optional
from openflexure_microscope_server.things.background_detect import ChannelDeviationLUV
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
from openflexure_microscope_server.things.camera.picamera import PiCameraV2
from ...shared_utils.lt_test_utils import LabThingsTestEnv
@ -21,7 +21,7 @@ def camera_test_env(settings_folder: Optional[str] = None):
temporary directory will be used as the settings folder.
"""
thing_conf = {
"camera": StreamingPiCamera2,
"camera": PiCameraV2,
"bg_channel_deviations_luv": ChannelDeviationLUV,
}
app_config = {

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

@ -9,7 +9,7 @@ from .cam_test_utils import camera_test_env
@pytest.fixture
def picamera_test_env() -> lt.ThingClient:
"""Initialise a test environment with only a StreamingPiCamera2 Thing."""
"""Initialise a test environment with only a PiCameraV2 Thing."""
with camera_test_env() as env:
yield env

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

@ -93,13 +93,13 @@ def test_exposure_time_on_start_and_stop_stream():
print(f"Starting simulation scan {i}")
# This will need updating if we start supporting other Picamera models
# It is currently used here to mimic the behaviour in in scanning.
client.start_streaming(main_resolution=(3280, 2464))
client.change_streaming_mode(mode="full_resolution")
time.sleep(0.5)
for _j in range(5):
client.capture_to_memory(buffer_max=1)
# Reset to main resolution
time.sleep(0.5)
client.start_streaming()
client.change_streaming_mode(mode="default")
# Check after all of this the exposure time is the same.
assert client.exposure_time == set_time

10
tests/hardware_specific_tests/picamera2/test_sensor_mode.py → tests/hardware_specific_tests/picamera2/test_streaming_mode.py
View file

@ -9,13 +9,13 @@ from .cam_test_utils import camera_test_client
logging.basicConfig(level=logging.DEBUG)
def test_sensor_mode():
def test_streaming_mode():
"""Test capturing raw arrays in two different sensor modes."""
with camera_test_client() as client:
for size in [(3280, 2464), (1640, 1232)]:
client.sensor_mode = {"output_size": size, "bit_depth": 10}
arr = np.array(client.capture_array(stream_name="raw"))
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"))
# 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.
assert arr.shape[0] == size[1]
assert arr.shape[0] == res[1]

19
tests/unit_tests/test_cameras.py
View file

@ -6,7 +6,6 @@ on camera functionality using the simulation camera are in "test_camera".
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.camera import BaseCamera
from openflexure_microscope_server.things.camera.opencv import OpenCVCamera
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
@ -57,11 +56,9 @@ def test_thing_description_equivalence(mock_picam_thing):
camera child classes. Any addition of actions must be accompanied by an update
to this test, prompting discussion of whether the action belongs in the subclass
or the base camera class.
"""
base_td = create_thing_without_server(BaseCamera).thing_description()
base_actions = set(base_td.actions.keys())
base_props = set(base_td.properties.keys())
The base camera class is not instantiated as it is an Abstract Base Class
"""
sim_camera = create_thing_without_server(SimulatedCamera)
sim_description = _get_clean_camera_description(sim_camera)
opencv_camera = create_thing_without_server(OpenCVCamera)
@ -82,8 +79,8 @@ def test_thing_description_equivalence(mock_picam_thing):
# Camera actions and properties should generally be equivalent except for exposed
# manual settings and calibration actions.
assert opencv_actions == sim_actions == base_actions
assert opencv_props == sim_props == base_props
assert opencv_actions == sim_actions
assert opencv_props == sim_props
# For now PiCamera has a number of extra actions and properties. These should be
# reduced over time by creating a way to use the functionality in a way as clearly
@ -92,7 +89,6 @@ def test_thing_description_equivalence(mock_picam_thing):
picamera_extra_actions = {
"set_static_green_equalisation",
"set_ce_enable_to_off",
"stop_streaming",
"reset_ccm",
}
picamera_extra_props = {
@ -103,15 +99,12 @@ def test_thing_description_equivalence(mock_picam_thing):
"sensor_resolution",
"capture_metadata",
"camera_configuration",
"stream_resolution",
"tuning",
"sensor_modes",
"sensor_mode",
}
# Note these are only the action not exposed as calibration actions.
for action in picamera_extra_actions:
assert action in picamera_actions
for props in picamera_extra_props:
assert props in picamera_props
assert picamera_actions - base_actions == picamera_extra_actions
assert picamera_props - base_props == picamera_extra_props
assert picamera_actions - sim_actions == picamera_extra_actions
assert picamera_props - sim_props == picamera_extra_props

9
tests/unit_tests/test_metadata.py
View file

@ -12,7 +12,6 @@ import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.background_detect import ChannelDeviationLUV
from openflexure_microscope_server.things.camera import BaseCamera
from openflexure_microscope_server.things.camera import (
picamera_tuning_file_utils as tf_utils,
)
@ -32,8 +31,8 @@ def temp_jpeg(tmp_path: Path) -> Path:
def test_add_metadata_to_capture(temp_jpeg):
"""Use a BaseCamera to add metadata to a tmp capture and test fields."""
base_cam = create_thing_without_server(BaseCamera)
"""Use a SimulatedCamera to add metadata to a tmp capture and test fields."""
cam = create_thing_without_server(SimulatedCamera)
metadata = {
"Dummy1": 1,
"Dummy2": "two",
@ -48,7 +47,7 @@ def test_add_metadata_to_capture(temp_jpeg):
"things_states": metadata,
}
base_cam._add_metadata_to_capture(str(temp_jpeg), capture_metadata)
cam._add_metadata_to_capture(str(temp_jpeg), capture_metadata)
# Reload EXIF
exif_dict = piexif.load(str(temp_jpeg))
@ -177,7 +176,7 @@ def test_picamera_metadata_written_to_exif(mock_picam_thing, temp_jpeg, mocker):
exif_dict = piexif.load(str(temp_jpeg))
user_comment = json.loads(exif_dict["Exif"][piexif.ExifIFD.UserComment].decode())
assert user_comment["camera"] == "StreamingPiCamera2"
assert user_comment["camera"] == "PiCameraV2"
assert user_comment["camera_board"] == "imx219"
# gamma_correction keys are cast to strings
assert user_comment["tuning"] == {

10
tests/unit_tests/test_smart_scan.py
View file

@ -492,7 +492,7 @@ def check_run_scan(scan_thing, caplog, expected_exception=None):
final_scan_data = scan_thing._ongoing_scan.save_scan_data.call_args[0][0]
calls = {
"cam_start_streaming_calls": scan_thing._cam.start_streaming.call_count,
"cam_change_streaming_mode_calls": scan_thing._cam.change_streaming_mode.call_count,
"main_scan_loop_calls": scan_thing._main_scan_loop.call_count,
"return_to_start_calls": scan_thing._return_to_starting_position.call_count,
"perform_final_stitch_calls": scan_thing._perform_final_stitch.call_count,
@ -509,7 +509,7 @@ def test_run_scan(scan_thing_mocked_for_run_scan, caplog):
assert len(logs) == 0
expected_calls_numbers = {
"cam_start_streaming_calls": 2,
"cam_change_streaming_mode_calls": 2,
"main_scan_loop_calls": 1,
"return_to_start_calls": 1,
"perform_final_stitch_calls": 1,
@ -534,7 +534,7 @@ def test_run_scan_err_in_main_loop(scan_thing_mocked_for_run_scan, caplog, mocke
# Main loop not run, nor are return to start, final stitch, or purging of empty
# scans. Save scan data is still called twice
expected_calls_numbers = {
"cam_start_streaming_calls": 2,
"cam_change_streaming_mode_calls": 2,
"main_scan_loop_calls": 1,
"return_to_start_calls": 0,
"perform_final_stitch_calls": 0,
@ -559,7 +559,7 @@ def test_run_scan_cancelled(scan_thing_mocked_for_run_scan, caplog, mocker):
# Main loop not run, nor are return to start, final stitch, or purging of empty
# scans. Save scan data is still called twice
expected_calls_numbers = {
"cam_start_streaming_calls": 2,
"cam_change_streaming_mode_calls": 2,
"main_scan_loop_calls": 1,
"return_to_start_calls": 1,
"perform_final_stitch_calls": 1,
@ -584,7 +584,7 @@ def test_run_scan_fill_disk(scan_thing_mocked_for_run_scan, caplog, mocker):
# Main loop not run, nor are return to start, final stitch, or purging of empty
# scans. Save scan data is still called twice
expected_calls_numbers = {
"cam_start_streaming_calls": 2,
"cam_change_streaming_mode_calls": 2,
"main_scan_loop_calls": 1,
"return_to_start_calls": 0,
"perform_final_stitch_calls": 0,