Source code for platypush.plugins.camera.pi

import os
import time
from typing import IO, Optional, Union

from platypush.plugins import action
from platypush.plugins.camera import CameraPlugin, Camera

from .model import PiCameraInfo, PiCamera


[docs] class CameraPiPlugin(CameraPlugin): """ Plugin to interact with a `Pi Camera <https://www.raspberrypi.com/documentation/accessories/camera.html>`_. This integration is intended to work with the `picamera2 <https://github.com/raspberrypi/picamera2>`_ module. If you are running a very old OS that only provides the deprecated `picamera <https://github.com/waveform80/picamera>`_ module, or you rely on features that are currently only supported by the old module, you should use :class:`platypush.plugins.camera.pi.legacy.CameraPiLegacyPlugin` instead. """ _camera_class = PiCamera _camera_info_class = PiCameraInfo _awb_modes = [ "Auto", "Incandescent", "Tungsten", "Fluorescent", "Indoor", "Daylight", "Cloudy", ]
[docs] def __init__( self, device: int = 0, fps: float = 30.0, warmup_seconds: float = 2.0, sharpness: float = 1.0, contrast: float = 1.0, brightness: float = 0.0, iso: int = 0, exposure_compensation: float = 0.0, awb_mode: str = 'Auto', stream_format: str = 'h264', **camera, ): """ :param device: Camera device number (default: 0). Only supported on devices with multiple camera slots. :param fps: Frames per second (default: 30.0). :param warmup_seconds: Seconds to wait for the camera to warm up before taking a photo (default: 2.0). :param sharpness: Sharpness level, as a float between 0.0 and 16.0, where 1.0 is the default value, and higher values are mapped to higher sharpness levels. :param contrast: Contrast level, as a float between 0.0 and 32.0, where 1.0 is the default value, and higher values are mapped to higher contrast levels. :param brightness: Brightness level, as a float between -1.0 and 1.0. :param iso: ISO level (default: 0). :param exposure_compensation: Exposure compensation level, as a float between -8.0 and 8.0. :param awb_mode: Auto white balance mode. Allowed values: - ``Auto`` (default) - ``Daylight`` - ``Cloudy`` - ``Indoor`` - ``Fluorescent`` :param stream_format: Default format for the output when streamed to a network device. Available: - ``h264`` (default) - ``mjpeg`` :param camera: Options for the base camera plugin (see :class:`platypush.plugins.camera.CameraPlugin`). """ super().__init__( device=device, fps=fps, warmup_seconds=warmup_seconds, stream_format=stream_format, **camera, ) self.camera_info.sharpness = sharpness # type: ignore self.camera_info.contrast = contrast # type: ignore self.camera_info.brightness = brightness # type: ignore self.camera_info.iso = iso # type: ignore self.camera_info.exposure_compensation = exposure_compensation # type: ignore self.camera_info.awb_mode = awb_mode # type: ignore
def _get_transform(self, device: Camera): from libcamera import Orientation, Transform # type: ignore from picamera2.utils import orientation_to_transform # type: ignore rot = device.info.rotate if not rot: return Transform( # It may seem counterintuitive, but the picamera2 library's flip # definition is the opposite of ours hflip=device.info.vertical_flip, vflip=device.info.horizontal_flip, ) if rot == 90: orient = ( Orientation.Rotate90Mirror if device.info.vertical_flip else Orientation.Rotate90 ) elif rot == 180: orient = ( Orientation.Rotate180Mirror if device.info.horizontal_flip else Orientation.Rotate180 ) elif rot == 270: orient = ( Orientation.Rotate270Mirror if device.info.vertical_flip else Orientation.Rotate270 ) else: raise AssertionError( f'Invalid rotation: {rot}. Supported values: 0, 90, 180, 270' ) return orientation_to_transform(orient)
[docs] def prepare_device( self, device: Camera, start: bool = True, video: bool = False, stream: bool = False, **_, ): from picamera2 import Picamera2 # type: ignore assert isinstance(device, PiCamera), f'Invalid device type: {type(device)}' camera = Picamera2(camera_num=device.info.device) still = not (video or stream) cfg_params = { 'main': { 'format': 'XBGR8888' if not still else 'BGR888', **( {'size': tuple(map(int, device.info.resolution))} if device.info.resolution else {} ), }, **( {'transform': self._get_transform(device)} if not still # We don't need to flip the image for individual frames, the base camera # class methods will take care of that else {} ), 'controls': { 'Brightness': float(device.info.brightness), 'Contrast': float(device.info.contrast), 'Sharpness': float(device.info.sharpness), 'AwbMode': self._awb_modes.index(device.info.awb_mode), }, } cfg = ( camera.create_video_configuration if not still else camera.create_still_configuration )(**cfg_params) camera.configure(cfg) if start: camera.start() time.sleep(max(1, device.info.warmup_seconds)) return camera
[docs] def release_device(self, device: Camera): if device.object: device.object.stop() device.object.close()
[docs] def capture_frame(self, device: Camera, *_, **__): assert device.object, 'Camera not open' return device.object.capture_image('main')
@property def _video_encoders_by_format(self) -> dict: from picamera2.encoders import H264Encoder, MJPEGEncoder # type: ignore return { 'h264': H264Encoder, 'mjpeg': MJPEGEncoder, }
[docs] @action def capture_video( self, device: Optional[int] = None, duration: Optional[float] = None, video_file: Optional[str] = None, preview: bool = False, **camera, ) -> Optional[Union[str, dict]]: """ Capture a video. :param device: 0-based index of the camera to capture from, if the device supports multiple cameras. Default: use the configured camera index or the first available camera. :param duration: Record duration in seconds (default: None, record until :meth:`.stop_capture``). :param video_file: If set, the stream will be recorded to the specified video file (default: None). :param camera: Camera parameters override - see constructors parameters. :param preview: Show a preview of the camera frames. :return: If duration is specified, the method will wait until the recording is done and return the local path to the recorded resource. Otherwise, it will return the status of the camera device after starting it. """ from picamera2 import Picamera2 # type: ignore from picamera2.encoders import H264Encoder # type: ignore assert video_file, 'Video file is required' camera = self.open_device( device=device, ctx={'start': False, 'video': True}, **camera ) encoder = H264Encoder() assert camera.object, 'Camera not open' assert isinstance( camera.object, Picamera2 ), f'Invalid camera object type: {type(camera.object)}' if preview: camera.object.start_preview() # Only H264 is supported for now camera.object.start_recording(encoder, os.path.expanduser(video_file)) if duration: self.wait_stop(duration) try: if preview: camera.object.stop_preview() finally: if camera.object: camera.object.stop_recording() camera.object.close() return video_file return self.status(camera.info.device).output
def _streaming_loop(self, camera: Camera, stream_format: str, sock: IO, *_, **__): from picamera2 import Picamera2 # type: ignore from picamera2.outputs import FileOutput # type: ignore encoder_cls = self._video_encoders_by_format.get(stream_format.lower()) assert ( encoder_cls ), f'Invalid stream format: {stream_format}. Supported formats: {", ".join(self._video_encoders_by_format)}' assert isinstance(camera, PiCamera), f'Invalid camera type: {type(camera)}' assert camera.object and isinstance( camera.object, Picamera2 ), f'Invalid camera object type: {type(camera.object)}' cam = camera.object encoder = encoder_cls() cam.encoders = encoder encoder.output = FileOutput(sock) cam.start_encoder(encoder) cam.start() def _prepare_stream_writer(self, *_, **__): """ Overrides the base method to do nothing - the stream writer is handled by the picamera2 library. """ def _cleanup_stream(self, camera: Camera, *_, **__): cam = camera.object if not cam: return cam.stop() cam.stop_encoder() cam.close()
# vim:sw=4:ts=4:et: