<s> """<STR_LIT>""" <EOL> import sys <EOL> import struct <EOL> from time import time as _time <EOL> from . debugging import ModuleLogger , DebugContents , bacpypes_debugging <EOL> from . udp import UDPDirector <EOL> from . task import OneShotTask , RecurringTask <EOL> from . comm import Client , Server , bind , ServiceAccessPoint , ApplicationServiceElement <EOL> from . pdu import Address , LocalBroadcast , LocalStation , PDU , unpack_ip_addr <EOL> from . bvll import BVLPDU , DeleteForeignDeviceTableEntry , DistributeBroadcastToNetwork , FDTEntry , ForwardedNPDU , OriginalBroadcastNPDU , OriginalUnicastNPDU , ReadBroadcastDistributionTable , ReadBroadcastDistributionTableAck , ReadForeignDeviceTable , ReadForeignDeviceTableAck , RegisterForeignDevice , Result , WriteBroadcastDistributionTable , bvl_pdu_types <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> class _MultiplexClient ( Client ) : <EOL> def __init__ ( self , mux ) : <EOL> Client . __init__ ( self ) <EOL> self . multiplexer = mux <EOL> def confirmation ( self , pdu ) : <EOL> self . multiplexer . confirmation ( self , pdu ) <EOL> class _MultiplexServer ( Server ) : <EOL> def __init__ ( self , mux ) : <EOL> Server . __init__ ( self ) <EOL> self . multiplexer = mux <EOL> def indication ( self , pdu ) : <EOL> self . multiplexer . indication ( self , pdu ) <EOL> class UDPMultiplexer : <EOL> def __init__ ( self , addr = None , noBroadcast = False ) : <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , addr , noBroadcast ) <EOL> specialBroadcast = False <EOL> if addr is None : <EOL> self . address = Address ( ) <EOL> self . addrTuple = ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . addrBroadcastTuple = ( '<STR_LIT>' , <NUM_LIT> ) <EOL> else : <EOL> if isinstance ( addr , Address ) : <EOL> self . address = addr <EOL> else : <EOL> self . address = Address ( addr ) <EOL> self . addrTuple = self . address . addrTuple <EOL> self . addrBroadcastTuple = self . address . addrBroadcastTuple <EOL> if ( self . addrTuple == self . addrBroadcastTuple ) : <EOL> self . addrBroadcastTuple = ( '<STR_LIT>' , self . addrTuple [ <NUM_LIT:1> ] ) <EOL> else : <EOL> specialBroadcast = True <EOL> if _debug : <EOL> UDPMultiplexer . _debug ( "<STR_LIT>" , self . address ) <EOL> UDPMultiplexer . _debug ( "<STR_LIT>" , self . addrTuple ) <EOL> UDPMultiplexer . _debug ( "<STR_LIT>" , self . addrBroadcastTuple ) <EOL> self . direct = _MultiplexClient ( self ) <EOL> self . directPort = UDPDirector ( self . addrTuple ) <EOL> bind ( self . direct , self . directPort ) <EOL> if specialBroadcast and ( not noBroadcast ) and sys . platform in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . broadcast = _MultiplexClient ( self ) <EOL> self . broadcastPort = UDPDirector ( self . addrBroadcastTuple , reuse = True ) <EOL> bind ( self . direct , self . broadcastPort ) <EOL> else : <EOL> self . broadcast = None <EOL> self . annexH = _MultiplexServer ( self ) <EOL> self . annexJ = _MultiplexServer ( self ) <EOL> def indication ( self , server , pdu ) : <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , server , pdu ) <EOL> if pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> dest = self . addrBroadcastTuple <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , dest ) <EOL> elif pdu . pduDestination . addrType == Address . localStationAddr : <EOL> dest = unpack_ip_addr ( pdu . pduDestination . addrAddr ) <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , dest ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . directPort . indication ( PDU ( pdu , destination = dest ) ) <EOL> def confirmation ( self , client , pdu ) : <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , client , pdu ) <EOL> if pdu . pduSource == self . addrTuple : <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" ) <EOL> return <EOL> src = Address ( pdu . pduSource ) <EOL> if client is self . direct : <EOL> dest = self . address <EOL> elif client is self . broadcast : <EOL> dest = LocalBroadcast ( ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if not pdu . pduData : <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" ) <EOL> return <EOL> msg_type = struct . unpack ( '<STR_LIT:B>' , pdu . pduData [ : <NUM_LIT:1> ] ) [ <NUM_LIT:0> ] <EOL> if _debug : UDPMultiplexer . _debug ( "<STR_LIT>" , msg_type ) <EOL> if msg_type == <NUM_LIT> : <EOL> if self . annexH . serverPeer : <EOL> self . annexH . response ( PDU ( pdu , source = src , destination = dest ) ) <EOL> elif msg_type == <NUM_LIT> : <EOL> if self . annexJ . serverPeer : <EOL> self . annexJ . response ( PDU ( pdu , source = src , destination = dest ) ) <EOL> else : <EOL> UDPMultiplexer . _warning ( "<STR_LIT>" ) <EOL> bacpypes_debugging ( UDPMultiplexer ) <EOL> class BTR ( Client , Server , DebugContents ) : <EOL> _debug_contents = ( '<STR_LIT>' , ) <EOL> def __init__ ( self , cid = None , sid = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BTR . _debug ( "<STR_LIT>" , cid , sid ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> self . peers = { } <EOL> def indication ( self , pdu ) : <EOL> if _debug : BTR . _debug ( "<STR_LIT>" , pdu ) <EOL> if pdu . pduDestination . addrType == Address . localStationAddr : <EOL> if pdu . pduDestination not in self . peers : <EOL> return <EOL> self . request ( pdu ) <EOL> elif pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> for peerAddr in self . peers . keys ( ) : <EOL> xpdu = PDU ( pdu . pduData , destination = peerAddr ) <EOL> self . request ( xpdu ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def confirmation ( self , pdu ) : <EOL> if _debug : BTR . _debug ( "<STR_LIT>" , pdu ) <EOL> if pdu . pduSource not in self . peers : <EOL> BTR . _warning ( "<STR_LIT>" , pdu . pduSource ) <EOL> return <EOL> self . response ( pdu ) <EOL> def add_peer ( self , peerAddr , networks = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BTR . _debug ( "<STR_LIT>" , peerAddr , networks ) <EOL> if peerAddr in self . peers : <EOL> if not networks : <EOL> networks = [ ] <EOL> else : <EOL> self . peers [ peerAddr ] . extend ( networks ) <EOL> else : <EOL> if not networks : <EOL> networks = [ ] <EOL> self . peers [ peerAddr ] = networks <EOL> def delete_peer ( self , peerAddr ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BTR . _debug ( "<STR_LIT>" , peerAddr ) <EOL> del self . peers [ peerAddr ] <EOL> bacpypes_debugging ( BTR ) <EOL> class AnnexJCodec ( Client , Server ) : <EOL> def __init__ ( self , cid = None , sid = None ) : <EOL> if _debug : AnnexJCodec . _debug ( "<STR_LIT>" , cid , sid ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> def indication ( self , rpdu ) : <EOL> if _debug : AnnexJCodec . _debug ( "<STR_LIT>" , rpdu ) <EOL> bvlpdu = BVLPDU ( ) <EOL> rpdu . encode ( bvlpdu ) <EOL> pdu = PDU ( ) <EOL> bvlpdu . encode ( pdu ) <EOL> self . request ( pdu ) <EOL> def confirmation ( self , pdu ) : <EOL> if _debug : AnnexJCodec . _debug ( "<STR_LIT>" , pdu ) <EOL> bvlpdu = BVLPDU ( ) <EOL> bvlpdu . decode ( pdu ) <EOL> rpdu = bvl_pdu_types [ bvlpdu . bvlciFunction ] ( ) <EOL> rpdu . decode ( bvlpdu ) <EOL> self . response ( rpdu ) <EOL> bacpypes_debugging ( AnnexJCodec ) <EOL> class BIPSAP ( ServiceAccessPoint ) : <EOL> def __init__ ( self , sap = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , sap ) <EOL> ServiceAccessPoint . __init__ ( self , sap ) <EOL> def sap_indication ( self , pdu ) : <EOL> if _debug : BIPSAP . _debug ( "<STR_LIT>" , pdu ) <EOL> self . request ( pdu ) <EOL> def sap_confirmation ( self , pdu ) : <EOL> if _debug : BIPSAP . _debug ( "<STR_LIT>" , pdu ) <EOL> self . request ( pdu ) <EOL> bacpypes_debugging ( BIPSAP ) <EOL> class BIPSimple ( BIPSAP , Client , Server ) : <EOL> def __init__ ( self , sapID = None , cid = None , sid = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , sapID , cid , sid ) <EOL> BIPSAP . __init__ ( self , sapID ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> def indication ( self , pdu ) : <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , pdu ) <EOL> if pdu . pduDestination . addrType == Address . localStationAddr : <EOL> xpdu = OriginalUnicastNPDU ( pdu , destination = pdu . pduDestination , user_data = pdu . pduUserData ) <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> xpdu = OriginalBroadcastNPDU ( pdu , destination = pdu . pduDestination , user_data = pdu . pduUserData ) <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> else : <EOL> BIPSimple . _warning ( "<STR_LIT>" , pdu . pduDestination ) <EOL> def confirmation ( self , pdu ) : <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , pdu ) <EOL> if isinstance ( pdu , Result ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , ReadBroadcastDistributionTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , ReadForeignDeviceTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , OriginalUnicastNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = pdu . pduDestination , user_data = pdu . pduUserData ) <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> elif isinstance ( pdu , OriginalBroadcastNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> elif isinstance ( pdu , ForwardedNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . bvlciAddress , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> if _debug : BIPSimple . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> else : <EOL> BIPSimple . _warning ( "<STR_LIT>" , type ( pdu ) ) <EOL> bacpypes_debugging ( BIPSimple ) <EOL> class BIPForeign ( BIPSAP , Client , Server , OneShotTask , DebugContents ) : <EOL> _debug_contents = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , addr = None , ttl = None , sapID = None , cid = None , sid = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" , addr , ttl , sapID , cid , sid ) <EOL> BIPSAP . __init__ ( self , sapID ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> OneShotTask . __init__ ( self ) <EOL> self . registrationStatus = - <NUM_LIT:1> <EOL> self . bbmdAddress = None <EOL> self . bbmdTimeToLive = None <EOL> if addr : <EOL> if ttl is None : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . register ( addr , ttl ) <EOL> def indication ( self , pdu ) : <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" , pdu ) <EOL> if self . registrationStatus != <NUM_LIT:0> : <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" ) <EOL> return <EOL> if pdu . pduDestination . addrType == Address . localStationAddr : <EOL> xpdu = OriginalUnicastNPDU ( pdu , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduDestination <EOL> self . request ( xpdu ) <EOL> elif pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> xpdu = DistributeBroadcastToNetwork ( pdu , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = self . bbmdAddress <EOL> self . request ( xpdu ) <EOL> else : <EOL> BIPForeign . _warning ( "<STR_LIT>" , pdu . pduDestination ) <EOL> def confirmation ( self , pdu ) : <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" , pdu ) <EOL> if isinstance ( pdu , Result ) : <EOL> if self . registrationStatus == - <NUM_LIT:2> : <EOL> return <EOL> if pdu . pduSource != self . bbmdAddress : <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" ) <EOL> return <EOL> self . registrationStatus = pdu . bvlciResultCode <EOL> if pdu . bvlciResultCode == <NUM_LIT:0> : <EOL> self . install_task ( _time ( ) + self . bbmdTimeToLive ) <EOL> return <EOL> if self . registrationStatus != <NUM_LIT:0> : <EOL> if _debug : BIPForeign . _debug ( "<STR_LIT>" ) <EOL> return <EOL> if isinstance ( pdu , ReadBroadcastDistributionTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , ReadForeignDeviceTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , OriginalUnicastNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = pdu . pduDestination , user_data = pdu . pduUserData ) <EOL> self . response ( xpdu ) <EOL> elif isinstance ( pdu , ForwardedNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . bvlciAddress , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> self . response ( xpdu ) <EOL> else : <EOL> BIPForeign . _warning ( "<STR_LIT>" , type ( pdu ) ) <EOL> def register ( self , addr , ttl ) : <EOL> """<STR_LIT>""" <EOL> if ttl <= <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if isinstance ( addr , Address ) : <EOL> self . bbmdAddress = addr <EOL> else : <EOL> self . bbmdAddress = Address ( addr ) <EOL> self . bbmdTimeToLive = ttl <EOL> self . install_task ( <NUM_LIT:0> ) <EOL> def unregister ( self ) : <EOL> """<STR_LIT>""" <EOL> pdu = RegisterForeignDevice ( <NUM_LIT:0> ) <EOL> pdu . pduDestination = self . bbmdAddress <EOL> self . request ( pdu ) <EOL> self . registrationStatus = - <NUM_LIT:2> <EOL> self . bbmdAddress = None <EOL> self . bbmdTimeToLive = None <EOL> def process_task ( self ) : <EOL> """<STR_LIT>""" <EOL> pdu = RegisterForeignDevice ( self . bbmdTimeToLive ) <EOL> pdu . pduDestination = self . bbmdAddress <EOL> self . request ( pdu ) <EOL> bacpypes_debugging ( BIPForeign ) <EOL> class BIPBBMD ( BIPSAP , Client , Server , RecurringTask , DebugContents ) : <EOL> _debug_contents = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , addr , sapID = None , cid = None , sid = None ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , addr , sapID , cid , sid ) <EOL> BIPSAP . __init__ ( self , sapID ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> RecurringTask . __init__ ( self , <NUM_LIT> ) <EOL> self . bbmdAddress = addr <EOL> self . bbmdBDT = [ ] <EOL> self . bbmdFDT = [ ] <EOL> self . install_task ( ) <EOL> def indication ( self , pdu ) : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , pdu ) <EOL> if pdu . pduDestination . addrType == Address . localStationAddr : <EOL> xpdu = OriginalUnicastNPDU ( pdu , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduDestination <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> xpdu = OriginalBroadcastNPDU ( pdu , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduDestination <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> xpdu = ForwardedNPDU ( self . bbmdAddress , pdu , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> for bdte in self . bbmdBDT : <EOL> if bdte != self . bbmdAddress : <EOL> xpdu . pduDestination = Address ( ( ( bdte . addrIP | ~ bdte . addrMask ) , bdte . addrPort ) ) <EOL> BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> for fdte in self . bbmdFDT : <EOL> xpdu . pduDestination = fdte . fdAddress <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> else : <EOL> BIPBBMD . _warning ( "<STR_LIT>" , pdu . pduDestination ) <EOL> def confirmation ( self , pdu ) : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , pdu ) <EOL> if isinstance ( pdu , Result ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , WriteBroadcastDistributionTable ) : <EOL> xpdu = Result ( code = <NUM_LIT> , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduSource <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , ReadBroadcastDistributionTable ) : <EOL> xpdu = ReadBroadcastDistributionTableAck ( self . bbmdBDT , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduSource <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , ReadBroadcastDistributionTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , ForwardedNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . bvlciAddress , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> xpdu = ForwardedNPDU ( pdu . bvlciAddress , pdu , destination = None , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> if self . bbmdBDT and ( self . bbmdBDT [ <NUM_LIT:0> ] == self . bbmdAddress ) : <EOL> xpdu . pduDestination = LocalBroadcast ( ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" ) <EOL> self . request ( xpdu ) <EOL> for fdte in self . bbmdFDT : <EOL> xpdu . pduDestination = fdte . fdAddress <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , RegisterForeignDevice ) : <EOL> stat = self . register_foreign_device ( pdu . pduSource , pdu . bvlciTimeToLive ) <EOL> xpdu = Result ( code = stat , destination = pdu . pduSource , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , ReadForeignDeviceTable ) : <EOL> xpdu = ReadForeignDeviceTableAck ( self . bbmdFDT , destination = pdu . pduSource , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , ReadForeignDeviceTableAck ) : <EOL> self . sap_response ( pdu ) <EOL> elif isinstance ( pdu , DeleteForeignDeviceTableEntry ) : <EOL> stat = self . delete_foreign_device_table_entry ( pdu . bvlciAddress ) <EOL> xpdu = Result ( code = stat , user_data = pdu . pduUserData ) <EOL> xpdu . pduDestination = pdu . pduSource <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , DistributeBroadcastToNetwork ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> xpdu = ForwardedNPDU ( pdu . pduSource , pdu , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> for bdte in self . bbmdBDT : <EOL> if bdte == self . bbmdAddress : <EOL> xpdu . pduDestination = LocalBroadcast ( ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" ) <EOL> self . request ( xpdu ) <EOL> else : <EOL> xpdu . pduDestination = Address ( ( ( bdte . addrIP | ~ bdte . addrMask ) , bdte . addrPort ) ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> for fdte in self . bbmdFDT : <EOL> if fdte . fdAddress != pdu . pduSource : <EOL> xpdu . pduDestination = fdte . fdAddress <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> elif isinstance ( pdu , OriginalUnicastNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = pdu . pduDestination , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> elif isinstance ( pdu , OriginalBroadcastNPDU ) : <EOL> xpdu = PDU ( pdu . pduData , source = pdu . pduSource , destination = LocalBroadcast ( ) , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> self . response ( xpdu ) <EOL> xpdu = ForwardedNPDU ( pdu . pduSource , pdu , user_data = pdu . pduUserData ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu ) <EOL> for bdte in self . bbmdBDT : <EOL> if bdte != self . bbmdAddress : <EOL> xpdu . pduDestination = Address ( ( ( bdte . addrIP | ~ bdte . addrMask ) , bdte . addrPort ) ) <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> for fdte in self . bbmdFDT : <EOL> xpdu . pduDestination = fdte . fdAddress <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , xpdu . pduDestination ) <EOL> self . request ( xpdu ) <EOL> else : <EOL> BIPBBMD . _warning ( "<STR_LIT>" , type ( pdu ) ) <EOL> def register_foreign_device ( self , addr , ttl ) : <EOL> """<STR_LIT>""" <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , addr , ttl ) <EOL> if isinstance ( addr , Address ) : <EOL> pass <EOL> elif isinstance ( addr , str ) : <EOL> addr = LocalStation ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> for fdte in self . bbmdFDT : <EOL> if addr == fdte . fdAddress : <EOL> break <EOL> else : <EOL> fdte = FDTEntry ( ) <EOL> fdte . fdAddress = addr <EOL> self . bbmdFDT . append ( fdte ) <EOL> fdte . fdTTL = ttl <EOL> fdte . fdRemain = ttl + <NUM_LIT:5> <EOL> return <NUM_LIT:0> <EOL> def delete_foreign_device_table_entry ( self , addr ) : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , addr ) <EOL> if isinstance ( addr , Address ) : <EOL> pass <EOL> elif isinstance ( addr , str ) : <EOL> addr = LocalStation ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> stat = <NUM_LIT:0> <EOL> for i in range ( len ( self . bbmdFDT ) - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> ) : <EOL> if addr == self . bbmdFDT [ i ] . fdAddress : <EOL> del self . bbmdFDT [ i ] <EOL> break <EOL> else : <EOL> stat = <NUM_LIT> <EOL> return stat <EOL> def process_task ( self ) : <EOL> for i in range ( len ( self . bbmdFDT ) - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> ) : <EOL> fdte = self . bbmdFDT [ i ] <EOL> fdte . fdRemain -= <NUM_LIT:1> <EOL> if fdte . fdRemain <= <NUM_LIT:0> : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , fdte . fdAddress ) <EOL> del self . bbmdFDT [ i ] <EOL> def add_peer ( self , addr ) : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , addr ) <EOL> if isinstance ( addr , Address ) : <EOL> pass <EOL> elif isinstance ( addr , str ) : <EOL> addr = LocalStation ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if self . bbmdBDT and ( addr == self . bbmdAddress ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> for bdte in self . bbmdBDT : <EOL> if addr == bdte : <EOL> break <EOL> else : <EOL> self . bbmdBDT . append ( addr ) <EOL> def delete_peer ( self , addr ) : <EOL> if _debug : BIPBBMD . _debug ( "<STR_LIT>" , addr ) <EOL> if isinstance ( addr , Address ) : <EOL> pass <EOL> elif isinstance ( addr , str ) : <EOL> addr = LocalStation ( addr ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> for i in range ( len ( self . bbmdBDT ) - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> ) : <EOL> if addr == self . bbmdBDT [ i ] : <EOL> del self . bbmdBDT [ i ] <EOL> break <EOL> else : <EOL> pass <EOL> bacpypes_debugging ( BIPBBMD ) <EOL> class BVLLServiceElement ( ApplicationServiceElement ) : <EOL> def __init__ ( self , aseID = None ) : <EOL> if _debug : BVLLServiceElement . _debug ( "<STR_LIT>" , aseID ) <EOL> ApplicationServiceElement . __init__ ( self , aseID ) <EOL> def indication ( self , npdu ) : <EOL> if _debug : BVLLServiceElement . _debug ( "<STR_LIT>" , npdu ) <EOL> fn = npdu . __class__ . __name__ <EOL> if hasattr ( self , fn ) : <EOL> getattr ( self , fn ) ( npdu ) <EOL> else : <EOL> BVLLServiceElement . _warning ( "<STR_LIT>" , fn ) <EOL> def confirmation ( self , npdu ) : <EOL> if _debug : BVLLServiceElement . _debug ( "<STR_LIT>" , npdu ) <EOL> fn = npdu . __class__ . __name__ <EOL> if hasattr ( self , fn ) : <EOL> getattr ( self , fn ) ( npdu ) <EOL> else : <EOL> BVLLServiceElement . _warning ( "<STR_LIT>" , fn ) <EOL> bacpypes_debugging ( BVLLServiceElement ) </s>
