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

Merge branch 'v3' into fast-stack, with conflict resolution but not fully

Browse source 
working yet
This commit is contained in:
Julian Stirling 2025-06-26 11:50:24 +01:00
commit fda0a6e1ab
20 changed files with 2167 additions and 366 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/openflexure_microscope_server/server/__init__.py
View file

@ -60,6 +60,7 @@ def serve_from_cli(argv: Optional[list[str]] = None):
host=args.host,
port=args.port,
log_config=log_config,
timeout_graceful_shutdown=2,
)
except BaseException as e:

146
src/openflexure_microscope_server/things/autofocus.py
View file

@ -6,7 +6,6 @@ camera together to perform an autofocus routine.
See repository root for licensing information.
"""
from __future__ import annotations
from contextlib import contextmanager
import logging
import time
@ -31,45 +30,64 @@ from .capture import RawCaptureDependency as CaptureDep
class StackParams(BaseModel):
"""Pydantic model for scan parameters"""
"""Pydantic model for scan parameters
# These are generic settings that seem to apply well to a standard scan
:param stack_dz: The number of motor steps between images
:param images_to_save: The number of images to save to disk
:param min_images_to_test: The minimum number of images in the stack before, the
stack is evaluated for focus. As more images are captured evaluation of the focus
is always evaluated with the same number of images. i.e. if min_images_to_test=9,
then 9 images are captured, if the stack is not well focussed, a 10th image is
captured and images 2 to 10 are evaluated for focus
:param autofocus_dz: The number of steps in a full autofocus (if required)
"""
# These parameters must be set on initialisation
stack_dz: int
images_to_save: int
min_images_to_test: int
autofocus_dz: int
# The following parameters have values that apply well to a standard scan
# and are not altered by default when running a scan
# time (in seconds) between moving and capturing an image
settling_time: float = 0.3
# distance (in steps) to overshoot a move and then undo, to account for backlash
backlash_correction: int = 250
# how many images can be appended to the stack after the predicted peak to test for focus
# before assuming the focus was passed, and restarting the stack
# how many images can be appended to the stack after the predicted peak to test
# for focus before assuming the focus was passed, and restarting the stack
stack_height_limit: int = 15
# how far below (in factors of stack_dz) the estimated optimal starting position to
# begin the stack. Better to start slightly too low and require many images, rather than
# too high and needing to autofocus and restart the stack
# begin the stack. Better to start slightly too low and require many images, rather
# than too high and needing to autofocus and restart the stack
img_undershoot: int = 5
# These we expect to be overwritten by AutofocusThing properties
stack_dz: int = 50
images_to_capture: int = 1
images_to_test: int = 5
autofocus_dz: int = 2000
# Per pydantic docs, even with the @property applied before @computed_field,
# mypy may throw a Decorated property not supported error (mypy issue #1362)
# To avoid this error message, add # type: ignore[prop-decorator] to the @computed_field line.
# To avoid this error message, add # type: ignore[prop-decorator] to the
# @computed_field line.
@computed_field
@property
def stack_z_range(self) -> int:
"""The range of the entire z stack, in steps"""
return self.stack_dz * (self.images_to_test - 1)
"""The range of the z stack, in steps
Note that this is the range of the minimum number of image captured,
which is also the range of the images sored in memory that can be
saved."""
return self.stack_dz * (self.min_images_to_test - 1)
@computed_field
@property
def steps_undershoot(self) -> int:
"""The distance to deliberately undershoot the estimated optimal starting point"""
"""
The distance to deliberately undershoot the estimated optimal starting point
"""
# Starting too low by "steps_undershoot" makes smart stacking faster.
# Starting a stack too high requires it to move to the start,
@ -77,10 +95,24 @@ class StackParams(BaseModel):
# requires extra +z movements and captures.
return self.stack_dz * self.img_undershoot
@computed_field
@property
def max_images_to_test(self) -> int
"""The maximum number of images that will be captured and tested in a stack
This is 15 images more then the minimum number that are captured.
"""
return self.min_images_to_test
class SharpnessDataArrays(BaseModel):
jpeg_times: NDArray
jpeg_sizes: NDArray
stage_times: NDArray
stage_positions: list[dict[str, int]]
class JPEGSharpnessMonitor:
__globals__ = globals() # Required for FastAPI dependency
def __init__(self, stage: Stage, camera: Camera, portal: BlockingPortal):
self.camera = camera
self.stage = stage
@ -156,9 +188,7 @@ class JPEGSharpnessMonitor:
stop = len(jpeg_times)
logging.debug("changing stop to %s", (stop))
jpeg_times = jpeg_times[start:stop]
jpeg_zs: np.ndarray = np.interp(
jpeg_times, stage_times, stage_zs
) # np.ndarray[float]
jpeg_zs: np.ndarray = np.interp(jpeg_times, stage_times, stage_zs)
return jpeg_times, jpeg_zs, jpeg_sizes[start:stop]
def sharpest_z_on_move(self, index: int) -> int:
@ -181,13 +211,6 @@ class JPEGSharpnessMonitor:
SharpnessMonitorDep = Annotated[JPEGSharpnessMonitor, Depends()]
class SharpnessDataArrays(BaseModel):
jpeg_times: NDArray
jpeg_sizes: NDArray
stage_times: NDArray
stage_positions: list[dict[str, int]]
class AutofocusThing(Thing):
"""The Thing concerned with combinations of z axis movements and the camera.
@ -298,24 +321,24 @@ class AutofocusThing(Thing):
return heights.tolist(), sizes.tolist()
@thing_property
def stack_images_to_capture(self) -> int:
def stack_images_to_save(self) -> int:
"""The number of images to capture and save in a stack
Defaults to 1 unless you need to see either side of focus"""
return self.thing_settings.get("stack_images_to_capture", 1)
return self.thing_settings.get("stack_images_to_save", 1)
@stack_images_to_capture.setter
def stack_images_to_capture(self, value: int) -> None:
self.thing_settings["stack_images_to_capture"] = value
@stack_images_to_save.setter
def stack_images_to_save(self, value: int) -> None:
self.thing_settings["stack_images_to_save"] = value
@thing_property
def stack_images_to_test(self) -> int:
def stack_min_images_to_test(self) -> int:
"""The number of images to test for successful focusing in a stack
Defaults to 9, which balances reliability and speed"""
return self.thing_settings.get("stack_images_to_test", 9)
return self.thing_settings.get("stack_min_images_to_test", 9)
@stack_images_to_test.setter
def stack_images_to_test(self, value: int) -> None:
self.thing_settings["stack_images_to_test"] = value
@stack_min_images_to_test.setter
def stack_min_images_to_test(self, value: int) -> None:
self.thing_settings["stack_min_images_to_test"] = value
@thing_property
def stack_dz(self) -> int:
@ -336,10 +359,10 @@ class AutofocusThing(Thing):
stage: Stage,
logger: InvocationLogger,
metadata_getter: GetThingStates,
capture: CaptureDep,
sharpness_monitor: SharpnessMonitorDep,
images_dir: str,
autofocus_dz: int,
capture_resolution: tuple[int, int],
) -> None:
"""Run a smart stack, which captures images offset in z, testing
whether the sharpest image is towards the centre of the stack.
@ -355,8 +378,8 @@ class AutofocusThing(Thing):
# Set the variables to prevent changes from the GUI or other windows
stack_parameters = StackParams(
stack_dz=self.stack_dz,
images_to_capture=self.stack_images_to_capture,
images_to_test=self.stack_images_to_test,
images_to_save=self.stack_images_to_save,
min_images_to_test=self.stack_min_images_to_test,
autofocus_dz=autofocus_dz,
)
@ -371,7 +394,6 @@ class AutofocusThing(Thing):
stack_parameters,
stage,
cam,
capture,
metadata_getter,
)
@ -393,11 +415,10 @@ class AutofocusThing(Thing):
captures,
stack_parameters,
logger,
capture,
)
# Return the z position of the sharpest image, for path planning and tracking
return heights[-stack_parameters.images_to_test :][sharpest_index]
return heights[-stack_parameters.min_images_to_test :][sharpest_index]
def reset_stack(
self,
@ -427,31 +448,26 @@ class AutofocusThing(Thing):
captures: list[list],
stack_parameters: StackParams,
logger: InvocationLogger,
capture: CaptureDep,
) -> int:
"""Save the required captures to disk. Will save the sharpest image,
and any images either side of focus.
Arguments:
sharpest_index: the index of the sharpest image, within the "images_to_test" slice
sharpest_index: the index of the sharpest image, within the "min_images_to_test" slice
captures: a list of captures, including file name, image data and metadata
stack_parameters: a StackParams Pydantic model with stack settings
variables logger and capture are Thing dependencies passed through from the calling action
"""
# Find the range of images from the stack to capture
stack_extent = int((stack_parameters.images_to_capture - 1) / 2)
stack_range = range(
sharpest_index - stack_extent, sharpest_index + stack_extent + 1
)
# Loop through the range, saving each capture to disk
for capture_index in stack_range:
capture._save_capture(
jpeg_path=captures[-stack_parameters.images_to_test :][capture_index][
jpeg_path=captures[-stack_parameters.min_images_to_test :][capture_index][
0
],
image=captures[-stack_parameters.images_to_test :][capture_index][1],
metadata=captures[-stack_parameters.images_to_test :][capture_index][2],
image=captures[-stack_parameters.min_images_to_test :][capture_index][1],
metadata=captures[-stack_parameters.min_images_to_test :][capture_index][2],
logger=logger,
)
return sharpest_index
@ -493,7 +509,7 @@ class AutofocusThing(Thing):
# If the sharpest image isn't found within the 15 images above the estimated point, break
# the loop and return "restart"
while len(captures) <= stack_parameters.images_to_test + 15:
while len(captures) <= stack_parameters.min_images_to_test + 15:
time.sleep(stack_parameters.settling_time)
# Append a new image to the stack
@ -509,14 +525,14 @@ class AutofocusThing(Thing):
)
# If the number of images is enough to test, test them
if len(captures) >= stack_parameters.images_to_test:
if len(captures) >= stack_parameters.min_images_to_test:
stack_result = self.check_stack_result(
sharpnesses[-stack_parameters.images_to_test :]
sharpnesses[-stack_parameters.min_images_to_test :]
)
if stack_result == "success":
sharpest_index = np.argmax(
sharpnesses[-stack_parameters.images_to_test :]
sharpnesses[-stack_parameters.min_images_to_test :]
)
return "success", heights, captures, sharpest_index
@ -567,18 +583,18 @@ class AutofocusThing(Thing):
"""Check the stack settings are appropriate, and raise an error if not
Arguments:
images_to_test: number of images in the stack to test for focus
images_to_capture: number of images to be captured around the focused image
min_images_to_test: number of images in the stack to test for focus
images_to_save: number of images to be captured around the focused image
"""
if stack_parameters.images_to_test < stack_parameters.images_to_capture:
if stack_parameters.min_images_to_test < stack_parameters.images_to_save:
raise RuntimeError(
"Can't capture more images than are tested. Please increase number to test, or decrease number to capture"
)
if stack_parameters.images_to_test % 2 == 0:
if stack_parameters.min_images_to_test % 2 == 0:
raise RuntimeError("Images to test should be odd")
if (
stack_parameters.images_to_test <= 0
or stack_parameters.images_to_capture <= 0
stack_parameters.min_images_to_test <= 0
or stack_parameters.images_to_save <= 0
):
raise RuntimeError("Stack parameters need to be at least 1")

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

@ -7,8 +7,12 @@ See repository root for licensing information.
"""
from __future__ import annotations
import logging
from typing import Literal, Protocol, runtime_checkable
from typing import Literal, Optional, Tuple
import json
from pydantic import RootModel
from PIL import Image
import piexif
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
@ -16,6 +20,7 @@ from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
from labthings_fastapi.dependencies.raw_thing import raw_thing_dependency
from labthings_fastapi.dependencies.invocation import InvocationLogger
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
from labthings_fastapi.outputs.blob import Blob
from labthings_fastapi.types.numpy import NDArray
@ -25,160 +30,243 @@ class JPEGBlob(Blob):
media_type: str = "image/jpeg"
@runtime_checkable
class CameraProtocol(Protocol):
"""A Thing representing a camera"""
class PNGBlob(Blob):
"""A class representing a PNG image as a LabThings FastAPI Blob"""
def __enter__(self) -> None: ...
media_type: str = "image/png"
def __exit__(self, _exc_type, _exc_value, _traceback) -> None: ...
@property
def stream_active(self) -> bool:
"Whether the MJPEG stream is active"
...
class ArrayModel(RootModel):
"""A model for an array"""
def snap_image(self) -> NDArray:
"""Acquire one image from the camera."""
...
root: NDArray
def capture_array(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
) -> NDArray: ...
def capture_jpeg(
self,
metadata_getter: GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob"""
...
class CaptureError(RuntimeError):
"""An error trying to capture from a CameraThing"""
def grab_jpeg(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as an array
This differs from `capture_jpeg` 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.
"""
...
def grab_jpeg_size(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size"""
...
class NoImageInMemoryError(RuntimeError):
"""An error called if no image in in memory when an method is called to use that image"""
class BaseCamera(Thing):
"""A Thing representing a camera
This is a concrete base class for `Thing`s implementing the `CameraProtocol`.
It provides the stream descriptors and actions to grab from the stream.
"""
"""The base class for all cameras. All cameras must directly inherit from this class"""
_memory_image: Optional[Image] = None
_memory_metadata: Optional[dict] = None
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
@thing_action
def snap_image(self) -> NDArray:
"""Acquire one image from the camera.
This action cannot run if the camera is in use by a background thread, for
example if a preview stream is running.
"""
return self.capture_array()
@thing_action
def grab_jpeg(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as an array
This differs from `capture_jpeg` 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.
"""
logging.info(
f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) starting"
)
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
frame = portal.call(stream.grab_frame)
logging.info(
f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) got frame"
)
return JPEGBlob.from_bytes(frame)
@thing_action
def grab_jpeg_size(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size"""
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
return portal.call(stream.next_frame_size)
class CameraStub(BaseCamera):
"""A stub for a camera, to allow dependencies
As of LabThings-FastAPI 0.0.7, we can't make a thing client dependency
based on a protocol, because the protocol is not a Thing, and its
methods/properties are not decorated as Affordances. This stub class
is a workaround for that limitation, and should not be used directly.
This stub class should be used for dependencies on the CameraProtocol.
"""
def __enter__(self) -> None:
raise NotImplementedError("Cameras must not inherit from CameraStub")
raise NotImplementedError("CameraThings must define their own __enter__ method")
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
raise NotImplementedError("Cameras must not inherit from CameraStub")
raise NotImplementedError("CameraThings must define their own __exit__ method")
@thing_property
def image_in_memory(self) -> bool:
"""True if an image is in memory ready to be saved"""
return self._memory_image is not None and self._memory_metadata is not None
@thing_action
def clear_image_memory(self) -> None:
"""Clear any image in memory"""
self._memory_image = None
self._memory_metadata = None
@thing_action
def start_streaming(
self, main_resolution: tuple[int, int], buffer_count: int
) -> None:
"""Start (or stop and restart) the camera with the given resolution
for the main stream, and buffer_count number of images in the buffer"""
raise NotImplementedError(
"CameraThings must define their own start_streaming method"
)
@thing_property
def stream_active(self) -> bool:
"Whether the MJPEG stream is active"
raise NotImplementedError("Cameras must not inherit from CameraStub")
@thing_action
def snap_image(self) -> NDArray:
"""Acquire one image from the camera."""
raise NotImplementedError("Cameras must not inherit from CameraStub")
raise NotImplementedError(
"CameraThings must define their own stream_active method"
)
@thing_action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 5,
) -> NDArray:
raise NotImplementedError("Cameras must not inherit from CameraStub")
raise NotImplementedError(
"CameraThings must define their own capture_array method"
)
@thing_action
def capture_jpeg(
self,
metadata_getter: GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 5,
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob"""
raise NotImplementedError("Cameras must not inherit from CameraStub")
raise NotImplementedError(
"CameraThings must define their own capture_jpeg method"
)
@thing_action
def grab_jpeg(
self,
portal: BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as an array
This differs from `capture_jpeg` 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.
"""
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
frame = portal.call(stream.grab_frame)
return JPEGBlob.from_bytes(frame)
@thing_action
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
wait: Optional[float],
) -> None:
"""Capture a PIL image from stream stream_name with timeout wait"""
raise NotImplementedError(
"CameraThings must define their own capture_image method"
)
@thing_action
def capture_and_save(
self,
jpeg_path: str,
logger: InvocationLogger,
metadata_getter: GetThingStates,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Capture an image and save it to disk
CameraDependency = direct_thing_client_dependency(CameraStub, "/camera/")
RawCameraDependency = raw_thing_dependency(CameraProtocol)
save_resolution can be set to resize the image before saving. By default this is None
meaning that the image is saved at original resoltion.
"""
image, metadata = self._robust_image_capture(
metadata_getter,
logger=logger,
)
self._save_capture(
jpeg_path,
image,
metadata,
logger,
save_resolution,
)
@thing_action
def capture_to_memory(
self,
logger: InvocationLogger,
metadata_getter: GetThingStates,
) -> None:
"""
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
in memory.
"""
self._memory_image, self._memory_metadata = self._robust_image_capture(
metadata_getter,
logger=logger,
)
@thing_action
def save_from_memory(
self,
jpeg_path: str,
logger: InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""
Save an image that has been captured to memory.
"""
if not self.image_in_memory:
raise NoImageInMemoryError("No image in memory to save.")
self._save_capture(
jpeg_path=jpeg_path,
image=self._memory_image,
metadata=self._memory_metadata,
logger=logger,
save_resolution=save_resolution,
)
self.clear_image_memory()
def _robust_image_capture(
self,
metadata_getter: GetThingStates,
logger: InvocationLogger,
) -> Image:
"""Capture an image in memory and return it with metadata
CaptureError raised if the capture fails for any reason
returns tuple with PIL Image, and dict of metadata.
This robust capturing method attempts to capture the image five times
each time with a 5 second timeout set.
"""
for capture_attempts in range(5):
try:
metadata = metadata_getter()
image = self.capture_image(stream_name="main", wait=5)
return image, metadata
except TimeoutError:
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 _save_capture(
self,
jpeg_path: str,
image: Image,
metadata: dict,
logger: InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Saving the captured image and metadata to disk
logger warning (via InvocationLogger) is raised if metadata is failed to be added
IOError is raised 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.BOX)
try:
# 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 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
image.save(jpeg_path, quality=95, subsampling=0)
try:
# Load EXIF metadata from image so it can be added to.
exif_dict = piexif.load(jpeg_path)
exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps(
metadata
).encode("utf-8")
piexif.insert(piexif.dump(exif_dict), jpeg_path)
except: # noqa: E722
# 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}")
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e
CameraDependency = direct_thing_client_dependency(BaseCamera, "/camera/")
RawCameraDependency = raw_thing_dependency(BaseCamera)

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

@ -19,7 +19,6 @@ import piexif
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
from labthings_fastapi.types.numpy import NDArray
from . import BaseCamera, JPEGBlob
@ -53,9 +52,6 @@ class OpenCVCamera(BaseCamera):
return self._capture_thread.is_alive()
return False
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
def _capture_frames(self):
portal = get_blocking_portal(self)
while self._capture_enabled:

919
src/openflexure_microscope_server/things/camera/picamera.py Normal file
View file

@ -0,0 +1,919 @@
"""
This SubModule interacts with a Raspberry Pi camera using the Picamera2 library.
The Picamera2 library uses LibCamera as the underlying camera stack. This gives us
some control of the GPU pipeline for the image.
The API documentation for PiCamera2 is unfortunately not in a standard auto-generated
website. For documentation of the PiCamera2 API there is a PDF called
"The Picamera2 Library" available at:
https://datasheets.raspberrypi.com/camera/picamera2-manual.pdf
For information on the algorithms used to tune/calibrate the Raspberry Pi Camera see
the guide called "Raspberry Pi Camera Algorithm and Tuning Guide"
Available at:
https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf
"""
from __future__ import annotations
from datetime import datetime
import json
import logging
import os
import tempfile
import time
from tempfile import TemporaryDirectory
from pydantic import BaseModel, BeforeValidator
from labthings_fastapi.descriptors.property import PropertyDescriptor
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStream
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from typing import Annotated, Any, Iterator, Literal, Mapping, Optional
from contextlib import contextmanager
import piexif
from threading import RLock
import picamera2
from picamera2 import Picamera2
from picamera2.encoders import MJPEGEncoder
from picamera2.outputs import Output
import numpy as np
from . import picamera_recalibrate_utils as recalibrate_utils
from . import BaseCamera, JPEGBlob, ArrayModel
class PicameraControl(PropertyDescriptor):
def __init__(
self, control_name: str, model: type = float, description: Optional[str] = None
):
"""A property descriptor controlling a picamera control"""
PropertyDescriptor.__init__(
self, model, observable=False, description=description
)
self.control_name = control_name
def _getter(self, obj: StreamingPiCamera2):
with obj.picamera() as cam:
return cam.capture_metadata()[self.control_name]
def _setter(self, obj: StreamingPiCamera2, value: Any):
with obj.picamera() as cam:
cam.set_controls({self.control_name: value})
class PicameraStreamOutput(Output):
"""An Output class that sends frames to a stream"""
def __init__(self, stream: MJPEGStream, portal: BlockingPortal):
"""Create an output that puts frames in an MJPEGStream
We need to pass the stream object, and also the blocking portal, because
new frame notifications happen in the anyio event loop and frames are
sent from a thread. The blocking portal enables thread-to-async
communication.
"""
Output.__init__(self)
self.stream = stream
self.portal = portal
def outputframe(
self, frame, _keyframe=True, _timestamp=None, _packet=None, _audio=False
):
"""Add a frame to the stream's ringbuffer"""
self.stream.add_frame(frame, self.portal)
class SensorMode(BaseModel):
"""
A Pydantic model holding all the information about a specific
sensor mode as reported by the PiCamera.
"""
unpacked: str
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)]
class SensorModeSelector(BaseModel):
"""
A Pydantic model holding the two values needed to select a PiCamera
Sensor mode. The output size and the bit depth.
This is a Pydantic modell so that it can be saved to the disk.
"""
output_size: tuple[int, int]
bit_depth: int
class LensShading(BaseModel):
"""
A Pydantic model holding the lens shading tables.
PiCamera needs three numpy arrays for lens shading correction. Each array is
(12, 16) in size. The arrays are luminance, red-difference chroma (Cr), and
blue-difference chroma (Cb).
This is a Pydantic modell so that it can be saved to the disk.
"""
luminance: list[list[float]]
Cr: list[list[float]]
Cb: list[list[float]]
class StreamingPiCamera2(BaseCamera):
"""A Thing that represents an OpenCV camera"""
def __init__(self, camera_num: int = 0):
self.camera_num = camera_num
self.camera_configs: dict[str, dict] = {}
# Note persistent controls will be updated with settings, in __enter__.
self.persistent_controls = {
"AeEnable": False,
"AnalogueGain": 1.0,
"AwbEnable": False,
"Brightness": 0,
"ColourGains": (1, 1),
"Contrast": 1,
"ExposureTime": 0,
"Saturation": 1,
"Sharpness": 1,
}
self.persistent_control_tolerances = {
"ExposureTime": 30,
}
def update_persistent_controls(self, discard_frames: int = 1):
"""Update the persistent controls dict from the camera
Query the camera and update the value of `persistent_controls` to
match the current state of the camera.
There is a work-around here, that will suppress small updates. There
appears to be a bug in the camera code that causes a slight drift in
`ExposureTime` each time the camera is reinitialised: this can
add up over time, particularly if the camera is reconfigured many
times. To get around this, we look in `self.persistent_control_tolerances`
and only update `self.persistent_controls` if the change is greater than
this tolerance.
"""
with self.picamera() as cam:
for i in range(discard_frames):
# Discard frames, so data is fresh
cam.capture_metadata()
for key, value in cam.capture_metadata().items():
if key not in self.persistent_controls:
# Skip keys that are not persistent controls
continue
if self._control_value_within_tolerance(key, value):
logging.debug(
"Ignoring a small change in persistent control %s from %s to "
"%s while updating persistent controls.",
key,
self.persistent_controls[key],
value,
)
else:
self.persistent_controls[key] = value
self.thing_settings.update(self.persistent_controls)
def _control_value_within_tolerance(self, key: str, value: float) -> bool:
"""Return True if the updated value for a persistent control is within
tolerance of current value. This allows ignoring small changes
"""
if key not in self.persistent_control_tolerances:
# Not tolerance range set, so it is not within tolerance
return False
difference = np.abs(self.persistent_controls[key] - value)
return difference < self.persistent_control_tolerances[key]
def settings_to_persistent_controls(self):
"""Update the persistent controls dict from the settings dict
NB this must be called **after** self.thing_settings is initialised,
i.e. during or after `__enter__`.
"""
try:
pc = self.thing_settings["persistent_controls"]
except KeyError:
return # If there are no saved settings, use defaults
for k in self.persistent_controls:
try:
self.persistent_controls[k] = pc[k]
except KeyError:
pass # If controls are missing, leave at default
stream_resolution = PropertyDescriptor(
tuple[int, int],
initial_value=(820, 616),
description="Resolution to use for the MJPEG stream",
)
mjpeg_bitrate = PropertyDescriptor(
Optional[int],
initial_value=100000000,
description="Bitrate for MJPEG stream (None for default)",
)
stream_active = PropertyDescriptor(
bool,
initial_value=False,
description="Whether the MJPEG stream is active",
observable=True,
readonly=True,
)
analogue_gain = PicameraControl("AnalogueGain", float)
colour_gains = PicameraControl("ColourGains", tuple[float, float])
exposure_time = PicameraControl(
"ExposureTime", int, description="The exposure time in microseconds"
)
@exposure_time.setter
def exposure_time(self, value: int):
"""
Custom setter for the above exposure_time PicameraControl.
This is overriding the standard setter for a PicameraControl, as
the value needs to be adjusted before setting to behave as expected.
See comment within the function for more detail.
"""
with self.picamera() as cam:
# Note: This set a value 1 higher than requested as picamera2 always sets
# a lower value than requested, even if the requested is allowed
cam.set_controls({"ExposureTime": value + 1})
_sensor_modes = None
@thing_property
def sensor_modes(self) -> list[SensorMode]:
"""All the available modes the current sensor supports"""
if not self._sensor_modes:
with self.picamera() as cam:
self._sensor_modes = cam.sensor_modes
return self._sensor_modes
@thing_property
def sensor_mode(self) -> Optional[SensorModeSelector]:
"""The intended sensor mode of the camera"""
return self.thing_settings["sensor_mode"]
@sensor_mode.setter
def sensor_mode(self, new_mode: Optional[SensorModeSelector]):
"""Change the sensor mode used"""
if isinstance(new_mode, SensorModeSelector):
new_mode = new_mode.model_dump()
with self.picamera(pause_stream=True):
self.thing_settings["sensor_mode"] = new_mode
@thing_property
def sensor_resolution(self) -> tuple[int, int]:
"""The native resolution of the camera's sensor"""
with self.picamera() as cam:
return cam.sensor_resolution
tuning = PropertyDescriptor(Optional[dict], None, readonly=True)
def settings_to_properties(self):
"""Set the values of properties based on the settings dict"""
try:
props = self.thing_settings["properties"]
except KeyError:
return
for k, v in props.items():
setattr(self, k, v)
def properties_to_settings(self):
"""Save certain properties to the settings dictionary"""
props = {}
for k in ["mjpeg_bitrate", "stream_resolution"]:
props[k] = getattr(self, k)
self.thing_settings["properties"] = props
def initialise_tuning(self):
"""Read the tuning from the settings, or load default tuning
NB this relies on `self.thing_settings` and `self.default_tuning`
so will fail if it's run before those are populated in `__enter__`.
"""
if "tuning" in self.thing_settings:
self.tuning = self.thing_settings["tuning"].dict
else:
logging.info("Did not find tuning in settings, reading from camera...")
self.tuning = self.default_tuning
def initialise_picamera(self):
"""Acquire the picamera device and store it as `self._picamera`"""
if hasattr(self, "_picamera_lock"):
# Don't close the camera if it's in use
self._picamera_lock.acquire()
with tempfile.NamedTemporaryFile("w") as tuning_file:
# This duplicates logic in `Picamera2.__init__` to provide a tuning file
# that will be read when the camera system initialises.
# This is a necessary work-around until `picamera2` better supports
# reinitialisation of the camera with new tuning.
json.dump(self.tuning, tuning_file)
tuning_file.flush() # but leave it open as closing it will delete it
os.environ["LIBCAMERA_RPI_TUNING_FILE"] = tuning_file.name
# NB even though we've put the tuning file in the environment, we will
# need to specify the filename in the `Picamera2` initialiser as otherwise
# it will be overwritten with None.
if hasattr(self, "_picamera") and self._picamera:
print("Closing picamera object for reinitialisation")
logging.info(
"Camera object already exists, closing for reinitialisation"
)
self._picamera.close()
print("closed, deleting picamera")
del self._picamera
recalibrate_utils.recreate_camera_manager()
print("[re]creating Picamera2 object")
self._picamera = picamera2.Picamera2(
camera_num=self.camera_num,
tuning=self.tuning,
)
self._picamera_lock = RLock()
def __enter__(self):
self.populate_default_tuning()
self.initialise_tuning()
self.initialise_picamera()
# populate sensor modes by reading the property
self.sensor_modes
self.settings_to_persistent_controls()
self.settings_to_properties()
self.start_streaming()
return self
@contextmanager
def picamera(self, pause_stream=False) -> Iterator[Picamera2]:
"""Return the underlying `Picamera2` instance, optionally pausing the stream.
If pause_stream is True (default is False), we will stop the MJPEG stream
before yielding control of the camera, and restart afterwards. If you make
changes to the camera settings, these may be ignored when the stream is
restarted: you may nened to call `update_persistent_controls()` to ensure
your changes persist after the stream restarts.
"""
already_streaming = self.stream_active
with self._picamera_lock:
if pause_stream and already_streaming:
self.update_persistent_controls()
self.stop_streaming(stop_web_stream=False)
try:
yield self._picamera
finally:
if pause_stream and already_streaming:
self.start_streaming()
def populate_default_tuning(self):
"""Sensor modes are enumerated and stored, once, on start-up (`__enter__`).
This opens and closes the camera - must be run before the camera is
initialised.
"""
logging.info("Starting & reconfiguring camera to populate sensor_modes.")
with Picamera2(camera_num=self.camera_num) as cam:
self.default_tuning = recalibrate_utils.load_default_tuning(cam)
logging.info("Done reading sensor modes & default tuning.")
def __exit__(self, exc_type, exc_value, traceback):
# Allow key controls to persist across restarts
self.update_persistent_controls()
self.thing_settings["persistent_controls"] = self.persistent_controls
self.thing_settings["tuning"] = self.tuning
self.properties_to_settings()
self.thing_settings.write_to_file()
# Shut down the camera
self.stop_streaming()
with self.picamera() as cam:
cam.close()
del self._picamera
@thing_action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 6
) -> None:
"""
Start the MJPEG stream
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.
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.
"""
# 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"
)
with self.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.thing_settings.get("sensor_mode", None),
controls=self.persistent_controls,
)
stream_config["buffer_count"] = buffer_count
picam.configure(stream_config)
logging.info("Starting picamera MJPEG stream...")
stream_name = "lores" if main_resolution[0] > 1280 else "main"
picam.start_recording(
MJPEGEncoder(self.mjpeg_bitrate),
PicameraStreamOutput(
self.mjpeg_stream,
get_blocking_portal(self),
),
name=stream_name,
)
picam.start_encoder(
MJPEGEncoder(100000000),
PicameraStreamOutput(
self.lores_mjpeg_stream,
get_blocking_portal(self),
),
name="lores",
)
except Exception as e:
logging.exception("Error while starting preview: {e}")
logging.exception(e)
else:
self.stream_active = True
logging.debug(
"Started MJPEG stream at %s on port %s", self.stream_resolution, 1
)
@thing_action
def stop_streaming(self, stop_web_stream: bool = True) -> None:
"""
Stop the MJPEG stream
"""
with self.picamera() as picam:
try:
picam.stop_recording() # This should also stop the extra lores encoder
except Exception as e:
logging.info("Stopping recording failed")
logging.exception(e)
else:
self.stream_active = False
if stop_web_stream:
self.mjpeg_stream.stop()
self.lores_mjpeg_stream.stop()
logging.info("Stopped MJPEG stream.")
# Adding a sleep to prevent camera getting confused by rapid commands
time.sleep(0.2)
@thing_action
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
wait: Optional[float] = 0.9,
) -> None:
"""Acquire one image from the camera.
Return it as a PIL Image
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
"""
with self.picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
@thing_action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 0.9,
) -> ArrayModel:
"""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.
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
"""
# 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.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.picamera() as cam:
return cam.capture_array(stream_name, wait=wait)
@thing_property
def camera_configuration(self) -> Mapping:
"""The "configuration" dictionary of the picamera2 object
The "configuration" sets the resolution and format of the camera's streams.
Together with the "tuning" it determines how the sensor is configured and
how the data is processed.
Note that the configuration may be modified when taking still images, and
this property refers to whatever configuration is currently in force -
usually the one used for the preview stream.
"""
with self.picamera() as cam:
return cam.camera_configuration()
@thing_action
def capture_jpeg(
self,
metadata_getter: 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 = 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.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.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)
@thing_property
def capture_metadata(self) -> dict:
"""Return the metadata from the camera"""
with self.picamera() as cam:
return cam.capture_metadata()
@thing_action
def auto_expose_from_minimum(
self,
target_white_level: int = 700,
percentile: float = 99.9,
):
"""Adjust exposure until a the target white level is reached
Starting from the minimum exposure, gradually increase exposure until
the image reaches the specified white level.
:param target_white_level: The target 10bit white level. 10-bit data has a
theoretical maximum of 1023, but with black level correction the true maxiumum
is about 950. Default is 700 as this is approximately 70% saturated.
:param percentile: The percentile to use instead of maximum. Default 99.9. When
calculating the brightest pixel, a percentile is used rather than the
maximum in order to be robust to a small number of noisy/bright pixels.
"""
with self.picamera(pause_stream=True) as cam:
recalibrate_utils.adjust_shutter_and_gain_from_raw(
cam,
target_white_level=target_white_level,
percentile=percentile,
)
self.update_persistent_controls()
@thing_action
def calibrate_white_balance(
self,
method: Literal["percentile", "centre"] = "centre",
luminance_power: float = 1.0,
):
"""Correct the white balance of the image
This calibration requires a neutral image, such that the 99th centile
of each colour channel should correspond to white. We calculate the
centiles and use this to set the colour gains. This is done on the raw
image with the lens shading correction applied, which should mean
that the image is uniform, rather than weighted towards the centre.
If `method` is `"centre"`, we will correct the mean of the central 10%
of the image.
"""
with self.picamera(pause_stream=True) as cam:
if self.lens_shading_is_static:
lst: LensShading = self.lens_shading_tables
recalibrate_utils.adjust_white_balance_from_raw(
cam,
percentile=99,
luminance=lst.luminance,
Cr=lst.Cr,
Cb=lst.Cb,
luminance_power=luminance_power,
method=method,
)
else:
recalibrate_utils.adjust_white_balance_from_raw(
cam, percentile=99, method=method
)
self.update_persistent_controls()
@thing_action
def calibrate_lens_shading(self) -> None:
"""Take an image and use it for flat-field correction.
This method requires an empty (i.e. bright) field of view. It will take
a raw image and effectively divide every subsequent image by the current
one. This uses the camera's "tuning" file to correct the preview and
the processed images. It should not affect raw images.
"""
with self.picamera(pause_stream=True) as cam:
# Suppress lint warning that L, Cr, and Cb are not lowercase, as these are
# the standard mathematical terms for:
# luminance (L), red-difference chroma (Cr), and blue-difference chroma
# (Cb).
L, Cr, Cb = recalibrate_utils.lst_from_camera(cam) # noqa: N806
recalibrate_utils.set_static_lst(self.tuning, L, Cr, Cb)
self.initialise_picamera()
@thing_property
def colour_correction_matrix(
self,
) -> tuple[float, float, float, float, float, float, float, float, float]:
"""An alias for `colour_correction_matrix` to fit the micromanager API"""
return self.thing_settings.get(
"colour_correction_matrix",
tuple(recalibrate_utils.get_static_ccm(self.tuning)[0]["ccm"]),
)
@colour_correction_matrix.setter # type: ignore
def colour_correction_matrix(self, value) -> None:
self.thing_settings["colour_correction_matrix"] = value
self.calibrate_colour_correction(value)
@thing_action
def reset_ccm(self):
"""
Overwrite the colour correction matrix in camera tuning with default values.
These values are from the Raspberry Pi Camera Algorithm and Tuning Guide, page
45.
"""
# This is flattened 3x3 matrix. See `calibrate_colour_correction`
col_corr_matrix = [
1.80439,
-0.73699,
-0.06739,
-0.36073,
1.83327,
-0.47255,
-0.08378,
-0.56403,
1.64781,
]
self.colour_correction_matrix = col_corr_matrix
@thing_action
def calibrate_colour_correction(
self,
col_corr_matrix: tuple[
float, float, float, float, float, float, float, float, float
],
) -> None:
"""Overwrite the colour correction matrix in camera tuning
col_corr_matrix: This is a 9 value tuple used to specify the 3x3
matrix that the GPU pipeline uses to convert from the camera R,G,B vector
to the standard R,G,B.
See page Raspberry Pi Camera Algorithm and Tuning Guide, page 45.
"""
with self.picamera(pause_stream=True):
recalibrate_utils.set_static_ccm(self.tuning, col_corr_matrix)
self.initialise_picamera()
@thing_action
def set_static_green_equalisation(self, offset: int = 65535) -> None:
"""Set the green equalisation to a static value.
Green equalisation avoids the debayering algorithm becoming confused
by the two green channels having different values, which is a problem
when the chief ray angle isn't what the sensor was designed for, and
that's the case in e.g. a microscope using camera module v2.
A value of 0 here does nothing, a value of 65535 is maximum correction.
"""
with self.picamera(pause_stream=True):
recalibrate_utils.set_static_geq(self.tuning, offset)
self.initialise_picamera()
@thing_action
def full_auto_calibrate(self) -> None:
"""Perform a full auto-calibration
This function will call the other calibration actions in sequence:
* `flat_lens_shading` to disable flat-field
* `auto_expose_from_minimum`
* `set_static_green_equalisation` to set geq offset to max
* `calibrate_lens_shading`
* `calibrate_white_balance`
"""
self.flat_lens_shading()
self.auto_expose_from_minimum()
self.set_static_green_equalisation()
self.calibrate_lens_shading()
self.calibrate_white_balance()
@thing_action
def flat_lens_shading(self) -> None:
"""Disable flat-field correction
This method will set a completely flat lens shading table. It is not the
same as the default behaviour, which is to use an adaptive lens shading
table.
This flat table is used to take an image wth no lens shading so that the
correct lens shading table can be calibrated.
"""
with self.picamera(pause_stream=True):
# Generate and array of ones of the correct size for each channel
flat_array = np.ones((12, 16))
recalibrate_utils.set_static_lst(
self.tuning, flat_array, flat_array, flat_array
)
self.initialise_picamera()
@thing_property
def lens_shading_tables(self) -> Optional[LensShading]:
"""The current lens shading (i.e. flat-field correction)
Return the current lens shading correction, as three 2D lists each with
dimensions 16x12, if a static lens shading table is in use.
Return None if:
- adaptive control is enabled
- multiple LSTs in use (for different colour temperatures),
"""
if not self.lens_shading_is_static:
return None
# Note "alsc" is the Picamera2 term for "Automatic Lens Shading Correction"
alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc")
# Check there is exactly 1 correction table for red-difference chroma (Cr)
# and blue-difference chroma (Cb)
if len(alsc["calibrations_Cr"]) != 1 or len(alsc["calibrations_Cb"]) != 1:
# If there is not exactly one table, then lens shading isn't static.
return None
def reshape_lst(lin: list[float]) -> list[list[float]]:
"""Reshape the 192 element list into a 2D 16x12 list"""
w, h = 16, 12
return [lin[w * i : w * (i + 1)] for i in range(h)]
return LensShading(
luminance=reshape_lst(alsc["luminance_lut"]),
Cr=reshape_lst(alsc["calibrations_Cr"][0]["table"]),
Cb=reshape_lst(alsc["calibrations_Cb"][0]["table"]),
)
@lens_shading_tables.setter
def lens_shading_tables(self, lst: LensShading) -> None:
"""Set the lens shading tables"""
with self.picamera(pause_stream=True):
recalibrate_utils.set_static_lst(
self.tuning,
luminance=lst.luminance,
cr=lst.Cr,
cb=lst.Cb,
)
self.initialise_picamera()
def correct_colour_gains_for_lens_shading(
self, colour_gains: tuple[float, float]
) -> tuple[float, float]:
"""Correct white balance gains for the effect of lens shading
The white balance algorithm we use assumes the brightest pixels
should be white, and that the only thing affecting the colour of
said pixels is the `colour_gains`.
The lens shading correction is normalised such that the *minimum*
gain in the `Cr` and `Cb` channels is 1. The white balance
assumption above requires that the gain for the brightest pixels
is 1. The solution might be that, when calibrating, we note which
pixels are brightest (usually the centre) and explicitly use
the LST values for there. However, for now I will assume that we
need to normalise by the **maximum** of the `Cr` and `Cb`
channels, which is correct the majority of the time.
"""
if not self.lens_shading_is_static:
return colour_gains
lst = self.lens_shading_tables
# The Cr and Cb corrections are normalised to have a minimum of 1,
# but the white balance algorithm normalises the brightest pixels
# to be white, assuming the brightest pixels have equal gain from
# the LST.
gain_r, gain_b = colour_gains
return (
float(gain_r / np.max(lst.Cr)),
float(gain_b / np.max(lst.Cb)),
)
@thing_action
def flat_lens_shading_chrominance(self) -> None:
"""Disable flat-field correction
This method will set the chrominance of the lens shading table to be
flat, i.e. we'll correct vignetting of intensity, but not any change in
colour across the image.
"""
with self.picamera(pause_stream=True):
alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc")
luminance = alsc["luminance_lut"]
flat = np.ones((12, 16))
recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat)
self.initialise_picamera()
@thing_action
def reset_lens_shading(self) -> None:
"""Revert to default lens shading settings
This method will restore the default "adaptive" lens shading method used
by the Raspberry Pi camera.
"""
with self.picamera(pause_stream=True):
recalibrate_utils.copy_alsc_section(self.default_tuning, self.tuning)
self.initialise_picamera()
@thing_property
def lens_shading_is_static(self) -> bool:
"""Whether the lens shading is static
This property is true if the lens shading correction has been set to use
a static table (i.e. the number of automatic correction iterations is zero).
The default LST is not static, but all the calibration controls will set it
to be static (except "reset")
"""
return recalibrate_utils.lst_is_static(self.tuning)

565
src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py Normal file
View file

@ -0,0 +1,565 @@
"""
Functions to set up a Raspberry Pi Camera v2 for scientific use
This module provides slower, simpler functions to set the
gain, exposure, and white balance of a Raspberry Pi camera, using
the `picamera2` Python library. It's mostly used by the OpenFlexure
Microscope, though it deliberately has no hard dependencies on
said software, so that it's useful on its own.
There are three main calibration steps:
* Setting exposure time and gain to get a reasonably bright
image.
* Fixing the white balance to get a neutral image
* Taking a uniform white image and using it to calibrate
the Lens Shading Table
The most reliable way to do this, avoiding any issues relating
to "memory" or nonlinearities in the camera's image processing
pipeline, is to use raw images. This is quite slow, but very
reliable. The three steps above can be accomplished by:
```
picamera = picamera2.Picamera2()
adjust_shutter_and_gain_from_raw(picamera)
adjust_white_balance_from_raw(picamera)
lst = lst_from_camera(picamera)
picamera.lens_shading_table = lst
```
"""
# Disable N806 & 803, which checks that all variables and args are lowercase.
# This is due to the number of matrix calculations and colour channel
# calculations that are clearer using the standard R, G, B, or L, Cr, Cb terms.
# ruff: noqa: N806 N803
from __future__ import annotations
import gc
import logging
import time
from typing import List, Literal, Optional, Tuple
from pydantic import BaseModel
import numpy as np
from scipy.ndimage import zoom
from picamera2 import Picamera2
import picamera2
LensShadingTables = tuple[np.ndarray, np.ndarray, np.ndarray]
def load_default_tuning(cam: Picamera2) -> dict:
"""Load the default tuning file for the camera
This will open and close the camera to determine its model. If you are
using a model that's supported by `picamera2` it should have a tuning
file built in. If not, this will probably crash with an error.
Error handling for unsupported cameras is not something we are likely
to test in the short term.
"""
cp = cam.camera_properties
fname = f"{cp['Model']}.json"
try:
return cam.load_tuning_file(fname)
except RuntimeError:
tuning_dir = "/usr/share/libcamera/ipa/raspberrypi"
# from picamera2 v0.3.9
# The directory above has been removed from the search path seems
# odd - as that's where the files currently are on a default
# Raspbian image. This may need updating if the files have moved
# in future updates to the system libcamera package
return cam.load_tuning_file(fname, dir=tuning_dir)
def set_minimum_exposure(camera: Picamera2) -> None:
"""Enable manual exposure, with low gain and shutter speed
We set exposure mode to manual, analog and digital gain
to 1, and shutter speed to the minimum (8us for Pi Camera v2)
Note ISO is left at auto, because this is needed for the gains
to be set correctly.
"""
# Disable Automatic exposure and gain algoritm (AeEnable), and set analogue
# gain and exposure time.
# Setting the shutter speed to 1us will result in it being set
# to the minimum possible, which is ~8us for PiCamera v2
camera.set_controls({"AeEnable": False, "AnalogueGain": 1, "ExposureTime": 1})
time.sleep(0.5)
class ExposureTest(BaseModel):
"""Record the results of testing the camera's current exposure settings"""
level: int
exposure_time: int
analog_gain: float
def test_exposure_settings(camera: Picamera2, percentile: float) -> ExposureTest:
"""Evaluate current exposure settings using a raw image
CAMERA SHOULD BE STARTED!
We will acquire a raw image and calculate the given percentile
of the pixel values. We return a dictionary containing the
percentile (which will be compared to the target), as well as
the camera's shutter and gain values.
"""
camera.capture_array("raw") # controls might not be updated for the first frame?
max_brightness = np.percentile(
channels_from_bayer_array(camera.capture_array("raw")),
percentile,
)
# The reported brightness can, theoretically, be negative or zero
# because of black level compensation. The line below forces a
# minimum value of 1 which will keep things well-behaved!
if max_brightness < 1:
logging.warning(
f"Measured brightness of {max_brightness}. "
"This should normally be >= 1, and may indicate the "
"camera's black level compensation has gone wrong."
)
max_brightness = 1
metadata = camera.capture_metadata()
result = ExposureTest(
level=max_brightness,
exposure_time=int(metadata["ExposureTime"]),
analog_gain=float(metadata["AnalogueGain"]),
)
logging.info(f"{result.model_dump()}")
return result
def check_convergence(test: ExposureTest, target: int, tolerance: float) -> bool:
"""Check whether the brightness is within the specified target range"""
return abs(test.level - target) < target * tolerance
def adjust_shutter_and_gain_from_raw(
camera: Picamera2,
target_white_level: int = 700,
max_iterations: int = 20,
tolerance: float = 0.05,
percentile: float = 99.9,
) -> float:
"""Adjust exposure and analog gain based on raw images.
This routine is slow but effective. It uses raw images, so we
are not affected by white balance or digital gain.
Arguments:
target_white_level:
The raw, 10-bit value we aim for. The brightest pixels
should be approximately this bright. Maximum possible
is about 900, 700 is reasonable.
max_iterations:
We will terminate once we perform this many iterations,
whether or not we converge. More than 10 shouldn't happen.
tolerance:
How close to the target value we consider "done". Expressed
as a fraction of the ``target_white_level`` so 0.05 means
+/- 5%
percentile:
Rather then use the maximum value for each channel, we
calculate a percentile. This makes us robust to single
pixels that are bright/noisy. 99.9% still picks the top
of the brightness range, but seems much more reliable
than just ``np.max()``.
"""
# TODO: read black level and bit depth from camera?
if target_white_level * (tolerance + 1) >= 959:
raise ValueError(
"The target level is too high - a saturated image would be "
"considered successful. target_white_level * (tolerance + 1) "
"must be less than 959."
)
config = camera.create_still_configuration(raw={"format": "SBGGR10"})
camera.configure(config)
camera.start()
set_minimum_exposure(camera)
# We start with very low exposure settings and work up
# until either the brightness is high enough, or we can't increase the
# shutter speed any more.
iterations = 0
while iterations < max_iterations:
test = test_exposure_settings(camera, percentile)
if check_convergence(test, target_white_level, tolerance):
break
iterations += 1
# Adjust shutter speed so that the brightness approximates the target
# NB we put a maximum of 8 on this, to stop it increasing too quickly.
new_time = int(test.exposure_time * min(target_white_level / test.level, 8))
camera.controls.ExposureTime = new_time
camera.controls.AeEnable = False
time.sleep(0.5)
# Check whether the shutter speed is still going up - if not, we've hit a maximum
if camera.capture_metadata()["ExposureTime"] == test.exposure_time:
logging.info(f"Shutter speed has maxed out at {test.exposure_time}")
break
# Now, if we've not converged, increase gain until we converge or run out of options.
while iterations < max_iterations:
test = test_exposure_settings(camera, percentile)
if check_convergence(test, target_white_level, tolerance):
break
iterations += 1
# Adjust gain to make the white level hit the target, again with a maximum
camera.controls.AnalogueGain = test.analog_gain * min(
target_white_level / test.level, 2
)
time.sleep(0.5)
# Check the gain is still changing - if not, we have probably hit the maximum
if camera.capture_metadata()["AnalogueGain"] == test.analog_gain:
logging.info(f"Gain has maxed out at {test.analog_gain}")
break
if check_convergence(test, target_white_level, tolerance):
logging.info(f"Brightness has converged to within {tolerance * 100:.0f}%.")
else:
logging.warning(
f"Failed to reach target brightness of {target_white_level}."
f"Brightness reached {test.level} after {iterations} iterations."
)
return test.level
def adjust_white_balance_from_raw(
camera: Picamera2,
percentile: float = 99,
luminance: Optional[np.ndarray] = None,
Cr: Optional[np.ndarray] = None,
Cb: Optional[np.ndarray] = None,
luminance_power: float = 1.0,
method: Literal["percentile", "centre"] = "centre",
) -> Tuple[float, float]:
"""Adjust the white balance in a single shot, based on the raw image.
NB if ``channels_from_raw_image`` is broken, this will go haywire.
We should probably have better logic to verify the channels really
are BGGR...
"""
config = camera.create_still_configuration(raw={"format": "SBGGR10"})
camera.configure(config)
camera.start()
channels = channels_from_bayer_array(camera.capture_array("raw"))
# TODO: read black level from camera rather than hard-coding 64
blacklevel = 64
if luminance is not None and Cr is not None and Cb is not None:
# Reconstruct a low-resolution image from the lens shading tables
# and use it to normalise the raw image, to compensate for
# the brightest pixels in each channel not coinciding.
grids = grids_from_lst(np.array(luminance) ** luminance_power, Cr, Cb)
channel_gains = 1 / grids
if channel_gains.shape[1:] != channels.shape[1:]:
channel_gains = upsample_channels(channel_gains, channels.shape[1:])
logging.info(
f"Before gains, channel maxima are {np.max(channels, axis=(1, 2))}"
)
channels = channels * channel_gains
logging.info(f"After gains, channel maxima are {np.max(channels, axis=(1, 2))}")
if method == "centre":
_, height, width = channels.shape
# Cut out the central 10% from 9/20 to 11/20...
low_y_range = 9 * height // 20
hi_y_range = 11 * height // 20
low_x_range = 9 * width // 20
hi_x_range = 11 * width // 20
# ... and then take the mean of each bayer channel.
centre_means = np.mean(
channels[:, low_y_range:hi_y_range, low_x_range:hi_x_range],
axis=(1, 2),
)
# Subtract blacklevel before splitting into channels
blue, g1, g2, red = centre_means - blacklevel
else:
blue, g1, g2, red = (
np.percentile(channels, percentile, axis=(1, 2)) - blacklevel
)
green = (g1 + g2) / 2.0
new_awb_gains = (green / red, green / blue)
if Cr is not None and Cb is not None:
# The LST algorithm normalises Cr and Cb by their minimum.
# The lens shading correction only ever boosts the red and blue values.
# Here, we decrease the gains by the minimum value of Cr and Cb.
new_awb_gains = (green / red * np.min(Cr), green / blue * np.min(Cb))
logging.info(
f"Raw white point is R: {red} G: {green} B: {blue}, "
f"setting AWB gains to ({new_awb_gains[0]:.2f}, "
f"{new_awb_gains[1]:.2f})."
)
camera.controls.AwbEnable = False
camera.controls.ColourGains = new_awb_gains
time.sleep(0.2)
m = camera.capture_metadata()
print(f"Camera confirms gains are now {m['ColourGains']}")
return new_awb_gains
def channels_from_bayer_array(bayer_array: np.ndarray) -> np.ndarray:
"""Given the 'array' from a PiBayerArray, return the 4 channels."""
bayer_pattern: List[Tuple[int, int]] = [(0, 0), (0, 1), (1, 0), (1, 1)]
bayer_array = bayer_array.view(np.uint16)
channels_shape: Tuple[int, int, int] = (
4,
bayer_array.shape[0] // 2,
bayer_array.shape[1] // 2,
)
channels: np.ndarray = np.zeros(channels_shape, dtype=bayer_array.dtype)
for i, offset in enumerate(bayer_pattern):
# We simplify life by dealing with only one channel at a time.
channels[i, :, :] = bayer_array[offset[0] :: 2, offset[1] :: 2]
return channels
def get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray:
"""Compresses channel down to a 16x12 grid - from libcamera
This is taken from
https://git.linuxtv.org/libcamera.git/tree/utils/raspberrypi/ctt/ctt_alsc.py
for consistency.
"""
grid = []
# since left and bottom border will not necessarily have rectangles of
# dimension dx x dy, the final iteration has to be handled separately.
for i in range(11):
for j in range(15):
grid.append(np.mean(chan[dy * i : dy * (1 + i), dx * j : dx * (1 + j)]))
grid.append(np.mean(chan[dy * i : dy * (1 + i), 15 * dx :]))
for j in range(15):
grid.append(np.mean(chan[11 * dy :, dx * j : dx * (1 + j)]))
grid.append(np.mean(chan[11 * dy :, 15 * dx :]))
# return as np.array, ready for further manipulation
return np.reshape(np.array(grid), (12, 16))
def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray:
"""Zoom an image in the last two dimensions
This is effectively the inverse operation of `get_16x12_grid`
"""
zoom_factors = [
1,
] + list(np.ceil(np.array(shape) / np.array(grids.shape[1:])))
return zoom(grids, zoom_factors, order=1)[:, : shape[0], : shape[1]]
def downsampled_channels(channels: np.ndarray, blacklevel=64) -> list[np.ndarray]:
"""Generate a downsampled, un-normalised image from which to calculate the LST
TODO: blacklevel probably ought to be determined from the camera...
"""
channel_shape = np.array(channels.shape[1:])
lst_shape = np.array([12, 16])
step = np.ceil(channel_shape / lst_shape).astype(int)
return np.stack(
[
get_16x12_grid(
channels[i, ...].astype(float) - blacklevel, step[1], step[0]
)
for i in range(channels.shape[0])
],
axis=0,
)
def lst_from_channels(channels: np.ndarray) -> LensShadingTables:
"""Given the 4 Bayer colour channels from a white image, generate a LST.
Internally, is just calls `downsampled_channels` and `lst_from_grids`.
"""
grids = downsampled_channels(channels)
return lst_from_grids(grids)
def lst_from_grids(grids: np.ndarray) -> LensShadingTables:
"""Given 4 downsampled grids, generate the luminance and chrominance tables
The grids are the 4 BAYER channels RGGB
The LST format has changed with `picamera2` and now uses a fixed resolution,
and is in luminance, Cr, Cb format. This function returns three ndarrays of
luminance, Cr, Cb, each with shape (12, 16).
"""
# Calculated red, green, and blue channels from Bayer data
r: np.ndarray = grids[3, ...]
g: np.ndarray = np.mean(grids[1:3, ...], axis=0)
b: np.ndarray = grids[0, ...]
# What we actually want to calculate is the gains needed to compensate for the
# lens shading - that's 1/lens_shading_table_float as we currently have it.
# Minimum luminance gain is 1
luminance_gains: np.ndarray = np.max(g) / g
cr_gains: np.ndarray = g / r
cb_gains: np.ndarray = g / b
return luminance_gains, cr_gains, cb_gains
def grids_from_lst(lum: np.ndarray, Cr: np.ndarray, Cb: np.ndarray) -> np.ndarray:
"""Convert form luminance/chrominance dict to four RGGB channels
Note that these will be normalised - the maximum green value is always 1.
Also, note that the channels are BGGR, to be consistent with the
`channels_from_raw_image` function. This should probably change in the
future.
"""
G = 1 / np.array(lum)
R = G / np.array(Cr)
B = G / np.array(Cb)
return np.stack([B, G, G, R], axis=0)
def set_static_lst(
tuning: dict,
luminance: np.ndarray,
cr: np.ndarray,
cb: np.ndarray,
) -> None:
"""Update the `rpi.alsc` section of a camera tuning dict to use a static correcton.
`tuning` will be updated in-place to set its shading to static, and disable any
adaptive tweaking by the algorithm.
"""
for table in luminance, cr, cb:
assert np.array(table).shape == (12, 16), "Lens shading tables must be 12x16!"
alsc = Picamera2.find_tuning_algo(tuning, "rpi.alsc")
alsc["n_iter"] = 0 # disable the adaptive part
alsc["luminance_strength"] = 1.0
alsc["calibrations_Cr"] = [
{"ct": 4500, "table": as_flat_rounded_list(cr, round_to=3)}
]
alsc["calibrations_Cb"] = [
{"ct": 4500, "table": as_flat_rounded_list(cb, round_to=3)}
]
alsc["luminance_lut"] = as_flat_rounded_list(luminance, round_to=3)
def set_static_ccm(
tuning: dict,
col_corr_matrix: tuple[
float, float, float, float, float, float, float, float, float
],
) -> None:
"""Update the `rpi.alsc` section of a camera tuning dict to use a static correcton.
`tuning` will be updated in-place to set its shading to static, and disable any
adaptive tweaking by the algorithm.
"""
ccm = Picamera2.find_tuning_algo(tuning, "rpi.ccm")
ccm["ccms"] = [{"ct": 2860, "ccm": col_corr_matrix}]
def get_static_ccm(tuning: dict) -> None:
"""Get the `rpi.ccm` section of a camera tuning dict"""
ccm = Picamera2.find_tuning_algo(tuning, "rpi.ccm")
return ccm["ccms"]
def lst_is_static(tuning: dict) -> bool:
"""Whether the lens shading table is set to static"""
alsc = Picamera2.find_tuning_algo(tuning, "rpi.alsc")
return alsc["n_iter"] == 0
def set_static_geq(
tuning: dict,
offset: int = 65535,
) -> None:
"""Update the `rpi.geq` section of a camera tuning dict to always use green
equalisation that averages the green pixels in the red and blue rows.
`tuning` will be updated in-place to set the geq offest to the given value.
The default 65535 is the maximum allowed value. This means
the brightness will always be below the threshold where averaging is used.
"""
geq = Picamera2.find_tuning_algo(tuning, "rpi.geq")
geq["offset"] = offset # max out offset to disable the adaptive green equalisation
def _geq_is_static(tuning: dict) -> bool:
"""Whether the green equalisation is set to static"""
geq = Picamera2.find_tuning_algo(tuning, "rpi.geq")
return geq["offset"] == 65535
def index_of_algorithm(algorithms: list[dict], algorithm: str) -> int:
"""Find the index of an algorithm's section in the tuning file"""
for i, a in enumerate(algorithms):
if algorithm in a:
return i
raise ValueError(f"Algorithm {algorithm} is not available.")
def copy_alsc_section(from_tuning: dict, to_tuning: dict) -> None:
"""Copy the `rpi.alsc` algorithm from one tuning to another.
This is done in-place, i.e. modifying to_tuning.
"""
# Using Picamera2 function to find the relevant sub-dict for each tuning file
from_i = index_of_algorithm(from_tuning["algorithms"], "rpi.alsc")
to_i = index_of_algorithm(to_tuning["algorithms"], "rpi.alsc")
# Updating the dictionary in place.
to_tuning["algorithms"][to_i] = from_tuning["algorithms"][from_i]
def lst_from_camera(camera: Picamera2) -> LensShadingTables:
"""Acquire a raw image and use it to calculate a lens shading table."""
channels = raw_channels_from_camera(camera)
return lst_from_channels(channels)
def raw_channels_from_camera(camera: Picamera2) -> LensShadingTables:
"""Acquire a raw image and return a 4xNxM array of the colour channels."""
if camera.started:
camera.stop_recording()
# We will acquire a raw image with unpacked pixels, which is what the
# format below requests. Bit depth and Bayer order may be overwritten.
# TODO: don't assume 10-bit - the high quality camera uses 12.
# TODO: what's the best mode to use here?
config = camera.create_still_configuration(raw={"format": "SBGGR10"})
camera.configure(config)
camera.start()
raw_image = camera.capture_array("raw")
camera.stop()
# Now we need to calculate a lens shading table that would make this flat.
# raw_image is a 3D array, with full resolution and 3 colour channels. No
# de-mosaicing has been done, so 2/3 of the values are zero (3/4 for R and B
# channels, 1/2 for green because there's twice as many green pixels).
raw_format = camera.camera_configuration()["raw"]["format"]
print(f"Acquired a raw image in format {raw_format}")
return channels_from_bayer_array(raw_image)
def recreate_camera_manager() -> None:
"""Delete and recreate the camera manager.
This is necessary to ensure the tuning file is re-read.
"""
del Picamera2._cm
gc.collect()
Picamera2._cm = picamera2.picamera2.CameraManager()
def as_flat_rounded_list(array: np.ndarray, round_to: int = 3) -> list[float]:
"""Flatten array, round, and then convert to list"""
return np.reshape(array, -1).round(round_to).tolist()

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

@ -22,12 +22,9 @@ from scipy.ndimage import gaussian_filter
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
from labthings_fastapi.types.numpy import NDArray
from labthings_fastapi.server import ThingServer
from pydantic import RootModel
from . import BaseCamera, JPEGBlob
from . import BaseCamera, JPEGBlob, ArrayModel
from ..stage import StageProtocol as Stage
# The ratio between "motor" steps and pixels
@ -35,12 +32,6 @@ from ..stage import StageProtocol as Stage
RATIO = 0.2
class ArrayModel(RootModel):
"""A model for an array"""
root: NDArray
class SimulatedCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
@ -160,9 +151,6 @@ class SimulatedCamera(BaseCamera):
return self._capture_thread.is_alive()
return False
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
def _capture_frames(self):
portal = get_blocking_portal(self)
while self._capture_enabled:

66
src/openflexure_microscope_server/things/capture.py
View file

@ -1,66 +0,0 @@
from PIL import Image
import piexif
import json
import numpy as np
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.raw_thing import raw_thing_dependency
from .camera import CameraDependency as CamDep
from labthings_fastapi.dependencies.invocation import (
InvocationLogger,
)
class CaptureError(RuntimeError):
"""An error trying to capture from Picamera"""
class CaptureThing(Thing):
"""A temporary Thing to handle capturing to disk with associated metadata
Will be moved to the camera Thing or dependency in due course"""
def _capture_image(
self, cam: CamDep, metadata_getter: GetThingStates
) -> tuple[np.ndarray, dict]:
"""Capture an image in memory and return it with metadata
CaptureError raised if the capture fails for any reason
returns tuple with numpy array of image data, and dict of metadata
"""
try:
metadata = metadata_getter()
image = cam.capture_array()[..., :3]
except Exception as e:
raise CaptureError("An error occurred while capturing") from e
return image, metadata
def _save_capture(
self,
jpeg_path: str,
image: Image.Image,
metadata: dict,
logger: InvocationLogger,
) -> None:
"""Saving the captured image and metadata to disk
logger warning (via InvocationLogger) is raised if metadata is failed to be added
IOError is raised if the file cannot be saved
nothing is returned on success"""
try:
Image.fromarray(image.astype("uint8"), "RGB").save(
jpeg_path, quality=95, subsampling=0
)
try:
exif_dict = piexif.load(jpeg_path)
exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps(
metadata
).encode("utf-8")
piexif.insert(piexif.dump(exif_dict), jpeg_path)
except: # noqa: E722
# 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}")
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e
RawCaptureDependency = raw_thing_dependency(CaptureThing)

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

