"""Submodule for interacting with a Raspberry Pi camera using the Picamera2 library. The Picamera2 library uses LibCamera as the underlying camera stack. This gives us some control of the GPU pipeline for the image. The API documentation for PiCamera2 is unfortunately not in a standard auto-generated website. For documentation of the PiCamera2 API there is a PDF called "The Picamera2 Library" available at: https://datasheets.raspberrypi.com/camera/picamera2-manual.pdf For information on the algorithms used to tune/calibrate the Raspberry Pi Camera see the guide called "Raspberry Pi Camera Algorithm and Tuning Guide" Available at: https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf """ from __future__ import annotations from typing import Annotated, Iterator, Literal, Mapping, Optional, overload, Any from types import TracebackType import json import logging import os import copy import tempfile import time from contextlib import contextmanager from threading import RLock from pydantic import BaseModel, BeforeValidator import numpy as np from PIL import Image from picamera2 import Picamera2 from picamera2.encoders import MJPEGEncoder from picamera2.outputs import Output import labthings_fastapi as lt from labthings_fastapi.exceptions import NotConnectedToServerError from openflexure_microscope_server.ui import ( ActionButton, PropertyControl, action_button_for, property_control_for, ) from . import picamera_recalibrate_utils as recalibrate_utils from . import picamera_tuning_file_utils as tf_utils from . import BaseCamera, ArrayModel LOGGER = logging.getLogger(__name__) SUPPORTED_CAMS_SENSOR_INFO = { "picamera_v2": recalibrate_utils.IMX219_SENSOR_INFO, "picamera_hq": recalibrate_utils.IMX477_SENSOR_INFO, } class PicameraModelError(RuntimeError): """There is a problem Picamera sensor model set by the configuration.""" class MissingCalibrationError(RuntimeError): """Picamera tuning file is missing or doesn't contain the requested algorithm.""" class PicameraStreamOutput(Output): """An Output class that sends frames to a stream.""" def __init__( self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal ) -> None: """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: bytes, _keyframe: Optional[bool] = True, _timestamp: Optional[int] = None, _packet: Any = None, _audio: bool = False, ) -> None: """Add a frame to the stream's ringbuffer.""" self.stream.add_frame(frame, self.portal) class SensorMode(BaseModel): """A Pydantic model holding all the information about a specific sensor mode. This data is as reported by the PiCamera2 module. """ 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): """A Pydantic model holding the two values needed to select a PiCamera Sensor mode. These values are the output size and the bit depth. This is a Pydantic model so that it can be saved to disk. """ output_size: tuple[int, int] bit_depth: int class LensShading(BaseModel): """A Pydantic model holding the lens shading tables. PiCamera needs three numpy arrays for lens shading correction. Each array is (12, 16) in size. The arrays are luminance, red-difference chroma (Cr), and blue-difference chroma (Cb). This is a Pydantic model so that it can be saved to the disk. """ luminance: list[list[float]] Cr: list[list[float]] Cb: list[list[float]] class StreamingPiCamera2(BaseCamera): """A Thing that provides and interface to the Raspberry Pi Camera. Currently the Thing only supports the PiCamera v2 board. This needs generalisation. """ def __init__(self, camera_num: int = 0, camera_board: str = "picamera_v2") -> None: """Initialise the camera with the given camera number. This makes no connection to the camera (except to get the default tuning file). :param camera_num: The number of the camera. This should generally be left as 0 as most Raspberry Pi boards only support 1 camera. :param camera_board: The camera board used. Supported options are "picamera_v2" and "picamera_hq". """ super().__init__() self._setting_save_in_progress = False self._camera_num = camera_num self._camera_board = camera_board if camera_board not in SUPPORTED_CAMS_SENSOR_INFO: raise PicameraModelError( f"The camera_board {camera_board} is not supported. Supported boards " f"are {SUPPORTED_CAMS_SENSOR_INFO.keys()}." ) self._sensor_info = SUPPORTED_CAMS_SENSOR_INFO[camera_board] self._picamera_lock = None self._picamera = None # Load the tuning file for the specified sensor mode. self.default_tuning = tf_utils.load_default_tuning( self._sensor_info.sensor_model ) # Set tuning to default tuning. This will be overwritten when the Thing is # connects to the server if tuning is saved to disk. try: self.tuning = copy.deepcopy(self.default_tuning) except NotConnectedToServerError: # This will throw an error after setting as we are not connected to # a server. But we know this, so we ignore the error. pass stream_resolution = lt.ThingProperty( tuple[int, int], initial_value=(820, 616), ) """Resolution to use for the MJPEG stream.""" mjpeg_bitrate = lt.ThingProperty( Optional[int], initial_value=100000000, ) """Bitrate for MJPEG stream (None for default).""" stream_active = lt.ThingProperty( bool, initial_value=False, observable=True, readonly=True, ) """Whether the MJPEG stream is active.""" def save_settings(self) -> None: """Override save_settings to ensure that camera properties don't recurse. This method is run by any Thing when a ThingSetting is saved. However, the method reads the thing_setting. As reading the thing setting talks to the camera and calls save_settings if the value is not as expected, this could cause recursion. Also this means that saving one setting causes all others to be read each time. """ try: self._setting_save_in_progress = True super().save_settings() finally: self._setting_save_in_progress = False ## Persistent controls! These are settings _analogue_gain: float = 1.0 @lt.thing_setting def analogue_gain(self) -> float: """The Analogue gain applied by the camera sensor.""" if not self._setting_save_in_progress and self.streaming: with self._streaming_picamera() as cam: cam_value = cam.capture_metadata()["AnalogueGain"] if cam_value != self._analogue_gain: self._analogue_gain = cam_value self.save_settings() return self._analogue_gain @analogue_gain.setter def analogue_gain(self, value: float) -> None: self._analogue_gain = value if self.streaming: with self._streaming_picamera() as cam: cam.set_controls({"AnalogueGain": value}) _colour_gains: tuple[float, float] = (1.0, 1.0) @lt.thing_setting def colour_gains(self) -> tuple[float, float]: """The red and blue colour gains, must be between 0.0 and 32.0.""" if not self._setting_save_in_progress and self.streaming: with self._streaming_picamera() as cam: cam_value = cam.capture_metadata()["ColourGains"] if cam_value != self._colour_gains: self._colour_gains = cam_value self.save_settings() return self._colour_gains @colour_gains.setter def colour_gains(self, value: tuple[float, float]) -> None: self._colour_gains = value if self.streaming: with self._streaming_picamera() as cam: cam.set_controls({"ColourGains": value}) _exposure_time: int = 0 @lt.thing_setting def exposure_time(self) -> int: """The camera exposure time in microseconds. When setting this property the camera will adjust the set value to the nearest allowed value that is lower than the current setting. """ if not self._setting_save_in_progress and self.streaming: with self._streaming_picamera() as cam: cam_value = cam.capture_metadata()["ExposureTime"] if cam_value != self._exposure_time: self._exposure_time = cam_value self.save_settings() return self._exposure_time @exposure_time.setter def exposure_time(self, value: int) -> None: self._exposure_time = value if self.streaming: with self._streaming_picamera() as cam: # Note: This set a value 1 higher than requested as picamera2 always # sets a lower value than requested, even if the requested is allowed cam.set_controls({"ExposureTime": value + 1}) def _get_persistent_controls(self) -> dict: if self.streaming: self.discard_frames() return { "AeEnable": False, "AnalogueGain": self.analogue_gain, "AwbEnable": False, "Brightness": 0, "ColourGains": self.colour_gains, "Contrast": 1, # Must also set plus 1 or the exposure drifts with start and stop stream. "ExposureTime": self.exposure_time + 1, "Saturation": 1, "Sharpness": 1, } _sensor_modes = None @lt.thing_property def sensor_modes(self) -> list[SensorMode]: """All the available modes the current sensor supports.""" if not self._sensor_modes: with self._streaming_picamera() as cam: self._sensor_modes = cam.sensor_modes return self._sensor_modes _sensor_mode: Optional[dict] = None @lt.thing_property def sensor_mode(self) -> Optional[SensorModeSelector]: """The intended sensor mode of the camera.""" if self._sensor_mode is None: return None return SensorModeSelector(**self._sensor_mode) @sensor_mode.setter def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None: """Change the sensor mode used.""" if new_mode is None: self._sensor_mode = None elif isinstance(new_mode, SensorModeSelector): self._sensor_mode = new_mode.model_dump() elif isinstance(new_mode, dict): self._sensor_mode = new_mode # By pausing the stream on when accessing, streaming_picamera # self._sensor_mode will be read and set when the stream restarts # after the context manager closes. with self._streaming_picamera(pause_stream=True): pass @lt.thing_property def sensor_resolution(self) -> Optional[tuple[int, int]]: """The native resolution of the camera's sensor.""" with self._streaming_picamera() as cam: return cam.sensor_resolution tuning = lt.ThingSetting(Optional[dict], None, readonly=True) """The Raspberry PiCamera Tuning File JSON.""" # Use overload to clarify that only a dictionary is returned if `raise_if_missing` # is True @overload def get_tuning_algo( self, algorithm_name: str, raise_if_missing: Literal[True] ) -> dict: ... # Otherwise may also be None @overload def get_tuning_algo( self, algorithm_name: str, raise_if_missing: bool ) -> Optional[dict]: ... def get_tuning_algo( self, algorithm_name: str, raise_if_missing: bool = True ) -> Optional[dict]: """Return the active tuning algorithm settings for the given algorithm. :returns: The algorithm dictionary if found, returns None if no tuning data is loaded or if the tuning algorithm is not found. :raises MissingCalibrationError: If raise_if_missing is true and there is no tuning file is available, or the requested algorithm is not present. """ if self.tuning is None: if raise_if_missing: raise MissingCalibrationError("No tuning data is set.") return None try: return Picamera2.find_tuning_algo(self.tuning, algorithm_name) except StopIteration as e: if raise_if_missing: raise MissingCalibrationError( f"No tuning algorithm with name {algorithm_name}." ) from e return None def _initialise_picamera(self, check_sensor_model: bool = False) -> None: """Acquire the picamera device and store it as ``self._picamera``. This duplicates logic in ``Picamera2.__init__`` to provide a tuning file that will be read when the camera system initialises. :param check_sensor_model: Set to true to check the sensor model is the expected sensor model. This is used on ``__enter__`` to confirm that the real camera matches the expected camera. :raises PicameraModelError: If check_sensor_model is True and the real camera sensor model doesn't match the expected sensor model. """ if self._picamera_lock is not None: # Don't close the camera if it's in use self._picamera_lock.acquire() with tempfile.NamedTemporaryFile("w") as tuning_file: 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 if self._picamera is not None: LOGGER.info("Closing picamera object for reinitialisation") LOGGER.info( "Camera object already exists, closing for reinitialisation" ) self._picamera.close() LOGGER.info("Picamera closed, deleting picamera") del self._picamera recalibrate_utils.recreate_camera_manager() LOGGER.info("Creating new Picamera2 object") # Specify tuning file otherwise it will be overwritten with None. self._picamera = Picamera2( camera_num=self._camera_num, tuning=self.tuning, ) if check_sensor_model: hw_sensor_model = self._picamera.camera_properties["Model"] if hw_sensor_model != self._sensor_info.sensor_model: raise PicameraModelError( f"Wrong Picamera model. Expecting {self._sensor_info.sensor_model}, " f"but found {hw_sensor_model}." ) self._picamera_lock = RLock() def __enter__(self) -> None: """Start streaming when the Thing context manager is opened. This opens the picamera connection, initialises the camera, sets the sensor_modes property, and then starts the streams. """ self._initialise_picamera(check_sensor_model=True) # Sensor modes is a cached property read it once after initialising the camera _modes = self.sensor_modes self.start_streaming() return self @property def streaming(self) -> bool: """True if the camera is streaming.""" return self._picamera is not None and self._picamera.started @contextmanager def _streaming_picamera(self, pause_stream: bool = False) -> Iterator[Picamera2]: """Lock access to picamera and return the underlying ``Picamera2`` instance. Optionally the stream can be paused to allow updating the camera settings. :param pause_stream: If False the ``Picamera2`` instance is simply yielded. If True: * Stop the MJPEG Stream * Yield the ``Picamera2`` instance for function calling the context manager to make changes. * On closing of the context manager the stream will restart. """ already_streaming = self.stream_active with self._picamera_lock: if pause_stream and already_streaming: self.stop_streaming(stop_web_stream=False) try: yield self._picamera finally: if pause_stream and already_streaming: self.start_streaming() def __exit__( self, _exc_type: type[BaseException], _exc_value: Optional[BaseException], _traceback: Optional[TracebackType], ) -> None: """Close the picamera connection when the Thing context manager is closed.""" self.stop_streaming() with self._streaming_picamera() as cam: cam.close() del self._picamera @lt.thing_action def start_streaming( self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 6 ) -> None: """Start the MJPEG stream. This is where persistent controls are sent to camera. 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. Value must be between 1 and 8, Defaults to 6. """ controls = self._get_persistent_controls() # Buffer count can't be negative, zero, or too high. if buffer_count < 1 or buffer_count > 8: # 8 is slightly arbitrary. 6 is the PiCamera default for video # and the documentation only says that setting values higher gives # diminishing returns, and that the true maximum is hardware dependent raise ValueError( f"Can't set a buffer count of {buffer_count}. " "Buffer count must be an integer from 1-8" ) with self._streaming_picamera() as picam: 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._sensor_mode, controls=controls, ) stream_config["buffer_count"] = buffer_count picam.configure(stream_config) LOGGER.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, lt.get_blocking_portal(self), ), name=stream_name, ) picam.start_encoder( MJPEGEncoder(100000000), PicameraStreamOutput( self.lores_mjpeg_stream, lt.get_blocking_portal(self), ), name="lores", ) except Exception as e: LOGGER.exception("Error while starting preview: {e}") LOGGER.exception(e) else: self.stream_active = True LOGGER.debug( "Started MJPEG stream at %s on port %s", self.stream_resolution, 1 ) @lt.thing_action def stop_streaming(self, stop_web_stream: bool = True) -> None: """Stop the MJPEG stream.""" with self._streaming_picamera() as picam: try: picam.stop_recording() # This should also stop the extra lores encoder except Exception as e: LOGGER.info("Stopping recording failed") LOGGER.exception(e) else: self.stream_active = False if stop_web_stream: portal = lt.get_blocking_portal(self) self.mjpeg_stream.stop(portal) self.lores_mjpeg_stream.stop(portal) LOGGER.info("Stopped MJPEG stream.") # Adding a sleep to prevent camera getting confused by rapid commands time.sleep(self._sensor_info.short_pause) @lt.thing_action def discard_frames(self) -> None: """Discard frames so that the next frame captured is fresh.""" with self._streaming_picamera() as cam: cam.capture_metadata() @contextmanager def _switch_to_still_capture_mode(self) -> Iterator[Picamera2]: """Get the picamera lock, pause stream and switch into still capture config. Restarts stream when complete. """ with self._streaming_picamera(pause_stream=True) as cam: LOGGER.debug("Reconfiguring camera for full resolution capture") cam.configure(cam.create_still_configuration(sensor=self._sensor_mode)) cam.start() time.sleep(self._sensor_info.short_pause) yield cam def capture_image( self, stream_name: Literal["main", "lores", "full"] = "main", wait: Optional[float] = 0.9, ) -> Image: """Acquire one image from the camera and return it as a PIL Image. If the ``stream_name`` 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. This will capture an image at the full resolution of the current sensor mode. If the current sensor mode bins or crops the image, this may not be the native resolution of the camera sensor. :param stream_name: (Optional) The PiCamera2 stream to use, should be one of ["main", "lores", "full"]. Default = "main". Note that "raw" images cannot be captured as PIL images. Use capture_array :param wait: (Optional, float) Set a timeout in seconds. Default = 0.9s, lower than the 1s timeout for the camera. This ensures that our code times out and returns before the camera times out. If None is set the default value of 0.9 will be used to prevent the possibility of the camera locking. :raises TimeoutError: if this time is exceeded during capture. """ if wait is None: wait = 0.9 if stream_name in ["main", "lores", "raw"]: with self._streaming_picamera() as cam: return cam.capture_image(stream_name, wait=wait) elif stream_name == "full": with self._switch_to_still_capture_mode() as cam: return cam.capture_image(name="main", wait=wait) else: raise ValueError(f'Unknown stream name "{stream_name}"') @lt.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. :param stream_name: (Optional) The PiCamera2 stream to use, should be one of ["main", "lores", "raw", "full"]. Default = "main" :param wait: (Optional, float) Set a timeout in seconds. Default = 0.9s, lower than the 1s timeout for the camera. This ensures that our code times out and returns before the camera times out. If None is set the default value of 0.9 will be used to prevent the possibility of the camera locking. :raises TimeoutError: if this time is exceeded during capture. """ if stream_name == "raw": # Raw cannot used capture_image. if wait is None: wait = 0.9 with self._switch_to_still_capture_mode() as cam: return cam.capture_array(name="raw", wait=wait) # Note that internally the PiCamera creates a PIL image and then converts to # numpy with ``np.array(Image.open(io.BytesIO(self.make_buffer(name))))``. # As such we use capture_image to get an Image from the picamera and return # as array return np.array(self.capture_image(stream_name, wait)) @lt.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._streaming_picamera() as cam: return cam.camera_configuration() @lt.thing_property def capture_metadata(self) -> dict: """Return the metadata from the camera.""" with self._streaming_picamera() as cam: return cam.capture_metadata() @lt.thing_action def auto_expose_from_minimum( self, target_white_level: Optional[int] = None, percentile: float = 99.9, ) -> None: """Adjust exposure until a the target white level is reached. Starting from the minimum exposure, gradually increase exposure until the image reaches the specified white level. :param target_white_level: Raw target white level, this should be an integer within the range set by the bit-depth of the camera sensor (10-bit for PiCamera v2, 12 Bit for Picamera HQ. If None the default will be used for the current sensor. This is approximately 70% saturated. :param percentile: The percentile to use instead of maximum. Default 99.9. When calculating the brightest pixel, a percentile is used rather than the maximum in order to be robust to a small number of noisy/bright pixels. """ if target_white_level is None: target_white_level = self._sensor_info.default_target_white_level with self._streaming_picamera(pause_stream=True) as cam: recalibrate_utils.adjust_shutter_and_gain_from_raw( cam, self._sensor_info, target_white_level=target_white_level, percentile=percentile, ) @lt.thing_action def calibrate_white_balance( self, method: Literal["percentile", "centre"] = "centre", luminance_power: float = 1.0, ) -> None: """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._streaming_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, self._sensor_info, 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, self._sensor_info, percentile=99, method=method ) @lt.thing_action def calibrate_lens_shading(self) -> None: """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._streaming_picamera(pause_stream=True) as cam: # Suppress lint warning that L, Cr, and Cb are not lowercase, as these are # the standard mathematical terms for: # 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._initialise_picamera() @lt.thing_property def colour_correction_matrix( self, ) -> tuple[float, float, float, float, float, float, float, float, float]: """The ``colour_correction_matrix`` from the tuning file. This is broken out into its own property for convenience and compatibility with the micromanager API Ir is a 9 value tuple used to specify the 3x3 matrix that the GPU pipeline uses to convert from the camera R,G,B vector to the standard R,G,B. See page Raspberry Pi Camera Algorithm and Tuning Guide, page 45. """ return tuple(tf_utils.get_static_ccm(self.tuning)[0]["ccm"]) @colour_correction_matrix.setter # type: ignore def colour_correction_matrix( self, value: tuple[float, float, float, float, float, float, float, float, float], ) -> None: tf_utils.set_static_ccm(self.tuning, value) if self._picamera is not None: with self._streaming_picamera(pause_stream=True): self._initialise_picamera() @lt.thing_action def reset_ccm(self) -> None: """Overwrite the colour correction matrix in camera tuning with default values. These values are from the Raspberry Pi Camera Algorithm and Tuning Guide, page 45. """ # This is flattened 3x3 matrix. See `colour_correction_matrix` col_corr_matrix = [ 1.80439, -0.73699, -0.06739, -0.36073, 1.83327, -0.47255, -0.08378, -0.56403, 1.64781, ] self.colour_correction_matrix = col_corr_matrix @lt.thing_action def set_static_green_equalisation(self, offset: int = 65535) -> None: """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._streaming_picamera(pause_stream=True): tf_utils.set_static_geq(self.tuning, offset) self._initialise_picamera() @lt.thing_action def full_auto_calibrate(self, portal: lt.deps.BlockingPortal) -> None: """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`` * ``reset_ccm`` * ``calibrate_white_balance`` * ``set_background`` """ self.flat_lens_shading() self.auto_expose_from_minimum() self.set_static_green_equalisation() self.calibrate_lens_shading() self.reset_ccm() self.calibrate_white_balance() self.set_background(portal) @lt.thing_action def flat_lens_shading(self) -> None: """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. This flat table is used to take an image with no lens shading so that the correct lens shading table can be calibrated. """ 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._initialise_picamera() @lt.thing_property def primary_calibration_actions(self) -> list[ActionButton]: """The calibration actions for both calibration wizard and settings panel.""" return [ action_button_for( self.full_auto_calibrate, submit_label="Full Auto-Calibrate", can_terminate=False, requires_confirmation=True, confirmation_message=( "Start recalibration? This may take a while, and the microscope " "will be locked during this time." ), notify_on_success=True, success_message="Finished recalibration.", ), ] @lt.thing_property def secondary_calibration_actions(self) -> list[ActionButton]: """The calibration actions that appear only in settings panel.""" return [ action_button_for( self.auto_expose_from_minimum, submit_label="Auto Gain & Shutter Speed", can_terminate=False, button_primary=False, ), action_button_for( self.calibrate_white_balance, submit_label="Auto White Balance", can_terminate=False, button_primary=False, ), action_button_for( self.calibrate_lens_shading, submit_label="Auto Flat Field Correction", can_terminate=False, button_primary=False, requires_confirmation=True, confirmation_message=( "Is the microscope looking at an evenly illuminated, empty field " "of view? If not, the current image will show through in any " "images captured afterwards." ), ), action_button_for( self.flat_lens_shading, submit_label="Disable Flat Field Correction", can_terminate=False, button_primary=False, ), action_button_for( self.reset_lens_shading, submit_label="Reset Flat Field Correction", can_terminate=False, button_primary=False, ), ] @lt.thing_property def manual_camera_settings(self) -> list[PropertyControl]: """The camera settings to expose as property controls in the settings panel.""" return [ property_control_for( self, "exposure_time", label="Exposure Time (0-33251)", read_back=True, read_back_delay=1000, ), property_control_for( self, "analogue_gain", label="Analogue Gain", read_back=True, read_back_delay=1000, ), property_control_for( self, "colour_gains", label="Colour Gains", read_back=True, read_back_delay=1000, ), ] @lt.thing_property def lens_shading_tables(self) -> Optional[LensShading]: """The current lens shading (i.e. flat-field correction). Return the current lens shading correction, as three 2D lists each with dimensions 16x12, if a static lens shading table is in use. Return None if: - adaptive control is enabled - multiple LSTs in use (for different colour temperatures), """ if not self.lens_shading_is_static: return None # Note "alsc" is the Picamera2 term for "Automatic Lens Shading Correction" alsc = self.get_tuning_algo("rpi.alsc") # Check there is exactly 1 correction table for red-difference chroma (Cr) # and blue-difference chroma (Cb) if len(alsc["calibrations_Cr"]) != 1 or len(alsc["calibrations_Cb"]) != 1: # If there is not exactly one table, then lens shading isn't static. return None def reshape_lst(lin: list[float]) -> list[list[float]]: """Reshape the 192 element list into a 2D 16x12 list.""" 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._streaming_picamera(pause_stream=True): tf_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)), ) @lt.thing_action def flat_lens_shading_chrominance(self) -> None: """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._streaming_picamera(pause_stream=True): 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._initialise_picamera() @lt.thing_action def reset_lens_shading(self) -> None: """Revert to default lens shading settings. This method will restore the default "adaptive" lens shading method used by the Raspberry Pi camera. """ with self._streaming_picamera(pause_stream=True): tf_utils.copy_alsc_section(self.default_tuning, self.tuning) self._initialise_picamera() @lt.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 tf_utils.lst_is_static(self.tuning)