Source code for platypush.plugins.camera.pi.legacy

import threading

from typing import IO, Optional, List, Tuple, Union

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

from .model import PiCameraInfo, PiCamera


[docs] class CameraPiLegacyPlugin(CameraPlugin): """ Plugin to interact with a `Pi Camera <https://www.raspberrypi.com/documentation/accessories/camera.html>`_. .. warning:: This plugin is **DEPRECATED**, as it relies on the old ``picamera`` module. The ``picamera`` module used in this plugin is deprecated and no longer maintained. The `picamera2 <https://github.com/raspberrypi/picamera2>`_ module is advised instead, which is used by :class:`platypush.plugins.camera.pi.CameraPiPlugin`. You may want to use this plugin if you are running an old OS that does not support the new ``picamera2`` module. Even in that case, you may probably consider using :class:`platypush.plugins.camera.ffmpeg.FfmpegCameraPlugin` or :class:`platypush.plugins.camera.gstreamer.GStreamerCameraPlugin`, as ``picamera`` is not maintained anymore and may not work properly. """ _camera_class = PiCamera _camera_info_class = PiCameraInfo _supported_encoders = ('h264', 'mjpeg')
[docs] def __init__( self, device: int = 0, fps: float = 30.0, warmup_seconds: float = 2.0, sharpness: int = 0, contrast: int = 0, brightness: int = 50, video_stabilization: bool = False, iso: int = 0, exposure_compensation: int = 0, exposure_mode: str = 'auto', meter_mode: str = 'average', awb_mode: str = 'auto', image_effect: str = 'none', led_pin: Optional[int] = None, color_effects: Optional[Union[str, List[str]]] = None, zoom: Tuple[float, float, float, float] = (0.0, 0.0, 1.0, 1.0), 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 an integer between -100 and 100. :param contrast: Contrast level, as an integer between -100 and 100. :param brightness: Brightness level, as an integer between 0 and 100. :param video_stabilization: Enable video stabilization (default: False). :param iso: ISO level (default: 0). :param exposure_compensation: Exposure compensation level, as an integer between -25 and 25. :param exposure_mode: Exposure mode. Allowed values: - ``off`` - ``auto`` (default) - ``night`` - ``nightpreview`` - ``backlight`` - ``spotlight`` - ``sports`` - ``snow`` - ``beach`` - ``verylong`` - ``fixedfps`` - ``antishake`` - ``fireworks`` :param meter_mode: Metering mode used for the exposure. Allowed values: - ``average`` (default) - ``spot`` - ``backlit`` - ``matrix`` :param awb_mode: Auto white balance mode. Allowed values: - ``off`` - ``auto`` (default) - ``sunlight`` - ``cloudy`` - ``shade`` - ``tungsten`` - ``fluorescent`` - ``incandescent`` - ``flash`` - ``horizon`` :param image_effect: Image effect applied to the camera. Allowed values: - ``none`` (default) - ``negative`` - ``solarize`` - ``sketch`` - ``denoise`` - ``emboss`` - ``oilpaint`` - ``hatch`` - ``gpen`` - ``pastel`` - ``watercolor`` - ``film`` - ``blur`` - ``saturation`` - ``colorswap`` - ``washedout`` - ``posterise`` - ``colorpoint`` - ``colorbalance`` - ``cartoon`` - ``deinterlace1`` - ``deinterlace2`` :param led_pin: LED PIN number, if the camera LED is wired to a GPIO PIN and you want to control it. :param zoom: Camera zoom, in the format ``(x, y, width, height)`` (default: ``(0.0, 0.0, 1.0, 1.0)``). :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.video_stabilization = video_stabilization # type: ignore self.camera_info.iso = iso # type: ignore self.camera_info.exposure_compensation = exposure_compensation # type: ignore self.camera_info.meter_mode = meter_mode # type: ignore self.camera_info.exposure_mode = exposure_mode # type: ignore self.camera_info.awb_mode = awb_mode # type: ignore self.camera_info.image_effect = image_effect # type: ignore self.camera_info.color_effects = color_effects # type: ignore self.camera_info.zoom = zoom # type: ignore self.camera_info.led_pin = led_pin # type: ignore
[docs] def prepare_device(self, device: Camera, **_): import picamera # type: ignore assert isinstance(device, PiCamera), f'Invalid camera type: {type(device)}' camera = picamera.PiCamera( camera_num=device.info.device, resolution=device.info.resolution, framerate=device.info.fps, led_pin=device.info.led_pin, ) camera.hflip = device.info.horizontal_flip camera.vflip = device.info.vertical_flip camera.sharpness = device.info.sharpness camera.contrast = device.info.contrast camera.brightness = device.info.brightness camera.video_stabilization = device.info.video_stabilization camera.iso = device.info.iso camera.exposure_compensation = device.info.exposure_compensation camera.exposure_mode = device.info.exposure_mode camera.meter_mode = device.info.meter_mode camera.awb_mode = device.info.awb_mode camera.image_effect = device.info.image_effect camera.color_effects = device.info.color_effects camera.rotation = device.info.rotate or 0 camera.zoom = device.info.zoom return camera
[docs] def release_device(self, device: Camera): import picamera # type: ignore assert isinstance(device, PiCamera), f'Invalid camera type: {type(device)}' if device.object: try: device.object.stop_recording() except (ConnectionError, picamera.PiCameraNotRecording): pass if device.object and not device.object.closed: try: device.object.close() except (ConnectionError, picamera.PiCameraClosed): pass
[docs] def capture_frame(self, device: Camera, *_, **__): import numpy as np from PIL import Image assert device.info.resolution, 'Invalid resolution' assert device.object, 'Camera not opened' shape = ( device.info.resolution[1] + (device.info.resolution[1] % 16), device.info.resolution[0] + (device.info.resolution[0] % 32), 3, ) frame = np.empty(shape, dtype=np.uint8) device.object.capture(frame, 'rgb') return Image.fromarray(frame)
[docs] def start_preview(self, camera: Camera): """ Start camera preview. """ assert camera.object, 'Camera not opened' camera.object.start_preview()
[docs] def stop_preview(self, camera: Camera): """ Stop camera preview. """ if not camera.object: return try: camera.object.stop_preview() except Exception as e: self.logger.warning(str(e))
[docs] @action def capture_preview( self, device: Optional[Union[str, int]] = None, duration: Optional[float] = None, n_frames: Optional[int] = None, **camera, ) -> dict: camera = self.open_device(device=device, **camera) self.start_preview(camera) if n_frames: duration = n_frames * (camera.info.fps or 0) if duration: threading.Timer(duration, lambda: self.stop_preview(camera)) return self.status() # type: ignore
def _streaming_loop(self, camera: Camera, stream_format: str, sock: IO, *_, **__): from picamera import PiCamera as PiCamera_ # type: ignore stream_format = stream_format.lower() assert ( stream_format in self._supported_encoders ), f'Invalid stream format: {stream_format}. Supported formats: {", ".join(self._supported_encoders)}' assert isinstance(camera, PiCamera), f'Invalid camera type: {type(camera)}' assert camera.object and isinstance( camera.object, PiCamera_ ), f'Invalid camera object type: {type(camera.object)}' cam = camera.object try: cam.start_recording(sock, format=stream_format) while not wait_for_either( camera.stop_stream_event, self._should_stop, timeout=1 ): cam.wait_recording(1) except ConnectionError: self.logger.info('Client closed connection') finally: try: cam.stop_recording() self.stop_streaming() except Exception as e: self.logger.warning('Could not stop streaming: %s', e) def _prepare_stream_writer(self, *_, **__): """ Overrides the base method to do nothing - the stream writer is handled by the picamera library. """
# vim:sw=4:ts=4:et: