From a050df73a61ac9c1abf3f12c41dba6c543f77c6a Mon Sep 17 00:00:00 2001 From: Joel Bender Date: Sat, 9 Sep 2017 22:56:17 -0400 Subject: [PATCH 1/3] offset assignment bug, working around floating point issues, added tests --- py25/bacpypes/task.py | 14 ++-- py27/bacpypes/task.py | 14 ++-- py34/bacpypes/task.py | 14 ++-- tests/test_utilities/test_time_machine.py | 87 ++++++++++++++++++----- tests/time_machine.py | 5 +- 5 files changed, 101 insertions(+), 33 deletions(-) diff --git a/py25/bacpypes/task.py b/py25/bacpypes/task.py index 135698a..319d1a3 100755 --- a/py25/bacpypes/task.py +++ b/py25/bacpypes/task.py @@ -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) diff --git a/py27/bacpypes/task.py b/py27/bacpypes/task.py index c59360d..8340ae5 100755 --- a/py27/bacpypes/task.py +++ b/py27/bacpypes/task.py @@ -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) diff --git a/py34/bacpypes/task.py b/py34/bacpypes/task.py index aa3634a..17fbf35 100755 --- a/py34/bacpypes/task.py +++ b/py34/bacpypes/task.py @@ -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) diff --git a/tests/test_utilities/test_time_machine.py b/tests/test_utilities/test_time_machine.py index 549c391..877af3b 100644 --- a/tests/test_utilities/test_time_machine.py +++ b/tests/test_utilities/test_time_machine.py @@ -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 + diff --git a/tests/time_machine.py b/tests/time_machine.py index dd936f6..fcde01f 100755 --- a/tests/time_machine.py +++ b/tests/time_machine.py @@ -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() + From 88e9afd2021629a9bffe2e3366f332d8337ce222 Mon Sep 17 00:00:00 2001 From: Joel Bender Date: Mon, 11 Sep 2017 10:29:42 -0400 Subject: [PATCH 2/3] fixes #138 --- py25/bacpypes/service/device.py | 4 ++-- py27/bacpypes/service/device.py | 4 ++-- py34/bacpypes/service/device.py | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/py25/bacpypes/service/device.py b/py25/bacpypes/service/device.py index 659f7eb..78bb82a 100644 --- a/py25/bacpypes/service/device.py +++ b/py25/bacpypes/service/device.py @@ -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') # diff --git a/py27/bacpypes/service/device.py b/py27/bacpypes/service/device.py index 3defb47..fc65568 100644 --- a/py27/bacpypes/service/device.py +++ b/py27/bacpypes/service/device.py @@ -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') # diff --git a/py34/bacpypes/service/device.py b/py34/bacpypes/service/device.py index cc257ae..e801354 100644 --- a/py34/bacpypes/service/device.py +++ b/py34/bacpypes/service/device.py @@ -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') # From 961fed7135a30eb8319f5fa78a16eda0ac8a0f49 Mon Sep 17 00:00:00 2001 From: Joel Bender Date: Tue, 19 Sep 2017 21:07:29 -0400 Subject: [PATCH 3/3] additional sample for helping with #140 --- samples/ThreadedReadProperty.py | 211 ++++++++++++++++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100755 samples/ThreadedReadProperty.py diff --git a/samples/ThreadedReadProperty.py b/samples/ThreadedReadProperty.py new file mode 100755 index 0000000..dfe3d9c --- /dev/null +++ b/samples/ThreadedReadProperty.py @@ -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() +