From e3361ebec830a0976dc9cca7e7779daed04d5088 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 27 May 2026 23:26:56 +0100 Subject: [PATCH] Update cameras inline with base camera rebase --- .../things/camera/opencv.py | 46 ++-- .../things/camera/picamera.py | 218 ++++++++++++------ .../things/camera/simulation.py | 34 ++- 3 files changed, 180 insertions(+), 118 deletions(-) diff --git a/src/openflexure_microscope_server/things/camera/opencv.py b/src/openflexure_microscope_server/things/camera/opencv.py index 00135bad..76d94502 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 cooerced 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 cooerced 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..fd3a9083 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,31 @@ 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, + mode_info: PiCamera2StreamingMode, + ) -> dict[str, Any]: + """Create the dictionary to pass to ``Picamera2.configure``.""" + controls = self._get_persistent_controls() + + 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. @@ -426,23 +466,16 @@ class StreamingPiCamera2(BaseCamera, ABC): self.streaming_mode = mode mode_info = self.streaming_modes[mode] - controls = self._get_persistent_controls() - - 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": mode_info.main_resolution}, - lores={"size": mode_info.lores_resolution, "format": "YUV420"}, - sensor=mode_info.sensor_mode_dict, - controls=controls, + stream_config = self._create_picam_config_from_mode_info( + picam=picam, + 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 +520,58 @@ 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 switch mode. Mode is reset and stream + restarts stream 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] + 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, + mode_info=streaming_mode_info, + ) + cam.configure(stream_config) + cam.start() + time.sleep(self._sensor_info.short_pause) + yield cam - def capture_image( - self, - stream_name: Literal["main", "lores", "full"] = "main", - wait: Optional[float] = 0.9, - ) -> Image.Image: + 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 for more detail 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. @@ -555,24 +581,22 @@ class StreamingPiCamera2(BaseCamera, ABC): :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. :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 used _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 +970,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 directly from the current stream.", + streaming_mode=None, + stream_name="main", + ), + } + class PiCameraHQ(StreamingPiCamera2): """A Thing that provides and interface to the Raspberry Pi Camera HQ.""" @@ -983,3 +1032,28 @@ 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, 3MP image created from downsampling an 12MP " + "capture." + ), + save_resolution=(1640, 1232), + streaming_mode="full_resolution", + stream_name="main", + ), + "full": PiCamera2CaptureMode( + description="A full resolution 12MP capture.", + streaming_mode="full_resolution", + stream_name="main", + ), + "quick": PiCamera2CaptureMode( + description="Capture directly from the current stream.", + 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..b3384367 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. @@ -491,28 +491,24 @@ class SimulatedCamera(BaseCamera): 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. """ - 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 camera doesn't support raw capture." + ) + # Warn if the capture mode is incorrect, but don't read the cooerced 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 cooerced value as + # this camera only supports one mode. + self._validate_capture_mode(capture_mode) return self.generate_frame() @lt.action