<s> from __future__ import ( <EOL> unicode_literals , <EOL> print_function , <EOL> division , <EOL> absolute_import , <EOL> ) <EOL> str = type ( '<STR_LIT>' ) <EOL> try : <EOL> range = xrange <EOL> except NameError : <EOL> pass <EOL> import io <EOL> import datetime <EOL> import threading <EOL> import warnings <EOL> import ctypes as ct <EOL> import picamera . mmal as mmal <EOL> from picamera . frames import PiVideoFrame , PiVideoFrameType <EOL> from picamera . streams import BufferIO <EOL> from picamera . exc import ( <EOL> mmal_check , <EOL> PiCameraError , <EOL> PiCameraMMALError , <EOL> PiCameraValueError , <EOL> PiCameraRuntimeError , <EOL> ) <EOL> def _debug_buffer ( buf ) : <EOL> f = buf [ <NUM_LIT:0> ] . flags <EOL> print ( '<STR_LIT>' . join ( ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:E>' if f & mmal . MMAL_BUFFER_HEADER_FLAG_FRAME_END else '<STR_LIT:_>' , <EOL> '<STR_LIT>' if f & mmal . MMAL_BUFFER_HEADER_FLAG_KEYFRAME else '<STR_LIT:_>' , <EOL> '<STR_LIT:C>' if f & mmal . MMAL_BUFFER_HEADER_FLAG_CONFIG else '<STR_LIT:_>' , <EOL> '<STR_LIT:M>' if f & mmal . MMAL_BUFFER_HEADER_FLAG_CODECSIDEINFO else '<STR_LIT:_>' , <EOL> '<STR_LIT:X>' if f & mmal . MMAL_BUFFER_HEADER_FLAG_EOS else '<STR_LIT:_>' , <EOL> '<STR_LIT:U+0020>' , <EOL> '<STR_LIT>' % buf [ <NUM_LIT:0> ] . length , <EOL> ) ) ) <EOL> def _encoder_callback ( port , buf ) : <EOL> encoder = ct . cast ( port [ <NUM_LIT:0> ] . userdata , ct . POINTER ( ct . py_object ) ) [ <NUM_LIT:0> ] <EOL> encoder . _callback ( port , buf ) <EOL> _encoder_callback = mmal . MMAL_PORT_BH_CB_T ( _encoder_callback ) <EOL> class PiEncoder ( object ) : <EOL> """<STR_LIT>""" <EOL> encoder_type = None <EOL> def __init__ ( <EOL> self , parent , camera_port , input_port , format , resize , ** options ) : <EOL> self . parent = parent <EOL> self . format = format <EOL> self . encoder = None <EOL> self . resizer = None <EOL> self . encoder_connection = None <EOL> self . resizer_connection = None <EOL> self . camera_port = camera_port <EOL> self . input_port = input_port <EOL> self . output_port = None <EOL> self . pool = None <EOL> self . started_capture = False <EOL> self . outputs_lock = threading . Lock ( ) <EOL> self . outputs = { } <EOL> self . exception = None <EOL> self . event = threading . Event ( ) <EOL> self . stopped = True <EOL> try : <EOL> if parent . closed : <EOL> raise PiCameraRuntimeError ( "<STR_LIT>" ) <EOL> if resize : <EOL> self . _create_resizer ( * resize ) <EOL> self . _create_encoder ( ** options ) <EOL> self . _create_pool ( ) <EOL> self . _create_connections ( ) <EOL> except : <EOL> self . close ( ) <EOL> raise <EOL> def _create_resizer ( self , width , height ) : <EOL> """<STR_LIT>""" <EOL> self . resizer = ct . POINTER ( mmal . MMAL_COMPONENT_T ) ( ) <EOL> mmal_check ( <EOL> mmal . mmal_component_create ( <EOL> mmal . MMAL_COMPONENT_DEFAULT_RESIZER , self . resizer ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if not self . resizer [ <NUM_LIT:0> ] . input_num : <EOL> raise PiCameraError ( "<STR_LIT>" ) <EOL> if not self . resizer [ <NUM_LIT:0> ] . output_num : <EOL> raise PiCameraError ( "<STR_LIT>" ) <EOL> mmal . mmal_format_copy ( <EOL> self . resizer [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format , self . input_port [ <NUM_LIT:0> ] . format ) <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( self . resizer [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal . mmal_format_copy ( <EOL> self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format , self . resizer [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format ) <EOL> fmt = self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . width = mmal . VCOS_ALIGN_UP ( width , <NUM_LIT:32> ) <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . height = mmal . VCOS_ALIGN_UP ( height , <NUM_LIT:16> ) <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . crop . x = <NUM_LIT:0> <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . crop . y = <NUM_LIT:0> <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . crop . width = width <EOL> fmt [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . crop . height = height <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] ) , <EOL> prefix = "<STR_LIT>" ) <EOL> def _create_encoder ( self ) : <EOL> """<STR_LIT>""" <EOL> assert not self . encoder <EOL> self . encoder = ct . POINTER ( mmal . MMAL_COMPONENT_T ) ( ) <EOL> mmal_check ( <EOL> mmal . mmal_component_create ( self . encoder_type , self . encoder ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if not self . encoder [ <NUM_LIT:0> ] . input_num : <EOL> raise PiCameraError ( "<STR_LIT>" ) <EOL> if not self . encoder [ <NUM_LIT:0> ] . output_num : <EOL> raise PiCameraError ( "<STR_LIT>" ) <EOL> self . output_port = self . encoder [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] <EOL> if self . resizer : <EOL> mmal . mmal_format_copy ( <EOL> self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format , self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format ) <EOL> else : <EOL> mmal . mmal_format_copy ( <EOL> self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format , self . input_port [ <NUM_LIT:0> ] . format ) <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal . mmal_format_copy ( <EOL> self . output_port [ <NUM_LIT:0> ] . format , self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . format ) <EOL> if self . format == '<STR_LIT>' : <EOL> self . output_port [ <NUM_LIT:0> ] . buffer_size = max ( <NUM_LIT> * <NUM_LIT> , self . output_port [ <NUM_LIT:0> ] . buffer_size_recommended ) <EOL> else : <EOL> self . output_port [ <NUM_LIT:0> ] . buffer_size = self . output_port [ <NUM_LIT:0> ] . buffer_size_recommended <EOL> self . output_port [ <NUM_LIT:0> ] . buffer_num = self . output_port [ <NUM_LIT:0> ] . buffer_num_recommended <EOL> def _create_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> assert not self . pool <EOL> self . pool = mmal . mmal_port_pool_create ( <EOL> self . output_port , <EOL> self . output_port [ <NUM_LIT:0> ] . buffer_num , <EOL> self . output_port [ <NUM_LIT:0> ] . buffer_size ) <EOL> if not self . pool : <EOL> raise PiCameraError ( <EOL> "<STR_LIT>" ) <EOL> def _create_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> assert not self . encoder_connection <EOL> if self . resizer : <EOL> self . resizer_connection = self . parent . _connect_ports ( <EOL> self . input_port , self . resizer [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) <EOL> self . encoder_connection = self . parent . _connect_ports ( <EOL> self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] , self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . encoder_connection = self . parent . _connect_ports ( <EOL> self . input_port , self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) <EOL> def _callback ( self , port , buf ) : <EOL> """<STR_LIT>""" <EOL> if self . stopped : <EOL> mmal . mmal_buffer_header_release ( buf ) <EOL> else : <EOL> stop = False <EOL> try : <EOL> try : <EOL> mmal_check ( <EOL> mmal . mmal_buffer_header_mem_lock ( buf ) , <EOL> prefix = "<STR_LIT>" ) <EOL> try : <EOL> stop = self . _callback_write ( buf ) <EOL> finally : <EOL> mmal . mmal_buffer_header_mem_unlock ( buf ) <EOL> finally : <EOL> mmal . mmal_buffer_header_release ( buf ) <EOL> self . _callback_recycle ( port , buf ) <EOL> except Exception as e : <EOL> stop = True <EOL> self . exception = e <EOL> if stop : <EOL> self . stopped = True <EOL> self . event . set ( ) <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> if buf [ <NUM_LIT:0> ] . length : <EOL> with self . outputs_lock : <EOL> try : <EOL> written = self . outputs [ key ] [ <NUM_LIT:0> ] . write ( <EOL> ct . string_at ( buf [ <NUM_LIT:0> ] . data , buf [ <NUM_LIT:0> ] . length ) ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> if ( written is not None ) and ( written != buf [ <NUM_LIT:0> ] . length ) : <EOL> raise PiCameraError ( <EOL> "<STR_LIT>" % key ) <EOL> return bool ( buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_EOS ) <EOL> def _callback_recycle ( self , port , buf ) : <EOL> """<STR_LIT>""" <EOL> new_buf = mmal . mmal_queue_get ( self . pool [ <NUM_LIT:0> ] . queue ) <EOL> if not new_buf : <EOL> raise PiCameraError ( <EOL> "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_port_send_buffer ( port , new_buf ) , <EOL> prefix = "<STR_LIT>" ) <EOL> def _open_output ( self , output , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> with self . outputs_lock : <EOL> opened = isinstance ( output , ( bytes , str ) ) <EOL> if opened : <EOL> output = io . open ( output , '<STR_LIT:wb>' , buffering = <NUM_LIT> ) <EOL> else : <EOL> try : <EOL> output . write <EOL> except AttributeError : <EOL> opened = True <EOL> output = BufferIO ( output ) <EOL> self . outputs [ key ] = ( output , opened ) <EOL> def _close_output ( self , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> with self . outputs_lock : <EOL> try : <EOL> ( output , opened ) = self . outputs . pop ( key ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> if opened : <EOL> output . close ( ) <EOL> else : <EOL> try : <EOL> output . flush ( ) <EOL> except AttributeError : <EOL> pass <EOL> @ property <EOL> def active ( self ) : <EOL> """<STR_LIT>""" <EOL> return bool ( self . encoder and self . output_port [ <NUM_LIT:0> ] . is_enabled ) <EOL> def start ( self , output ) : <EOL> """<STR_LIT>""" <EOL> self . event . clear ( ) <EOL> self . stopped = False <EOL> self . exception = None <EOL> self . _open_output ( output ) <EOL> self . output_port [ <NUM_LIT:0> ] . userdata = ct . cast ( <EOL> ct . pointer ( ct . py_object ( self ) ) , <EOL> ct . c_void_p ) <EOL> with self . parent . _encoders_lock : <EOL> mmal_check ( <EOL> mmal . mmal_port_enable ( self . output_port , _encoder_callback ) , <EOL> prefix = "<STR_LIT>" ) <EOL> for q in range ( mmal . mmal_queue_length ( self . pool [ <NUM_LIT:0> ] . queue ) ) : <EOL> buf = mmal . mmal_queue_get ( self . pool [ <NUM_LIT:0> ] . queue ) <EOL> if not buf : <EOL> raise PiCameraRuntimeError ( <EOL> "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_port_send_buffer ( self . output_port , buf ) , <EOL> prefix = "<STR_LIT>" ) <EOL> self . parent . _start_capture ( self . camera_port ) <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> result = self . event . wait ( timeout ) <EOL> if result : <EOL> self . stop ( ) <EOL> if self . exception : <EOL> raise self . exception <EOL> return result <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . active : <EOL> with self . parent . _encoders_lock : <EOL> self . parent . _stop_capture ( self . camera_port ) <EOL> try : <EOL> mmal_check ( <EOL> mmal . mmal_port_disable ( self . output_port ) , <EOL> prefix = "<STR_LIT>" ) <EOL> except PiCameraMMALError as e : <EOL> if e . status != mmal . MMAL_EINVAL : <EOL> raise <EOL> self . stopped = True <EOL> self . event . set ( ) <EOL> self . _close_output ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stop ( ) <EOL> if self . encoder_connection : <EOL> mmal . mmal_connection_destroy ( self . encoder_connection ) <EOL> self . encoder_connection = None <EOL> if self . pool : <EOL> mmal . mmal_port_pool_destroy ( self . output_port , self . pool ) <EOL> self . pool = None <EOL> if self . resizer_connection : <EOL> mmal . mmal_connection_destroy ( self . resizer_connection ) <EOL> if self . encoder : <EOL> mmal . mmal_component_destroy ( self . encoder ) <EOL> self . encoder = None <EOL> if self . resizer : <EOL> mmal . mmal_component_destroy ( self . resizer ) <EOL> self . resizer = None <EOL> self . output_port = None <EOL> class PiRawMixin ( PiEncoder ) : <EOL> """<STR_LIT>""" <EOL> RAW_ENCODINGS = { <EOL> '<STR_LIT>' : ( mmal . MMAL_ENCODING_I420 , <NUM_LIT> ) , <EOL> '<STR_LIT>' : ( mmal . MMAL_ENCODING_RGBA , <NUM_LIT:3> ) , <EOL> '<STR_LIT>' : ( mmal . MMAL_ENCODING_RGBA , <NUM_LIT:4> ) , <EOL> '<STR_LIT>' : ( mmal . MMAL_ENCODING_BGRA , <NUM_LIT:3> ) , <EOL> '<STR_LIT>' : ( mmal . MMAL_ENCODING_BGRA , <NUM_LIT:4> ) , <EOL> } <EOL> def __init__ ( <EOL> self , parent , camera_port , input_port , format , resize , ** options ) : <EOL> if not resize : <EOL> if parent . RAW_FORMATS [ format ] != input_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . encoding . value : <EOL> resize = parent . resolution <EOL> if resize : <EOL> width , height = resize <EOL> self . _strip_alpha = format in ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> else : <EOL> width , height = parent . resolution <EOL> self . _strip_alpha = False <EOL> width = mmal . VCOS_ALIGN_UP ( width , <NUM_LIT:32> ) <EOL> height = mmal . VCOS_ALIGN_UP ( height , <NUM_LIT:16> ) <EOL> if resize : <EOL> resize = ( width , height ) <EOL> self . _frame_size = int ( width * height * self . RAW_ENCODINGS [ format ] [ <NUM_LIT:1> ] ) <EOL> super ( PiRawMixin , self ) . __init__ ( <EOL> parent , camera_port , input_port , format , resize , ** options ) <EOL> def _create_resizer ( self , width , height ) : <EOL> """<STR_LIT>""" <EOL> super ( PiRawMixin , self ) . _create_resizer ( width , height ) <EOL> encoding = self . RAW_ENCODINGS [ self . format ] [ <NUM_LIT:0> ] <EOL> port = self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] <EOL> port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . encoding = encoding <EOL> port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . encoding_variant = encoding <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( port ) , <EOL> prefix = "<STR_LIT>" ) <EOL> def _create_encoder ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . resizer : <EOL> self . output_port = self . resizer [ <NUM_LIT:0> ] . output [ <NUM_LIT:0> ] <EOL> else : <EOL> self . output_port = self . input_port <EOL> def _create_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . resizer : <EOL> self . resizer_connection = self . parent . _connect_ports ( <EOL> self . input_port , self . resizer [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] ) <EOL> @ property <EOL> def active ( self ) : <EOL> return bool ( self . output_port [ <NUM_LIT:0> ] . is_enabled ) <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> if self . _strip_alpha : <EOL> s = ct . string_at ( buf [ <NUM_LIT:0> ] . data , buf [ <NUM_LIT:0> ] . length ) <EOL> s = bytearray ( s ) <EOL> del s [ <NUM_LIT:3> : : <NUM_LIT:4> ] <EOL> new_buf = mmal . MMAL_BUFFER_HEADER_T . from_buffer_copy ( buf [ <NUM_LIT:0> ] ) <EOL> new_buf . length = len ( s ) <EOL> new_buf . data = ct . pointer ( ct . c_uint8 . from_buffer ( s ) ) <EOL> return super ( PiRawMixin , self ) . _callback_write ( ct . pointer ( new_buf ) , key ) <EOL> else : <EOL> return super ( PiRawMixin , self ) . _callback_write ( buf , key ) <EOL> class PiVideoEncoder ( PiEncoder ) : <EOL> """<STR_LIT>""" <EOL> encoder_type = mmal . MMAL_COMPONENT_DEFAULT_VIDEO_ENCODER <EOL> def __init__ ( <EOL> self , parent , camera_port , input_port , format , resize , ** options ) : <EOL> super ( PiVideoEncoder , self ) . __init__ ( <EOL> parent , camera_port , input_port , format , resize , ** options ) <EOL> self . _next_output = [ ] <EOL> self . frame = None <EOL> def _create_encoder ( <EOL> self , bitrate = <NUM_LIT> , intra_period = None , profile = '<STR_LIT>' , <EOL> quantization = <NUM_LIT:0> , quality = <NUM_LIT:0> , inline_headers = True , sei = False , <EOL> motion_output = None , intra_refresh = None ) : <EOL> """<STR_LIT>""" <EOL> super ( PiVideoEncoder , self ) . _create_encoder ( ) <EOL> quality = quality or quantization <EOL> try : <EOL> self . output_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . encoding = { <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_H264 , <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_MJPEG , <EOL> } [ self . format ] <EOL> except KeyError : <EOL> raise PiCameraValueError ( '<STR_LIT>' % self . format ) <EOL> if not ( <NUM_LIT:0> <= bitrate <= <NUM_LIT> ) : <EOL> raise PiCameraValueError ( '<STR_LIT>' ) <EOL> self . output_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . bitrate = bitrate <EOL> self . output_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . frame_rate . num = <NUM_LIT:0> <EOL> self . output_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . es [ <NUM_LIT:0> ] . video . frame_rate . den = <NUM_LIT:1> <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( self . output_port ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if self . format == '<STR_LIT>' : <EOL> mp = mmal . MMAL_PARAMETER_VIDEO_PROFILE_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_PROFILE , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_VIDEO_PROFILE_T ) , <EOL> ) , <EOL> ) <EOL> try : <EOL> mp . profile [ <NUM_LIT:0> ] . profile = { <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_PROFILE_H264_BASELINE , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_PROFILE_H264_MAIN , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_PROFILE_H264_HIGH , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_PROFILE_H264_CONSTRAINED_BASELINE , <EOL> } [ profile ] <EOL> except KeyError : <EOL> raise PiCameraValueError ( "<STR_LIT>" % profile ) <EOL> mp . profile [ <NUM_LIT:0> ] . level = mmal . MMAL_VIDEO_LEVEL_H264_4 <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if inline_headers : <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_boolean ( <EOL> self . output_port , <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_INLINE_HEADER , <EOL> mmal . MMAL_TRUE ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if sei : <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_boolean ( <EOL> self . output_port , <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_SEI_ENABLE , <EOL> mmal . MMAL_TRUE ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if motion_output : <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_boolean ( <EOL> self . output_port , <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_INLINE_VECTORS , <EOL> mmal . MMAL_TRUE ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if intra_period is not None : <EOL> mp = mmal . MMAL_PARAMETER_UINT32_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_INTRAPERIOD , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_UINT32_T ) , <EOL> ) , <EOL> intra_period <EOL> ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> self . _intra_period = intra_period <EOL> else : <EOL> mp = mmal . MMAL_PARAMETER_UINT32_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_INTRAPERIOD , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_UINT32_T ) , <EOL> ) ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_get ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> self . _intra_period = mp . value <EOL> if intra_refresh is not None : <EOL> mp = mmal . MMAL_PARAMETER_VIDEO_INTRA_REFRESH_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_VIDEO_INTRA_REFRESH , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_VIDEO_INTRA_REFRESH_T ) , <EOL> ) ) <EOL> mmal . mmal_port_parameter_get ( self . output_port , mp . hdr ) <EOL> try : <EOL> mp . refresh_mode = { <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_INTRA_REFRESH_CYCLIC , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_INTRA_REFRESH_ADAPTIVE , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_INTRA_REFRESH_BOTH , <EOL> '<STR_LIT>' : mmal . MMAL_VIDEO_INTRA_REFRESH_CYCLIC_MROWS , <EOL> } [ intra_refresh ] <EOL> except KeyError : <EOL> raise PiCameraValueError ( <EOL> "<STR_LIT>" % intra_refresh ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> elif self . format == '<STR_LIT>' : <EOL> self . _intra_period = <NUM_LIT:1> <EOL> if quality : <EOL> mp = mmal . MMAL_PARAMETER_UINT32_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_INITIAL_QUANT , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_UINT32_T ) , <EOL> ) , <EOL> quality <EOL> ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mp = mmal . MMAL_PARAMETER_UINT32_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_MIN_QUANT , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_UINT32_T ) , <EOL> ) , <EOL> quality , <EOL> ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mp = mmal . MMAL_PARAMETER_UINT32_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_VIDEO_ENCODE_MAX_QUANT , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_UINT32_T ) , <EOL> ) , <EOL> quality , <EOL> ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_boolean ( <EOL> self . encoder [ <NUM_LIT:0> ] . input [ <NUM_LIT:0> ] , <EOL> mmal . MMAL_PARAMETER_VIDEO_IMMUTABLE_INPUT , <EOL> <NUM_LIT:1> ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_component_enable ( self . encoder ) , <EOL> prefix = "<STR_LIT>" ) <EOL> def start ( self , output , motion_output = None ) : <EOL> """<STR_LIT>""" <EOL> self . frame = PiVideoFrame ( <EOL> index = <NUM_LIT:0> , <EOL> frame_type = None , <EOL> frame_size = <NUM_LIT:0> , <EOL> video_size = <NUM_LIT:0> , <EOL> split_size = <NUM_LIT:0> , <EOL> timestamp = <NUM_LIT:0> , <EOL> complete = False , <EOL> ) <EOL> if motion_output is not None : <EOL> self . _open_output ( motion_output , PiVideoFrameType . motion_data ) <EOL> super ( PiVideoEncoder , self ) . start ( output ) <EOL> def stop ( self ) : <EOL> super ( PiVideoEncoder , self ) . stop ( ) <EOL> self . _close_output ( PiVideoFrameType . motion_data ) <EOL> def split ( self , output , motion_output = None ) : <EOL> """<STR_LIT>""" <EOL> with self . outputs_lock : <EOL> outputs = { } <EOL> if output is not None : <EOL> outputs [ PiVideoFrameType . frame ] = output <EOL> if motion_output is not None : <EOL> outputs [ PiVideoFrameType . motion_data ] = motion_output <EOL> self . _next_output . append ( outputs ) <EOL> timeout = max ( <NUM_LIT> , float ( self . _intra_period / self . parent . framerate ) * <NUM_LIT> ) <EOL> if not self . event . wait ( timeout ) : <EOL> raise PiCameraRuntimeError ( <EOL> '<STR_LIT>' ) <EOL> self . event . clear ( ) <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> self . frame = PiVideoFrame ( <EOL> index = <EOL> self . frame . index + <NUM_LIT:1> <EOL> if self . frame . complete else <EOL> self . frame . index , <EOL> frame_type = <EOL> PiVideoFrameType . key_frame <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_KEYFRAME else <EOL> PiVideoFrameType . sps_header <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CONFIG else <EOL> PiVideoFrameType . motion_data <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CODECSIDEINFO else <EOL> PiVideoFrameType . frame , <EOL> frame_size = <EOL> buf [ <NUM_LIT:0> ] . length <EOL> if self . frame . complete else <EOL> self . frame . frame_size + buf [ <NUM_LIT:0> ] . length , <EOL> video_size = <EOL> self . frame . video_size <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CODECSIDEINFO else <EOL> self . frame . video_size + buf [ <NUM_LIT:0> ] . length , <EOL> split_size = <EOL> self . frame . split_size <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CODECSIDEINFO else <EOL> self . frame . split_size + buf [ <NUM_LIT:0> ] . length , <EOL> timestamp = <EOL> None <EOL> if buf [ <NUM_LIT:0> ] . pts in ( <NUM_LIT:0> , mmal . MMAL_TIME_UNKNOWN ) else <EOL> buf [ <NUM_LIT:0> ] . pts , <EOL> complete = <EOL> bool ( buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_FRAME_END ) , <EOL> ) <EOL> if self . format != '<STR_LIT>' or ( buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CONFIG ) : <EOL> with self . outputs_lock : <EOL> try : <EOL> new_outputs = self . _next_output . pop ( <NUM_LIT:0> ) <EOL> except IndexError : <EOL> new_outputs = None <EOL> if new_outputs : <EOL> for new_key , new_output in new_outputs . items ( ) : <EOL> self . _close_output ( new_key ) <EOL> self . _open_output ( new_output , new_key ) <EOL> if new_key == PiVideoFrameType . frame : <EOL> self . frame = PiVideoFrame ( <EOL> index = self . frame . index , <EOL> frame_type = self . frame . frame_type , <EOL> frame_size = self . frame . frame_size , <EOL> video_size = self . frame . video_size , <EOL> split_size = <NUM_LIT:0> , <EOL> timestamp = self . frame . timestamp , <EOL> complete = self . frame . complete , <EOL> ) <EOL> self . event . set ( ) <EOL> if buf [ <NUM_LIT:0> ] . flags & mmal . MMAL_BUFFER_HEADER_FLAG_CODECSIDEINFO : <EOL> key = PiVideoFrameType . motion_data <EOL> return super ( PiVideoEncoder , self ) . _callback_write ( buf , key ) <EOL> class PiCookedVideoEncoder ( PiVideoEncoder ) : <EOL> """<STR_LIT>""" <EOL> class PiRawVideoEncoder ( PiRawMixin , PiVideoEncoder ) : <EOL> """<STR_LIT>""" <EOL> def _create_encoder ( self ) : <EOL> super ( PiRawVideoEncoder , self ) . _create_encoder ( ) <EOL> self . _intra_period = <NUM_LIT:1> <EOL> class PiImageEncoder ( PiEncoder ) : <EOL> """<STR_LIT>""" <EOL> encoder_type = mmal . MMAL_COMPONENT_DEFAULT_IMAGE_ENCODER <EOL> def _create_encoder ( self , quality = <NUM_LIT> , thumbnail = ( <NUM_LIT:64> , <NUM_LIT> , <NUM_LIT> ) , bayer = False ) : <EOL> """<STR_LIT>""" <EOL> super ( PiImageEncoder , self ) . _create_encoder ( ) <EOL> try : <EOL> self . output_port [ <NUM_LIT:0> ] . format [ <NUM_LIT:0> ] . encoding = { <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_JPEG , <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_PNG , <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_GIF , <EOL> '<STR_LIT>' : mmal . MMAL_ENCODING_BMP , <EOL> } [ self . format ] <EOL> except KeyError : <EOL> raise PiCameraValueError ( "<STR_LIT>" % self . format ) <EOL> mmal_check ( <EOL> mmal . mmal_port_format_commit ( self . output_port ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if self . format == '<STR_LIT>' : <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_uint32 ( <EOL> self . output_port , <EOL> mmal . MMAL_PARAMETER_JPEG_Q_FACTOR , <EOL> quality ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set_boolean ( <EOL> self . camera_port , <EOL> mmal . MMAL_PARAMETER_ENABLE_RAW_CAPTURE , <EOL> int ( bool ( bayer ) ) ) , <EOL> prefix = "<STR_LIT>" ) <EOL> if thumbnail is None : <EOL> mp = mmal . MMAL_PARAMETER_THUMBNAIL_CONFIG_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_THUMBNAIL_CONFIGURATION , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_THUMBNAIL_CONFIG_T ) <EOL> ) , <EOL> <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> else : <EOL> mp = mmal . MMAL_PARAMETER_THUMBNAIL_CONFIG_T ( <EOL> mmal . MMAL_PARAMETER_HEADER_T ( <EOL> mmal . MMAL_PARAMETER_THUMBNAIL_CONFIGURATION , <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_THUMBNAIL_CONFIG_T ) <EOL> ) , <EOL> <NUM_LIT:1> , * thumbnail ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . encoder [ <NUM_LIT:0> ] . control , mp . hdr ) , <EOL> prefix = "<STR_LIT>" ) <EOL> mmal_check ( <EOL> mmal . mmal_component_enable ( self . encoder ) , <EOL> prefix = "<STR_LIT>" ) <EOL> class PiOneImageEncoder ( PiImageEncoder ) : <EOL> """<STR_LIT>""" <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> return ( <EOL> super ( PiOneImageEncoder , self ) . _callback_write ( buf , key ) <EOL> ) or bool ( <EOL> buf [ <NUM_LIT:0> ] . flags & ( <EOL> mmal . MMAL_BUFFER_HEADER_FLAG_FRAME_END | <EOL> mmal . MMAL_BUFFER_HEADER_FLAG_TRANSMISSION_FAILED ) <EOL> ) <EOL> class PiMultiImageEncoder ( PiImageEncoder ) : <EOL> """<STR_LIT>""" <EOL> def _open_output ( self , outputs , key = PiVideoFrameType . frame ) : <EOL> self . _output_iter = iter ( outputs ) <EOL> self . _next_output ( key ) <EOL> def _next_output ( self , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> self . _close_output ( key ) <EOL> super ( PiMultiImageEncoder , self ) . _open_output ( next ( self . _output_iter ) , key ) <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> try : <EOL> if ( <EOL> super ( PiMultiImageEncoder , self ) . _callback_write ( buf , key ) <EOL> ) or bool ( <EOL> buf [ <NUM_LIT:0> ] . flags & ( <EOL> mmal . MMAL_BUFFER_HEADER_FLAG_FRAME_END | <EOL> mmal . MMAL_BUFFER_HEADER_FLAG_TRANSMISSION_FAILED ) <EOL> ) : <EOL> self . _next_output ( key ) <EOL> return False <EOL> except StopIteration : <EOL> return True <EOL> class PiCookedOneImageEncoder ( PiOneImageEncoder ) : <EOL> """<STR_LIT>""" <EOL> exif_encoding = '<STR_LIT:ascii>' <EOL> def _add_exif_tag ( self , tag , value ) : <EOL> if isinstance ( tag , str ) : <EOL> tag = tag . encode ( self . exif_encoding ) <EOL> if isinstance ( value , str ) : <EOL> value = value . encode ( self . exif_encoding ) <EOL> elif isinstance ( value , datetime . datetime ) : <EOL> value = value . strftime ( '<STR_LIT>' ) . encode ( self . exif_encoding ) <EOL> buf = ct . create_string_buffer ( <EOL> ct . sizeof ( mmal . MMAL_PARAMETER_EXIF_T ) + len ( tag ) + len ( value ) + <NUM_LIT:1> ) <EOL> mp = ct . cast ( buf , ct . POINTER ( mmal . MMAL_PARAMETER_EXIF_T ) ) <EOL> mp [ <NUM_LIT:0> ] . hdr . id = mmal . MMAL_PARAMETER_EXIF <EOL> mp [ <NUM_LIT:0> ] . hdr . size = len ( buf ) <EOL> if ( b'<STR_LIT:=>' in tag or b'<STR_LIT:\x00>' in value ) : <EOL> data = tag + value <EOL> mp [ <NUM_LIT:0> ] . keylen = len ( tag ) <EOL> mp [ <NUM_LIT:0> ] . value_offset = len ( tag ) <EOL> mp [ <NUM_LIT:0> ] . valuelen = len ( value ) <EOL> else : <EOL> data = tag + b'<STR_LIT:=>' + value <EOL> ct . memmove ( mp [ <NUM_LIT:0> ] . data , data , len ( data ) ) <EOL> mmal_check ( <EOL> mmal . mmal_port_parameter_set ( self . output_port , mp [ <NUM_LIT:0> ] . hdr ) , <EOL> prefix = "<STR_LIT>" % tag ) <EOL> def start ( self , output ) : <EOL> timestamp = datetime . datetime . now ( ) <EOL> timestamp_tags = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> for tag in timestamp_tags : <EOL> self . _add_exif_tag ( tag , self . parent . exif_tags . get ( tag , timestamp ) ) <EOL> for tag , value in self . parent . exif_tags . items ( ) : <EOL> if not tag in timestamp_tags : <EOL> self . _add_exif_tag ( tag , value ) <EOL> super ( PiCookedOneImageEncoder , self ) . start ( output ) <EOL> class PiCookedMultiImageEncoder ( PiMultiImageEncoder ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class PiRawImageMixin ( PiRawMixin , PiImageEncoder ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , parent , camera_port , input_port , format , resize , ** options ) : <EOL> super ( PiRawImageMixin , self ) . __init__ ( <EOL> parent , camera_port , input_port , format , resize , ** options ) <EOL> self . _image_size = <NUM_LIT:0> <EOL> def _callback_write ( self , buf , key = PiVideoFrameType . frame ) : <EOL> """<STR_LIT>""" <EOL> if self . _image_size > <NUM_LIT:0> : <EOL> super ( PiRawImageMixin , self ) . _callback_write ( buf , key ) <EOL> self . _image_size -= buf [ <NUM_LIT:0> ] . length <EOL> return self . _image_size <= <NUM_LIT:0> <EOL> def start ( self , output ) : <EOL> self . _image_size = self . _frame_size <EOL> super ( PiRawImageMixin , self ) . start ( output ) <EOL> class PiRawOneImageEncoder ( PiOneImageEncoder , PiRawImageMixin ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class PiRawMultiImageEncoder ( PiMultiImageEncoder , PiRawImageMixin ) : <EOL> """<STR_LIT>""" <EOL> def _next_output ( self , key = PiVideoFrameType . frame ) : <EOL> super ( PiRawMultiImageEncoder , self ) . _next_output ( key ) <EOL> self . _image_size = self . _frame_size </s>
