diff --git a/hardware-specific-tests/picamera2/test_tuning.py b/hardware-specific-tests/picamera2/test_tuning.py index 35a67676..b193131d 100644 --- a/hardware-specific-tests/picamera2/test_tuning.py +++ b/hardware-specific-tests/picamera2/test_tuning.py @@ -8,20 +8,17 @@ from picamera2 import Picamera2 from openflexure_microscope_server.things.camera import ( picamera_recalibrate_utils as recalibrate_utils, ) +from openflexure_microscope_server.things.camera import ( + picamera_tuning_file_utils as tf_utils, +) MODEL = Picamera2.global_camera_info()[0]["Model"] -def load_default_tuning(): - """Return the default tuning file for the connected camera.""" - fname = f"{MODEL}.json" - return Picamera2.load_tuning_file(fname) - - def generate_bad_tuning(): """Return a tuning file with an invalid version number to force an error when loaded.""" - default_tuning = load_default_tuning() + default_tuning = tf_utils.load_default_tuning("imx219") bad_tuning = default_tuning.copy() bad_tuning["version"] = 999 return bad_tuning @@ -58,7 +55,7 @@ def _test_bad_tuning_after_good_tuning(configure: bool = False): PiCamera2 behaviour does not expect the tuning file to be reloaded. """ bad_tuning = generate_bad_tuning() - default_tuning = load_default_tuning() + default_tuning = tf_utils.load_default_tuning("imx219") print_tuning() print("opening camera with default tuning") with Picamera2(tuning=default_tuning) as cam: diff --git a/ofm_config_full.json b/ofm_config_full.json index 7f4c4077..60b85a66 100644 --- a/ofm_config_full.json +++ b/ofm_config_full.json @@ -1,6 +1,11 @@ { "things": { - "/camera/": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2", + "/camera/": { + "class": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2", + "kwargs": { + "camera_board": "picamera_v2" + } + }, "/stage/": "openflexure_microscope_server.things.stage.sangaboard:SangaboardThing", "/autofocus/": "openflexure_microscope_server.things.autofocus:AutofocusThing", "/camera_stage_mapping/": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper", diff --git a/picamera_coverage.zip b/picamera_coverage.zip index 42e05dcf..d20a51c0 100644 Binary files a/picamera_coverage.zip and b/picamera_coverage.zip differ diff --git a/picamera_tests.py b/picamera_tests.py index a1bd38d0..892491eb 100755 --- a/picamera_tests.py +++ b/picamera_tests.py @@ -21,6 +21,7 @@ HASHED_FILES = [ os.path.join(CAM_DIR, "__init__.py"), os.path.join(CAM_DIR, "picamera.py"), os.path.join(CAM_DIR, "picamera_recalibrate_utils.py"), + os.path.join(CAM_DIR, "picamera_tuning_file_utils.py"), ] diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py index e8ab2e2f..be01b337 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -43,8 +43,19 @@ from openflexure_microscope_server.ui import ( property_control_for, ) from . import picamera_recalibrate_utils as recalibrate_utils +from . import picamera_tuning_file_utils as tf_utils + from . import BaseCamera, ArrayModel +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.""" @@ -128,24 +139,34 @@ class StreamingPiCamera2(BaseCamera): generalisation. """ - def __init__(self, camera_num: int = 0) -> None: + 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_configs: dict[str, dict] = {} + 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 - 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.") + + # 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: @@ -354,11 +375,18 @@ class StreamingPiCamera2(BaseCamera): ) from e return None - def _initialise_picamera(self) -> 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 @@ -381,9 +409,16 @@ class StreamingPiCamera2(BaseCamera): logging.info("Creating new Picamera2 object") # Specify tuning file otherwise it will be overwritten with None. self._picamera = Picamera2( - camera_num=self.camera_num, + 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: @@ -392,7 +427,7 @@ class StreamingPiCamera2(BaseCamera): This opens the picamera connection, initialises the camera, sets the sensor_modes property, and then starts the streams. """ - self._initialise_picamera() + 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() @@ -529,7 +564,7 @@ class StreamingPiCamera2(BaseCamera): logging.info("Stopped MJPEG stream.") # Adding a sleep to prevent camera getting confused by rapid commands - time.sleep(0.2) + time.sleep(self._sensor_info.short_pause) @lt.thing_action def discard_frames(self) -> None: @@ -547,7 +582,7 @@ class StreamingPiCamera2(BaseCamera): logging.debug("Reconfiguring camera for full resolution capture") cam.configure(cam.create_still_configuration(sensor=self._sensor_mode)) cam.start() - time.sleep(0.2) + time.sleep(self._sensor_info.short_pause) yield cam def capture_image( @@ -646,7 +681,7 @@ class StreamingPiCamera2(BaseCamera): @lt.thing_action def auto_expose_from_minimum( self, - target_white_level: int = 700, + target_white_level: Optional[int] = None, percentile: float = 99.9, ) -> None: """Adjust exposure until a the target white level is reached. @@ -654,17 +689,21 @@ class StreamingPiCamera2(BaseCamera): Starting from the minimum exposure, gradually increase exposure until the image reaches the specified white level. - :param target_white_level: The target 10bit white level. 10-bit data has a - theoretical maximum of 1023, but with black level correction the true - maximum is about 950. Default is 700 as this is approximately 70% - saturated. + :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, ) @@ -691,6 +730,7 @@ class StreamingPiCamera2(BaseCamera): 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, @@ -700,7 +740,7 @@ class StreamingPiCamera2(BaseCamera): ) else: recalibrate_utils.adjust_white_balance_from_raw( - cam, percentile=99, method=method + cam, self._sensor_info, percentile=99, method=method ) @lt.thing_action @@ -717,8 +757,8 @@ class StreamingPiCamera2(BaseCamera): # 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) # noqa: N806 - recalibrate_utils.set_static_lst(self.tuning, L, Cr, 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 @@ -735,14 +775,14 @@ class StreamingPiCamera2(BaseCamera): See page Raspberry Pi Camera Algorithm and Tuning Guide, page 45. """ - return tuple(recalibrate_utils.get_static_ccm(self.tuning)[0]["ccm"]) + 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: - recalibrate_utils.set_static_ccm(self.tuning, value) + tf_utils.set_static_ccm(self.tuning, value) if self._picamera is not None: with self._streaming_picamera(pause_stream=True): @@ -781,7 +821,7 @@ class StreamingPiCamera2(BaseCamera): A value of 0 here does nothing, a value of 65535 is maximum correction. """ with self._streaming_picamera(pause_stream=True): - recalibrate_utils.set_static_geq(self.tuning, offset) + tf_utils.set_static_geq(self.tuning, offset) self._initialise_picamera() @lt.thing_action @@ -820,9 +860,7 @@ class StreamingPiCamera2(BaseCamera): with self._streaming_picamera(pause_stream=True): # Generate and array of ones of the correct size for each channel flat_array = np.ones((12, 16)) - recalibrate_utils.set_static_lst( - self.tuning, flat_array, flat_array, flat_array - ) + tf_utils.set_static_lst(self.tuning, flat_array, flat_array, flat_array) self._initialise_picamera() @lt.thing_property @@ -945,7 +983,7 @@ class StreamingPiCamera2(BaseCamera): def lens_shading_tables(self, lst: LensShading) -> None: """Set the lens shading tables.""" with self._streaming_picamera(pause_stream=True): - recalibrate_utils.set_static_lst( + tf_utils.set_static_lst( self.tuning, luminance=lst.luminance, cr=lst.Cr, @@ -996,7 +1034,7 @@ class StreamingPiCamera2(BaseCamera): alsc = self.get_tuning_algo("rpi.alsc") luminance = alsc["luminance_lut"] flat = np.ones((12, 16)) - recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat) + tf_utils.set_static_lst(self.tuning, luminance, flat, flat) self._initialise_picamera() @lt.thing_action @@ -1007,7 +1045,7 @@ class StreamingPiCamera2(BaseCamera): by the Raspberry Pi camera. """ with self._streaming_picamera(pause_stream=True): - recalibrate_utils.copy_alsc_section(self.default_tuning, self.tuning) + tf_utils.copy_alsc_section(self.default_tuning, self.tuning) self._initialise_picamera() @lt.thing_property @@ -1019,4 +1057,4 @@ class StreamingPiCamera2(BaseCamera): 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) + return tf_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 index 44f269f6..5f38ee6e 100644 --- a/src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py +++ b/src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py @@ -1,4 +1,4 @@ -"""Functions to set up a Raspberry Pi Camera v2 for scientific use. +"""Functions to set up a Raspberry Pi Camera (v2 and HQ) for scientific use. This module provides slower, simpler functions to set the gain, exposure, and white balance of a Raspberry Pi camera, using @@ -22,10 +22,15 @@ reliable. The three steps above can be accomplished by: .. code-block:: python picamera = picamera2.Picamera2() + sensor_info = IMX219_SENSOR_INFO - adjust_shutter_and_gain_from_raw(picamera) - adjust_white_balance_from_raw(picamera) - lst = lst_from_camera(picamera) + adjust_shutter_and_gain_from_raw( + picamera, + sensor_info, + target_white_level=sensor_info.default_target_white_level, + ) + adjust_white_balance_from_raw(picamera, sensor_info) + lst = lst_from_camera(picamera, sensor_info) picamera.lens_shading_table = lst """ @@ -48,101 +53,59 @@ from picamera2 import Picamera2 import picamera2 +class SensorInfo(BaseModel): + """Information about the sensor used for calibration and property setting.""" + + sensor_model: str + """The model of the sensor, as specified by the Picamera2 library.""" + + unpacked_pixel_format: str + """The format of the unpacked pixels.""" + + bit_depth: int + """The bit depth of each pixel.""" + + blacklevel: int + """The sensor black level.""" + + default_target_white_level: int + """The default target white level during exposure setting.""" + + short_pause: float + """The time to pause for actions that update quickly.""" + + long_pause: float + """Time to pause for actions that are known to update slowly.""" + + +IMX219_SENSOR_INFO = SensorInfo( + sensor_model="imx219", + unpacked_pixel_format="SBGGR10", + bit_depth=10, + blacklevel=64, + default_target_white_level=700, + short_pause=0.2, + long_pause=0.5, +) + +IMX477_SENSOR_INFO = SensorInfo( + sensor_model="imx477", + unpacked_pixel_format="SBGGR12", + bit_depth=12, + blacklevel=256, + default_target_white_level=2800, + short_pause=0.2, + long_pause=1.0, +) + + LensShadingTables = tuple[np.ndarray, np.ndarray, np.ndarray] -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: - tuning_dir = "/usr/share/libcamera/ipa/raspberrypi" - # from picamera2 v0.3.9 - # The directory above has been removed from the search path seems - # 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=tuning_dir) - - -def set_minimum_exposure(camera: Picamera2) -> None: - """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) - - Note ISO is left at auto, because this is needed for the gains - to be set correctly. - """ - # Disable Automatic exposure and gain algorithm (AeEnable), and set analogue - # gain and exposure time. - # Setting the shutter speed to 1us will result in it being set - # to the minimum possible, which is ~8us for PiCamera v2 - camera.set_controls({"AeEnable": False, "AnalogueGain": 1, "ExposureTime": 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) -> bool: - """Check whether the brightness is within the specified target range.""" - return abs(test.level - target) < target * tolerance - - def adjust_shutter_and_gain_from_raw( camera: Picamera2, - target_white_level: int = 700, + sensor_info: SensorInfo, + target_white_level: int, max_iterations: int = 20, tolerance: float = 0.05, percentile: float = 99.9, @@ -153,9 +116,12 @@ def adjust_shutter_and_gain_from_raw( are not affected by white balance or digital gain. :param camera: A Picamera2 object. - :param 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. + :param target_white_level: The raw value we aim for, the raw value of the brightest + pixels should be approximately this bright. The value to set depends on the + sensor bit depth. We recommend values of 700 for 10-bit sensors and 2800 for + 12-bit sensors. This is about 70% of saturated once the blacklevel is + subtracted. The maximum possible value depends on the sensor bit depth, the + sensor blacklevel and the tolerance argument. :param max_iterations: We will terminate once we perform this many iterations, whether or not we converge. More than 10 shouldn't happen. :param tolerance: How close to the target value we consider "done". Expressed as a @@ -166,26 +132,29 @@ def adjust_shutter_and_gain_from_raw( than just ``np.max()``. """ - # TODO: read black level and bit depth from camera? - if target_white_level * (tolerance + 1) >= 959: + # Calculate the maximum possible pixel value once blacklevel is subtracted. + max_level = 2**sensor_info.bit_depth - 1 - sensor_info.blacklevel + if target_white_level * (tolerance + 1) >= max_level: 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." + f"must be less than {max_level}." ) - config = camera.create_still_configuration(raw={"format": "SBGGR10"}) + config = camera.create_still_configuration( + raw={"format": sensor_info.unpacked_pixel_format} + ) camera.configure(config) camera.start() - set_minimum_exposure(camera) + _set_minimum_exposure(camera, sensor_info) # 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): + test = _test_exposure_settings(camera, percentile) + if _check_convergence(test, target_white_level, tolerance): break iterations += 1 @@ -194,7 +163,7 @@ def adjust_shutter_and_gain_from_raw( 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) + time.sleep(sensor_info.long_pause) # Check whether the shutter speed is still going up - if not, we've hit a maximum if camera.capture_metadata()["ExposureTime"] == test.exposure_time: @@ -203,8 +172,8 @@ def adjust_shutter_and_gain_from_raw( # 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): + test = _test_exposure_settings(camera, percentile) + if _check_convergence(test, target_white_level, tolerance): break iterations += 1 @@ -212,14 +181,14 @@ def adjust_shutter_and_gain_from_raw( camera.controls.AnalogueGain = test.analog_gain * min( target_white_level / test.level, 2 ) - time.sleep(0.5) + time.sleep(sensor_info.long_pause) # 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): + if _check_convergence(test, target_white_level, tolerance): logging.info(f"Brightness has converged to within {tolerance * 100:.0f}%.") else: logging.warning( @@ -232,6 +201,7 @@ def adjust_shutter_and_gain_from_raw( def adjust_white_balance_from_raw( camera: Picamera2, + sensor_info: SensorInfo, percentile: float = 99, luminance: Optional[np.ndarray] = None, Cr: Optional[np.ndarray] = None, @@ -245,20 +215,21 @@ def adjust_white_balance_from_raw( We should probably have better logic to verify the channels really are BGGR... """ - config = camera.create_still_configuration(raw={"format": "SBGGR10"}) + config = camera.create_still_configuration( + raw={"format": sensor_info.unpacked_pixel_format} + ) camera.configure(config) camera.start() - channels = channels_from_bayer_array(camera.capture_array("raw")) - # TODO: read black level from camera rather than hard-coding 64 - blacklevel = 64 + channels = _channels_from_bayer_array(camera.capture_array("raw")) + 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) + 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:]) + channel_gains = _upsample_channels(channel_gains, channels.shape[1:]) logging.info( f"Before gains, channel maxima are {np.max(channels, axis=(1, 2))}" ) @@ -277,10 +248,10 @@ def adjust_white_balance_from_raw( axis=(1, 2), ) # Subtract blacklevel before splitting into channels - blue, g1, g2, red = centre_means - blacklevel + blue, g1, g2, red = centre_means - sensor_info.blacklevel else: blue, g1, g2, red = ( - np.percentile(channels, percentile, axis=(1, 2)) - blacklevel + np.percentile(channels, percentile, axis=(1, 2)) - sensor_info.blacklevel ) green = (g1 + g2) / 2.0 new_awb_gains = (green / red, green / blue) @@ -297,13 +268,94 @@ def adjust_white_balance_from_raw( ) camera.controls.AwbEnable = False camera.controls.ColourGains = new_awb_gains - time.sleep(0.2) + time.sleep(sensor_info.long_pause) 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: +def lst_from_camera(camera: Picamera2, sensor_info: SensorInfo) -> LensShadingTables: + """Acquire a raw image and use it to calculate a lens shading table.""" + channels = _raw_channels_from_camera(camera, sensor_info) + return _lst_from_channels(channels, sensor_info.blacklevel) + + +def recreate_camera_manager() -> None: + """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() + + +class _ExposureTest(BaseModel): + """Record the results of testing the camera's current exposure settings.""" + + level: int + exposure_time: int + analog_gain: float + + +def _set_minimum_exposure(camera: Picamera2, sensor_info: SensorInfo) -> None: + """Enable manual exposure, with low gain and shutter speed. + + Set exposure mode to manual, analog and digital gain to 1, and + shutter speed to the minimum (8us for Pi Camera v2) + + Note ISO is left at auto, because this is needed for the gains + to be set correctly. + """ + # Disable Automatic exposure and gain algorithm (AeEnable), and set analogue + # gain and exposure time. + # Setting the shutter speed to 1us will result in it being set + # to the minimum possible, which is ~8us for PiCamera v2 + camera.set_controls({"AeEnable": False, "AnalogueGain": 1, "ExposureTime": 1}) + time.sleep(sensor_info.long_pause) + + +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) -> bool: + """Check whether the brightness is within the specified target range.""" + return abs(test.level - target) < target * tolerance + + +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) @@ -320,7 +372,7 @@ def channels_from_bayer_array(bayer_array: np.ndarray) -> np.ndarray: return channels -def get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray: +def _get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray: """Compresses channel down to a 16x12 grid - from libcamera. This is taken from @@ -342,10 +394,10 @@ def get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray: return np.reshape(np.array(grid), (12, 16)) -def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray: +def _upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray: """Zoom an image in the last two dimensions. - This is effectively the inverse operation of ``get_16x12_grid`` + This is effectively the inverse operation of ``_get_16x12_grid`` """ zoom_factors = [ 1, @@ -353,19 +405,14 @@ def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray: return zoom(grids, zoom_factors, order=1)[:, : shape[0], : shape[1]] -def downsampled_channels( - channels: np.ndarray, blacklevel: int = 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... - """ +def _downsampled_channels(channels: np.ndarray, blacklevel: int) -> list[np.ndarray]: + """Generate a downsampled, un-normalised image from which to calculate the LST.""" 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( + _get_16x12_grid( channels[i, ...].astype(float) - blacklevel, step[1], step[0] ) for i in range(channels.shape[0]) @@ -374,16 +421,16 @@ def downsampled_channels( ) -def lst_from_channels(channels: np.ndarray) -> LensShadingTables: +def _lst_from_channels(channels: np.ndarray, blacklevel: int) -> LensShadingTables: """Given the 4 Bayer colour channels from a white image, generate a LST. - Internally, is just calls ``downsampled_channels`` and ``lst_from_grids``. + Internally, is just calls ``_downsampled_channels`` and ``_lst_from_grids``. """ - grids = downsampled_channels(channels) - return lst_from_grids(grids) + grids = _downsampled_channels(channels, blacklevel) + return _lst_from_grids(grids) -def lst_from_grids(grids: np.ndarray) -> LensShadingTables: +def _lst_from_grids(grids: np.ndarray) -> LensShadingTables: """Given 4 downsampled grids, generate the luminance and chrominance tables. The grids are the 4 BAYER channels RGGB @@ -409,7 +456,7 @@ def lst_from_grids(grids: np.ndarray) -> LensShadingTables: return luminance_gains, cr_gains, cb_gains -def grids_from_lst(lum: np.ndarray, Cr: np.ndarray, Cb: np.ndarray) -> np.ndarray: +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. @@ -423,118 +470,17 @@ def grids_from_lst(lum: np.ndarray, Cr: np.ndarray, Cb: np.ndarray) -> np.ndarra 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 correction. - - ``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": as_flat_rounded_list(cr, round_to=3)} - ] - alsc["calibrations_Cb"] = [ - {"ct": 4500, "table": as_flat_rounded_list(cb, round_to=3)} - ] - alsc["luminance_lut"] = as_flat_rounded_list(luminance, round_to=3) - - -def set_static_ccm( - tuning: dict, - col_corr_matrix: tuple[ - float, float, float, float, float, float, float, float, float - ], -) -> None: - """Update the ``rpi.alsc`` section of a camera tuning dict to use a static correction. - - ``tuning`` will be updated in-place to set its shading to static, and disable any - adaptive tweaking by the algorithm. - """ - ccm = Picamera2.find_tuning_algo(tuning, "rpi.ccm") - ccm["ccms"] = [{"ct": 2860, "ccm": col_corr_matrix}] - - -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. - - :param tuning: the raspberry pi tuning file. This will be updated in-place to - set the geq offset to the given value. - :param offset: The desired green equalisation offset. Default 65535. The default is - the maximum allowed value. This means the brightness will always be below the - threshold where averaging is used. This is default as we always need the green - equalisation to averages the green pixels in the red and blue rows due to the - chief ray angle compensation issue when the the stock lens is replaced by an - objective. - """ - geq = Picamera2.find_tuning_algo(tuning, "rpi.geq") - geq["offset"] = offset # max out offset to disable the adaptive green equalisation - - -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) -> int: - """Find the index of an algorithm's section in the tuning file.""" - for i, a in enumerate(algorithms): - if algorithm in a: - return i - raise ValueError(f"Algorithm {algorithm} is not available.") - - -def copy_alsc_section(from_tuning: dict, to_tuning: dict) -> None: - """Copy the ``rpi.alsc`` algorithm from one tuning to another. - - This is done in-place, i.e. modifying to_tuning. - """ - # Using Picamera2 function to find the relevant sub-dict for each tuning file - from_i = index_of_algorithm(from_tuning["algorithms"], "rpi.alsc") - to_i = index_of_algorithm(to_tuning["algorithms"], "rpi.alsc") - # Updating the dictionary in place. - 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: +def _raw_channels_from_camera( + camera: Picamera2, sensor_info: SensorInfo +) -> 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"}) + config = camera.create_still_configuration( + raw={"format": sensor_info.unpacked_pixel_format} + ) camera.configure(config) camera.start() raw_image = camera.capture_array("raw") @@ -545,19 +491,4 @@ def raw_channels_from_camera(camera: Picamera2) -> LensShadingTables: # channels, 1/2 for green because there's twice as many green pixels). raw_format = camera.camera_configuration()["raw"]["format"] print(f"Acquired a raw image in format {raw_format}") - return channels_from_bayer_array(raw_image) - - -def recreate_camera_manager() -> None: - """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() - - -def as_flat_rounded_list(array: np.ndarray, round_to: int = 3) -> list[float]: - """Flatten array, round, and then convert to list.""" - return np.reshape(array, -1).round(round_to).tolist() + return _channels_from_bayer_array(raw_image) diff --git a/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py b/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py new file mode 100644 index 00000000..00e5a0d3 --- /dev/null +++ b/src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py @@ -0,0 +1,128 @@ +"""Functions for loading, adjusting, or reading from the Picamera2 tuning file. + +The functions that edit the tuning files edit them in place. This will change in +the future. +""" + +from picamera2 import Picamera2 +import numpy as np + + +def load_default_tuning(sensor_model: str) -> dict: + """Load the default tuning file for the camera. + + This will loat the tuning file based on the specified sensor model. + """ + fname = f"{sensor_model}.json" + try: + return Picamera2.load_tuning_file(fname) + except RuntimeError: + tuning_dir = "/usr/share/libcamera/ipa/raspberrypi" + # from picamera2 v0.3.9 + # The directory above has been removed from the search path seems + # 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 Picamera2.load_tuning_file(fname, dir=tuning_dir) + + +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 correction. + + ``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": _as_flat_rounded_list(cr, round_to=3)} + ] + alsc["calibrations_Cb"] = [ + {"ct": 4500, "table": _as_flat_rounded_list(cb, round_to=3)} + ] + alsc["luminance_lut"] = _as_flat_rounded_list(luminance, round_to=3) + + +def set_static_ccm( + tuning: dict, + col_corr_matrix: tuple[ + float, float, float, float, float, float, float, float, float + ], +) -> None: + """Update the ``rpi.alsc`` section of a camera tuning dict to use a static correction. + + ``tuning`` will be updated in-place to set its shading to static, and disable any + adaptive tweaking by the algorithm. + """ + ccm = Picamera2.find_tuning_algo(tuning, "rpi.ccm") + ccm["ccms"] = [{"ct": 2860, "ccm": col_corr_matrix}] + + +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. + + :param tuning: the raspberry pi tuning file. This will be updated in-place to + set the geq offset to the given value. + :param offset: The desired green equalisation offset. Default 65535. The default is + the maximum allowed value. This means the brightness will always be below the + threshold where averaging is used. This is default as we always need the green + equalisation to averages the green pixels in the red and blue rows due to the + chief ray angle compensation issue when the the stock lens is replaced by an + objective. + """ + geq = Picamera2.find_tuning_algo(tuning, "rpi.geq") + geq["offset"] = offset # max out offset to disable the adaptive green equalisation + + +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 copy_alsc_section(from_tuning: dict, to_tuning: dict) -> None: + """Copy the ``rpi.alsc`` algorithm from one tuning to another. + + This is done in-place, i.e. modifying to_tuning. + """ + # Using Picamera2 function to find the relevant sub-dict for each tuning file + from_i = _index_of_algorithm(from_tuning["algorithms"], "rpi.alsc") + to_i = _index_of_algorithm(to_tuning["algorithms"], "rpi.alsc") + # Updating the dictionary in place. + to_tuning["algorithms"][to_i] = from_tuning["algorithms"][from_i] + + +def _index_of_algorithm(algorithms: list[dict], algorithm: str) -> int: + """Find the index of an algorithm's section in the tuning file.""" + for i, a in enumerate(algorithms): + if algorithm in a: + return i + raise ValueError(f"Algorithm {algorithm} is not available.") + + +def _as_flat_rounded_list(array: np.ndarray, round_to: int = 3) -> list[float]: + """Flatten array, round, and then convert to list.""" + return np.reshape(array, -1).round(round_to).tolist() diff --git a/tests/test_cameras.py b/tests/test_cameras.py index 06d282bf..27de7bd4 100644 --- a/tests/test_cameras.py +++ b/tests/test_cameras.py @@ -27,10 +27,7 @@ def mock_picam_thing(mocker): "picamera2.outputs": mocker.Mock(), }, ) - mocker.patch( - "openflexure_microscope_server.things.camera.picamera.recalibrate_utils.load_default_tuning", - return_value={"mock": "tuning"}, - ) + from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2 return StreamingPiCamera2()