Smarter application of picamera settings (ea253f80) · Commits · OpenFlexure / openflexure-microscope-server

openflexure_microscope/camera/pi.py

+67 −73

Original line number	Diff line number	Diff line
		@@ -12,7 +12,7 @@ Video port:
		Splitter port 2: Video capture
		Splitter port 3: [Currently unused]

		StreamingCamera streams at stream_resolution
		StreamingCamera streams at video_resolution
		Camera capture resolution set to stream_resolution in frames()
		Video port uses that resolution for everything. If a different resolution
		is specified for video capture, this is handled by the resizer.
		@@ -39,29 +39,21 @@ from .base import BaseCamera, CaptureObject
		# Richard's fix gain
		from .set_picamera_gain import set_analog_gain, set_digital_gain

		# Handle config and picamera settings
		CONFIG_KEYS = {
		'stream_resolution': tuple,
		'image_resolution': tuple,
		'numpy_resolution': tuple,
		'analog_gain': float,
		'digital_gain': float,
		'jpeg_quality': int,
		'picamera_settings': {
		'exposure_mode': str,
		'awb_mode': str,
		'awb_gains': tuple,
		'framerate': int,
		'shutter_speed': float,
		'saturation': float,
		'lens_shading_table': np.ndarray,
		},
		}


		# MAIN CLASS
		class StreamingCamera(BaseCamera):
		"""Raspberry Pi camera implementation of StreamingCamera."""
		picamera_settings_keys = [
		'exposure_mode',
		'analog_gain',
		'digital_gain',
		'shutter_speed',
		'awb_gains',
		'awb_mode',
		'framerate',
		'saturation',
		'lens_shading_table'
		]

		def __init__(self):
		# Run BaseCamera init
		@@ -103,61 +95,28 @@ class StreamingCamera(BaseCamera):
		self.camera.close()

		# HANDLE SETTINGS
		@property
		def supports_lens_shading(self):
		"""Determine whether the picamera module supports lens shading.

		As of March 2018, picamera did not wrap the necessary MMAL commands to
		set the lens shading table, or to write the value of analog or digital
		gain. I have a forked version of the library that does support these.
		For ease of use by people who don't want those features, this library
		does not have a hard dependency on lens shading. However, we need to
		check in some places whether it's available.
		"""
		return hasattr(self.camera, "lens_shading_table")

		@property
		def digital_gain(self):
		# TODO: Handle missing picamera?
		return self.camera.digital_gain

		@digital_gain.setter
		def digital_gain(self, value):
		set_digital_gain(self.camera, value)

		@property
		def analog_gain(self):
		# TODO: Handle missing picamera?
		return self.camera.analog_gain

		@analog_gain.setter
		def analog_gain(self, value):
		set_analog_gain(self.camera, value)

		def read_config(self):
		"""
		Return config dictionary of the StreamingCamera.
		"""
		global CONFIG_KEYS

		conf_dict = {
		'stream_resolution': self.stream_resolution,
		'image_resolution': self.image_resolution,
		'numpy_resolution': self.numpy_resolution,
		'jpeg_quality': self.jpeg_quality,
		'picamera_settings': {},
		}

		# PiCamera parameters
		for key in CONFIG_KEYS['picamera_settings']:
		for key in StreamingCamera.picamera_settings_keys:
		try:
		value = getattr(self.camera, key)
		logging.debug("{}: {}".format(key, value))
		logging.debug("Reading PiCamera().{}: {}".format(key, value))
		conf_dict['picamera_settings'][key] = value
		except AttributeError:
		logging.debug("Unable to read PiCamera attribute {}".format(key))

		# StreamingCamera parameters
		for key in CONFIG_KEYS:
		if (key != 'picamera_settings') and hasattr(self, key):
		conf_dict[key] = getattr(self, key)

		return conf_dict

		def apply_config(self, config: dict) -> None:
		@@ -170,11 +129,10 @@ class StreamingCamera(BaseCamera):
		Args:
		config (dict): Dictionary of config parameters.
		"""
		global CONFIG_KEYS
		# TODO: Include timing and batching logic when applying PiCamera settings

		paused_stream = False
		logging.debug("Applying config:")
		logging.debug("StreamingCamera: Applying config:")
		logging.debug(config)

		with self.lock:
		@@ -182,9 +140,6 @@ class StreamingCamera(BaseCamera):
		# Apply valid config params to Picamera object
		if not self.state['record_active']: # If not recording a video

		logging.debug("Applying config:")
		logging.debug(config)

		# Pause stream while changing settings
		if self.state['stream_active']: # If stream is active
		logging.info("Pausing stream to update config.")
		@@ -193,19 +148,11 @@ class StreamingCamera(BaseCamera):

		# PiCamera parameters
		if 'picamera_settings' in config: # If new settings are given
		for key, value in config['picamera_settings'].items(): # For each given setting
		if hasattr(self.camera, key):
		if key not in CONFIG_KEYS['picamera_settings']:
		CONFIG_KEYS['picamera_settings'][key] = type(value)
		logging.debug("Setting parameter {}: {}".format(key, config['picamera_settings'][key]))
		setattr(self.camera, key, value) # Write setting to camera
		self.apply_picamera_settings(config['picamera_settings'], pause_for_effect=True)

		# StreamingCamera parameters
		for key, value in config.items(): # For each provided setting
		if (key != 'picamera_settings') and hasattr(self, key):
		if key not in CONFIG_KEYS:
		CONFIG_KEYS[key] = type(value)
		logging.debug("Setting parameter {}: {}".format(key, value))
		setattr(self, key, value)

		# If stream was paused to update config, unpause
		@@ -217,6 +164,53 @@ class StreamingCamera(BaseCamera):
		raise Exception(
		"Cannot update camera config while recording is active.")

		def apply_picamera_settings(self, settings_dict: dict, pause_for_effect: bool=True):
		# Set exposure mode
		if 'exposure_mode' in settings_dict:
		logging.debug("Applying exposure_mode: {}".format(settings_dict['exposure_mode']))
		self.camera.exposure_mode = settings_dict['exposure_mode']

		# Apply gains and let them settle
		if 'analog_gain' in settings_dict:
		logging.debug("Applying analog_gain: {}".format(settings_dict['analog_gain']))
		set_analog_gain(self.camera, settings_dict['analog_gain'])
		if 'digital_gain' in settings_dict:
		logging.debug("Applying digital_gain: {}".format(settings_dict['digital_gain']))
		set_digital_gain(self.camera, settings_dict['digital_gain'])

		# Apply shutter speed
		if 'shutter_speed' in settings_dict:
		logging.debug("Applying shutter_speed: {}".format(settings_dict['shutter_speed']))
		self.camera.shutter_speed = settings_dict['shutter_speed']

		time.sleep(0.2) # Let gains settle

		# Handle AWB in a half-smart way
		if 'awb_gains' in settings_dict:
		logging.debug("Applying awb_mode: off")
		self.camera.awb_mode = 'off'
		logging.debug("Applying awb_gains: {}".format(settings_dict['awb_gains']))
		self.camera.awb_gains = settings_dict['awb_gains']
		elif 'awb_mode' in settings_dict:
		logging.debug("Applying awb_mode: {}".format(settings_dict['awb_mode']))
		self.camera.awb_mode = settings_dict['awb_mode']

		# Handle some properties that can be quickly applied
		batched_keys = ['framerate', 'saturation']
		for key in batched_keys:
		if (key in settings_dict) and hasattr(self.camera, key):
		logging.debug("Applying {}: {}".format(key, settings_dict[key]))
		setattr(self.camera, key, settings_dict[key])

		# Handle lens shading if camera supports it
		if ('lens_shading_table' in settings_dict) and hasattr(self.camera, 'lens_shading_table'):
		logging.debug("Applying lens_shading_table: {}".format(settings_dict['lens_shading_table']))
		self.camera.lens_shading_table = settings_dict['lens_shading_table']

		# Final optional pause to settle
		if pause_for_effect:
		time.sleep(0.2)

		def set_zoom(self, zoom_value: float = 1.) -> None:
		"""
		Change the camera zoom, handling re-centering and scaling.