mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-09-28 22:15:23 +08:00
Code Issues Releases Wiki Activity

update to the new IOCB client API

Browse Source
This commit is contained in:
Joel Bender 2016-09-01 13:47:00 -04:00
parent 368dc50852
commit 40804b67b5
1 changed files with 59 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
samples/ReadWriteProperty.py
View File

@ -19,7 +19,7 @@ from bacpypes.core import run, enable_sleeping
from bacpypes.pdu import Address
from bacpypes.object import get_object_class, get_datatype
from bacpypes.apdu import Error, AbortPDU, SimpleAckPDU, \
from bacpypes.apdu import SimpleAckPDU, \
ReadPropertyRequest, ReadPropertyACK, WritePropertyRequest
from bacpypes.primitivedata import Null, Atomic, Integer, Unsigned, Real
from bacpypes.constructeddata import Array, Any
@ -34,63 +34,6 @@ _log = ModuleLogger(globals())
# globals
this_application = 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
#
@ -129,7 +72,45 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request)
# give it to the application
this_application.request(request)
iocb = this_application.request(request)
if _debug: ReadWritePropertyConsoleCmd._debug(" - iocb: %r", 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)
@ -208,7 +189,24 @@ class ReadWritePropertyConsoleCmd(ConsoleCmd):
if _debug: ReadWritePropertyConsoleCmd._debug(" - request: %r", request)
# give it to the application
this_application.request(request)
iocb = this_application.request(request)
if _debug: ReadWritePropertyConsoleCmd._debug(" - iocb: %r", 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)
@ -253,7 +251,7 @@ def main():
)
# make a simple application
this_application = ReadPropertyApplication(this_device, args.ini.address)
this_application = BIPSimpleApplication(this_device, args.ini.address)
# get the services supported
services_supported = this_application.get_services_supported()