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

Merge branch 'unify-jpeg-capture' into 'v3'

Browse source 
Deduplicating camera functionality

Closes #299, #429, and #355

See merge request openflexure/openflexure-microscope-server!378
This commit is contained in:
Julian Stirling 2025-08-27 17:14:41 +00:00
commit 154f063ab3
8 changed files with 131 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

2
hardware-specific-tests/picamera2/test_acquisition.py
View file

@ -40,7 +40,7 @@ def test_jpeg_and_array(client):
assert mjpeg_frame.format == "JPEG"
# Capture a jpeg
blob = client.capture_jpeg(resolution="main")
blob = client.capture_jpeg(stream_name="main")
jpeg_capture = Image.open(blob.open())
jpeg_capture.verify()
assert jpeg_capture.format == "JPEG"

8
hardware-specific-tests/picamera2/test_exposure_time_drift.py
View file

@ -63,7 +63,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(resolution="full")
client.capture_jpeg(stream_name="full")
if i == 0:
# Exposure can update on first capture, due to frame rate restrictions
first_et = client.exposure_time
@ -80,7 +80,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(resolution="full")
client.capture_jpeg(stream_name="full")
assert client.exposure_time == frame_et
print("Exposure time didn't change!!")
print(f"End of test for exposure target {desired_time}")
@ -106,7 +106,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(resolution="full")
client.capture_jpeg(stream_name="full")
# Save this time.
set_time = client.exposure_time
assert abs(set_time - desired_time) < EXPOSURE_TOL
@ -136,7 +136,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(resolution="full")
client.capture_jpeg(stream_name="full")
# Save this time.
return client.exposure_time

BIN
picamera_coverage.zip
View file

Binary file not shown.

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

@ -12,6 +12,9 @@ import json
import io
import time
import logging
from datetime import datetime
import tempfile
import os
import numpy as np
from pydantic import RootModel
@ -257,14 +260,37 @@ class BaseCamera(lt.Thing):
def capture_jpeg(
self,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 5,
logger: lt.deps.InvocationLogger,
stream_name: str = "main",
wait: Optional[float] = None,
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob."""
raise NotImplementedError(
"CameraThings must define their own capture_jpeg method"
"""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 metadata_getter: LabThings GetThingStates dependency, automatically
injected.
:param logger: LabThings InvocationLogger dependency, automatically injected.
: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)
self._save_capture(
jpeg_path=jpeg_path,
image=img,
metadata=metadata_getter(),
logger=logger,
)
return JPEGBlob.from_temporary_directory(directory, fname)
@lt.thing_action
def grab_jpeg(
self,
@ -331,7 +357,7 @@ class BaseCamera(lt.Thing):
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
wait: Optional[float],
wait: Optional[float] = None,
) -> Image:
"""Capture a PIL image from stream stream_name with timeout wait."""
raise NotImplementedError(
@ -486,10 +512,10 @@ class BaseCamera(lt.Thing):
metadata
).encode("utf-8")
piexif.insert(piexif.dump(exif_dict), jpeg_path)
except: # noqa: E722
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.
logger.warning(f"Failed to add metadata to {jpeg_path}")
logger.exception(f"Failed to add metadata to {jpeg_path}")
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e

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

@ -7,19 +7,18 @@ See repository root for licensing information.
"""
from __future__ import annotations
import io
import json
import logging
from typing import Literal, Optional
from threading import Thread
import cv2
import piexif
from PIL import Image
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from . import BaseCamera, JPEGBlob
from . import BaseCamera
class OpenCVCamera(BaseCamera):
@ -84,7 +83,8 @@ class OpenCVCamera(BaseCamera):
@lt.thing_action
def capture_array(
self,
resolution: Literal["main", "full"] = "full",
stream_name: Literal["main", "full"] = "full",
wait: Optional[float] = None,
) -> NDArray:
"""Acquire one image from the camera and return as an array.
@ -92,7 +92,9 @@ class OpenCVCamera(BaseCamera):
It's likely to be highly inefficient - raw and/or uncompressed captures using
binary image formats will be added in due course.
"""
logging.warning(f"OpenCV camera doesn't respect {resolution} setting")
if wait is not None:
logging.warning("OpenCV camera has no wait option. Use None.")
logging.warning(f"OpenCV camera doesn't respect {stream_name=}")
ret, frame = self.cap.read()
if not ret:
raise RuntimeError(
@ -100,30 +102,16 @@ class OpenCVCamera(BaseCamera):
)
return frame
@lt.thing_action
def capture_jpeg(
def capture_image(
self,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob.
stream_name: Literal["main", "full"] = "main",
wait: Optional[float] = None,
) -> Image:
"""Acquire one image from the camera and return as a PIL image.
This function will produce a JPEG image.
"""
logging.warning(f"OpenCV camera doesn't respect {resolution} setting")
frame = self.capture_array()
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
exif_dict = {
"Exif": {
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode(
"utf-8"
)
},
"GPS": {},
"Interop": {},
"1st": {},
"thumbnail": None,
}
output = io.BytesIO()
piexif.insert(piexif.dump(exif_dict), jpeg, output)
return JPEGBlob.from_bytes(output.getvalue())
if wait is not None:
logging.warning("OpenCV camera has no wait option. Use None.")
logging.warning(f"OpenCV camera doesn't respect {stream_name=}")
return Image.fromarray(self.capture_array())

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

@ -16,7 +16,6 @@ https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf
from __future__ import annotations
from typing import Annotated, Iterator, Literal, Mapping, Optional, overload
from datetime import datetime
import json
import logging
import os
@ -27,7 +26,6 @@ from contextlib import contextmanager
from threading import RLock
from pydantic import BaseModel, BeforeValidator
import piexif
import numpy as np
from PIL import Image
from picamera2 import Picamera2
@ -44,7 +42,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from . import picamera_recalibrate_utils as recalibrate_utils
from . import BaseCamera, JPEGBlob, ArrayModel
from . import BaseCamera, ArrayModel
class MissingCalibrationError(RuntimeError):
@ -526,22 +524,57 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera() as cam:
cam.capture_metadata()
@contextmanager
def _switch_to_still_capture_mode(self) -> Iterator[Picamera2]:
"""Get the picamera lock, pause stream and switch into still capture config.
Restarts stream when complete.
"""
with self._streaming_picamera(pause_stream=True) as cam:
logging.debug("Reconfiguring camera for full resolution capture")
cam.configure(cam.create_still_configuration(sensor=self._sensor_mode))
cam.start()
time.sleep(0.2)
yield cam
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = 0.9,
) -> Image:
"""Acquire one image from the camera.
"""Acquire one image from the camera and return it as a PIL Image.
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).
stream_name: (Optional) The PiCamera2 stream to use, should be one of ["main", "lores", "raw"]. Default = "main"
wait: (Optional, float) Set a timeout in seconds.
A TimeoutError is raised if this time is exceeded during capture.
Default = 0.9s, lower than the 1s timeout default in picamera yaml settings
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.
:raises TimeoutError: if this time is exceeded during capture.
"""
with self._streaming_picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
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_still_capture_mode() as cam:
return cam.capture_image(name="main", wait=wait)
else:
raise ValueError(f'Unknown stream name "{stream_name}"')
@lt.thing_action
def capture_array(
@ -555,19 +588,26 @@ class StreamingPiCamera2(BaseCamera):
It's likely to be highly inefficient - raw and/or uncompressed captures using
binary image formats will be added in due course.
stream_name: (Optional) The PiCamera2 stream to use, should be one of ["main", "lores", "raw", "full"]. Default = "main"
wait: (Optional, float) Set a timeout in seconds.
A TimeoutError is raised if this time is exceeded during capture.
Default = 0.9s, lower than the 1s timeout default in picamera yaml settings
: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.
"""
# This was slower than capture_image for our use case, but directly returning
# an image as an array is still a useful feature
if stream_name == "full":
with self._streaming_picamera(pause_stream=True) as picam2:
capture_config = picam2.create_still_configuration()
return picam2.switch_mode_and_capture_array(capture_config, wait=wait)
with self._streaming_picamera() as cam:
return cam.capture_array(stream_name, wait=wait)
if stream_name == "raw":
# Raw cannot used capture_image.
if wait is None:
wait = 0.9
with self._switch_to_still_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))))``.
# 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))
@lt.thing_property
def camera_configuration(self) -> Mapping:
@ -584,62 +624,6 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera() as cam:
return cam.camera_configuration()
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 0.9,
) -> JPEGBlob:
"""Acquire one image from the camera as a JPEG.
The JPEG will be acquired using ``Picamera2.capture_file``. If the
``resolution`` 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.
wait: (Optional, float) Set a timeout in seconds.
A TimeoutError is raised if this time is exceeded during capture.
Default = 0.9s, lower than the 1s timeout default in picamera yaml settings
Note that this always uses the image processing pipeline - to
bypass this, you must use a raw capture.
"""
fname = datetime.now().strftime("%Y-%m-%d-%H%M%S.jpeg")
folder = tempfile.TemporaryDirectory()
path = os.path.join(folder.name, fname)
config = self.camera_configuration
# Low-res and main streams are running already - so we don't need
# to reconfigure for these
if resolution in ("lores", "main") and config[resolution]:
with self._streaming_picamera() as cam:
cam.capture_file(path, name=resolution, format="jpeg", wait=wait)
else:
if resolution != "full":
logging.warning(
f"There was no {resolution} stream, capturing full resolution"
)
with self._streaming_picamera(pause_stream=True) as cam:
logging.info("Reconfiguring camera for full resolution capture")
cam.configure(cam.create_still_configuration())
cam.start()
cam.options["quality"] = 95
logging.info("capturing")
cam.capture_file(path, name="main", format="jpeg", wait=wait)
logging.info("done")
# After the file is written, add metadata about the current Things
exif_dict = piexif.load(path)
exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps(
metadata_getter()
).encode("utf-8")
piexif.insert(piexif.dump(exif_dict), path)
return JPEGBlob.from_temporary_directory(folder, fname)
@lt.thing_property
def capture_metadata(self) -> dict:
"""Return the metadata from the camera."""

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

@ -7,8 +7,6 @@ See repository root for licensing information.
"""
from __future__ import annotations
import io
import json
import logging
from typing import Literal, Optional
from threading import Thread
@ -17,7 +15,6 @@ import time
import cv2
import numpy as np
from PIL import Image
import piexif
from scipy.ndimage import gaussian_filter
import labthings_fastapi as lt
@ -29,7 +26,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from . import BaseCamera, JPEGBlob, ArrayModel
from . import BaseCamera, ArrayModel
from ..stage import BaseStage
# The ratio between "motor" steps and pixels
@ -280,7 +277,8 @@ class SimulatedCamera(BaseCamera):
@lt.thing_action
def capture_array(
self,
resolution: Literal["main", "full"] = "full",
stream_name: Literal["main", "full"] = "full",
wait: Optional[float] = None,
) -> ArrayModel:
"""Acquire one image from the camera and return as an array.
@ -288,37 +286,11 @@ 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.
"""
logging.warning(f"Simulation camera doesn't respect {resolution=} setting")
if wait is not None:
logging.warning("Simulation camera has no wait option. Use None.")
logging.warning(f"Simulation camera camera doesn't respect {stream_name=}")
return self.generate_frame()
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob.
This function will produce a JPEG image.
"""
logging.warning(f"Simulation camera doesn't respect {resolution=} setting")
frame = self.capture_array()
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
exif_dict = {
"Exif": {
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode(
"utf-8"
)
},
"GPS": {},
"Interop": {},
"1st": {},
"thumbnail": None,
}
output = io.BytesIO()
piexif.insert(piexif.dump(exif_dict), jpeg, output)
return JPEGBlob.from_bytes(output.getvalue())
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
@ -328,9 +300,9 @@ class SimulatedCamera(BaseCamera):
It is used for capture to memory.
"""
logging.warning(
f"Simulation camera doesn't respect {stream_name=} or {wait=} arguments."
)
if wait is not None:
logging.warning("Simulation camera has no wait option. Use None.")
logging.warning(f"Simulation camera camera doesn't respect {stream_name=}")
return Image.fromarray(self.generate_frame())
@lt.thing_action

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

@ -141,7 +141,7 @@
<action-button
thing="camera"
action="capture_jpeg"
:submit-data="{ resolution: 'main' }"
:submit-data="{ stream_name: 'main' }"
:submit-label="'Low Resolution'"
:submit-on-event="'globalCaptureEvent'"
@response="handleCaptureResponse"
@ -153,7 +153,7 @@
<action-button
thing="camera"
action="capture_jpeg"
:submit-data="{ resolution: 'full' }"
:submit-data="{ stream_name: 'full' }"
submit-label="Full Resolution"
:submit-on-event="'globalCaptureEvent'"
@response="handleCaptureResponse"