diff --git a/hardware-specific-tests/picamera2/test_acquisition.py b/hardware-specific-tests/picamera2/test_acquisition.py index b8e2fe88..62b41058 100644 --- a/hardware-specific-tests/picamera2/test_acquisition.py +++ b/hardware-specific-tests/picamera2/test_acquisition.py @@ -5,14 +5,13 @@ from PIL import Image import numpy as np from pytest import fixture -from labthings_fastapi.server import ThingServer -from labthings_fastapi.client import ThingClient +import labthings_fastapi as lt from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 @fixture(scope="module") -def client(): +def client() -> lt.ThingClient: """Initialise a test client for the StreamingPiCamera2 Thing. This fixture: @@ -21,18 +20,13 @@ def client(): * Registers a StreamingPiCamera2 instance at the "/camera/" endpoint * Provides a ThingClient for interacting with it during tests. """ - server = ThingServer() + server = lt.ThingServer() server.add_thing(StreamingPiCamera2(), "/camera/") 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 -def test_calibration(client): - """Check that full auto calibrate completes without an exception.""" - client.full_auto_calibrate() - - def test_jpeg_and_array(client): """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 blob = client.grab_jpeg() 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 blob = client.capture_jpeg(resolution="main") jpeg_capture = Image.open(blob.open()) - assert jpeg_capture + jpeg_capture.verify() + assert jpeg_capture.format == "JPEG" # Capture an array arrlist = client.capture_array(stream_name="main") diff --git a/hardware-specific-tests/picamera2/test_calibration.py b/hardware-specific-tests/picamera2/test_calibration.py new file mode 100644 index 00000000..7c56a9c8 --- /dev/null +++ b/hardware-specific-tests/picamera2/test_calibration.py @@ -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 diff --git a/picamera_coverage.zip b/picamera_coverage.zip index 784d83d9..2090ce5f 100644 Binary files a/picamera_coverage.zip and b/picamera_coverage.zip differ diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py index 130537de..f3282c61 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -15,11 +15,12 @@ https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf """ 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 import json import logging import os +import copy import tempfile import time from contextlib import contextmanager @@ -46,6 +47,10 @@ from . import picamera_recalibrate_utils as recalibrate_utils from . import BaseCamera, JPEGBlob, ArrayModel +class MissingCalibrationError(RuntimeError): + """Picamera tuning file is missing or doesn't contain the requested algorithm.""" + + class PicameraStreamOutput(Output): """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 # connects to the server if tuning is saved to disk. try: - self.tuning = self.default_tuning + self.tuning = copy.deepcopy(self.default_tuning) except NotConnectedToServerError: # This will throw an error after setting as we are not connected to # 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) """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): """Acquire the picamera device and store it as ``self._picamera``. @@ -753,6 +794,7 @@ class StreamingPiCamera2(BaseCamera): * ``auto_expose_from_minimum`` * ``set_static_green_equalisation`` to set geq offset to max * ``calibrate_lens_shading`` + * ``reset_ccm`` * ``calibrate_white_balance`` * ``set_background`` """ @@ -760,8 +802,8 @@ class StreamingPiCamera2(BaseCamera): self.auto_expose_from_minimum() self.set_static_green_equalisation() self.calibrate_lens_shading() - self.calibrate_white_balance() self.reset_ccm() + self.calibrate_white_balance() self.set_background(portal) @lt.thing_action @@ -880,7 +922,7 @@ class StreamingPiCamera2(BaseCamera): return None # 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) # and blue-difference chroma (Cb) @@ -951,7 +993,7 @@ class StreamingPiCamera2(BaseCamera): colour across the image. """ 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"] flat = np.ones((12, 16)) recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat)