<s> """<STR_LIT>""" <EOL> import random <EOL> from copy import deepcopy <EOL> from . errors import ConfigurationError <EOL> from . debugging import ModuleLogger , bacpypes_debugging <EOL> from . core import deferred <EOL> from . pdu import Address <EOL> from . comm import Server <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> @ bacpypes_debugging <EOL> class Network : <EOL> def __init__ ( self , dropPercent = <NUM_LIT:0.0> ) : <EOL> if _debug : Network . _debug ( "<STR_LIT>" , dropPercent ) <EOL> self . nodes = [ ] <EOL> self . dropPercent = dropPercent <EOL> def add_node ( self , node ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Network . _debug ( "<STR_LIT>" , node ) <EOL> self . nodes . append ( node ) <EOL> node . lan = self <EOL> def remove_node ( self , node ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Network . _debug ( "<STR_LIT>" , node ) <EOL> self . nodes . remove ( node ) <EOL> node . lan = None <EOL> def process_pdu ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Network . _debug ( "<STR_LIT>" , pdu ) <EOL> if self . dropPercent != <NUM_LIT:0.0> : <EOL> if ( random . random ( ) * <NUM_LIT> ) < self . dropPercent : <EOL> if _debug : Network . _debug ( "<STR_LIT>" ) <EOL> return <EOL> if not pdu . pduDestination or not isinstance ( pdu . pduDestination , Address ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> elif pdu . pduDestination . addrType == Address . localBroadcastAddr : <EOL> for n in self . nodes : <EOL> if ( pdu . pduSource != n . address ) : <EOL> n . response ( deepcopy ( pdu ) ) <EOL> elif pdu . pduDestination . addrType == Address . localStationAddr : <EOL> for n in self . nodes : <EOL> if n . promiscuous or ( pdu . pduDestination == n . address ) : <EOL> n . response ( deepcopy ( pdu ) ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def __len__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Network . _debug ( "<STR_LIT>" ) <EOL> return len ( self . nodes ) <EOL> @ bacpypes_debugging <EOL> class Node ( Server ) : <EOL> def __init__ ( self , addr , lan = None , promiscuous = False , spoofing = False , sid = None ) : <EOL> if _debug : <EOL> Node . _debug ( "<STR_LIT>" , <EOL> addr , lan , promiscuous , spoofing , sid <EOL> ) <EOL> Server . __init__ ( self , sid ) <EOL> if not isinstance ( addr , Address ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . lan = None <EOL> self . address = addr <EOL> if lan : <EOL> self . bind ( lan ) <EOL> self . promiscuous = promiscuous <EOL> self . spoofing = spoofing <EOL> def bind ( self , lan ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Node . _debug ( "<STR_LIT>" , lan ) <EOL> lan . add_node ( self ) <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : Node . _debug ( "<STR_LIT>" , pdu ) <EOL> if not self . lan : <EOL> raise ConfigurationError ( "<STR_LIT>" ) <EOL> if pdu . pduSource is None : <EOL> pdu . pduSource = self . address <EOL> elif ( not self . spoofing ) and ( pdu . pduSource != self . address ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> deferred ( self . lan . process_pdu , pdu ) </s>
