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 API

Browse Source
This commit is contained in:
Joel Bender 2016-09-01 22:27:29 -04:00
parent 40804b67b5
commit a93604fe89
1 changed files with 22 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
samples/RecurringMultipleReadProperty.py
View File

@ -1,10 +1,10 @@
#!/usr/bin/env python
"""
Mutliple Read Property
Recurring Read Property
This application has a static list of points that it would like to read. It reads the
values of each of them in turn and then quits.
This application has a static list of points that it would like to read. It
reads the values of each of them in turn and then quits.
"""
from collections import deque
@ -43,19 +43,14 @@ point_list = [
class PrairieDog(BIPSimpleApplication, RecurringTask):
def __init__(self, interval, *args):
if _debug: PrairieDog._debug("__init__ %r, %r", interval, args)
if _debug: PrairieDog._debug("__init__ %r %r", interval, args)
BIPSimpleApplication.__init__(self, *args)
RecurringTask.__init__(self, interval * 1000)
# keep track of requests to line up responses
self._request = None
# start out idle
# no longer busy
self.is_busy = False
self.point_queue = deque()
self.response_values = []
# install it
# install the task
self.install_task()
def process_task(self):
@ -99,28 +94,26 @@ class PrairieDog(BIPSimpleApplication, RecurringTask):
addr, obj_type, obj_inst, prop_id = self.point_queue.popleft()
# build a request
self._request = ReadPropertyRequest(
request = ReadPropertyRequest(
objectIdentifier=(obj_type, obj_inst),
propertyIdentifier=prop_id,
)
self._request.pduDestination = Address(addr)
if _debug: PrairieDog._debug(" - request: %r", self._request)
request.pduDestination = Address(addr)
if _debug: PrairieDog._debug(" - request: %r", request)
# forward it along
BIPSimpleApplication.request(self, self._request)
# send the request
iocb = self.request(request)
if _debug: PrairieDog._debug(" - iocb: %r", iocb)
def confirmation(self, apdu):
if _debug: PrairieDog._debug("confirmation %r", apdu)
# set a callback for the response
iocb.add_callback(self.complete_request)
if isinstance(apdu, Error):
if _debug: PrairieDog._debug(" - error: %r", apdu)
self.response_values.append(apdu)
def complete_request(self, iocb):
if _debug: PrairieDog._debug("complete_request %r", iocb)
elif isinstance(apdu, AbortPDU):
if _debug: PrairieDog._debug(" - abort: %r", apdu)
self.response_values.append(apdu)
if iocb.ioResponse:
apdu = iocb.ioResponse
elif (isinstance(self._request, ReadPropertyRequest)) and (isinstance(apdu, ReadPropertyACK)):
# find the datatype
datatype = get_datatype(apdu.objectIdentifier[0], apdu.propertyIdentifier)
if _debug: PrairieDog._debug(" - datatype: %r", datatype)
@ -140,6 +133,10 @@ class PrairieDog(BIPSimpleApplication, RecurringTask):
# save the value
self.response_values.append(value)
if iocb.ioError:
if _debug: PrairieDog._debug(" - error: %r", iocb.ioError)
self.response_values.append(iocb.ioError)
# fire off another request
deferred(self.next_request)