<s> """<STR_LIT>""" <EOL> import asyncore <EOL> import socket <EOL> import cPickle as pickle <EOL> from time import time as _time , sleep as _sleep <EOL> from StringIO import StringIO <EOL> from . debugging import ModuleLogger , DebugContents , bacpypes_debugging <EOL> from . core import deferred <EOL> from . task import FunctionTask , OneShotFunction <EOL> from . comm import PDU , Client , Server <EOL> from . comm import ServiceAccessPoint , ApplicationServiceElement <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> REBIND_SLEEP_INTERVAL = <NUM_LIT> <EOL> class PickleActorMixIn : <EOL> def __init__ ( self , * args ) : <EOL> if _debug : PickleActorMixIn . _debug ( "<STR_LIT>" , args ) <EOL> super ( PickleActorMixIn , self ) . __init__ ( * args ) <EOL> self . pickleBuffer = '<STR_LIT>' <EOL> def indication ( self , pdu ) : <EOL> if _debug : PickleActorMixIn . _debug ( "<STR_LIT>" , pdu ) <EOL> pdu . pduData = pickle . dumps ( pdu . pduData ) <EOL> super ( PickleActorMixIn , self ) . indication ( pdu ) <EOL> def response ( self , pdu ) : <EOL> if _debug : PickleActorMixIn . _debug ( "<STR_LIT>" , pdu ) <EOL> self . pickleBuffer += pdu . pduData <EOL> strm = StringIO ( self . pickleBuffer ) <EOL> pos = <NUM_LIT:0> <EOL> while ( pos < strm . len ) : <EOL> try : <EOL> msg = pickle . load ( strm ) <EOL> except : <EOL> break <EOL> rpdu = PDU ( msg ) <EOL> rpdu . update ( pdu ) <EOL> super ( PickleActorMixIn , self ) . response ( rpdu ) <EOL> pos = strm . tell ( ) <EOL> if ( pos < strm . len ) : <EOL> self . pickleBuffer = self . pickleBuffer [ pos : ] <EOL> else : <EOL> self . pickleBuffer = '<STR_LIT>' <EOL> bacpypes_debugging ( PickleActorMixIn ) <EOL> class TCPClient ( asyncore . dispatcher ) : <EOL> def __init__ ( self , peer ) : <EOL> if _debug : TCPClient . _debug ( "<STR_LIT>" , peer ) <EOL> asyncore . dispatcher . __init__ ( self ) <EOL> self . create_socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . peer = peer <EOL> self . request = '<STR_LIT>' <EOL> self . socketError = None <EOL> if _debug : TCPClient . _debug ( "<STR_LIT>" ) <EOL> self . connect ( peer ) <EOL> if _debug : TCPClient . _debug ( "<STR_LIT>" ) <EOL> def handle_connect ( self ) : <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" ) <EOL> def handle_expt ( self ) : <EOL> pass <EOL> def readable ( self ) : <EOL> return <NUM_LIT:1> <EOL> def handle_read ( self ) : <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" ) <EOL> try : <EOL> msg = self . recv ( <NUM_LIT> ) <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" , len ( msg ) ) <EOL> self . socketError = None <EOL> if not self . socket : <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" ) <EOL> else : <EOL> deferred ( self . response , PDU ( msg ) ) <EOL> except socket . error , err : <EOL> if ( err . args [ <NUM_LIT:0> ] == <NUM_LIT> ) : <EOL> deferred ( TCPClient . _error , "<STR_LIT>" , self . peer ) <EOL> else : <EOL> deferred ( TCPClient . _error , "<STR_LIT>" , err ) <EOL> self . socketError = err <EOL> def writable ( self ) : <EOL> return ( len ( self . request ) != <NUM_LIT:0> ) <EOL> def handle_write ( self ) : <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" ) <EOL> try : <EOL> sent = self . send ( self . request ) <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" , sent , len ( self . request ) - sent ) <EOL> self . socketError = None <EOL> self . request = self . request [ sent : ] <EOL> except socket . error , err : <EOL> if ( err . args [ <NUM_LIT:0> ] == <NUM_LIT> ) : <EOL> deferred ( TCPClient . _error , "<STR_LIT>" , self . peer ) <EOL> else : <EOL> deferred ( TCPClient . _error , "<STR_LIT>" , err ) <EOL> self . socketError = err <EOL> def handle_close ( self ) : <EOL> if _debug : deferred ( TCPClient . _debug , "<STR_LIT>" ) <EOL> self . close ( ) <EOL> self . socket = None <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPClient . _debug ( "<STR_LIT>" , pdu ) <EOL> self . request += pdu . pduData <EOL> bacpypes_debugging ( TCPClient ) <EOL> class TCPClientActor ( TCPClient ) : <EOL> def __init__ ( self , director , peer ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" , director , peer ) <EOL> TCPClient . __init__ ( self , peer ) <EOL> self . director = director <EOL> self . timeout = director . timeout <EOL> if self . timeout > <NUM_LIT:0> : <EOL> self . timer = FunctionTask ( self . idle_timeout ) <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> else : <EOL> self . timer = None <EOL> self . flushTask = None <EOL> self . director . add_actor ( self ) <EOL> def handle_close ( self ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" ) <EOL> if self . flushTask : <EOL> self . flushTask . suspend_task ( ) <EOL> if self . timer : <EOL> self . timer . suspend_task ( ) <EOL> self . director . remove_actor ( self ) <EOL> TCPClient . handle_close ( self ) <EOL> def idle_timeout ( self ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" ) <EOL> self . handle_close ( ) <EOL> def indication ( self , pdu ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" , pdu ) <EOL> if self . flushTask : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> return <EOL> if self . timer : <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> TCPClient . indication ( self , pdu ) <EOL> def response ( self , pdu ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" , pdu ) <EOL> pdu . pduSource = self . peer <EOL> if self . timer : <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> self . director . response ( pdu ) <EOL> def flush ( self ) : <EOL> if _debug : TCPClientActor . _debug ( "<STR_LIT>" ) <EOL> self . flushTask = None <EOL> if self . request : <EOL> self . flushTask = OneShotFunction ( self . flush ) <EOL> return <EOL> self . handle_close ( ) <EOL> bacpypes_debugging ( TCPClientActor ) <EOL> class TCPPickleClientActor ( PickleActorMixIn , TCPClientActor ) : <EOL> pass <EOL> class TCPClientDirector ( Server , ServiceAccessPoint , DebugContents ) : <EOL> _debug_contents = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , timeout = <NUM_LIT:0> , actorClass = TCPClientActor , sid = None , sapID = None ) : <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , timeout , actorClass , sid , sapID ) <EOL> Server . __init__ ( self , sid ) <EOL> ServiceAccessPoint . __init__ ( self , sapID ) <EOL> if not issubclass ( actorClass , TCPClientActor ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . actorClass = actorClass <EOL> self . timeout = timeout <EOL> self . clients = { } <EOL> self . reconnect = { } <EOL> def add_actor ( self , actor ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , actor ) <EOL> self . clients [ actor . peer ] = actor <EOL> if self . serviceElement : <EOL> self . sap_request ( addPeer = actor . peer ) <EOL> def remove_actor ( self , actor ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , actor ) <EOL> del self . clients [ actor . peer ] <EOL> if self . serviceElement : <EOL> self . sap_request ( delPeer = actor . peer ) <EOL> if actor . peer in self . reconnect : <EOL> connect_task = FunctionTask ( self . connect , actor . peer ) <EOL> connect_task . install_task ( _time ( ) + self . reconnect [ actor . peer ] ) <EOL> def get_actor ( self , address ) : <EOL> """<STR_LIT>""" <EOL> return self . clients . get ( address , None ) <EOL> def connect ( self , address , reconnect = <NUM_LIT:0> ) : <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , address , reconnect ) <EOL> if address in self . clients : <EOL> return <EOL> client = self . actorClass ( self , address ) <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , client ) <EOL> if reconnect : <EOL> self . reconnect [ address ] = reconnect <EOL> def disconnect ( self , address ) : <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , address ) <EOL> if address not in self . clients : <EOL> return <EOL> if address in self . reconnect : <EOL> del self . reconnect [ address ] <EOL> self . clients [ address ] . handle_close ( ) <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPClientDirector . _debug ( "<STR_LIT>" , pdu ) <EOL> addr = pdu . pduDestination <EOL> client = self . clients . get ( addr , None ) <EOL> if not client : <EOL> client = self . actorClass ( self , addr ) <EOL> client . indication ( pdu ) <EOL> bacpypes_debugging ( TCPClientDirector ) <EOL> class TCPServer ( asyncore . dispatcher ) : <EOL> def __init__ ( self , sock , peer ) : <EOL> if _debug : TCPServer . _debug ( "<STR_LIT>" , sock , peer ) <EOL> asyncore . dispatcher . __init__ ( self , sock ) <EOL> self . peer = peer <EOL> self . request = '<STR_LIT>' <EOL> self . socketError = None <EOL> def handle_connect ( self ) : <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" ) <EOL> def readable ( self ) : <EOL> return <NUM_LIT:1> <EOL> def handle_read ( self ) : <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" ) <EOL> try : <EOL> msg = self . recv ( <NUM_LIT> ) <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" , len ( msg ) ) <EOL> self . socketError = None <EOL> if not self . socket : <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" ) <EOL> else : <EOL> deferred ( self . response , PDU ( msg ) ) <EOL> except socket . error , err : <EOL> if ( err . args [ <NUM_LIT:0> ] == <NUM_LIT> ) : <EOL> deferred ( TCPServer . _error , "<STR_LIT>" , self . peer ) <EOL> else : <EOL> deferred ( TCPServer . _error , "<STR_LIT>" , err ) <EOL> self . socketError = err <EOL> def writable ( self ) : <EOL> return ( len ( self . request ) != <NUM_LIT:0> ) <EOL> def handle_write ( self ) : <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" ) <EOL> try : <EOL> sent = self . send ( self . request ) <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" , sent , len ( self . request ) - sent ) <EOL> self . socketError = None <EOL> self . request = self . request [ sent : ] <EOL> except socket . error , why : <EOL> if ( why . args [ <NUM_LIT:0> ] == <NUM_LIT> ) : <EOL> deferred ( TCPServer . _error , "<STR_LIT>" , self . peer ) <EOL> else : <EOL> deferred ( TCPServer . _error , "<STR_LIT>" , why ) <EOL> self . socketError = why <EOL> def handle_close ( self ) : <EOL> if _debug : deferred ( TCPServer . _debug , "<STR_LIT>" ) <EOL> if not self : <EOL> deferred ( TCPServer . _warning , "<STR_LIT>" ) <EOL> return <EOL> if not self . socket : <EOL> deferred ( TCPServer . _warning , "<STR_LIT>" ) <EOL> return <EOL> self . close ( ) <EOL> self . socket = None <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPServer . _debug ( "<STR_LIT>" , pdu ) <EOL> self . request += pdu . pduData <EOL> bacpypes_debugging ( TCPServer ) <EOL> class TCPServerActor ( TCPServer ) : <EOL> def __init__ ( self , director , sock , peer ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" , director , sock , peer ) <EOL> TCPServer . __init__ ( self , sock , peer ) <EOL> self . director = director <EOL> self . timeout = director . timeout <EOL> if self . timeout > <NUM_LIT:0> : <EOL> self . timer = FunctionTask ( self . idle_timeout ) <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> else : <EOL> self . timer = None <EOL> self . flushTask = None <EOL> self . director . add_actor ( self ) <EOL> def handle_close ( self ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> if self . flushTask : <EOL> self . flushTask . suspend_task ( ) <EOL> self . director . remove_actor ( self ) <EOL> TCPServer . handle_close ( self ) <EOL> def idle_timeout ( self ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> self . handle_close ( ) <EOL> def indication ( self , pdu ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" , pdu ) <EOL> if self . flushTask : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> return <EOL> if self . timer : <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> TCPServer . indication ( self , pdu ) <EOL> def response ( self , pdu ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" , pdu ) <EOL> if self . flushTask : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> return <EOL> pdu . pduSource = self . peer <EOL> if self . timer : <EOL> self . timer . install_task ( _time ( ) + self . timeout ) <EOL> self . director . response ( pdu ) <EOL> def flush ( self ) : <EOL> if _debug : TCPServerActor . _debug ( "<STR_LIT>" ) <EOL> self . flushTask = None <EOL> if self . request : <EOL> self . flushTask = OneShotFunction ( self . flush ) <EOL> return <EOL> self . handle_close ( ) <EOL> bacpypes_debugging ( TCPServerActor ) <EOL> class TCPPickleServerActor ( PickleActorMixIn , TCPServerActor ) : <EOL> pass <EOL> class TCPServerDirector ( asyncore . dispatcher , Server , ServiceAccessPoint , DebugContents ) : <EOL> _debug_contents = ( '<STR_LIT:port>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , address , listeners = <NUM_LIT:5> , timeout = <NUM_LIT:0> , reuse = False , actorClass = TCPServerActor , cid = None , sapID = None ) : <EOL> if _debug : <EOL> TCPServerDirector . _debug ( "<STR_LIT>" <EOL> , address , listeners , timeout , reuse , actorClass , cid , sapID <EOL> ) <EOL> Server . __init__ ( self , cid ) <EOL> ServiceAccessPoint . __init__ ( self , sapID ) <EOL> self . port = address <EOL> self . timeout = timeout <EOL> if not issubclass ( actorClass , TCPServerActor ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . actorClass = actorClass <EOL> self . servers = { } <EOL> asyncore . dispatcher . __init__ ( self ) <EOL> self . create_socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> if reuse : <EOL> self . set_reuse_addr ( ) <EOL> hadBindErrors = False <EOL> for i in range ( <NUM_LIT:30> ) : <EOL> try : <EOL> self . bind ( address ) <EOL> break <EOL> except socket . error , err : <EOL> hadBindErrors = True <EOL> TCPServerDirector . _warning ( '<STR_LIT>' , err ) <EOL> _sleep ( REBIND_SLEEP_INTERVAL ) <EOL> else : <EOL> TCPServerDirector . _error ( '<STR_LIT>' ) <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if hadBindErrors : <EOL> TCPServerDirector . _info ( '<STR_LIT>' ) <EOL> self . listen ( listeners ) <EOL> def handle_accept ( self ) : <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" ) <EOL> try : <EOL> client , addr = self . accept ( ) <EOL> except socket . error : <EOL> TCPServerDirector . _warning ( '<STR_LIT>' ) <EOL> return <EOL> except TypeError : <EOL> TCPServerDirector . _warning ( '<STR_LIT>' ) <EOL> return <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" , client , addr ) <EOL> server = self . actorClass ( self , client , addr ) <EOL> self . servers [ addr ] = server <EOL> return server <EOL> def handle_close ( self ) : <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" ) <EOL> self . close ( ) <EOL> def add_actor ( self , actor ) : <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" , actor ) <EOL> self . servers [ actor . peer ] = actor <EOL> if self . serviceElement : <EOL> self . sap_request ( addPeer = actor . peer ) <EOL> def remove_actor ( self , actor ) : <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" , actor ) <EOL> try : <EOL> del self . servers [ actor . peer ] <EOL> except KeyError : <EOL> TCPServerDirector . _warning ( "<STR_LIT>" , actor ) <EOL> if self . serviceElement : <EOL> self . sap_request ( delPeer = actor . peer ) <EOL> def get_actor ( self , address ) : <EOL> """<STR_LIT>""" <EOL> return self . servers . get ( address , None ) <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TCPServerDirector . _debug ( "<STR_LIT>" , pdu ) <EOL> addr = pdu . pduDestination <EOL> server = self . servers . get ( addr , None ) <EOL> if not server : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> server . indication ( pdu ) <EOL> bacpypes_debugging ( TCPServerDirector ) <EOL> class StreamToPacket ( Client , Server ) : <EOL> def __init__ ( self , fn , cid = None , sid = None ) : <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , fn , cid , sid ) <EOL> Client . __init__ ( self , cid ) <EOL> Server . __init__ ( self , sid ) <EOL> self . packetFn = fn <EOL> self . upstreamBuffer = { } <EOL> self . downstreamBuffer = { } <EOL> def packetize ( self , pdu , streamBuffer ) : <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , pdu ) <EOL> def chop ( addr ) : <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , addr ) <EOL> buff = streamBuffer . get ( addr , '<STR_LIT>' ) + pdu . pduData <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , buff ) <EOL> while <NUM_LIT:1> : <EOL> packet = self . packetFn ( buff ) <EOL> if packet is None : <EOL> break <EOL> yield PDU ( packet [ <NUM_LIT:0> ] , <EOL> source = pdu . pduSource , <EOL> destination = pdu . pduDestination , <EOL> user_data = pdu . pduUserData , <EOL> ) <EOL> buff = packet [ <NUM_LIT:1> ] <EOL> streamBuffer [ addr ] = buff <EOL> if pdu . pduSource : <EOL> for pdu in chop ( pdu . pduSource ) : <EOL> yield pdu <EOL> if pdu . pduDestination : <EOL> for pdu in chop ( pdu . pduDestination ) : <EOL> yield pdu <EOL> def indication ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , pdu ) <EOL> for packet in self . packetize ( pdu , self . downstreamBuffer ) : <EOL> self . request ( packet ) <EOL> def confirmation ( self , pdu ) : <EOL> """<STR_LIT>""" <EOL> if _debug : StreamToPacket . _debug ( "<STR_LIT>" , pdu ) <EOL> for packet in self . packetize ( pdu , self . upstreamBuffer ) : <EOL> self . response ( packet ) <EOL> bacpypes_debugging ( StreamToPacket ) <EOL> class StreamToPacketSAP ( ApplicationServiceElement , ServiceAccessPoint ) : <EOL> def __init__ ( self , stp , aseID = None , sapID = None ) : <EOL> if _debug : StreamToPacketSAP . _debug ( "<STR_LIT>" , stp , aseID , sapID ) <EOL> ApplicationServiceElement . __init__ ( self , aseID ) <EOL> ServiceAccessPoint . __init__ ( self , sapID ) <EOL> self . stp = stp <EOL> def indication ( self , addPeer = None , delPeer = None ) : <EOL> if _debug : StreamToPacketSAP . _debug ( "<STR_LIT>" , addPeer , delPeer ) <EOL> if addPeer : <EOL> self . stp . upstreamBuffer [ addPeer ] = '<STR_LIT>' <EOL> self . stp . downstreamBuffer [ addPeer ] = '<STR_LIT>' <EOL> if delPeer : <EOL> del self . stp . upstreamBuffer [ delPeer ] <EOL> del self . stp . downstreamBuffer [ delPeer ] <EOL> if self . serviceElement : <EOL> self . sap_request ( addPeer = addPeer , delPeer = delPeer ) <EOL> bacpypes_debugging ( StreamToPacketSAP ) </s>
