diff --git a/hardware-specific-tests/picamera2/test_acquisition.py b/hardware-specific-tests/picamera2/test_acquisition.py new file mode 100644 index 00000000..9c8ce20f --- /dev/null +++ b/hardware-specific-tests/picamera2/test_acquisition.py @@ -0,0 +1,32 @@ +from labthings_picamera2 import StreamingPiCamera2 +from labthings_fastapi.server import ThingServer +from labthings_fastapi.client import ThingClient +from fastapi.testclient import TestClient +from PIL import Image +import numpy as np +from pytest import fixture + +@fixture(scope="module") +def client(): + server = ThingServer() + server.add_thing(StreamingPiCamera2(), "/camera/") + with TestClient(server.app) as test_client: + client = ThingClient.from_url("/camera/", client=test_client) + yield client + +def test_calibration(client): + client.full_auto_calibrate() + +def test_jpeg_and_array(client): + blob = client.grab_jpeg() + mjpeg_frame = Image.open(blob.open()) + assert mjpeg_frame + blob = client.capture_jpeg(resolution="main") + jpeg_capture = Image.open(blob.open()) + assert jpeg_capture + arrlist = client.capture_array(stream_name="main") + array_main = np.array(arrlist) + assert mjpeg_frame.size == jpeg_capture.size + assert array_main.shape[1::-1] == jpeg_capture.size + + diff --git a/hardware-specific-tests/picamera2/test_exposure_time_drift.py b/hardware-specific-tests/picamera2/test_exposure_time_drift.py new file mode 100644 index 00000000..ee75d179 --- /dev/null +++ b/hardware-specific-tests/picamera2/test_exposure_time_drift.py @@ -0,0 +1,32 @@ +import logging +import time + +from fastapi.testclient import TestClient + +from labthings_fastapi.server import ThingServer +from labthings_fastapi.client import ThingClient +from labthings_picamera2.thing import StreamingPiCamera2 + +logging.basicConfig(level=logging.DEBUG) + + +def test_exposure_time_drift(): + cam = StreamingPiCamera2() + server = ThingServer() + server.add_thing(cam, "/camera/") + + with TestClient(server.app) as test_client: + client = ThingClient.from_url("/camera/", client=test_client) + client.exposure_time = 50000 + time.sleep(0.1) + initial_et = client.exposure_time + print(f"Before capture, et is {client.exposure_time}") + for i in range(10): + client.capture_jpeg(resolution="full") + print(f"After capture, et is {client.exposure_time}") + final_et = client.exposure_time + assert initial_et == final_et + + +if __name__ == "__main__": + test_exposure_time_drift() diff --git a/hardware-specific-tests/picamera2/test_sensor_mode.py b/hardware-specific-tests/picamera2/test_sensor_mode.py new file mode 100644 index 00000000..91694a89 --- /dev/null +++ b/hardware-specific-tests/picamera2/test_sensor_mode.py @@ -0,0 +1,27 @@ +import logging + +from fastapi.testclient import TestClient +import numpy as np + +from labthings_fastapi.server import ThingServer +from labthings_fastapi.client import ThingClient +from labthings_picamera2.thing import StreamingPiCamera2 + +logging.basicConfig(level=logging.DEBUG) + + +def test_sensor_mode(): + cam = StreamingPiCamera2() + server = ThingServer() + server.add_thing(cam, "/camera/") + + with TestClient(server.app) as test_client: + client = ThingClient.from_url("/camera/", client=test_client) + for size in [(3280, 2464), (1640, 1232)]: + client.sensor_mode = {"output_size": size, "bit_depth": 10} + arr = np.array(client.capture_array(stream_name="raw")) + assert arr.shape[0] == size[1] + + +if __name__ == "__main__": + test_sensor_mode() diff --git a/hardware-specific-tests/picamera2/test_tuning.py b/hardware-specific-tests/picamera2/test_tuning.py new file mode 100644 index 00000000..f284badc --- /dev/null +++ b/hardware-specific-tests/picamera2/test_tuning.py @@ -0,0 +1,81 @@ +import os +from picamera2 import Picamera2 +from labthings_picamera2 import recalibrate_utils + +import pytest + + +MODEL = Picamera2.global_camera_info()[0]['Model'] + + +def check_camera_available(): + assert len(Picamera2.global_camera_info()) >= 1 + + +def load_default_tuning(): + fname = f"{MODEL}.json" + return Picamera2.load_tuning_file(fname) + + +def generate_bad_tuning(): + default_tuning = load_default_tuning() + bad_tuning = default_tuning.copy() + bad_tuning["version"] = 999 + return bad_tuning + + +def print_tuning(read_file=False): + key = "LIBCAMERA_RPI_TUNING_FILE" + if key in os.environ: + print(f"Tuning file environment variable: {os.environ[key]}") + if read_file: + with open(os.environ[key], "r") as f: + print(f.read()) + else: + print("Tuning file environment variable not set") + + +def _test_bad_tuning_after_good_tuning(configure): + bad_tuning = generate_bad_tuning() + default_tuning = load_default_tuning() + print_tuning() + print("opening camera with explicitly specified tuning") + with Picamera2(tuning=default_tuning) as cam: + print_tuning() + if configure: + cam.configure(cam.create_preview_configuration()) + del cam + recalibrate_utils.recreate_camera_manager() + print(f"Opening camera with tuning['version'] = {bad_tuning['version']}") + with pytest.raises(IndexError): + # The bad version should cause a problem + cam = Picamera2(tuning=bad_tuning) + print_tuning() + print("Success (not expected)!") + del cam + recalibrate_utils.recreate_camera_manager() + with Picamera2(tuning=default_tuning) as cam: + # Reload the camera with working tuning, or it will stop responding + # and fail future tests + pass + del cam + + +@pytest.mark.filterwarnings("ignore: Exception ignored") +def test_bad_tuning_after_good_tuning_noconfigure(): + _test_bad_tuning_after_good_tuning(False) + + +@pytest.mark.filterwarnings("ignore: Exception ignored") +def test_bad_tuning_after_good_tuning_configure(): + _test_bad_tuning_after_good_tuning(True) + + +@pytest.mark.filterwarnings("ignore: Exception ignored") +def test_bad_tuning_after_good_tuning_noconfigure2(): + _test_bad_tuning_after_good_tuning(False) + + +@pytest.mark.filterwarnings("ignore: Exception ignored") +def test_bad_tuning_after_good_tuning_configure2(): + _test_bad_tuning_after_good_tuning(True) diff --git a/ofm_config_full.json b/ofm_config_full.json index a73e4e40..bf9e29cc 100644 --- a/ofm_config_full.json +++ b/ofm_config_full.json @@ -1,6 +1,6 @@ { "things": { - "/camera/": "labthings_picamera2.thing:StreamingPiCamera2", + "/camera/": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2", "/stage/": "labthings_sangaboard:SangaboardThing", "/auto_recentre_stage/": "openflexure_microscope_server.things.auto_recentre_stage:RecentringThing", "/autofocus/": "openflexure_microscope_server.things.autofocus:AutofocusThing", diff --git a/pyproject.toml b/pyproject.toml index bfa7b2c5..5b00f07b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,7 +40,7 @@ dev = [ "matplotlib~=3.10" ] pi = [ - "labthings-picamera2 == 0.0.2", + "picamera2~=0.3.12", ] [project.scripts] diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py index c1fb9a17..71c7406d 100644 --- a/src/openflexure_microscope_server/things/camera/__init__.py +++ b/src/openflexure_microscope_server/things/camera/__init__.py @@ -10,6 +10,8 @@ from __future__ import annotations import logging from typing import Literal, Protocol, runtime_checkable, Optional +from pydantic import RootModel + from labthings_fastapi.thing import Thing from labthings_fastapi.decorators import thing_action, thing_property from labthings_fastapi.dependencies.metadata import GetThingStates @@ -24,129 +26,14 @@ from labthings_fastapi.types.numpy import NDArray class JPEGBlob(Blob): media_type: str = "image/jpeg" +class PNGBlob(Blob): + media_type: str = "image/png" -@runtime_checkable -class CameraProtocol(Protocol): - """A Thing representing a camera""" - - def __enter__(self) -> None: ... - - def __exit__(self, _exc_type, _exc_value, _traceback) -> None: ... - - @property - def stream_active(self) -> bool: - "Whether the MJPEG stream is active" - ... - - def snap_image(self) -> NDArray: - """Acquire one image from the camera.""" - ... - - def capture_array( - self, - stream_name: Literal["main", "lores", "raw", "full"] = "main", - wait: Optional[float] = 5, - ) -> NDArray: ... - - 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""" - ... - - 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""" - ... - - def start_streaming(self, main_resolution, buffer_count) -> 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""" - ... - - def capture_image(self, stream_name, wait): - """Capture a PIL image from stream stream_name with timeout wait""" - ... - +class ArrayModel(RootModel): + """A model for an array""" + root: NDArray 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. - """ - - 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 @@ -158,8 +45,11 @@ class CameraStub(BaseCamera): This stub class should be used for dependencies on the CameraProtocol. """ + mjpeg_stream = MJPEGStreamDescriptor() + lores_mjpeg_stream = MJPEGStreamDescriptor() + 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") @@ -169,11 +59,6 @@ class CameraStub(BaseCamera): "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") - @thing_action def capture_array( self, @@ -204,5 +89,5 @@ class CameraStub(BaseCamera): raise NotImplementedError("Cameras must not inherit from CameraStub") -CameraDependency = direct_thing_client_dependency(CameraStub, "/camera/") -RawCameraDependency = raw_thing_dependency(CameraProtocol) +CameraDependency = direct_thing_client_dependency(BaseCamera, "/camera/") +RawCameraDependency = raw_thing_dependency(BaseCamera) diff --git a/src/openflexure_microscope_server/things/camera/opencv.py b/src/openflexure_microscope_server/things/camera/opencv.py index abb50252..2278e2bb 100644 --- a/src/openflexure_microscope_server/things/camera/opencv.py +++ b/src/openflexure_microscope_server/things/camera/opencv.py @@ -53,9 +53,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: diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py new file mode 100644 index 00000000..81434918 --- /dev/null +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -0,0 +1,868 @@ +from __future__ import annotations +from dataclasses import dataclass +from datetime import datetime +import io +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.thing import Thing +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.types.numpy import NDArray +from labthings_fastapi.dependencies.metadata import GetThingStates +from labthings_fastapi.dependencies.blocking_portal import BlockingPortal +from typing import Annotated, Any, Iterator, Literal, Mapping, Optional, Self +from contextlib import contextmanager +import piexif +from scipy.ndimage import zoom +from scipy.interpolate import interp1d +from PIL import Image +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, PNGBlob, 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: + ret = cam.capture_metadata()[self.control_name] + return ret + + 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): + 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): + output_size: tuple[int, int] + bit_depth: int + + +class LensShading(BaseModel): + 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] = {} + # NB 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 we know our data is fresh + cam.capture_metadata() + for k, v in cam.capture_metadata().items(): + if k in self.persistent_controls: + if k in self.persistent_control_tolerances: + if ( + np.abs(self.persistent_controls[k] - v) + < self.persistent_control_tolerances[k] + ): + logging.debug( + f"Ignoring a small change in persistent control {k}" + f"from {self.persistent_controls[k]} to {v}" + "while updating persistent controls." + ) + continue # Ignore small changes, to avoid drift + self.persistent_controls[k] = v + self.thing_settings.update( + self.persistent_controls + ) # TODO: Is this saving to the wrong place? + + 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)", + ) + + @mjpeg_bitrate.setter + def mjpeg_bitrate(self, value: Optional[int]): + """Restart the stream when we set the bitrate""" + with self.picamera(pause_stream=True): + pass # just pausing and restarting the stream is enough. + + 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" + ) + + _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: + # TODO: should this be a separate file? + 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() + 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. Defaults to 6. + """ + with self.picamera() as picam: + # TODO: Filip: can we use the lores output to keep preview stream going + # while recording? According to picamera2 docs 4.2.1.6 this should work + 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, + ) + # Set buffer count - can't be negative + 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=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.") + + # Increase the resolution for taking an image + time.sleep( + 0.2 + ) # Sprinkled a sleep to prevent camera getting confused by rapid commands + + @thing_action + def capture_image( + self, + stream_name: Literal["main", "lores", "raw", "full"] = "main", + wait: Optional[float] = 0.9, + ): + """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", "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 + """ + 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_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.debug( + 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.debug( + 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) + + @thing_property + def exposure(self) -> float: + """An alias for `exposure_time` to fit the micromanager API""" + return self.exposure_time + + @exposure.setter # type: ignore + def exposure(self, value): + self.exposure_time = value + + @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 to hit the target white level + + Starting from the minimum exposure, we gradually increase exposure until + we hit the specified white level. We use a percentile 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): + """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: + L, Cr, Cb = recalibrate_utils.lst_from_camera(cam) + 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): + 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 from the documentation""" + c = [ + 1.80439, + -0.73699, + -0.06739, + -0.36073, + 1.83327, + -0.47255, + -0.08378, + -0.56403, + 1.64781, + ] + self.colour_correction_matrix = c + + @thing_action + def calibrate_colour_correction(self, c: tuple): + """Overwrite the colour correction matrix in camera tuning""" + with self.picamera(pause_stream=True): + recalibrate_utils.set_static_ccm(self.tuning, c) + self.initialise_picamera() + + @thing_action + def set_static_green_equalisation(self, offset: int = 65535): + """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): + """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): + """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. + """ + with self.picamera(pause_stream=True): + f = np.ones((12, 16)) + recalibrate_utils.set_static_lst(self.tuning, f, f, f) + self.initialise_picamera() + + @thing_property + def lens_shading_tables(self) -> Optional[LensShading]: + """The current lens shading (i.e. flat-field correction) + + This returns the current lens shading correction, as three 2D lists + each with dimensions 16x12. This assumes that we are using a static + lens shading table - if adaptive control is enabled, or if there + are multiple LSTs in use for different colour temperatures, + we return a null value to avoid confusion. + """ + if not self.lens_shading_is_static: + return None + alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc") + if any(len(alsc[f"calibrations_C{c}"]) != 1 for c in ("r", "b")): + return None + + def reshape_lst(lin: list[float]) -> list[list[float]]: + 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): + """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): + """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) diff --git a/src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py b/src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py new file mode 100644 index 00000000..0370ba33 --- /dev/null +++ b/src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py @@ -0,0 +1,561 @@ +""" +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 +``` +""" + +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 + + +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: + dir = "/usr/share/libcamera/ipa/raspberrypi" # from picamera2 v0.3.9 + # The directory above has been removed from the search path, which I + # find 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=dir) + + +def set_minimum_exposure(camera: Picamera2): + """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) + NB ISO is left at auto, because this is needed for the gains + to be set correctly. + """ + camera.set_controls({"AeEnable": False, "AnalogueGain": 1, "ExposureTime": 1}) + # camera.iso = 0 # We must set ISO=0 (auto) or we can't set gain + # camera.analog_gain = 1 + # camera.digital_gain = 1 (not configurable) + # Setting the shutter speed to 1us will result in it being set + # to the minimum possible, which is probably 8us for PiCamera v2 + # camera.shutter_speed = 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): + """Check whether the brightness is within the specified target range""" + converged = abs(test.level - target) < target * tolerance + return converged + + +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")) + # logging.info(f"White balance: channels were retrieved with shape {channels.shape}.") + 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": + _, h, w = channels.shape + blue, g1, g2, red = ( + np.mean( + channels[:, 9 * h // 20 : 11 * h // 20, 9 * w // 20 : 11 * w // 20], + axis=(1, 2), + ) + - 64 + ) + else: + # TODO: read black level from camera rather than hard-coding 64 + blue, g1, g2, red = np.percentile(channels, percentile, axis=(1, 2)) - 64 + 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 + + +LensShadingTables = tuple[np.ndarray, np.ndarray, np.ndarray] + + +def get_16x12_grid(chan: np.ndarray, dx: int, dy: int): + """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 32nd 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]): + """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 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). + + # TODO: make consistent with + https://git.linuxtv.org/libcamera.git/tree/utils/raspberrypi/ctt/ctt_alsc.py + """ + 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. + luminance_gains: np.ndarray = np.max(g) / g # Minimum luminance gain is 1 + cr_gains: np.ndarray = g / r + # cr_gains /= cr_gains[5, 7] # Normalise so the central colour doesn't change + cb_gains: np.ndarray = g / b + # cb_gains /= cb_gains[5, 7] + 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": np.reshape(cr, (-1)).round(3).tolist()} + ] + alsc["calibrations_Cb"] = [ + {"ct": 4500, "table": np.reshape(cb, (-1)).round(3).tolist()} + ] + alsc["luminance_lut"] = np.reshape(luminance, (-1)).round(3).tolist() + + +def set_static_ccm(tuning: dict, c: list) -> 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": c}] + + +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): + """Find the index of an algorithm's section in the tuning file""" + for i, a in enumerate(algorithms): + if algorithm in a: + return i + + +def copy_alsc_section(from_tuning: dict, to_tuning: dict): + """Copy the `rpi.alsc` algorithm from one tuning to another. + + This is done in-place, i.e. modifying to_tuning. + """ + from_i = index_of_algorithm(from_tuning["algorithms"], "rpi.alsc") + to_i = index_of_algorithm(to_tuning["algorithms"], "rpi.alsc") + # Please excuse the clumsy update-and-delete - this lets us use + # the nice Picamera2 function to find the relevant sub-dict. + 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). + format = camera.camera_configuration()["raw"]["format"] + print(f"Acquired a raw image in format {format}") + return channels_from_bayer_array(raw_image) + + +def recreate_camera_manager(): + """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() + + +if __name__ == "__main__": + """This block is untested but has been updated.""" + with Picamera2() as cam: + tuning = load_default_tuning(cam) + f = np.ones((12, 16)) + set_static_lst(tuning, f, f, f) + set_static_geq(tuning) + with Picamera2(tuning=tuning) as cam: + cam.start_preview() + time.sleep(3) + logging.info("Recalibrating...") + adjust_shutter_and_gain_from_raw(cam) + adjust_white_balance_from_raw(cam) + lst = lst_from_camera(cam) + set_static_lst(tuning, *lst) + logging.info("Done.") + with Picamera2(tuning=tuning) as cam: + cam.start_preview() + time.sleep(2) diff --git a/src/openflexure_microscope_server/things/camera/simulation.py b/src/openflexure_microscope_server/things/camera/simulation.py index ad0ad1c2..723b52cc 100644 --- a/src/openflexure_microscope_server/things/camera/simulation.py +++ b/src/openflexure_microscope_server/things/camera/simulation.py @@ -22,12 +22,11 @@ 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,11 +34,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 +154,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: