diff --git a/hardware-specific-tests/picamera2/cam_test_utils/__init__.py b/hardware-specific-tests/picamera2/cam_test_utils/__init__.py new file mode 100644 index 00000000..32019040 --- /dev/null +++ b/hardware-specific-tests/picamera2/cam_test_utils/__init__.py @@ -0,0 +1,41 @@ +"""Utilities to help with testing the camera.""" + +from typing import Optional +import tempfile +from contextlib import contextmanager + +from fastapi.testclient import TestClient + +import labthings_fastapi as lt + +from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 + + +@contextmanager +def camera_test_client( + cam: Optional[StreamingPiCamera2] = None, settings_folder: Optional[str] = None +): + """Yield a camera ThingClient on a camera server. + + This is a context manager, not a pytest fixture, as it needs to be created + multiple times in some tests. + + :param cam: The camera Thing to be used. If not supplied a new one will be created. + :param settings_folder: The settings folder for the camera, if none is supplied, new + temporary directory will be used as the settings folder. + """ + if cam is None: + cam = StreamingPiCamera2() + # Create a temp dir, if the setting folder is set it isn't really needed + # but doesn't add much overhead. + with tempfile.TemporaryDirectory() as tmpdir: + if settings_folder is None: + settings_folder = tmpdir + server = lt.ThingServer(settings_folder=settings_folder) + server.add_thing(cam, "/camera/") + + with TestClient(server.app) as test_client: + client = lt.ThingClient.from_url("/camera/", client=test_client) + yield client + del server + del cam diff --git a/hardware-specific-tests/picamera2/conftest.py b/hardware-specific-tests/picamera2/conftest.py new file mode 100644 index 00000000..f62b22a1 --- /dev/null +++ b/hardware-specific-tests/picamera2/conftest.py @@ -0,0 +1,33 @@ +"""Shared fixtures for picamera tests.""" + +import pytest + +import labthings_fastapi as lt + +from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 + +from .cam_test_utils import camera_test_client + + +@pytest.fixture +def picamera_thing() -> StreamingPiCamera2: + """Return a StreamingPiCamera2 Thing. + + This is the Thing that the fixture picamera_client uses. It can be used to probe + the Thing directly to check actions had the expected response. + """ + return StreamingPiCamera2() + + +@pytest.fixture +def picamera_client(picamera_thing) -> lt.ThingClient: + """Initialise a test picamera_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. + """ + with camera_test_client(cam=picamera_thing) as picamera_client: + yield picamera_client diff --git a/hardware-specific-tests/picamera2/test_acquisition.py b/hardware-specific-tests/picamera2/test_acquisition.py index caf8849a..7c8bd8cf 100644 --- a/hardware-specific-tests/picamera2/test_acquisition.py +++ b/hardware-specific-tests/picamera2/test_acquisition.py @@ -1,52 +1,29 @@ """Test data collection from the Raspberry Picamera.""" -from fastapi.testclient import TestClient from PIL import Image import numpy as np -import pytest - -import labthings_fastapi as lt - -from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 -@pytest.fixture(scope="module") -def client() -> 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. - """ - server = lt.ThingServer() - server.add_thing(StreamingPiCamera2(), "/camera/") - with TestClient(server.app) as test_client: - client = lt.ThingClient.from_url("/camera/", client=test_client) - yield client - - -def test_jpeg_and_array(client): +def test_jpeg_and_array(picamera_client): """Check that a jpeg grabbed from the stream is the same size as other captures. Compare it to an array capture and a jpeg capture. """ # Grab a jpeg from the stream - blob = client.grab_jpeg() + blob = picamera_client.grab_jpeg() mjpeg_frame = Image.open(blob.open()) # 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(stream_name="main") + blob = picamera_client.capture_jpeg(stream_name="main") jpeg_capture = Image.open(blob.open()) jpeg_capture.verify() assert jpeg_capture.format == "JPEG" # Capture an array - arrlist = client.capture_array(stream_name="main") + arrlist = picamera_client.capture_array(stream_name="main") array_main = np.array(arrlist) # Verify image sizes are the same diff --git a/hardware-specific-tests/picamera2/test_calibration.py b/hardware-specific-tests/picamera2/test_calibration.py index 909b533e..622405e7 100644 --- a/hardware-specific-tests/picamera2/test_calibration.py +++ b/hardware-specific-tests/picamera2/test_calibration.py @@ -1,45 +1,17 @@ """Test data collection from the Raspberry Picamera.""" -import tempfile from copy import deepcopy +import tempfile -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 +from .cam_test_utils import camera_test_client def test_get_tuning_algo(picamera_thing): @@ -73,7 +45,7 @@ def test_get_tuning_algo(picamera_thing): assert picamera_thing.get_tuning_algo("rpi.geq", raise_if_missing=False) is None -def test_calibration(picamera_thing, client): +def test_calibration(picamera_thing, picamera_client): """Check that full auto calibrate completes and set the expected values.""" # Check the calibration_required property used by the calibration wizard assert picamera_thing.calibration_required @@ -88,7 +60,7 @@ def test_calibration(picamera_thing, client): assert not picamera_thing.background_detector_status.ready # Run full auto calibrate - client.full_auto_calibrate() + picamera_client.full_auto_calibrate() # After calibration it should report that calibration is no longer required assert not picamera_thing.calibration_required @@ -104,3 +76,41 @@ def test_calibration(picamera_thing, client): # 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 + + +def test_tuning_is_persistent(): + """Check that tuning file adjustments are persistent.""" + # First check full auto-calibrate is persistent if same save dir is used. + with tempfile.TemporaryDirectory() as tmpdir: + with camera_test_client(settings_folder=tmpdir) as client: + initial_tuning = deepcopy(client.tuning) + client.full_auto_calibrate() + calibrated_tuning = deepcopy(client.tuning) + # Full auto calibrate should have saved the tuning. + assert initial_tuning != calibrated_tuning + + # Reload the camera and server from scratch + with camera_test_client(settings_folder=tmpdir) as client: + initial_tuning = deepcopy(client.tuning) + # On reload the initial tuning is as it was when calibrated last run + assert initial_tuning == calibrated_tuning + # Set the lens shading to be flat + client.flat_lens_shading() + flat_tuning = deepcopy(client.tuning) + assert flat_tuning != calibrated_tuning + + # Reload to check the flat lst persists + with camera_test_client(settings_folder=tmpdir) as client: + initial_tuning = deepcopy(client.tuning) + # On reload the initial tuning the flat tuning + assert initial_tuning == flat_tuning + # Reset the lens shading to what is in the default tuning file + client.reset_lens_shading() + flat_tuning = deepcopy(client.tuning) + assert flat_tuning != calibrated_tuning + + # Reload the one more time to check the reset lst persists + with camera_test_client(settings_folder=tmpdir) as client: + initial_tuning = deepcopy(client.tuning) + # On reload the initial tuning the flat tuning + assert initial_tuning == flat_tuning diff --git a/hardware-specific-tests/picamera2/test_exposure_time_drift.py b/hardware-specific-tests/picamera2/test_exposure_time_drift.py index 75700436..792aa861 100644 --- a/hardware-specific-tests/picamera2/test_exposure_time_drift.py +++ b/hardware-specific-tests/picamera2/test_exposure_time_drift.py @@ -4,20 +4,15 @@ This can get very tedious. Recommend running pytest with -s option to monitor progress. """ -from typing import Optional, Any +from typing import Any import logging import time import tempfile import os import json -from contextlib import contextmanager -from fastapi.testclient import TestClient -from labthings_fastapi.server import ThingServer -from labthings_fastapi.client import ThingClient - -from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 +from .cam_test_utils import camera_test_client logging.basicConfig(level=logging.DEBUG) @@ -29,24 +24,6 @@ logging.basicConfig(level=logging.DEBUG) EXPOSURE_TOL = 30 -@contextmanager -def camera_test_client(settings_folder: Optional[str] = None): - """Yield a camera ThingClient on a camera server. - - This is a context manager not a pytest fixture as it needs to be created - multiple times in some tests. - """ - cam = StreamingPiCamera2() - server = ThingServer(settings_folder=settings_folder) - server.add_thing(cam, "/camera/") - - with TestClient(server.app) as test_client: - client = ThingClient.from_url("/camera/", client=test_client) - yield client - del server - del cam - - def _test_exposure_time_drift(desired_time: int) -> None: """Capture 10 full res images and check that the exposure time remains constant. diff --git a/picamera_coverage.zip b/picamera_coverage.zip index 2e0aa752..169800f5 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 430e7a40..1c646be5 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -374,7 +374,7 @@ class StreamingPiCamera2(BaseCamera): raise MissingCalibrationError("No tuning data is set.") return None try: - return Picamera2.find_tuning_algo(self.tuning, algorithm_name) + return tf_utils.find_tuning_algo(self.tuning, algorithm_name) except StopIteration as e: if raise_if_missing: raise MissingCalibrationError( @@ -730,7 +730,7 @@ class StreamingPiCamera2(BaseCamera): # luminance (L), red-difference chroma (Cr), and blue-difference chroma # (Cb). L, Cr, Cb = recalibrate_utils.lst_from_camera(cam, self._sensor_info) # noqa: N806 - tf_utils.set_static_lst(self.tuning, L, Cr, Cb) + self.tuning = tf_utils.set_static_lst(self.tuning, L, Cr, Cb) # Re-initialise the picamera to reload the tuning file. self._initialise_picamera() @@ -760,7 +760,7 @@ class StreamingPiCamera2(BaseCamera): self, value: tuple[float, float, float, float, float, float, float, float, float], ) -> None: - tf_utils.set_static_ccm(self.tuning, value) + self.tuning = tf_utils.set_static_ccm(self.tuning, value) if self._picamera is not None: with self._streaming_picamera(pause_stream=True): @@ -813,7 +813,7 @@ class StreamingPiCamera2(BaseCamera): A value of 0 here does nothing, a value of 65535 is maximum correction. """ with self._streaming_picamera(pause_stream=True): - tf_utils.set_static_geq(self.tuning, offset) + self.tuning = tf_utils.set_static_geq(self.tuning, offset) self._initialise_picamera() @lt.thing_action @@ -824,7 +824,7 @@ class StreamingPiCamera2(BaseCamera): of view, causing inconsistent settings when capturing. """ with self._streaming_picamera(pause_stream=True): - tf_utils.set_ce_to_disabled(self.tuning) + self.tuning = tf_utils.set_ce_to_disabled(self.tuning) self._initialise_picamera() @lt.thing_action @@ -863,7 +863,9 @@ class StreamingPiCamera2(BaseCamera): with self._streaming_picamera(pause_stream=True): # Generate and array of ones of the correct size for each channel flat_array = np.ones((12, 16)) - tf_utils.set_static_lst(self.tuning, flat_array, flat_array, flat_array) + self.tuning = tf_utils.set_static_lst( + self.tuning, flat_array, flat_array, flat_array + ) self._initialise_picamera() @lt.thing_property @@ -985,7 +987,7 @@ class StreamingPiCamera2(BaseCamera): def lens_shading_tables(self, lst: LensShading) -> None: """Set the lens shading tables.""" with self._streaming_picamera(pause_stream=True): - tf_utils.set_static_lst( + self.tuning = tf_utils.set_static_lst( self.tuning, luminance=lst.luminance, cr=lst.Cr, @@ -1005,7 +1007,7 @@ class StreamingPiCamera2(BaseCamera): alsc = self.get_tuning_algo("rpi.alsc") luminance = alsc["luminance_lut"] flat = np.ones((12, 16)) - tf_utils.set_static_lst(self.tuning, luminance, flat, flat) + self.tuning = tf_utils.set_static_lst(self.tuning, luminance, flat, flat) self._initialise_picamera() @lt.thing_action @@ -1016,7 +1018,9 @@ class StreamingPiCamera2(BaseCamera): by the Raspberry Pi camera. """ with self._streaming_picamera(pause_stream=True): - tf_utils.copy_alsc_section(self.default_tuning, self.tuning) + self.tuning = tf_utils.copy_tuning_with_alsc_section_from_other( + base_tuning_file=self.tuning, copy_alsc_from=self.default_tuning + ) self._initialise_picamera() @lt.thing_property diff --git a/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py b/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py index b93217b4..e68b7681 100644 --- a/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py +++ b/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py @@ -4,6 +4,9 @@ The functions that edit the tuning files edit them in place. This will change in the future. """ +from typing import Any +from copy import deepcopy + from picamera2 import Picamera2 import numpy as np @@ -11,7 +14,7 @@ import numpy as np def load_default_tuning(sensor_model: str) -> dict: """Load the default tuning file for the camera. - This will loat the tuning file based on the specified sensor model. + This will load the tuning file based on the specified sensor model. """ fname = f"{sensor_model}.json" try: @@ -26,21 +29,46 @@ def load_default_tuning(sensor_model: str) -> dict: return Picamera2.load_tuning_file(fname, dir=tuning_dir) +def find_tuning_algo(tuning: dict[str, dict], name: str) -> dict[str, Any]: + """Return the parameters for the named algorithm in the given camera tuning dict. + + This is the same methodolgy used in the PiCamera2 library but is provided here so + it can be tested independently of installing picamera2 + + :param tuning: The camera tuning dictionary + :param name: The key for the algorithm in the tuning file + :return: The algorithm from the tuning dictionary. Editing this will edit the + original dictionary. + """ + version = tuning.get("version", 1) + # Version 1 of the tuning files was simply a dictionary of algorithms. Later + # versions have an "algorithms" key, the value of which is a list of algorithms. + if version == 1: + return tuning[name] + # The tuning file "algorithms" is a list of dictionaries + algorithms = tuning["algorithms"] + # The list is a list of dictionaries that have 1 key: the algorithm name + algo_dict = next(algo for algo in algorithms if name in algo) + # We want the value for that key, which is a dictionary of algorithm parameters + return algo_dict[name] + + def set_static_lst( tuning: dict, luminance: np.ndarray, cr: np.ndarray, cb: np.ndarray, -) -> None: +) -> dict: """Update the ``rpi.alsc`` section of a camera tuning dict to use a static correction. ``tuning`` will be updated in-place to set its shading to static, and disable any adaptive tweaking by the algorithm. """ + output_tuning = deepcopy(tuning) for table in luminance, cr, cb: if np.array(table).shape != (12, 16): raise ValueError("Lens shading tables must be 12x16!") - alsc = Picamera2.find_tuning_algo(tuning, "rpi.alsc") + alsc = find_tuning_algo(output_tuning, "rpi.alsc") alsc["n_iter"] = 0 # disable the adaptive part alsc["luminance_strength"] = 1.0 alsc["calibrations_Cr"] = [ @@ -50,6 +78,7 @@ def set_static_lst( {"ct": 4500, "table": _as_flat_rounded_list(cb, round_to=3)} ] alsc["luminance_lut"] = _as_flat_rounded_list(luminance, round_to=3) + return output_tuning def set_static_ccm( @@ -57,32 +86,34 @@ def set_static_ccm( col_corr_matrix: tuple[ float, float, float, float, float, float, float, float, float ], -) -> None: +) -> dict: """Update the ``rpi.alsc`` section of a camera tuning dict to use a static correction. ``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") + output_tuning = deepcopy(tuning) + ccm = find_tuning_algo(output_tuning, "rpi.ccm") ccm["ccms"] = [{"ct": 5000, "ccm": col_corr_matrix}] + return output_tuning 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"] + """Get a copy of the the ``rpi.ccm`` section of a camera tuning dict.""" + ccm = find_tuning_algo(tuning, "rpi.ccm") + return deepcopy(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") + alsc = find_tuning_algo(tuning, "rpi.alsc") return alsc["n_iter"] == 0 def set_static_geq( tuning: dict, offset: int = 65535, -) -> None: +) -> dict: """Update the ``rpi.geq`` section of a camera tuning dict. :param tuning: the raspberry pi tuning file. This will be updated in-place to @@ -94,44 +125,58 @@ def set_static_geq( chief ray angle compensation issue when the the stock lens is replaced by an objective. """ - geq = Picamera2.find_tuning_algo(tuning, "rpi.geq") - geq["offset"] = offset # max out offset to disable the adaptive green equalisation + output_tuning = deepcopy(tuning) + geq = find_tuning_algo(output_tuning, "rpi.geq") + # max out offset to disable the adaptive green equalisation + geq["offset"] = offset + return output_tuning def geq_is_static(tuning: dict) -> bool: """Whether the green equalisation is set to static.""" - geq = Picamera2.find_tuning_algo(tuning, "rpi.geq") + geq = find_tuning_algo(tuning, "rpi.geq") return geq["offset"] == 65535 def set_ce_to_disabled( tuning: dict, -) -> None: +) -> dict: """Set ``ce_enable`` in ``rpi.contrast`` to zero to disable adaptive contrast enhancement. - :param tuning: the raspberry pi tuning file. This will be updated in-place to - set ce_enable to 0. + :param tuning: The raspberry pi camera tuning file. + :returns: A deepcopy of the input file with ce_enable set to 0. """ - contrast = Picamera2.find_tuning_algo(tuning, "rpi.contrast") + output_tuning = deepcopy(tuning) + contrast = find_tuning_algo(output_tuning, "rpi.contrast") contrast["ce_enable"] = 0 + return output_tuning def ce_enable_is_static(tuning: dict) -> bool: """Whether the ce_enable flag is disabled.""" - contrast = Picamera2.find_tuning_algo(tuning, "rpi.contrast") + contrast = find_tuning_algo(tuning, "rpi.contrast") return contrast["ce_enable"] == 0 -def copy_alsc_section(from_tuning: dict, to_tuning: dict) -> None: - """Copy the ``rpi.alsc`` algorithm from one tuning to another. +def copy_tuning_with_alsc_section_from_other( + *, base_tuning_file: dict, copy_alsc_from: dict +) -> dict: + """Return a copy of tuning_file with the lens shading correction from another file. - This is done in-place, i.e. modifying to_tuning. + All parameters are keyword only for clarity. + + :param base_tuning_file: The tuning file to copy. + :param copy_alsc_from: The tuning file to take the alsc section from. + :return: A deep copy of base_tuning_file with the alsc section copied in from the + other tuning file. """ - # 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") + output_tuning = deepcopy(base_tuning_file) + # Find the relevant sub-dict for each tuning file + from_i = _index_of_algorithm(copy_alsc_from["algorithms"], "rpi.alsc") + to_i = _index_of_algorithm(base_tuning_file["algorithms"], "rpi.alsc") # Updating the dictionary in place. - to_tuning["algorithms"][to_i] = from_tuning["algorithms"][from_i] + output_tuning["algorithms"][to_i] = deepcopy(copy_alsc_from["algorithms"][from_i]) + return output_tuning def _index_of_algorithm(algorithms: list[dict], algorithm: str) -> int: