Merge branch 'tuning-tests-and-fixes' into 'v3'

Fix picamera default tuning being updated by calibrations, add tests

See merge request openflexure/openflexure-microscope-server!373
This commit is contained in:
Julian Stirling 2025-08-21 12:23:47 +00:00
commit 46be2c770e
4 changed files with 158 additions and 17 deletions

View file

@ -5,14 +5,13 @@ from PIL import Image
import numpy as np import numpy as np
from pytest import fixture from pytest import fixture
from labthings_fastapi.server import ThingServer import labthings_fastapi as lt
from labthings_fastapi.client import ThingClient
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
@fixture(scope="module") @fixture(scope="module")
def client(): def client() -> lt.ThingClient:
"""Initialise a test client for the StreamingPiCamera2 Thing. """Initialise a test client for the StreamingPiCamera2 Thing.
This fixture: This fixture:
@ -21,18 +20,13 @@ def client():
* Registers a StreamingPiCamera2 instance at the "/camera/" endpoint * Registers a StreamingPiCamera2 instance at the "/camera/" endpoint
* Provides a ThingClient for interacting with it during tests. * Provides a ThingClient for interacting with it during tests.
""" """
server = ThingServer() server = lt.ThingServer()
server.add_thing(StreamingPiCamera2(), "/camera/") server.add_thing(StreamingPiCamera2(), "/camera/")
with TestClient(server.app) as test_client: with TestClient(server.app) as test_client:
client = ThingClient.from_url("/camera/", client=test_client) client = lt.ThingClient.from_url("/camera/", client=test_client)
yield client yield client
def test_calibration(client):
"""Check that full auto calibrate completes without an exception."""
client.full_auto_calibrate()
def test_jpeg_and_array(client): def test_jpeg_and_array(client):
"""Check that a jpeg grabbed from the stream is the same size as other captures. """Check that a jpeg grabbed from the stream is the same size as other captures.
@ -41,12 +35,15 @@ def test_jpeg_and_array(client):
# Grab a jpeg from the stream # Grab a jpeg from the stream
blob = client.grab_jpeg() blob = client.grab_jpeg()
mjpeg_frame = Image.open(blob.open()) mjpeg_frame = Image.open(blob.open())
assert mjpeg_frame # Verify throws an error if there are issues with the image
mjpeg_frame.verify()
assert mjpeg_frame.format == "JPEG"
# Capture a jpeg # Capture a jpeg
blob = client.capture_jpeg(resolution="main") blob = client.capture_jpeg(resolution="main")
jpeg_capture = Image.open(blob.open()) jpeg_capture = Image.open(blob.open())
assert jpeg_capture jpeg_capture.verify()
assert jpeg_capture.format == "JPEG"
# Capture an array # Capture an array
arrlist = client.capture_array(stream_name="main") arrlist = client.capture_array(stream_name="main")

View file

@ -0,0 +1,102 @@
"""Test data collection from the Raspberry Picamera."""
import tempfile
from copy import deepcopy
from fastapi.testclient import TestClient
import pytest
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera.picamera import (
StreamingPiCamera2,
MissingCalibrationError,
)
@pytest.fixture()
def picamera_thing() -> StreamingPiCamera2:
"""Return a StreamingPiCamera2 Thing.
This is the Thing that the fixture client uses. It can be used to probe the
Thing directly to check actions had the expected response.
"""
return StreamingPiCamera2()
@pytest.fixture()
def client(picamera_thing) -> lt.ThingClient:
"""Initialise a test client for the StreamingPiCamera2 Thing.
This fixture:
* Sets up a ThingServer,
* Registers a StreamingPiCamera2 instance at the "/camera/" endpoint
* Provides a ThingClient for interacting with it during tests.
"""
temp_folder = tempfile.TemporaryDirectory()
server = lt.ThingServer(settings_folder=temp_folder.name)
server.add_thing(picamera_thing, "/camera/")
with TestClient(server.app) as test_client:
client = lt.ThingClient.from_url("/camera/", client=test_client)
yield client
def test_get_tuning_algo(picamera_thing):
"""Test that get_tuning algorithm retrieves tuning algorithms."""
# Missing algorithm raises an error.
with pytest.raises(MissingCalibrationError):
picamera_thing.get_tuning_algo("foo")
# Or returns None if explicitly told not to error.
assert picamera_thing.get_tuning_algo("foo", raise_if_missing=False) is None
# Real algorithm is a dict and contains expected keys
assert isinstance(picamera_thing.get_tuning_algo("rpi.geq"), dict)
assert "offset" in picamera_thing.get_tuning_algo("rpi.geq")
# Set the tuning to None as it is technically optional. And check this
# is handled gracefully. This needs to be done in a try block as LabThings
# tries and fails to emit an event.
try:
picamera_thing.tuning = None
except lt.exceptions.NotConnectedToServerError:
# Labthings will complain that it is not connected to a server
pass
# Check it is set to None despite the above LabThings issue.
assert picamera_thing.tuning is None
# Raises an error as there is no tuning file set.
with pytest.raises(MissingCalibrationError):
picamera_thing.get_tuning_algo("rpi.geq")
# Or returns None if explicitly told not to error.
assert picamera_thing.get_tuning_algo("rpi.geq", raise_if_missing=False) is None
def test_calibration(picamera_thing, client):
"""Check that full auto calibrate completes and set the expected values."""
# Save copy of default tuning file for end of test
original_default = deepcopy(picamera_thing.default_tuning)
# Tuning should start the same as the server is loading with no settings.
assert picamera_thing.default_tuning == picamera_thing.tuning
# lens_shading tables should be None when loaded on default tuning as it is
# adaptive
assert picamera_thing.lens_shading_tables is None
# The background detector isn't ready as there is no background image.
assert not picamera_thing.background_detector_status.ready
# Run full auto calibrate
client.full_auto_calibrate()
# After calibration the tuning files should be different
assert picamera_thing.default_tuning != picamera_thing.tuning
# The default should be unchanged
assert picamera_thing.default_tuning == original_default
# Lens shading tables should no longer be None
assert picamera_thing.lens_shading_tables is not None
# There should be exactly one colour correction matrix
# (no variations by colour temp)
assert len(picamera_thing.get_tuning_algo("rpi.ccm")["ccms"]) == 1
# Green equalisation offset should be set to maximum.
assert picamera_thing.get_tuning_algo("rpi.geq")["offset"] == 65535
assert picamera_thing.background_detector_status.ready

Binary file not shown.

View file

@ -15,11 +15,12 @@ https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf
""" """
from __future__ import annotations from __future__ import annotations
from typing import Annotated, Iterator, Literal, Mapping, Optional from typing import Annotated, Iterator, Literal, Mapping, Optional, overload
from datetime import datetime from datetime import datetime
import json import json
import logging import logging
import os import os
import copy
import tempfile import tempfile
import time import time
from contextlib import contextmanager from contextlib import contextmanager
@ -46,6 +47,10 @@ from . import picamera_recalibrate_utils as recalibrate_utils
from . import BaseCamera, JPEGBlob, ArrayModel from . import BaseCamera, JPEGBlob, ArrayModel
class MissingCalibrationError(RuntimeError):
"""Picamera tuning file is missing or doesn't contain the requested algorithm."""
class PicameraStreamOutput(Output): class PicameraStreamOutput(Output):
"""An Output class that sends frames to a stream.""" """An Output class that sends frames to a stream."""
@ -138,7 +143,7 @@ class StreamingPiCamera2(BaseCamera):
# Set tuning to default tuning. This will be overwritten when the Thing is # Set tuning to default tuning. This will be overwritten when the Thing is
# connects to the server if tuning is saved to disk. # connects to the server if tuning is saved to disk.
try: try:
self.tuning = self.default_tuning self.tuning = copy.deepcopy(self.default_tuning)
except NotConnectedToServerError: except NotConnectedToServerError:
# This will throw an error after setting as we are not connected to # This will throw an error after setting as we are not connected to
# a server. But we know this, so we ignore the error. # a server. But we know this, so we ignore the error.
@ -307,6 +312,42 @@ class StreamingPiCamera2(BaseCamera):
tuning = lt.ThingSetting(Optional[dict], None, readonly=True) tuning = lt.ThingSetting(Optional[dict], None, readonly=True)
"""The Raspberry PiCamera Tuning File JSON.""" """The Raspberry PiCamera Tuning File JSON."""
# Use overload to clarify that only a dictionary is returned if `raise_if_missing`
# is True
@overload
def get_tuning_algo(
self, algorithm_name: str, raise_if_missing: Literal[True]
) -> dict: ...
# Otherwise may also be None
@overload
def get_tuning_algo(
self, algorithm_name: str, raise_if_missing: bool
) -> Optional[dict]: ...
def get_tuning_algo(
self, algorithm_name: str, raise_if_missing: bool = True
) -> Optional[dict]:
"""Return the active tuning algorithm settings for the given algorithm.
:returns: The algorithm dictionary if found, returns None if no tuning data
is loaded or if the tuning algorithm is not found.
:raises MissingCalibrationError: If raise_if_missing is true and there is no
tuning file is available, or the requested algorithm is not present.
"""
if self.tuning is None:
if raise_if_missing:
raise MissingCalibrationError("No tuning data is set.")
return None
try:
return Picamera2.find_tuning_algo(self.tuning, algorithm_name)
except StopIteration as e:
if raise_if_missing:
raise MissingCalibrationError(
f"No tuning algorithm with name {algorithm_name}."
) from e
return None
def _initialise_picamera(self): def _initialise_picamera(self):
"""Acquire the picamera device and store it as ``self._picamera``. """Acquire the picamera device and store it as ``self._picamera``.
@ -753,6 +794,7 @@ class StreamingPiCamera2(BaseCamera):
* ``auto_expose_from_minimum`` * ``auto_expose_from_minimum``
* ``set_static_green_equalisation`` to set geq offset to max * ``set_static_green_equalisation`` to set geq offset to max
* ``calibrate_lens_shading`` * ``calibrate_lens_shading``
* ``reset_ccm``
* ``calibrate_white_balance`` * ``calibrate_white_balance``
* ``set_background`` * ``set_background``
""" """
@ -760,8 +802,8 @@ class StreamingPiCamera2(BaseCamera):
self.auto_expose_from_minimum() self.auto_expose_from_minimum()
self.set_static_green_equalisation() self.set_static_green_equalisation()
self.calibrate_lens_shading() self.calibrate_lens_shading()
self.calibrate_white_balance()
self.reset_ccm() self.reset_ccm()
self.calibrate_white_balance()
self.set_background(portal) self.set_background(portal)
@lt.thing_action @lt.thing_action
@ -880,7 +922,7 @@ class StreamingPiCamera2(BaseCamera):
return None return None
# Note "alsc" is the Picamera2 term for "Automatic Lens Shading Correction" # Note "alsc" is the Picamera2 term for "Automatic Lens Shading Correction"
alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc") alsc = self.get_tuning_algo("rpi.alsc")
# Check there is exactly 1 correction table for red-difference chroma (Cr) # Check there is exactly 1 correction table for red-difference chroma (Cr)
# and blue-difference chroma (Cb) # and blue-difference chroma (Cb)
@ -951,7 +993,7 @@ class StreamingPiCamera2(BaseCamera):
colour across the image. colour across the image.
""" """
with self._streaming_picamera(pause_stream=True): with self._streaming_picamera(pause_stream=True):
alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc") alsc = self.get_tuning_algo("rpi.alsc")
luminance = alsc["luminance_lut"] luminance = alsc["luminance_lut"]
flat = np.ones((12, 16)) flat = np.ones((12, 16))
recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat) recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat)