<s> try : <EOL> from functools import singledispatch <EOL> except ImportError : <EOL> from singledispatch import singledispatch <EOL> from . import ffi , librtmp <EOL> from . aval import AVal <EOL> from . compat import is_py2 , bytes , range <EOL> from . exceptions import AMFError <EOL> AMF_STRING_TYPES = ( librtmp . AMF_STRING , librtmp . AMF_LONG_STRING ) <EOL> AMF_OBJECT_DICT_TYPES = ( librtmp . AMF_OBJECT , librtmp . AMF_ECMA_ARRAY ) <EOL> __all__ = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> class AMFObject ( dict ) : <EOL> pass <EOL> def _create_buffer ( size ) : <EOL> pbuf = ffi . new ( "<STR_LIT>" , size ) <EOL> pend = pbuf + size <EOL> buf = ffi . buffer ( pbuf , size ) <EOL> return pbuf , pend , buf <EOL> def _encode_key_name ( key ) : <EOL> key = bytes ( key , "<STR_LIT:utf8>" ) <EOL> key_len = len ( key ) <EOL> pbuf , pend , buf = _create_buffer ( key_len + <NUM_LIT:2> ) <EOL> librtmp . AMF_EncodeInt16 ( pbuf , pend , key_len ) <EOL> buf [ <NUM_LIT:2> : key_len + <NUM_LIT:2> ] = key <EOL> return buf [ : ] <EOL> @ singledispatch <EOL> def encoder ( val ) : <EOL> raise AMFError ( "<STR_LIT>" . format ( type ( val ) . __name__ ) ) <EOL> @ encoder . register ( type ( None ) ) <EOL> def _encode_none ( val ) : <EOL> return bytes ( ( librtmp . AMF_NULL , ) ) <EOL> @ encoder . register ( str ) <EOL> def _encode_str ( val ) : <EOL> val = AVal ( val ) <EOL> pbuf , pend , buf = _create_buffer ( val . aval . av_len + <NUM_LIT:1> + <NUM_LIT:4> ) <EOL> res = librtmp . AMF_EncodeString ( pbuf , pend , val . aval ) <EOL> size = res - pbuf <EOL> return buf [ : size ] <EOL> if is_py2 : <EOL> encoder . register ( unicode , _encode_str ) <EOL> @ encoder . register ( float ) <EOL> @ encoder . register ( int ) <EOL> def _encode_number ( val ) : <EOL> val = float ( val ) <EOL> pbuf , pend , buf = _create_buffer ( <NUM_LIT:9> ) <EOL> librtmp . AMF_EncodeNumber ( pbuf , pend , val ) <EOL> return buf [ : ] <EOL> if is_py2 : <EOL> encoder . register ( long , _encode_number ) <EOL> @ encoder . register ( bool ) <EOL> def _encode_boolean ( val ) : <EOL> pbuf , pend , buf = _create_buffer ( <NUM_LIT:2> ) <EOL> librtmp . AMF_EncodeBoolean ( pbuf , pend , int ( val ) ) <EOL> return buf [ : ] <EOL> @ encoder . register ( AMFObject ) <EOL> def _encode_object ( val ) : <EOL> phead , headend , head = _create_buffer ( <NUM_LIT:4> ) <EOL> head [ <NUM_LIT:0> ] = bytes ( ( librtmp . AMF_OBJECT , ) ) <EOL> librtmp . AMF_EncodeInt24 ( phead + <NUM_LIT:1> , headend , librtmp . AMF_OBJECT_END ) <EOL> body = bytearray ( ) <EOL> for key , value in val . items ( ) : <EOL> body += _encode_key_name ( key ) <EOL> body += encoder ( value ) <EOL> return head [ : <NUM_LIT:1> ] + bytes ( body ) + head [ <NUM_LIT:1> : ] <EOL> @ encoder . register ( dict ) <EOL> def _encode_ecma_array ( val ) : <EOL> phead , headend , head = _create_buffer ( <NUM_LIT:8> ) <EOL> head [ <NUM_LIT:0> ] = bytes ( ( librtmp . AMF_ECMA_ARRAY , ) ) <EOL> librtmp . AMF_EncodeInt32 ( phead + <NUM_LIT:1> , headend , len ( val ) ) <EOL> librtmp . AMF_EncodeInt24 ( phead + <NUM_LIT:5> , headend , librtmp . AMF_OBJECT_END ) <EOL> body = bytearray ( ) <EOL> for key , value in val . items ( ) : <EOL> body += _encode_key_name ( key ) <EOL> body += encoder ( value ) <EOL> return head [ : <NUM_LIT:5> ] + bytes ( body ) + head [ <NUM_LIT:5> : ] <EOL> @ encoder . register ( list ) <EOL> def _encode_array ( val ) : <EOL> phead , headend , head = _create_buffer ( <NUM_LIT:5> ) <EOL> head [ <NUM_LIT:0> ] = bytes ( ( librtmp . AMF_STRICT_ARRAY , ) ) <EOL> librtmp . AMF_EncodeInt32 ( phead + <NUM_LIT:1> , headend , len ( val ) ) <EOL> body = bytearray ( ) <EOL> for value in val : <EOL> body += encoder ( value ) <EOL> return head [ : ] + bytes ( body ) <EOL> def _decode_prop ( prop ) : <EOL> prop_type = librtmp . AMFProp_GetType ( prop ) <EOL> if prop_type == librtmp . AMF_NUMBER : <EOL> val = librtmp . AMFProp_GetNumber ( prop ) <EOL> elif prop_type in AMF_STRING_TYPES : <EOL> aval = AVal ( ) <EOL> librtmp . AMFProp_GetString ( prop , aval . aval ) <EOL> val = aval . value . decode ( "<STR_LIT:utf8>" , "<STR_LIT:ignore>" ) <EOL> elif prop_type == librtmp . AMF_BOOLEAN : <EOL> val = bool ( librtmp . AMFProp_GetBoolean ( prop ) ) <EOL> elif prop_type in AMF_OBJECT_DICT_TYPES : <EOL> if prop_type == librtmp . AMF_OBJECT : <EOL> val = AMFObject ( ) <EOL> else : <EOL> val = dict ( ) <EOL> for key , value in _decode_prop_obj ( prop ) : <EOL> val [ key ] = value <EOL> elif prop_type == librtmp . AMF_STRICT_ARRAY : <EOL> val = [ ] <EOL> for key , value in _decode_prop_obj ( prop ) : <EOL> val . append ( value ) <EOL> else : <EOL> val = None <EOL> return val <EOL> def _decode_prop_obj ( prop ) : <EOL> obj = ffi . new ( "<STR_LIT>" ) <EOL> librtmp . AMFProp_GetObject ( prop , obj ) <EOL> prop_count = librtmp . AMF_CountProp ( obj ) <EOL> for i in range ( prop_count ) : <EOL> prop = librtmp . AMF_GetProp ( obj , ffi . NULL , i ) <EOL> key = AVal ( ) <EOL> librtmp . AMFProp_GetName ( prop , key . aval ) <EOL> key = key . value . decode ( "<STR_LIT:utf8>" , "<STR_LIT:ignore>" ) <EOL> value = _decode_prop ( prop ) <EOL> yield key , value <EOL> def encode_amf ( value ) : <EOL> return encoder ( value ) <EOL> def decode_amf ( body ) : <EOL> obj = ffi . new ( "<STR_LIT>" ) <EOL> res = librtmp . AMF_Decode ( obj , body , len ( body ) , <NUM_LIT:0> ) <EOL> if res == ffi . NULL : <EOL> raise AMFError ( "<STR_LIT>" ) <EOL> rval = [ ] <EOL> prop_count = librtmp . AMF_CountProp ( obj ) <EOL> for i in range ( prop_count ) : <EOL> prop = librtmp . AMF_GetProp ( obj , ffi . NULL , i ) <EOL> val = _decode_prop ( prop ) <EOL> rval . append ( val ) <EOL> return rval </s>
