<s> """<STR_LIT>""" <EOL> import struct <EOL> from jasmin . vendor . smpp . pdu import gsm_constants , gsm_types <EOL> from jasmin . vendor . smpp . pdu . encoding import IEncoder <EOL> from jasmin . vendor . smpp . pdu . error import PDUParseError <EOL> class UDHParseError ( Exception ) : <EOL> pass <EOL> class UDHInformationElementIdentifierUnknownError ( UDHParseError ) : <EOL> pass <EOL> class Int8Encoder ( IEncoder ) : <EOL> def encode ( self , value ) : <EOL> return struct . pack ( '<STR_LIT>' , value ) <EOL> def decode ( self , file ) : <EOL> byte = self . read ( file , <NUM_LIT:1> ) <EOL> return struct . unpack ( '<STR_LIT>' , byte ) [ <NUM_LIT:0> ] <EOL> class Int16Encoder ( IEncoder ) : <EOL> def encode ( self , value ) : <EOL> return struct . pack ( '<STR_LIT>' , value ) <EOL> def decode ( self , file ) : <EOL> bytes = self . read ( file , <NUM_LIT:2> ) <EOL> return struct . unpack ( '<STR_LIT>' , bytes ) [ <NUM_LIT:0> ] <EOL> class InformationElementIdentifierEncoder ( IEncoder ) : <EOL> int8Encoder = Int8Encoder ( ) <EOL> nameMap = gsm_constants . information_element_identifier_name_map <EOL> valueMap = gsm_constants . information_element_identifier_value_map <EOL> def encode ( self , value ) : <EOL> name = str ( value ) <EOL> if name not in self . nameMap : <EOL> raise ValueError ( "<STR_LIT>" % name ) <EOL> return self . int8Encoder . encode ( self . nameMap [ name ] ) <EOL> def decode ( self , file ) : <EOL> intVal = self . int8Encoder . decode ( file ) <EOL> if intVal not in self . valueMap : <EOL> errStr = "<STR_LIT>" % intVal <EOL> raise UDHInformationElementIdentifierUnknownError ( errStr ) <EOL> name = self . valueMap [ intVal ] <EOL> return getattr ( gsm_types . InformationElementIdentifier , name ) <EOL> class IEConcatenatedSMEncoder ( IEncoder ) : <EOL> int8Encoder = Int8Encoder ( ) <EOL> int16Encoder = Int16Encoder ( ) <EOL> def __init__ ( self , is16bitRefNum ) : <EOL> self . is16bitRefNum = is16bitRefNum <EOL> def encode ( self , cms ) : <EOL> bytes = '<STR_LIT>' <EOL> if self . is16bitRefNum : <EOL> bytes += self . int16Encoder . encode ( cms . referenceNum ) <EOL> else : <EOL> bytes += self . int8Encoder . encode ( cms . referenceNum ) <EOL> bytes += self . int8Encoder . encode ( cms . maximumNum ) <EOL> bytes += self . int8Encoder . encode ( cms . sequenceNum ) <EOL> return bytes <EOL> def decode ( self , file ) : <EOL> refNum = None <EOL> if self . is16bitRefNum : <EOL> refNum = self . int16Encoder . decode ( file ) <EOL> else : <EOL> refNum = self . int8Encoder . decode ( file ) <EOL> maxNum = self . int8Encoder . decode ( file ) <EOL> seqNum = self . int8Encoder . decode ( file ) <EOL> return gsm_types . IEConcatenatedSM ( refNum , maxNum , seqNum ) <EOL> class InformationElementEncoder ( IEncoder ) : <EOL> int8Encoder = Int8Encoder ( ) <EOL> iEIEncoder = InformationElementIdentifierEncoder ( ) <EOL> dataEncoders = { <EOL> gsm_types . InformationElementIdentifier . CONCATENATED_SM_8BIT_REF_NUM : IEConcatenatedSMEncoder ( False ) , <EOL> gsm_types . InformationElementIdentifier . CONCATENATED_SM_16BIT_REF_NUM : IEConcatenatedSMEncoder ( True ) , <EOL> } <EOL> def encode ( self , iElement ) : <EOL> dataBytes = None <EOL> if iElement . identifier in self . dataEncoders : <EOL> dataBytes = self . dataEncoders [ iElement . identifier ] . encode ( iElement . data ) <EOL> else : <EOL> dataBytes = iElement . data <EOL> length = len ( dataBytes ) <EOL> bytes = '<STR_LIT>' <EOL> bytes += self . iEIEncoder . encode ( iElement . identifier ) <EOL> bytes += self . int8Encoder . encode ( length ) <EOL> bytes += dataBytes <EOL> return bytes <EOL> def decode ( self , file ) : <EOL> fStart = file . tell ( ) <EOL> identifier = None <EOL> try : <EOL> identifier = self . iEIEncoder . decode ( file ) <EOL> except UDHInformationElementIdentifierUnknownError : <EOL> pass <EOL> length = self . int8Encoder . decode ( file ) <EOL> data = None <EOL> if identifier in self . dataEncoders : <EOL> data = self . dataEncoders [ identifier ] . decode ( file ) <EOL> elif length > <NUM_LIT:0> : <EOL> data = self . read ( file , length ) <EOL> parsed = file . tell ( ) - fStart <EOL> if parsed != length + <NUM_LIT:2> : <EOL> raise UDHParseError ( "<STR_LIT>" % ( length + <NUM_LIT:2> , parsed ) ) <EOL> if identifier is None : <EOL> return None <EOL> return gsm_types . InformationElement ( identifier , data ) <EOL> class UserDataHeaderEncoder ( IEncoder ) : <EOL> iEEncoder = InformationElementEncoder ( ) <EOL> int8Encoder = Int8Encoder ( ) <EOL> def encode ( self , udh ) : <EOL> nonRepeatable = { } <EOL> iEBytes = '<STR_LIT>' <EOL> for iElement in udh : <EOL> if not self . isIdentifierRepeatable ( iElement . identifier ) : <EOL> if iElement . identifier in nonRepeatable : <EOL> raise ValueError ( "<STR_LIT>" % str ( iElement . identifier ) ) <EOL> for identifier in self . getIdentifierExclusionList ( iElement . identifier ) : <EOL> if identifier in nonRepeatable : <EOL> raise ValueError ( "<STR_LIT>" % ( str ( iElement . identifier ) , str ( identifier ) ) ) <EOL> nonRepeatable [ iElement . identifier ] = None <EOL> iEBytes += self . iEEncoder . encode ( iElement ) <EOL> headerLen = len ( iEBytes ) <EOL> return self . int8Encoder . encode ( headerLen ) + iEBytes <EOL> def decode ( self , file ) : <EOL> repeatable = [ ] <EOL> nonRepeatable = { } <EOL> headerLen = self . int8Encoder . decode ( file ) <EOL> while file . tell ( ) < headerLen + <NUM_LIT:1> : <EOL> iStart = file . tell ( ) <EOL> iElement = self . iEEncoder . decode ( file ) <EOL> if iElement is not None : <EOL> if self . isIdentifierRepeatable ( iElement . identifier ) : <EOL> repeatable . append ( iElement ) <EOL> else : <EOL> nonRepeatable [ iElement . identifier ] = iElement <EOL> for identifier in self . getIdentifierExclusionList ( iElement . identifier ) : <EOL> if identifier in nonRepeatable : <EOL> del nonRepeatable [ identifier ] <EOL> bytesRead = file . tell ( ) - iStart <EOL> return repeatable + nonRepeatable . values ( ) <EOL> def isIdentifierRepeatable ( self , identifier ) : <EOL> return gsm_constants . information_element_identifier_full_value_map [ gsm_constants . information_element_identifier_name_map [ str ( identifier ) ] ] [ '<STR_LIT>' ] <EOL> def getIdentifierExclusionList ( self , identifier ) : <EOL> nameList = gsm_constants . information_element_identifier_full_value_map [ gsm_constants . information_element_identifier_name_map [ str ( identifier ) ] ] [ '<STR_LIT>' ] <EOL> return [ getattr ( gsm_types . InformationElementIdentifier , name ) for name in nameList ] </s>