@ -59,11 +59,11 @@ def ensure_free_disk_space(path: str, min_space: int = 500000000) -> None:
Raises:
NotEnoughFreeSpaceError if the remaining storage is below min_space
"""
du = shutil.disk_usage(path)
if du.free < min_space:
disk_usage = shutil.disk_usage(path)
if disk_usage.free < min_space:
raise NotEnoughFreeSpaceError(
"There is not enough free disk space to continue. "
f"(Required: {min_space}, {du})."
f"(Required: {min_space >> 20}MB, free: {disk_usage.free >> 20}MB)."
)
@ -90,6 +90,7 @@ SCAN_DATA_FILENAME = "scan_data.json"
STITCHING_CMD = "openflexure-stitch"
SCAN_ZERO_PAD_DIGITS = 4
STITCHING_RESOLUTION = (820, 616)
def _scan_running(method):
@ -374,28 +375,28 @@ class SmartScanThing(Thing):
return (next_point[0], next_point[1], z_estimate)
@_scan_running
def _take_test_image_to_calc_displacement(self, overlap):
def _calc_displacement_from_test_image(self, overlap: int) -> tuple[int, int]:
"""
Take a test image and use camera stage mapping to calculate x and y displacement
:param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means
that each image should overlap its nearest neighbour by 50%.
Return (dx, dy) - the x and y displacments in steps
"""
test_jpg = self._cam.grab_jpeg()
test_image = np.array(Image.open(test_jpg.open()))
test_image_res = list(test_image.shape[:2])
test_image_res = list(test_image.shape)
csm_image_res = [int(i) for i in self._csm.image_resolution]
if test_image_res != csm_image_res:
raise RuntimeError(
"Cannot start scan as it is set up to capture with a resolution that "
"has not been mapped.\n"
f"Scan resolution: {test_image_res}\n"
f"camera-stage-mapping resolution {csm_image_res}."
)
# If current stream width is different to csm calibration width,
# perform the conversion here
res_ratio = csm_image_res[0] / test_image_res[0]
# get displacement matrix. note it is for (y, x) not (x, y) coordinates
csm_disp_matrix = self._csm.image_to_stage_displacement_matrix
csm_disp_matrix = np.array(self._csm.image_to_stage_displacement_matrix)
csm_disp_matrix *= res_ratio
# Calculate displacements in image coordinates
dx_img = test_image.shape[1] * (1 - overlap)
@ -419,7 +420,13 @@ class SmartScanThing(Thing):
dataclass.
"""
overlap = self.overlap
dx, dy = self._take_test_image_to_calc_displacement(overlap)
dx, dy = self._calc_displacement_from_test_image(overlap)
stitch_resize = STITCHING_RESOLUTION[0] / self.capture_resolution[0]
self._scan_logger.debug(
f"Resizing images when stitching by a factor of {stitch_resize}"
)
self._scan_logger.info(
f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}"
)
@ -446,6 +453,8 @@ class SmartScanThing(Thing):
"start_time": time.strftime("%H_%M_%S-%d_%m_%Y"),
"skip_background": self.skip_background,
"stitch_automatically": self.stitch_automatically,
"stitch_resize": stitch_resize,
"capture_resolution": self.capture_resolution,
}
@_scan_running
@ -464,6 +473,7 @@ class SmartScanThing(Thing):
"dy": self._scan_data["dy"],
"start time": self._scan_data["start_time"],
"skipping background": self._scan_data["skip_background"],
"capture resolution": self._scan_data["capture_resolution"],
}
scan_inputs_fname = os.path.join(
@ -533,6 +543,7 @@ class SmartScanThing(Thing):
scan_successful = True
try:
self._cam.start_streaming(main_resolution=(3280, 2464))
self._set_scan_data()
self._save_scan_inputs_json()
if self._scan_images_taken != 0:
@ -564,6 +575,8 @@ class SmartScanThing(Thing):
)
raise e
finally:
# Start streaming in the default resolution again as soon as possible
self._cam.start_streaming()
if self._capture_thread:
# If the capture thread had an error, we capture it here
try:
@ -652,7 +665,19 @@ class SmartScanThing(Thing):
)
route_planner.mark_location_visited(
current_pos_xyz, imaged=True, focused=True
current_pos_xyz, imaged=True, focused=focused
)
site_folder = os.path.join(
self._ongoing_scan_images_dir,
"stacks",
f"{new_pos_xyz[0]}_{new_pos_xyz[1]}",
)
os.makedirs(site_folder, exist_ok=True)
self._autofocus.run_z_stack(
images_dir=self._ongoing_scan_images_dir,
stack_dir=site_folder,
capture_resolution=self._scan_data["capture_resolution"],
)
# increment capure counter as thread has completed
@ -692,6 +717,7 @@ class SmartScanThing(Thing):
self.stitch_scan(
logger=self._scan_logger,
scan_name=self._ongoing_scan_name,
stitch_resize=self._scan_data["stitch_resize"],
overlap=self._scan_data["overlap"],
)
self.promote_stitch_files(self._scan_logger)
@ -721,6 +747,16 @@ class SmartScanThing(Thing):
raise HTTPException(404, "File not found")
return FileResponse(path)
@thing_property
def capture_resolution(self) -> tuple[int, int]:
"""A tuple of the image resolution to capture. Should be in a
4:3 aspect ratio"""
return self.thing_settings.get("capture_resolution", ((1640, 1232)))
@capture_resolution.setter
def capture_resolution(self, value: tuple[int, int]) -> None:
self.thing_settings["capture_resolution"] = value
@thing_property
def max_range(self) -> int:
"""The maximum distance from the centre of the scan before we break in steps"""
@ -973,6 +1009,8 @@ class SmartScanThing(Thing):
"preview_stitch",
"--minimum_overlap",
f"{min_overlap}",
"--resize",
f"{self._scan_data['stitch_resize']}",
self._ongoing_scan_images_dir,
]
)
@ -1043,6 +1081,7 @@ class SmartScanThing(Thing):
self,
logger: InvocationLogger,
scan_name: str,
stitch_resize: Optional[float] = None,
overlap: float = 0.0,
) -> None:
"""Generate a stitched image based on stage position metadata
@ -1062,7 +1101,6 @@ class SmartScanThing(Thing):
try:
with open(json_fpath, "r", encoding="utf-8") as data_file:
data_loaded = json.load(data_file)
logger.info(data_loaded)
overlap = data_loaded["overlap"]
except (json.decoder.JSONDecodeError, FileNotFoundError, TypeError):
# As there is no schema or pydantic model this should handle
@ -1079,6 +1117,29 @@ class SmartScanThing(Thing):
"Attempting stitch with overlap value of 0.1"
)
overlap = 0.1
if stitch_resize is None:
try:
with open(json_fpath, "r", encoding="utf-8") as data_file:
data_loaded = json.load(data_file)
capture_resolution = data_loaded["capture resolution"]
stitch_resize = STITCHING_RESOLUTION[0] / capture_resolution[0]
except (json.decoder.JSONDecodeError, FileNotFoundError, TypeError):
# As there is no schema or pydantic model this should handle
# the file not being there, it not being json in the file,
# or the imported data not being indexable
logger.warning(
f"Couldn't read scan data, is {SCAN_DATA_FILENAME} missing or corrupt? "
"Attempting stitch with resize value of 0.5"
)
stitch_resize = 0.5
except KeyError:
logger.warning(
"Value for capture resolution not found in scan data. "
"Attempting stitch with resize value of 0.5"
)
stitch_resize = 0.5
self.run_subprocess(
logger,
[
@ -1089,6 +1150,8 @@ class SmartScanThing(Thing):
"--stitch_dzi",
"--minimum_overlap",
f"{round(overlap * 0.9, 2)}",
"--resize",
f"{stitch_resize}",
images_folder,
],
)