rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'reorganise-picamera-tuning-and-recal' into 'v3'

Browse source 
Allow both HQ camera or PiCamera v2

See merge request openflexure/openflexure-microscope-server!394
This commit is contained in:
Julian Stirling 2025-09-18 14:50:39 +00:00
commit 6d8cb299aa
8 changed files with 399 additions and 302 deletions
Download patch file Download diff file Expand all files Collapse all files

13
hardware-specific-tests/picamera2/test_tuning.py
View file

@ -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:

7
ofm_config_full.json
View file

@ -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",

BIN
picamera_coverage.zip
View file

Binary file not shown.

1
picamera_tests.py
View file

@ -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"),
]

98
src/openflexure_microscope_server/things/camera/picamera.py
View file

@ -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)

449
src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py
View file

@ -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)

128
src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py Normal file
View file

@ -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()

5
tests/test_cameras.py
View file

@ -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()