<s> """<STR_LIT>""" <EOL> import sys <EOL> from bacpypes . debugging import bacpypes_debugging , ModuleLogger <EOL> from bacpypes . consolelogging import ConfigArgumentParser <EOL> from bacpypes . consolecmd import ConsoleCmd <EOL> from bacpypes . core import run <EOL> from bacpypes . pdu import Address <EOL> from bacpypes . app import LocalDeviceObject , BIPSimpleApplication <EOL> from bacpypes . apdu import Error , AbortPDU , AtomicReadFileRequest , AtomicReadFileRequestAccessMethodChoice , AtomicReadFileRequestAccessMethodChoiceRecordAccess , AtomicReadFileRequestAccessMethodChoiceStreamAccess , AtomicReadFileACK , AtomicWriteFileRequest , AtomicWriteFileRequestAccessMethodChoice , AtomicWriteFileRequestAccessMethodChoiceRecordAccess , AtomicWriteFileRequestAccessMethodChoiceStreamAccess , AtomicWriteFileACK <EOL> from bacpypes . basetypes import ServicesSupported <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> this_application = None <EOL> @ bacpypes_debugging <EOL> class TestApplication ( BIPSimpleApplication ) : <EOL> def request ( self , apdu ) : <EOL> if _debug : TestApplication . _debug ( "<STR_LIT>" , apdu ) <EOL> self . _request = apdu <EOL> BIPSimpleApplication . request ( self , apdu ) <EOL> def confirmation ( self , apdu ) : <EOL> if _debug : TestApplication . _debug ( "<STR_LIT>" , apdu ) <EOL> if isinstance ( apdu , Error ) : <EOL> sys . stdout . write ( "<STR_LIT>" % ( apdu . errorCode , ) ) <EOL> sys . stdout . flush ( ) <EOL> elif isinstance ( apdu , AbortPDU ) : <EOL> apdu . debug_contents ( ) <EOL> elif ( isinstance ( self . _request , AtomicReadFileRequest ) ) and ( isinstance ( apdu , AtomicReadFileACK ) ) : <EOL> if apdu . accessMethod . recordAccess : <EOL> value = apdu . accessMethod . recordAccess . fileRecordData <EOL> elif apdu . accessMethod . streamAccess : <EOL> value = apdu . accessMethod . streamAccess . fileData <EOL> TestApplication . _debug ( "<STR_LIT>" , value ) <EOL> sys . stdout . write ( repr ( value ) + '<STR_LIT:\n>' ) <EOL> sys . stdout . flush ( ) <EOL> elif ( isinstance ( self . _request , AtomicWriteFileRequest ) ) and ( isinstance ( apdu , AtomicWriteFileACK ) ) : <EOL> if apdu . fileStartPosition is not None : <EOL> value = apdu . fileStartPosition <EOL> elif apdu . fileStartRecord is not None : <EOL> value = apdu . fileStartRecord <EOL> TestApplication . _debug ( "<STR_LIT>" , value ) <EOL> sys . stdout . write ( repr ( value ) + '<STR_LIT:\n>' ) <EOL> sys . stdout . flush ( ) <EOL> @ bacpypes_debugging <EOL> class TestConsoleCmd ( ConsoleCmd ) : <EOL> def do_readrecord ( self , args ) : <EOL> """<STR_LIT>""" <EOL> args = args . split ( ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , args ) <EOL> try : <EOL> addr , obj_inst , start_record , record_count = args <EOL> obj_type = '<STR_LIT:file>' <EOL> obj_inst = int ( obj_inst ) <EOL> start_record = int ( start_record ) <EOL> record_count = int ( record_count ) <EOL> request = AtomicReadFileRequest ( <EOL> fileIdentifier = ( obj_type , obj_inst ) , <EOL> accessMethod = AtomicReadFileRequestAccessMethodChoice ( <EOL> recordAccess = AtomicReadFileRequestAccessMethodChoiceRecordAccess ( <EOL> fileStartRecord = start_record , <EOL> requestedRecordCount = record_count , <EOL> ) , <EOL> ) , <EOL> ) <EOL> request . pduDestination = Address ( addr ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , request ) <EOL> this_application . request ( request ) <EOL> except Exception , e : <EOL> TestConsoleCmd . _exception ( "<STR_LIT>" , e ) <EOL> def do_readstream ( self , args ) : <EOL> """<STR_LIT>""" <EOL> args = args . split ( ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , args ) <EOL> try : <EOL> addr , obj_inst , start_position , octet_count = args <EOL> obj_type = '<STR_LIT:file>' <EOL> obj_inst = int ( obj_inst ) <EOL> start_position = int ( start_position ) <EOL> octet_count = int ( octet_count ) <EOL> request = AtomicReadFileRequest ( <EOL> fileIdentifier = ( obj_type , obj_inst ) , <EOL> accessMethod = AtomicReadFileRequestAccessMethodChoice ( <EOL> streamAccess = AtomicReadFileRequestAccessMethodChoiceStreamAccess ( <EOL> fileStartPosition = start_position , <EOL> requestedOctetCount = octet_count , <EOL> ) , <EOL> ) , <EOL> ) <EOL> request . pduDestination = Address ( addr ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , request ) <EOL> this_application . request ( request ) <EOL> except Exception , e : <EOL> TestConsoleCmd . _exception ( "<STR_LIT>" , e ) <EOL> def do_writerecord ( self , args ) : <EOL> """<STR_LIT>""" <EOL> args = args . split ( ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , args ) <EOL> try : <EOL> addr , obj_inst , start_record , record_count = args [ <NUM_LIT:0> : <NUM_LIT:4> ] <EOL> obj_type = '<STR_LIT:file>' <EOL> obj_inst = int ( obj_inst ) <EOL> start_record = int ( start_record ) <EOL> record_count = int ( record_count ) <EOL> record_data = list ( args [ <NUM_LIT:4> : ] ) <EOL> request = AtomicWriteFileRequest ( <EOL> fileIdentifier = ( obj_type , obj_inst ) , <EOL> accessMethod = AtomicWriteFileRequestAccessMethodChoice ( <EOL> recordAccess = AtomicWriteFileRequestAccessMethodChoiceRecordAccess ( <EOL> fileStartRecord = start_record , <EOL> recordCount = record_count , <EOL> fileRecordData = record_data , <EOL> ) , <EOL> ) , <EOL> ) <EOL> request . pduDestination = Address ( addr ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , request ) <EOL> this_application . request ( request ) <EOL> except Exception , e : <EOL> TestConsoleCmd . _exception ( "<STR_LIT>" , e ) <EOL> def do_writestream ( self , args ) : <EOL> """<STR_LIT>""" <EOL> args = args . split ( ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , args ) <EOL> try : <EOL> addr , obj_inst , start_position , data = args <EOL> obj_type = '<STR_LIT:file>' <EOL> obj_inst = int ( obj_inst ) <EOL> start_position = int ( start_position ) <EOL> request = AtomicWriteFileRequest ( <EOL> fileIdentifier = ( obj_type , obj_inst ) , <EOL> accessMethod = AtomicWriteFileRequestAccessMethodChoice ( <EOL> streamAccess = AtomicWriteFileRequestAccessMethodChoiceStreamAccess ( <EOL> fileStartPosition = start_position , <EOL> fileData = data , <EOL> ) , <EOL> ) , <EOL> ) <EOL> request . pduDestination = Address ( addr ) <EOL> if _debug : TestConsoleCmd . _debug ( "<STR_LIT>" , request ) <EOL> this_application . request ( request ) <EOL> except Exception , e : <EOL> TestConsoleCmd . _exception ( "<STR_LIT>" , e ) <EOL> try : <EOL> args = ConfigArgumentParser ( description = __doc__ ) . parse_args ( ) <EOL> if _debug : _log . debug ( "<STR_LIT>" ) <EOL> if _debug : _log . debug ( "<STR_LIT>" , args ) <EOL> this_device = LocalDeviceObject ( <EOL> objectName = args . ini . objectname , <EOL> objectIdentifier = int ( args . ini . objectidentifier ) , <EOL> maxApduLengthAccepted = int ( args . ini . maxapdulengthaccepted ) , <EOL> segmentationSupported = args . ini . segmentationsupported , <EOL> vendorIdentifier = int ( args . ini . vendoridentifier ) , <EOL> ) <EOL> this_application = TestApplication ( this_device , args . ini . address ) <EOL> services_supported = this_application . get_services_supported ( ) <EOL> if _debug : _log . debug ( "<STR_LIT>" , services_supported ) <EOL> this_device . protocolServicesSupported = services_supported . value <EOL> this_console = TestConsoleCmd ( ) <EOL> _log . debug ( "<STR_LIT>" ) <EOL> run ( ) <EOL> except Exception , e : <EOL> _log . exception ( "<STR_LIT>" , e ) <EOL> finally : <EOL> _log . debug ( "<STR_LIT>" ) </s>
