<s> """<STR_LIT>""" <EOL> import re <EOL> import socket <EOL> import struct <EOL> from . debugging import ModuleLogger , bacpypes_debugging , btox , xtob <EOL> from . comm import PCI as _PCI , PDUData <EOL> _short_mask = <NUM_LIT> L <EOL> _long_mask = <NUM_LIT> L <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> ip_address_mask_port_re = re . compile ( r'<STR_LIT>' ) <EOL> ethernet_re = re . compile ( r'<STR_LIT>' ) <EOL> class Address : <EOL> nullAddr = <NUM_LIT:0> <EOL> localBroadcastAddr = <NUM_LIT:1> <EOL> localStationAddr = <NUM_LIT:2> <EOL> remoteBroadcastAddr = <NUM_LIT:3> <EOL> remoteStationAddr = <NUM_LIT:4> <EOL> globalBroadcastAddr = <NUM_LIT:5> <EOL> def __init__ ( self , * args ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" , args ) <EOL> self . addrType = Address . nullAddr <EOL> self . addrNet = None <EOL> self . addrLen = <NUM_LIT:0> <EOL> self . addrAddr = '<STR_LIT>' <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> self . decode_address ( args [ <NUM_LIT:0> ] ) <EOL> elif len ( args ) == <NUM_LIT:2> : <EOL> self . decode_address ( args [ <NUM_LIT:1> ] ) <EOL> if self . addrType == Address . localStationAddr : <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = args [ <NUM_LIT:0> ] <EOL> elif self . addrType == Address . localBroadcastAddr : <EOL> self . addrType = Address . remoteBroadcastAddr <EOL> self . addrNet = args [ <NUM_LIT:0> ] <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def decode_address ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Address . _debug ( "<STR_LIT>" , addr , type ( addr ) ) <EOL> self . addrType = Address . localStationAddr <EOL> self . addrNet = None <EOL> if addr == "<STR_LIT:*>" : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrType = Address . localBroadcastAddr <EOL> self . addrNet = None <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> elif addr == "<STR_LIT>" : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrType = Address . globalBroadcastAddr <EOL> self . addrNet = None <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> elif isinstance ( addr , int ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> if ( addr < <NUM_LIT:0> ) or ( addr >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrAddr = struct . pack ( '<STR_LIT:B>' , addr ) <EOL> self . addrLen = <NUM_LIT:1> <EOL> elif isinstance ( addr , str ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> m = ip_address_mask_port_re . match ( addr ) <EOL> if m : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> net , addr , mask , port = m . groups ( ) <EOL> if not mask : mask = '<STR_LIT>' <EOL> if not port : port = '<STR_LIT>' <EOL> if _debug : Address . _debug ( "<STR_LIT>" , net , addr , mask , port ) <EOL> if net : <EOL> net = int ( net ) <EOL> if ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = net <EOL> self . addrPort = int ( port ) <EOL> self . addrTuple = ( addr , self . addrPort ) <EOL> addrstr = socket . inet_aton ( addr ) <EOL> self . addrIP = struct . unpack ( '<STR_LIT>' , addrstr ) [ <NUM_LIT:0> ] <EOL> self . addrMask = ( _long_mask << ( <NUM_LIT:32> - int ( mask ) ) ) & _long_mask <EOL> self . addrHost = ( self . addrIP & ~ self . addrMask ) <EOL> self . addrSubnet = ( self . addrIP & self . addrMask ) <EOL> bcast = ( self . addrSubnet | ~ self . addrMask ) <EOL> self . addrBroadcastTuple = ( socket . inet_ntoa ( struct . pack ( '<STR_LIT>' , bcast & _long_mask ) ) , self . addrPort ) <EOL> self . addrAddr = addrstr + struct . pack ( '<STR_LIT>' , self . addrPort & _short_mask ) <EOL> self . addrLen = <NUM_LIT:6> <EOL> elif ethernet_re . match ( addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrAddr = xtob ( addr , '<STR_LIT::>' ) <EOL> self . addrLen = len ( self . addrAddr ) <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> addr = int ( addr ) <EOL> if ( addr > <NUM_LIT:255> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrAddr = struct . pack ( '<STR_LIT:B>' , addr ) <EOL> self . addrLen = <NUM_LIT:1> <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> addr = int ( addr [ : - <NUM_LIT:2> ] ) <EOL> if ( addr >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteBroadcastAddr <EOL> self . addrNet = addr <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> net , addr = addr . split ( '<STR_LIT::>' ) <EOL> net = int ( net ) <EOL> addr = int ( addr ) <EOL> if ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if ( addr > <NUM_LIT:255> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = net <EOL> self . addrAddr = struct . pack ( '<STR_LIT:B>' , addr ) <EOL> self . addrLen = <NUM_LIT:1> <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrAddr = xtob ( addr [ <NUM_LIT:2> : ] ) <EOL> self . addrLen = len ( self . addrAddr ) <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrAddr = xtob ( addr [ <NUM_LIT:2> : - <NUM_LIT:1> ] ) <EOL> self . addrLen = len ( self . addrAddr ) <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> net , addr = addr . split ( '<STR_LIT::>' ) <EOL> net = int ( net ) <EOL> if ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = net <EOL> self . addrAddr = xtob ( addr [ <NUM_LIT:2> : ] ) <EOL> self . addrLen = len ( self . addrAddr ) <EOL> elif re . match ( r"<STR_LIT>" , addr ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> net , addr = addr . split ( '<STR_LIT::>' ) <EOL> net = int ( net ) <EOL> if ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = net <EOL> self . addrAddr = xtob ( addr [ <NUM_LIT:2> : - <NUM_LIT:1> ] ) <EOL> self . addrLen = len ( self . addrAddr ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif isinstance ( addr , tuple ) : <EOL> addr , port = addr <EOL> self . addrPort = int ( port ) <EOL> if isinstance ( addr , str ) : <EOL> if not addr : <EOL> addrstr = '<STR_LIT>' <EOL> else : <EOL> addrstr = socket . inet_aton ( addr ) <EOL> self . addrTuple = ( addr , self . addrPort ) <EOL> elif isinstance ( addr , ( int , long ) ) : <EOL> addrstr = struct . pack ( '<STR_LIT>' , addr & _long_mask ) <EOL> self . addrTuple = ( socket . inet_ntoa ( addrstr ) , self . addrPort ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if _debug : Address . _debug ( "<STR_LIT>" , addrstr ) <EOL> self . addrIP = struct . unpack ( '<STR_LIT>' , addrstr ) [ <NUM_LIT:0> ] <EOL> self . addrMask = _long_mask <EOL> self . addrHost = None <EOL> self . addrSubnet = None <EOL> self . addrBroadcastTuple = self . addrTuple <EOL> self . addrAddr = addrstr + struct . pack ( '<STR_LIT>' , self . addrPort & _short_mask ) <EOL> self . addrLen = <NUM_LIT:6> <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> def __str__ ( self ) : <EOL> if self . addrType == Address . nullAddr : <EOL> return '<STR_LIT>' <EOL> elif self . addrType == Address . localBroadcastAddr : <EOL> return '<STR_LIT:*>' <EOL> elif self . addrType == Address . localStationAddr : <EOL> rslt = '<STR_LIT>' <EOL> if self . addrLen == <NUM_LIT:1> : <EOL> rslt += str ( ord ( self . addrAddr ) ) <EOL> else : <EOL> port = struct . unpack ( '<STR_LIT>' , self . addrAddr [ - <NUM_LIT:2> : ] ) [ <NUM_LIT:0> ] <EOL> if ( len ( self . addrAddr ) == <NUM_LIT:6> ) and ( port >= <NUM_LIT> ) and ( port <= <NUM_LIT> ) : <EOL> rslt += '<STR_LIT:.>' . join ( [ "<STR_LIT>" % ord ( x ) for x in self . addrAddr [ <NUM_LIT:0> : <NUM_LIT:4> ] ] ) <EOL> if port != <NUM_LIT> : <EOL> rslt += '<STR_LIT::>' + str ( port ) <EOL> else : <EOL> rslt += '<STR_LIT>' + btox ( self . addrAddr ) <EOL> return rslt <EOL> elif self . addrType == Address . remoteBroadcastAddr : <EOL> return '<STR_LIT>' % ( self . addrNet , ) <EOL> elif self . addrType == Address . remoteStationAddr : <EOL> rslt = '<STR_LIT>' % ( self . addrNet , ) <EOL> if self . addrLen == <NUM_LIT:1> : <EOL> rslt += str ( ord ( self . addrAddr [ <NUM_LIT:0> ] ) ) <EOL> else : <EOL> port = struct . unpack ( '<STR_LIT>' , self . addrAddr [ - <NUM_LIT:2> : ] ) [ <NUM_LIT:0> ] <EOL> if ( len ( self . addrAddr ) == <NUM_LIT:6> ) and ( port >= <NUM_LIT> ) and ( port <= <NUM_LIT> ) : <EOL> rslt += '<STR_LIT:.>' . join ( [ "<STR_LIT>" % ord ( x ) for x in self . addrAddr [ <NUM_LIT:0> : <NUM_LIT:4> ] ] ) <EOL> if port != <NUM_LIT> : <EOL> rslt += '<STR_LIT::>' + str ( port ) <EOL> else : <EOL> rslt += '<STR_LIT>' + btox ( self . addrAddr ) <EOL> return rslt <EOL> elif self . addrType == Address . globalBroadcastAddr : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % self . addrType ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . __class__ . __name__ , self . __str__ ( ) ) <EOL> def __hash__ ( self ) : <EOL> return hash ( ( self . addrType , self . addrNet , self . addrAddr ) ) <EOL> def __eq__ ( self , arg ) : <EOL> if not isinstance ( arg , Address ) : <EOL> arg = Address ( arg ) <EOL> return ( self . addrType == arg . addrType ) and ( self . addrNet == arg . addrNet ) and ( self . addrAddr == arg . addrAddr ) <EOL> def __ne__ ( self , arg ) : <EOL> return not self . __eq__ ( arg ) <EOL> def dict_contents ( self , use_dict = None , as_class = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : _log . debug ( "<STR_LIT>" , use_dict , as_class ) <EOL> return str ( self ) <EOL> bacpypes_debugging ( Address ) <EOL> def pack_ip_addr ( addr ) : <EOL> """<STR_LIT>""" <EOL> addr , port = addr <EOL> return socket . inet_aton ( addr ) + struct . pack ( '<STR_LIT>' , port & _short_mask ) <EOL> def unpack_ip_addr ( addr ) : <EOL> """<STR_LIT>""" <EOL> return ( socket . inet_ntoa ( addr [ <NUM_LIT:0> : <NUM_LIT:4> ] ) , struct . unpack ( '<STR_LIT>' , addr [ <NUM_LIT:4> : <NUM_LIT:6> ] ) [ <NUM_LIT:0> ] ) <EOL> class LocalStation ( Address ) : <EOL> def __init__ ( self , addr ) : <EOL> self . addrType = Address . localStationAddr <EOL> self . addrNet = None <EOL> if isinstance ( addr , int ) : <EOL> if ( addr < <NUM_LIT:0> ) or ( addr >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrAddr = struct . pack ( '<STR_LIT:B>' , addr ) <EOL> self . addrLen = <NUM_LIT:1> <EOL> elif isinstance ( addr , str ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrAddr = addr <EOL> self . addrLen = len ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> class RemoteStation ( Address ) : <EOL> def __init__ ( self , net , addr ) : <EOL> if not isinstance ( net , int ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if ( net < <NUM_LIT:0> ) or ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteStationAddr <EOL> self . addrNet = net <EOL> if isinstance ( addr , int ) : <EOL> if ( addr < <NUM_LIT:0> ) or ( addr >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrAddr = struct . pack ( '<STR_LIT:B>' , addr ) <EOL> self . addrLen = <NUM_LIT:1> <EOL> elif isinstance ( addr , str ) : <EOL> if _debug : Address . _debug ( "<STR_LIT>" ) <EOL> self . addrAddr = addr <EOL> self . addrLen = len ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> class LocalBroadcast ( Address ) : <EOL> def __init__ ( self ) : <EOL> self . addrType = Address . localBroadcastAddr <EOL> self . addrNet = None <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> class RemoteBroadcast ( Address ) : <EOL> def __init__ ( self , net ) : <EOL> if not isinstance ( net , int ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if ( net < <NUM_LIT:0> ) or ( net >= <NUM_LIT> ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . addrType = Address . remoteBroadcastAddr <EOL> self . addrNet = net <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> class GlobalBroadcast ( Address ) : <EOL> def __init__ ( self ) : <EOL> self . addrType = Address . globalBroadcastAddr <EOL> self . addrNet = None <EOL> self . addrAddr = None <EOL> self . addrLen = None <EOL> class PCI ( _PCI ) : <EOL> _debug_contents = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> if _debug : PCI . _debug ( "<STR_LIT>" , args , kwargs ) <EOL> my_kwargs = { } <EOL> other_kwargs = { } <EOL> for element in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if element in kwargs : <EOL> my_kwargs [ element ] = kwargs [ element ] <EOL> for kw in kwargs : <EOL> if kw not in my_kwargs : <EOL> other_kwargs [ kw ] = kwargs [ kw ] <EOL> if _debug : PCI . _debug ( "<STR_LIT>" , my_kwargs ) <EOL> if _debug : PCI . _debug ( "<STR_LIT>" , other_kwargs ) <EOL> super ( PCI , self ) . __init__ ( * args , ** other_kwargs ) <EOL> self . pduExpectingReply = my_kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . pduNetworkPriority = my_kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> def update ( self , pci ) : <EOL> """<STR_LIT>""" <EOL> _PCI . update ( self , pci ) <EOL> self . pduExpectingReply = pci . pduExpectingReply <EOL> self . pduNetworkPriority = pci . pduNetworkPriority <EOL> def pci_contents ( self , use_dict = None , as_class = dict ) : <EOL> """<STR_LIT>""" <EOL> if _debug : PCI . _debug ( "<STR_LIT>" , use_dict , as_class ) <EOL> if use_dict is None : <EOL> use_dict = as_class ( ) <EOL> _PCI . pci_contents ( self , use_dict = use_dict , as_class = as_class ) <EOL> use_dict . __setitem__ ( '<STR_LIT>' , self . pduExpectingReply ) <EOL> use_dict . __setitem__ ( '<STR_LIT>' , self . pduNetworkPriority ) <EOL> return use_dict <EOL> def dict_contents ( self , use_dict = None , as_class = dict ) : <EOL> """<STR_LIT>""" <EOL> if _debug : PCI . _debug ( "<STR_LIT>" , use_dict , as_class ) <EOL> return self . pci_contents ( use_dict = use_dict , as_class = as_class ) <EOL> bacpypes_debugging ( PCI ) <EOL> class PDU ( PCI , PDUData ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> if _debug : PDU . _debug ( "<STR_LIT>" , args , kwargs ) <EOL> super ( PDU , self ) . __init__ ( * args , ** kwargs ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . pduSource , self . pduDestination , btox ( self . pduData , '<STR_LIT:.>' ) ) <EOL> def dict_contents ( self , use_dict = None , as_class = dict ) : <EOL> """<STR_LIT>""" <EOL> if _debug : PDU . _debug ( "<STR_LIT>" , use_dict , as_class ) <EOL> if use_dict is None : <EOL> use_dict = as_class ( ) <EOL> self . pci_contents ( use_dict = use_dict , as_class = as_class ) <EOL> self . pdudata_contents ( use_dict = use_dict , as_class = as_class ) <EOL> return use_dict <EOL> bacpypes_debugging ( PDU ) </s>
