<s> from coapthon . utils import parse_blockwise <EOL> from coapthon import defines <EOL> from coapthon . messages . option import Option <EOL> __author__ = '<STR_LIT>' <EOL> __version__ = "<STR_LIT>" <EOL> class Message ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _type = None <EOL> self . _mid = None <EOL> self . _token = None <EOL> self . _options = [ ] <EOL> self . _payload = None <EOL> self . _destination = None <EOL> self . _source = None <EOL> self . _code = None <EOL> self . _acknowledged = None <EOL> self . _rejected = None <EOL> self . _timeouted = None <EOL> self . _cancelled = None <EOL> self . _duplicated = None <EOL> self . _timestamp = None <EOL> self . _version = <NUM_LIT:1> <EOL> @ property <EOL> def version ( self ) : <EOL> return self . _version <EOL> @ version . setter <EOL> def version ( self , v ) : <EOL> if not isinstance ( v , int ) or v != <NUM_LIT:1> : <EOL> raise AttributeError <EOL> self . _version = v <EOL> @ property <EOL> def type ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _type <EOL> @ type . setter <EOL> def type ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if value not in defines . Types . values ( ) : <EOL> raise AttributeError <EOL> self . _type = value <EOL> @ property <EOL> def mid ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _mid <EOL> @ mid . setter <EOL> def mid ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( value , int ) or value > <NUM_LIT> : <EOL> raise AttributeError <EOL> self . _mid = value <EOL> @ mid . deleter <EOL> def mid ( self ) : <EOL> self . _mid = None <EOL> @ property <EOL> def token ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _token <EOL> @ token . setter <EOL> def token ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( value , str ) : <EOL> value = str ( value ) <EOL> if len ( value ) > <NUM_LIT> : <EOL> raise AttributeError <EOL> self . _token = value <EOL> @ token . deleter <EOL> def token ( self ) : <EOL> self . _token = None <EOL> @ property <EOL> def options ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _options <EOL> @ options . setter <EOL> def options ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if value is None : <EOL> value = [ ] <EOL> assert isinstance ( value , list ) <EOL> self . _options = value <EOL> @ property <EOL> def payload ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _payload <EOL> @ payload . setter <EOL> def payload ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( value , tuple ) : <EOL> content_type , payload = value <EOL> self . content_type = content_type <EOL> self . _payload = payload <EOL> else : <EOL> self . _payload = value <EOL> @ property <EOL> def destination ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _destination <EOL> @ destination . setter <EOL> def destination ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if value is not None and ( not isinstance ( value , tuple ) or len ( value ) ) != <NUM_LIT:2> : <EOL> raise AttributeError <EOL> self . _destination = value <EOL> @ property <EOL> def source ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _source <EOL> @ source . setter <EOL> def source ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( value , tuple ) or len ( value ) != <NUM_LIT:2> : <EOL> raise AttributeError <EOL> self . _source = value <EOL> @ property <EOL> def code ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _code <EOL> @ code . setter <EOL> def code ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if value not in defines . Codes . LIST . keys ( ) and value is not None : <EOL> raise AttributeError <EOL> self . _code = value <EOL> @ property <EOL> def acknowledged ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _acknowledged <EOL> @ acknowledged . setter <EOL> def acknowledged ( self , value ) : <EOL> """<STR_LIT>""" <EOL> assert ( isinstance ( value , bool ) ) <EOL> self . _acknowledged = value <EOL> if value : <EOL> self . _timeouted = False <EOL> self . _rejected = False <EOL> self . _cancelled = False <EOL> @ property <EOL> def rejected ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _rejected <EOL> @ rejected . setter <EOL> def rejected ( self , value ) : <EOL> """<STR_LIT>""" <EOL> assert ( isinstance ( value , bool ) ) <EOL> self . _rejected = value <EOL> if value : <EOL> self . _timeouted = False <EOL> self . _acknowledged = False <EOL> self . _cancelled = True <EOL> @ property <EOL> def timeouted ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _timeouted <EOL> @ timeouted . setter <EOL> def timeouted ( self , value ) : <EOL> """<STR_LIT>""" <EOL> assert ( isinstance ( value , bool ) ) <EOL> self . _timeouted = value <EOL> if value : <EOL> self . _acknowledged = False <EOL> self . _rejected = False <EOL> self . _cancelled = True <EOL> @ property <EOL> def duplicated ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _duplicated <EOL> @ duplicated . setter <EOL> def duplicated ( self , value ) : <EOL> """<STR_LIT>""" <EOL> assert ( isinstance ( value , bool ) ) <EOL> self . _duplicated = value <EOL> @ property <EOL> def timestamp ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _timestamp <EOL> @ timestamp . setter <EOL> def timestamp ( self , value ) : <EOL> """<STR_LIT>""" <EOL> self . _timestamp = value <EOL> def _already_in ( self , option ) : <EOL> """<STR_LIT>""" <EOL> for opt in self . _options : <EOL> if option . number == opt . number : <EOL> return True <EOL> return False <EOL> def add_option ( self , option ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( option , Option ) <EOL> repeatable = defines . OptionRegistry . LIST [ option . number ] . repeatable <EOL> if not repeatable : <EOL> ret = self . _already_in ( option ) <EOL> if ret : <EOL> raise TypeError ( "<STR_LIT>" , option . name ) <EOL> else : <EOL> self . _options . append ( option ) <EOL> else : <EOL> self . _options . append ( option ) <EOL> def del_option ( self , option ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( option , Option ) <EOL> while option in list ( self . _options ) : <EOL> self . _options . remove ( option ) <EOL> def del_option_by_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> for o in list ( self . _options ) : <EOL> assert isinstance ( o , Option ) <EOL> if o . name == name : <EOL> self . _options . remove ( o ) <EOL> def del_option_by_number ( self , number ) : <EOL> """<STR_LIT>""" <EOL> for o in list ( self . _options ) : <EOL> assert isinstance ( o , Option ) <EOL> if o . number == number : <EOL> self . _options . remove ( o ) <EOL> @ property <EOL> def etag ( self ) : <EOL> """<STR_LIT>""" <EOL> value = [ ] <EOL> for option in self . options : <EOL> if option . number == defines . OptionRegistry . ETAG . number : <EOL> value . append ( option . value ) <EOL> return value <EOL> @ etag . setter <EOL> def etag ( self , etag ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( etag , list ) : <EOL> etag = [ etag ] <EOL> for e in etag : <EOL> option = Option ( ) <EOL> option . number = defines . OptionRegistry . ETAG . number <EOL> option . value = e <EOL> self . add_option ( option ) <EOL> @ etag . deleter <EOL> def etag ( self ) : <EOL> """<STR_LIT>""" <EOL> self . del_option_by_number ( defines . OptionRegistry . ETAG . number ) <EOL> @ property <EOL> def content_type ( self ) : <EOL> """<STR_LIT>""" <EOL> value = <NUM_LIT:0> <EOL> for option in self . options : <EOL> if option . number == defines . OptionRegistry . CONTENT_TYPE . number : <EOL> value = int ( option . value ) <EOL> return value <EOL> @ content_type . setter <EOL> def content_type ( self , content_type ) : <EOL> """<STR_LIT>""" <EOL> option = Option ( ) <EOL> option . number = defines . OptionRegistry . CONTENT_TYPE . number <EOL> option . value = int ( content_type ) <EOL> self . add_option ( option ) <EOL> @ content_type . deleter <EOL> def content_type ( self ) : <EOL> self . del_option_by_number ( defines . OptionRegistry . CONTENT_TYPE . number ) <EOL> @ property <EOL> def observe ( self ) : <EOL> """<STR_LIT>""" <EOL> for option in self . options : <EOL> if option . number == defines . OptionRegistry . OBSERVE . number : <EOL> if option . value is None : <EOL> return <NUM_LIT:0> <EOL> return option . value <EOL> return None <EOL> @ observe . setter <EOL> def observe ( self , ob ) : <EOL> """<STR_LIT>""" <EOL> option = Option ( ) <EOL> option . number = defines . OptionRegistry . OBSERVE . number <EOL> option . value = ob <EOL> self . del_option_by_number ( defines . OptionRegistry . OBSERVE . number ) <EOL> self . add_option ( option ) <EOL> @ observe . deleter <EOL> def observe ( self ) : <EOL> self . del_option_by_number ( defines . OptionRegistry . OBSERVE . number ) <EOL> @ property <EOL> def block1 ( self ) : <EOL> """<STR_LIT>""" <EOL> value = None <EOL> for option in self . options : <EOL> if option . number == defines . OptionRegistry . BLOCK1 . number : <EOL> value = parse_blockwise ( option . value ) <EOL> return value <EOL> @ block1 . setter <EOL> def block1 ( self , value ) : <EOL> """<STR_LIT>""" <EOL> option = Option ( ) <EOL> option . number = defines . OptionRegistry . BLOCK1 . number <EOL> num , m , size = value <EOL> if size > <NUM_LIT> : <EOL> szx = <NUM_LIT:6> <EOL> elif <NUM_LIT> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:5> <EOL> elif <NUM_LIT> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:4> <EOL> elif <NUM_LIT:64> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:3> <EOL> elif <NUM_LIT:32> < size <= <NUM_LIT:64> : <EOL> szx = <NUM_LIT:2> <EOL> elif <NUM_LIT:16> < size <= <NUM_LIT:32> : <EOL> szx = <NUM_LIT:1> <EOL> else : <EOL> szx = <NUM_LIT:0> <EOL> value = ( num << <NUM_LIT:4> ) <EOL> value |= ( m << <NUM_LIT:3> ) <EOL> value |= szx <EOL> option . value = value <EOL> self . add_option ( option ) <EOL> @ block1 . deleter <EOL> def block1 ( self ) : <EOL> self . del_option_by_number ( defines . OptionRegistry . BLOCK1 . number ) <EOL> @ property <EOL> def block2 ( self ) : <EOL> """<STR_LIT>""" <EOL> value = None <EOL> for option in self . options : <EOL> if option . number == defines . OptionRegistry . BLOCK2 . number : <EOL> value = parse_blockwise ( option . value ) <EOL> return value <EOL> @ block2 . setter <EOL> def block2 ( self , value ) : <EOL> option = Option ( ) <EOL> option . number = defines . OptionRegistry . BLOCK2 . number <EOL> num , m , size = value <EOL> if size > <NUM_LIT> : <EOL> szx = <NUM_LIT:6> <EOL> elif <NUM_LIT> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:5> <EOL> elif <NUM_LIT> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:4> <EOL> elif <NUM_LIT:64> < size <= <NUM_LIT> : <EOL> szx = <NUM_LIT:3> <EOL> elif <NUM_LIT:32> < size <= <NUM_LIT:64> : <EOL> szx = <NUM_LIT:2> <EOL> elif <NUM_LIT:16> < size <= <NUM_LIT:32> : <EOL> szx = <NUM_LIT:1> <EOL> else : <EOL> szx = <NUM_LIT:0> <EOL> value = ( num << <NUM_LIT:4> ) <EOL> value |= ( m << <NUM_LIT:3> ) <EOL> value |= szx <EOL> option . value = value <EOL> self . add_option ( option ) <EOL> @ block2 . deleter <EOL> def block2 ( self ) : <EOL> self . del_option_by_number ( defines . OptionRegistry . BLOCK2 . number ) <EOL> @ property <EOL> def line_print ( self ) : <EOL> inv_types = { v : k for k , v in defines . Types . iteritems ( ) } <EOL> if self . _code is None : <EOL> self . _code = defines . Codes . EMPTY . number <EOL> msg = "<STR_LIT>" . format ( source = self . _source , destination = self . _destination , type = inv_types [ self . _type ] , mid = self . _mid , <EOL> code = defines . Codes . LIST [ self . _code ] . name , token = self . _token ) <EOL> for opt in self . _options : <EOL> msg += "<STR_LIT>" . format ( name = opt . name , value = opt . value ) <EOL> msg += "<STR_LIT:]>" <EOL> if self . payload is not None : <EOL> msg += "<STR_LIT>" . format ( payload = self . payload [ <NUM_LIT:0> : <NUM_LIT:20> ] , length = len ( self . payload ) ) <EOL> else : <EOL> msg += "<STR_LIT>" <EOL> return msg <EOL> def __str__ ( self ) : <EOL> return self . line_print <EOL> def pretty_print ( self ) : <EOL> """<STR_LIT>""" <EOL> msg = "<STR_LIT>" + str ( self . _source ) + "<STR_LIT:\n>" <EOL> msg += "<STR_LIT>" + str ( self . _destination ) + "<STR_LIT:\n>" <EOL> inv_types = { v : k for k , v in defines . Types . iteritems ( ) } <EOL> msg += "<STR_LIT>" + str ( inv_types [ self . _type ] ) + "<STR_LIT:\n>" <EOL> msg += "<STR_LIT>" + str ( self . _mid ) + "<STR_LIT:\n>" <EOL> if self . _code is None : <EOL> self . _code = <NUM_LIT:0> <EOL> msg += "<STR_LIT>" + str ( defines . Codes . LIST [ self . _code ] . name ) + "<STR_LIT:\n>" <EOL> msg += "<STR_LIT>" + str ( self . _token ) + "<STR_LIT:\n>" <EOL> for opt in self . _options : <EOL> msg += str ( opt ) <EOL> msg += "<STR_LIT>" + "<STR_LIT:\n>" <EOL> msg += str ( self . _payload ) + "<STR_LIT:\n>" <EOL> return msg </s>
