<s> from . error import AMFError <EOL> from . packet import Packet <EOL> from . types import AMF0String , AMF0Value , U8 , U16BE , U32BE <EOL> class AMFHeader ( Packet ) : <EOL> exception = AMFError <EOL> def __init__ ( self , name , value , must_understand = False ) : <EOL> self . name = name <EOL> self . value = value <EOL> self . must_understand = must_understand <EOL> @ property <EOL> def size ( self ) : <EOL> size = <NUM_LIT:4> + <NUM_LIT:1> <EOL> size += AMF0String . size ( self . name ) <EOL> size += AMF0Value . size ( self . value ) <EOL> return size <EOL> def _serialize ( self , packet ) : <EOL> packet += AMF0String ( self . name ) <EOL> packet += U8 ( int ( self . must_understand ) ) <EOL> packet += U32BE ( self . size ) <EOL> packet += AMF0Value ( self . value ) <EOL> @ classmethod <EOL> def _deserialize ( cls , io ) : <EOL> name = AMF0String . read ( io ) <EOL> must_understand = bool ( U8 . read ( io ) ) <EOL> length = U32BE . read ( io ) <EOL> value = AMF0Value . read ( io ) <EOL> return cls ( name , value , must_understand ) <EOL> class AMFMessage ( Packet ) : <EOL> exception = AMFError <EOL> def __init__ ( self , target_uri , response_uri , value ) : <EOL> self . target_uri = target_uri <EOL> self . response_uri = response_uri <EOL> self . value = value <EOL> @ property <EOL> def size ( self ) : <EOL> size = <NUM_LIT:4> <EOL> size += AMF0String . size ( self . target_uri ) <EOL> size += AMF0String . size ( self . response_uri ) <EOL> size += AMF0Value . size ( self . value ) <EOL> return size <EOL> def _serialize ( self , packet ) : <EOL> packet += AMF0String ( self . target_uri ) <EOL> packet += AMF0String ( self . response_uri ) <EOL> packet += U32BE ( self . size ) <EOL> packet += AMF0Value . pack ( self . value ) <EOL> @ classmethod <EOL> def _deserialize ( cls , io ) : <EOL> target_uri = AMF0String . read ( io ) <EOL> response_uri = AMF0String . read ( io ) <EOL> length = U32BE . read ( io ) <EOL> value = AMF0Value . read ( io ) <EOL> return cls ( target_uri , response_uri , value ) <EOL> class AMFPacket ( Packet ) : <EOL> exception = AMFError <EOL> def __init__ ( self , version , headers = None , messages = None ) : <EOL> if headers is None : <EOL> headers = [ ] <EOL> if messages is None : <EOL> messages = [ ] <EOL> self . version = version <EOL> self . headers = headers <EOL> self . messages = messages <EOL> @ property <EOL> def size ( self ) : <EOL> size = <NUM_LIT:2> + <NUM_LIT:2> + <NUM_LIT:2> <EOL> for header in self . headers : <EOL> size += header . size <EOL> for message in self . messages : <EOL> size += message . size <EOL> return size <EOL> def _serialize ( self , packet ) : <EOL> packet += U16BE ( self . version ) <EOL> packet += U16BE ( len ( self . headers ) ) <EOL> for header in self . headers : <EOL> header . serialize ( packet ) <EOL> packet += U16BE ( len ( self . messages ) ) <EOL> for message in self . messages : <EOL> message . serialize ( packet ) <EOL> @ classmethod <EOL> def _deserialize ( cls , io ) : <EOL> version = U16BE . read ( io ) <EOL> if not version in ( <NUM_LIT:0> , <NUM_LIT:3> ) : <EOL> raise AMFError ( "<STR_LIT>" ) <EOL> headers = [ ] <EOL> header_count = U16BE . read ( io ) <EOL> for i in range ( header_count ) : <EOL> header = AMFHeader . deserialize ( io ) <EOL> headers . append ( header ) <EOL> messages = [ ] <EOL> message_count = U16BE . read ( io ) <EOL> for i in range ( message_count ) : <EOL> message = AMFMessage . deserialize ( io ) <EOL> messages . append ( message ) <EOL> return cls ( version , headers , messages ) <EOL> __all__ = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] </s>
