#!/usr/bin/env python """ This application is similar to ReadWriteProperty but it is just for beating on the event message texts, an array of character strings. """ import sys from bacpypes.debugging import bacpypes_debugging, ModuleLogger from bacpypes.consolelogging import ConfigArgumentParser from bacpypes.consolecmd import ConsoleCmd from bacpypes.core import run, enable_sleeping from bacpypes.iocb import IOCB from bacpypes.pdu import Address from bacpypes.object import get_datatype from bacpypes.apdu import SimpleAckPDU, \ ReadPropertyRequest, ReadPropertyACK, WritePropertyRequest from bacpypes.primitivedata import Unsigned, CharacterString from bacpypes.constructeddata import Array, ArrayOf, Any from bacpypes.app import BIPSimpleApplication from bacpypes.service.device import LocalDeviceObject # some debugging _debug = 0 _log = ModuleLogger(globals()) # globals this_application = None context = None # # ReadWritePropertyConsoleCmd # @bacpypes_debugging class ReadWritePropertyConsoleCmd(ConsoleCmd): def do_read(self, args): """read [ ]""" args = args.split() if _debug: ReadWritePropertyConsoleCmd._debug("do_read %r", args) global context try: addr, obj_type, obj_inst = context prop_id = 'eventMessageTexts' datatype = get_datatype(obj_type, prop_id) if not datatype: raise ValueError("invalid property for object type") # build a request request = ReadPropertyRequest( objectIdentifier=(obj_type, obj_inst), propertyIdentifier=prop_id, ) request.pduDestination = Address(addr) if len(args) == 1: request.propertyArrayIndex = int(args[0]) if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request) # make an IOCB iocb = IOCB(request) if _debug: ReadWritePropertyConsoleCmd._debug(" - iocb: %r", iocb) # give it to the application this_application.request_io(iocb) # wait for it to complete iocb.wait() # do something for success if iocb.ioResponse: apdu = iocb.ioResponse # should be an ack if not isinstance(apdu, ReadPropertyACK): if _debug: ReadWritePropertyConsoleCmd._debug(" - not an ack") return # find the datatype datatype = get_datatype(apdu.objectIdentifier[0], apdu.propertyIdentifier) if _debug: ReadWritePropertyConsoleCmd._debug(" - datatype: %r", datatype) if not datatype: raise TypeError("unknown datatype") # special case for array parts, others are managed by cast_out if issubclass(datatype, Array) and (apdu.propertyArrayIndex is not None): if apdu.propertyArrayIndex == 0: value = apdu.propertyValue.cast_out(Unsigned) else: value = apdu.propertyValue.cast_out(datatype.subtype) else: value = apdu.propertyValue.cast_out(datatype) if _debug: ReadWritePropertyConsoleCmd._debug(" - value: %r", value) sys.stdout.write(str(value) + '\n') if hasattr(value, 'debug_contents'): value.debug_contents(file=sys.stdout) sys.stdout.flush() # do something for error/reject/abort if iocb.ioError: sys.stdout.write(str(iocb.ioError) + '\n') except Exception as error: ReadWritePropertyConsoleCmd._exception("exception: %r", error) def do_write(self, args): """ write write 0 write [ ]... """ args = args.split() ReadWritePropertyConsoleCmd._debug("do_write %r", args) try: addr, obj_type, obj_inst = context prop_id = 'eventMessageTexts' indx = None if args and args[0].isdigit(): indx = int(args[0]) if indx == 0: value = Unsigned(int(args[1])) else: value = CharacterString(args[1]) else: value = ArrayOf(CharacterString)(args[0:]) # build a request request = WritePropertyRequest( objectIdentifier=(obj_type, obj_inst), propertyIdentifier=prop_id ) request.pduDestination = Address(addr) # save the value request.propertyValue = Any() try: request.propertyValue.cast_in(value) except Exception as error: ReadWritePropertyConsoleCmd._exception("WriteProperty cast error: %r", error) # optional array index if indx is not None: request.propertyArrayIndex = indx if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request) # make an IOCB iocb = IOCB(request) if _debug: ReadWritePropertyConsoleCmd._debug(" - iocb: %r", iocb) # give it to the application this_application.request_io(iocb) # wait for it to complete iocb.wait() # do something for success if iocb.ioResponse: # should be an ack if not isinstance(iocb.ioResponse, SimpleAckPDU): if _debug: ReadWritePropertyConsoleCmd._debug(" - not an ack") return sys.stdout.write("ack\n") # do something for error/reject/abort if iocb.ioError: sys.stdout.write(str(iocb.ioError) + '\n') except Exception as error: ReadWritePropertyConsoleCmd._exception("exception: %r", error) def do_rtn(self, args): """rtn ... """ args = args.split() if _debug: ReadWritePropertyConsoleCmd._debug("do_rtn %r", args) # safe to assume only one adapter adapter = this_application.nsap.adapters[0] if _debug: ReadWritePropertyConsoleCmd._debug(" - adapter: %r", adapter) # provide the address and a list of network numbers router_address = Address(args[0]) network_list = [int(arg) for arg in args[1:]] # pass along to the service access point this_application.nsap.add_router_references(adapter, router_address, network_list) # # __main__ # def main(): global this_application, context # parse the command line arguments parser = ConfigArgumentParser(description=__doc__) parser.add_argument( "address", help="address of server", ) parser.add_argument( "objtype", help="object type", ) parser.add_argument( "objinst", type=int, help="object instance", ) args = parser.parse_args() if _debug: _log.debug("initialization") if _debug: _log.debug(" - args: %r", args) # set the context, the collection of the above parameters context = args.address, args.objtype, args.objinst if _debug: _log.debug(" - context: %r", context) # make a device object this_device = LocalDeviceObject( objectName=args.ini.objectname, objectIdentifier=int(args.ini.objectidentifier), maxApduLengthAccepted=int(args.ini.maxapdulengthaccepted), segmentationSupported=args.ini.segmentationsupported, vendorIdentifier=int(args.ini.vendoridentifier), ) # make a simple application this_application = BIPSimpleApplication(this_device, args.ini.address) # get the services supported services_supported = this_application.get_services_supported() if _debug: _log.debug(" - services_supported: %r", services_supported) # let the device object know this_device.protocolServicesSupported = services_supported.value # make a console this_console = ReadWritePropertyConsoleCmd() if _debug: _log.debug(" - this_console: %r", this_console) # enable sleeping will help with threads enable_sleeping() _log.debug("running") run() _log.debug("fini") if __name__ == "__main__": main()