Merge branch 'dont-mutate-tuning-files' into 'v3'

Update the picamera tuning file utilities to not modify the dictionaries in place

Closes #474 and #579

See merge request openflexure/openflexure-microscope-server!418

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

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

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

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

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.
 

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

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: