mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-10-05 22:18:16 +08:00
Code Issues Releases Wiki Activity

bring branch up to date with stage again

Browse Source
This commit is contained in:
Joel Bender 2017-09-19 21:35:30 -04:00
commit 1592589fc3
9 changed files with 318 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
py25/bacpypes/service/device.py
View File

@ -36,7 +36,7 @@ class CurrentDateProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
@ -58,7 +58,7 @@ class CurrentTimeProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#

14
py25/bacpypes/task.py
View File

@ -183,7 +183,8 @@ class RecurringTask(_Task):
if interval is not None:
self.taskInterval = interval
if offset is not None:
self.taskIntervalOffset = interval
self.taskIntervalOffset = offset
if self.taskInterval is None:
raise RuntimeError("interval unset, use ctor or install_task parameter")
if self.taskInterval <= 0.0:
@ -195,15 +196,18 @@ class RecurringTask(_Task):
_unscheduled_tasks.append(self)
else:
# offset is also in milliseconds to be consistent
# get ready for the next interval plus a jitter
now = _task_manager.get_time() + 0.000001
# interval and offset are in milliseconds to be consistent
interval = self.taskInterval / 1000.0
if self.taskIntervalOffset:
offset = self.taskIntervalOffset / 1000.0
else:
offset = 0.0
if _debug: RecurringTask._debug(" - now, interval, offset: %r, %r, %r", now, interval, offset)
# get ready for the next interval (aligned)
now = _task_manager.get_time()
interval = self.taskInterval / 1000.0
# compute the time
self.taskTime = (now - offset) + interval - ((now - offset) % interval) + offset
if _debug: RecurringTask._debug(" - task time: %r", self.taskTime)

4
py27/bacpypes/service/device.py
View File

@ -36,7 +36,7 @@ class CurrentDateProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
@ -58,7 +58,7 @@ class CurrentTimeProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#

14
py27/bacpypes/task.py
View File

@ -183,7 +183,8 @@ class RecurringTask(_Task):
if interval is not None:
self.taskInterval = interval
if offset is not None:
self.taskIntervalOffset = interval
self.taskIntervalOffset = offset
if self.taskInterval is None:
raise RuntimeError("interval unset, use ctor or install_task parameter")
if self.taskInterval <= 0.0:
@ -195,15 +196,18 @@ class RecurringTask(_Task):
_unscheduled_tasks.append(self)
else:
# offset is also in milliseconds to be consistent
# get ready for the next interval plus a jitter
now = _task_manager.get_time() + 0.000001
# interval and offset are in milliseconds to be consistent
interval = self.taskInterval / 1000.0
if self.taskIntervalOffset:
offset = self.taskIntervalOffset / 1000.0
else:
offset = 0.0
if _debug: RecurringTask._debug(" - now, interval, offset: %r, %r, %r", now, interval, offset)
# get ready for the next interval (aligned)
now = _task_manager.get_time()
interval = self.taskInterval / 1000.0
# compute the time
self.taskTime = (now - offset) + interval - ((now - offset) % interval) + offset
if _debug: RecurringTask._debug(" - task time: %r", self.taskTime)

4
py34/bacpypes/service/device.py
View File

@ -36,7 +36,7 @@ class CurrentDateProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#
@ -58,7 +58,7 @@ class CurrentTimeProperty(Property):
now.now()
return now.value
def WriteProperty(self, obj, value, arrayIndex=None, priority=None):
def WriteProperty(self, obj, value, arrayIndex=None, priority=None, direct=False):
raise ExecutionError(errorClass='property', errorCode='writeAccessDenied')
#

14
py34/bacpypes/task.py
View File

@ -183,7 +183,8 @@ class RecurringTask(_Task):
if interval is not None:
self.taskInterval = interval
if offset is not None:
self.taskIntervalOffset = interval
self.taskIntervalOffset = offset
if self.taskInterval is None:
raise RuntimeError("interval unset, use ctor or install_task parameter")
if self.taskInterval <= 0.0:
@ -195,15 +196,18 @@ class RecurringTask(_Task):
_unscheduled_tasks.append(self)
else:
# offset is also in milliseconds to be consistent
# get ready for the next interval plus a jitter
now = _task_manager.get_time() + 0.000001
# interval and offset are in milliseconds to be consistent
interval = self.taskInterval / 1000.0
if self.taskIntervalOffset:
offset = self.taskIntervalOffset / 1000.0
else:
offset = 0.0
if _debug: RecurringTask._debug(" - now, interval, offset: %r, %r, %r", now, interval, offset)
# get ready for the next interval (aligned)
now = _task_manager.get_time()
interval = self.taskInterval / 1000.0
# compute the time
self.taskTime = (now - offset) + interval - ((now - offset) % interval) + offset
if _debug: RecurringTask._debug(" - task time: %r", self.taskTime)

211
samples/ThreadedReadProperty.py Executable file
View File

@ -0,0 +1,211 @@
#!/usr/bin/env python
"""
Threaded Read Property
This application has a static list of points that it would like to read. It
starts a thread for each unique device address and reads the points for that
device.
"""
from threading import Thread
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.consolelogging import ConfigArgumentParser
from bacpypes.core import run, stop, deferred
from bacpypes.iocb import IOCB
from bacpypes.pdu import Address
from bacpypes.object import get_datatype
from bacpypes.apdu import ReadPropertyRequest
from bacpypes.primitivedata import Unsigned
from bacpypes.constructeddata import Array
from bacpypes.app import BIPSimpleApplication
from bacpypes.service.device import LocalDeviceObject
# some debugging
_debug = 0
_log = ModuleLogger(globals())
# globals
this_application = None
# point list, set according to your devices
point_list = [
('10.0.1.14', [
('analogValue', 1, 'presentValue'),
('analogValue', 2, 'presentValue'),
]),
('10.0.1.15', [
('analogValue', 1, 'presentValue'),
('analogValue', 2, 'presentValue'),
]),
]
#
# ReadPointListThread
#
@bacpypes_debugging
class ReadPointListThread(Thread):
def __init__(self, device_address, point_list):
if _debug: ReadPointListThread._debug("__init__ %r %r", device_address, point_list)
Thread.__init__(self)
# save the address
self.device_address = Address(device_address)
# turn the point list into a queue
self.point_list = point_list
# make a list of the response values
self.response_values = []
def run(self):
if _debug: ReadPointListThread._debug("run")
global this_application
# loop through the points
for obj_type, obj_inst, prop_id in self.point_list:
# build a request
request = ReadPropertyRequest(
destination=self.device_address,
objectIdentifier=(obj_type, obj_inst),
propertyIdentifier=prop_id,
)
if _debug: ReadPointListThread._debug(" - request: %r", request)
# make an IOCB
iocb = IOCB(request)
if _debug: ReadPointListThread._debug(" - iocb: %r", iocb)
# give it to the application
this_application.request_io(iocb)
# wait for the response
iocb.wait()
if iocb.ioResponse:
apdu = iocb.ioResponse
# find the datatype
datatype = get_datatype(apdu.objectIdentifier[0], apdu.propertyIdentifier)
if _debug: ReadPointListThread._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: ReadPointListThread._debug(" - value: %r", value)
# save the value
self.response_values.append(value)
if iocb.ioError:
if _debug: ReadPointListThread._debug(" - error: %r", iocb.ioError)
self.response_values.append(iocb.ioError)
if _debug: ReadPointListThread._debug(" - fini")
#
# ThreadSupervisor
#
@bacpypes_debugging
class ThreadSupervisor(Thread):
def __init__(self, thread_list):
if _debug: ThreadSupervisor._debug("__init__ ...")
Thread.__init__(self)
self.thread_list = thread_list
def run(self):
if _debug: ThreadSupervisor._debug("run")
# start them up
for read_thread in self.thread_list:
read_thread.start()
if _debug: ThreadSupervisor._debug(" - all started")
# wait for them to finish
for read_thread in self.thread_list:
read_thread.join()
if _debug: ThreadSupervisor._debug(" - all finished")
# stop the core
stop()
#
# __main__
#
def main():
global this_application
# 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 = 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
thread_list = []
# loop through the address and point lists
for addr, points in point_list:
# create a thread
read_thread = ReadPointListThread(addr, points)
if _debug: _log.debug(" - read_thread: %r", read_thread)
thread_list.append(read_thread)
# create a thread supervisor
thread_supervisor = ThreadSupervisor(thread_list)
# start it running when the core is running
deferred(thread_supervisor.start)
_log.debug("running")
run()
# dump out the results
for read_thread in thread_list:
for request, response in zip(read_thread.point_list, read_thread.response_values):
print(request, response)
_log.debug("fini")
if __name__ == "__main__":
main()

87
tests/test_utilities/test_time_machine.py
View File

@ -10,8 +10,7 @@ import unittest
from bacpypes.debugging import bacpypes_debugging, ModuleLogger
from bacpypes.task import OneShotTask, FunctionTask, \
RecurringTask, RecurringFunctionTask
from bacpypes.task import OneShotTask, FunctionTask, RecurringTask
from ..time_machine import TimeMachine, reset_time_machine, run_time_machine
# some debugging
@ -21,6 +20,23 @@ _log = ModuleLogger(globals())
# reference to time machine
time_machine = None
@bacpypes_debugging
def almost_equal(x, y):
"""Compare two arrays of floats."""
# must be the same length
if len(x) != len(y):
return False
# absolute value of the difference is tollerable
for xx, yy in zip(x, y):
if abs(xx - yy) > 0.000001:
return False
# good to go
return True
@bacpypes_debugging
def setup_module(module):
if _debug: setup_module._debug("setup_module %r", module)
@ -50,23 +66,26 @@ class SampleOneShotTask(OneShotTask):
if _debug: SampleOneShotTask._debug("__init__")
OneShotTask.__init__(self)
self.process_task_called = 0
self.process_task_called = []
def process_task(self):
if _debug: SampleOneShotTask._debug("process_task @ %r", time_machine.current_time)
self.process_task_called += 1
global time_machine
# add the current time
self.process_task_called.append(time_machine.current_time)
# flag to make sure the function was called
sample_task_function_called = 0
sample_task_function_called = []
@bacpypes_debugging
def sample_task_function(*args, **kwargs):
if _debug: sample_task_function._debug("sample_task_function %r %r @ %r", args, kwargs, time_machine.current_time)
global sample_task_function_called
global sample_task_function_called, time_machine
# bump the counter
sample_task_function_called += 1
sample_task_function_called.append(time_machine.current_time)
@bacpypes_debugging
@ -76,11 +95,14 @@ class SampleRecurringTask(RecurringTask):
if _debug: SampleRecurringTask._debug("__init__")
RecurringTask.__init__(self)
self.process_task_called = 0
self.process_task_called = []
def process_task(self):
if _debug: SampleRecurringTask._debug("process_task @ %r", time_machine.current_time)
self.process_task_called += 1
global time_machine
# add the current time
self.process_task_called.append(time_machine.current_time)
@bacpypes_debugging
@ -116,7 +138,7 @@ class TestTimeMachine(unittest.TestCase):
run_time_machine(60.0)
# function called, no time has passed
assert ft.process_task_called == 1
assert almost_equal(ft.process_task_called, [0.0])
assert time_machine.current_time == 0.0
def test_function_task_immediate(self):
@ -125,7 +147,7 @@ class TestTimeMachine(unittest.TestCase):
# create a function task
ft = FunctionTask(sample_task_function)
sample_task_function_called = 0
sample_task_function_called = []
# reset the time machine, install the task, let it run
reset_time_machine()
@ -133,7 +155,7 @@ class TestTimeMachine(unittest.TestCase):
run_time_machine(60.0)
# function called, no time has passed
assert sample_task_function_called == 1
assert almost_equal(sample_task_function_called, [0.0])
assert time_machine.current_time == 0.0
def test_function_task_delay(self):
@ -144,7 +166,7 @@ class TestTimeMachine(unittest.TestCase):
# create a function task
ft = FunctionTask(sample_task_function)
sample_task_function_called = 0
sample_task_function_called = []
# reset the time machine, install the task, let it run
reset_time_machine()
@ -152,7 +174,7 @@ class TestTimeMachine(unittest.TestCase):
run_time_machine(60.0)
# function called, no time has passed
assert sample_task_function_called == 1
assert almost_equal(sample_task_function_called, [sample_delay])
assert time_machine.current_time == sample_delay
def test_recurring_task_1(self):
@ -167,7 +189,7 @@ class TestTimeMachine(unittest.TestCase):
run_time_machine(5.0)
# function called, no time has passed
assert ft.process_task_called == 4
assert almost_equal(ft.process_task_called, [1.0, 2.0, 3.0, 4.0])
assert time_machine.current_time == 5.0
def test_recurring_task_2(self):
@ -184,6 +206,37 @@ class TestTimeMachine(unittest.TestCase):
run_time_machine(5.0)
# function called, no time has passed
assert ft1.process_task_called == 4
assert ft2.process_task_called == 3
assert almost_equal(ft1.process_task_called, [1.0, 2.0, 3.0, 4.0])
assert almost_equal(ft2.process_task_called, [1.5, 3.0, 4.5])
assert time_machine.current_time == 5.0
def test_recurring_task_3(self):
if _debug: TestTimeMachine._debug("test_recurring_task_3")
# create a function task
ft = SampleRecurringTask()
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(1000.0, offset=100.0)
run_time_machine(5.0)
# function called, no time has passed
assert almost_equal(ft.process_task_called, [0.1, 1.1, 2.1, 3.1, 4.1])
assert time_machine.current_time == 5.0
def test_recurring_task_4(self):
if _debug: TestTimeMachine._debug("test_recurring_task_4")
# create a function task
ft = SampleRecurringTask()
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(1000.0, offset=-100.0)
run_time_machine(5.0)
# function called, no time has passed
assert almost_equal(ft.process_task_called, [0.9, 1.9, 2.9, 3.9, 4.9])
assert time_machine.current_time == 5.0

5
tests/time_machine.py
View File

@ -67,7 +67,8 @@ class TimeMachine(_TaskManager):
and return how long it will be until the next one should be
processed."""
if _debug: TimeMachine._debug("get_next_task @ %r", self.current_time)
if _debug: TimeMachine._debug(" - self.tasks: %r", self.tasks)
if _debug: TimeMachine._debug(" - time_limit: %r", self.time_limit)
if _debug: TimeMachine._debug(" - tasks: %r", self.tasks)
task = None
delta = None
@ -123,6 +124,7 @@ def reset_time_machine():
raise RuntimeError("no time machine")
# begin time at the beginning
time_machine.tasks = []
time_machine.current_time = 0.0
time_machine.time_limit = None
@ -146,3 +148,4 @@ def run_time_machine(time_limit):
# run until there is nothing left to do
run_once()