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

Unify JPEG capture, metadata, and saving; reducing duplication

Browse source
This commit is contained in:
Julian Stirling 2025-08-22 10:24:16 +01:00
parent 4450fda945
commit 5484b51a1e
5 changed files with 83 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -8,10 +8,14 @@ See repository root for licensing information.
from __future__ import annotations
from typing import Literal, Optional, Tuple, Any
from types import EllipsisType
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 +261,42 @@ 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 | EllipsisType] = ...,
) -> 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 if
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 not set 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)
# Using Ellipsis to specify no input specified. As `None` set for wait may
# have a meaning as it does for the Picamera. If wait is Ellipsis then
# do not specify.
if wait is Ellipsis:
img = self.capture_image(stream_name)
else:
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 +363,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 +518,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

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

@ -7,14 +7,13 @@ 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
@ -100,30 +99,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",
stream_name: Literal["main", "full"] = "main",
wait: Optional[float] = None,
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob.
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())
logging.warning(
f"Simulation camera doesn't respect {stream_name=} or {wait=} arguments."
)
return Image.fromarray(self.capture_array())

100
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):
@ -528,20 +526,39 @@ class StreamingPiCamera2(BaseCamera):
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
stream_name: Literal["main", "lores", "raw", "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.
: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.
:rasises TimeoutError: if this time is exceeded during capture. 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.
"""
with self._streaming_picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
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._streaming_picamera(pause_stream=True) as cam:
logging.debug("Reconfiguring camera for full resolution capture")
cam.configure(cam.create_still_configuration())
cam.start()
time.sleep(0.2)
return cam.capture_image(name="main", wait=wait)
else:
raise ValueError(f'Unknown stream name "{stream_name}"')
@lt.thing_action
def capture_array(
@ -555,7 +572,8 @@ 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"
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
@ -584,62 +602,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."""

33
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
@ -291,34 +288,6 @@ class SimulatedCamera(BaseCamera):
logging.warning(f"Simulation camera doesn't respect {resolution=} setting")
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"],

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"