#!/usr/bin/python """ This application presents a 'console' prompt to the user asking for commands. For 'read' commands it will create ReadPropertyRequest PDUs, then lines up the coorresponding ReadPropertyACK and prints the value. For 'write' commands it will create WritePropertyRequst PDUs and prints out a simple acknowledgement. """ 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.pdu import Address from bacpypes.app import LocalDeviceObject, BIPSimpleApplication from bacpypes.object import get_object_class, get_datatype from bacpypes.apdu import Error, AbortPDU, SimpleAckPDU, \ ReadPropertyRequest, ReadPropertyACK, WritePropertyRequest from bacpypes.primitivedata import Null, Atomic, Integer, Unsigned, Real from bacpypes.constructeddata import Array, Any from bacpypes.basetypes import ServicesSupported # some debugging _debug = 0 _log = ModuleLogger(globals()) # globals this_device = None this_application = None this_console = None # # ReadPropertyApplication # @bacpypes_debugging class ReadPropertyApplication(BIPSimpleApplication): def __init__(self, *args): if _debug: ReadPropertyApplication._debug("__init__ %r", args) BIPSimpleApplication.__init__(self, *args) # keep track of requests to line up responses self._request = None def request(self, apdu): if _debug: ReadPropertyApplication._debug("request %r", apdu) # save a copy of the request self._request = apdu # forward it along BIPSimpleApplication.request(self, apdu) def confirmation(self, apdu): if _debug: ReadPropertyApplication._debug("confirmation %r", apdu) if isinstance(apdu, Error): sys.stdout.write("error: %s\n" % (apdu.errorCode,)) sys.stdout.flush() elif isinstance(apdu, AbortPDU): apdu.debug_contents() if isinstance(apdu, SimpleAckPDU): sys.stdout.write("ack\n") sys.stdout.flush() elif (isinstance(self._request, ReadPropertyRequest)) and (isinstance(apdu, ReadPropertyACK)): # find the datatype datatype = get_datatype(apdu.objectIdentifier[0], apdu.propertyIdentifier) if _debug: ReadPropertyApplication._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: ReadPropertyApplication._debug(" - value: %r", value) sys.stdout.write(str(value) + '\n') sys.stdout.flush() # # ReadWritePropertyConsoleCmd # @bacpypes_debugging class ReadWritePropertyConsoleCmd(ConsoleCmd): def do_read(self, args): """read [ ]""" args = args.split() if _debug: ReadWritePropertyConsoleCmd._debug("do_read %r", args) try: addr, obj_type, obj_inst, prop_id = args[:4] if obj_type.isdigit(): obj_type = int(obj_type) elif not get_object_class(obj_type): raise ValueError("unknown object type") obj_inst = int(obj_inst) 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) == 5: request.propertyArrayIndex = int(args[4]) if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request) # give it to the application this_application.request(request) except Exception as error: ReadWritePropertyConsoleCmd._exception("exception: %r", error) def do_write(self, args): """write [ ] [ ]""" args = args.split() ReadWritePropertyConsoleCmd._debug("do_write %r", args) try: addr, obj_type, obj_inst, prop_id = args[:4] if obj_type.isdigit(): obj_type = int(obj_type) obj_inst = int(obj_inst) value = args[4] indx = None if len(args) >= 6: if args[5] != "-": indx = int(args[5]) if _debug: ReadWritePropertyConsoleCmd._debug(" - indx: %r", indx) priority = None if len(args) >= 7: priority = int(args[6]) if _debug: ReadWritePropertyConsoleCmd._debug(" - priority: %r", priority) # get the datatype datatype = get_datatype(obj_type, prop_id) if _debug: ReadWritePropertyConsoleCmd._debug(" - datatype: %r", datatype) # change atomic values into something encodeable, null is a special case if (value == 'null'): value = Null() elif issubclass(datatype, Atomic): if datatype is Integer: value = int(value) elif datatype is Real: value = float(value) elif datatype is Unsigned: value = int(value) value = datatype(value) elif issubclass(datatype, Array) and (indx is not None): if indx == 0: value = Integer(value) elif issubclass(datatype.subtype, Atomic): value = datatype.subtype(value) elif not isinstance(value, datatype.subtype): raise TypeError("invalid result datatype, expecting %s" % (datatype.subtype.__name__,)) elif not isinstance(value, datatype): raise TypeError("invalid result datatype, expecting %s" % (datatype.__name__,)) if _debug: ReadWritePropertyConsoleCmd._debug(" - encodeable value: %r %s", value, type(value)) # 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 # optional priority if priority is not None: request.priority = priority if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request) # give it to the application this_application.request(request) except Exception as error: ReadWritePropertyConsoleCmd._exception("exception: %r", error) # # __main__ # try: # parse the command line arguments args = ConfigArgumentParser(description=__doc__).parse_args() if _debug: _log.debug("initialization") if _debug: _log.debug(" - args: %r", args) # 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 = ReadPropertyApplication(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() _log.debug("running") # enable sleeping will allow handling of threads enable_sleeping() run() except Exception as error: _log.exception("an error has occurred: %s", error) finally: _log.debug("finally")