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

make the time limit manditory, and make sure the machine runs up to, but does not exceed, the limit

Browse Source
This commit is contained in:
Joel Bender 2015-08-09 12:28:46 -04:00
parent 523e462fbf
commit 46249dd55a
1 changed files with 24 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
tests/time_machine.py
View File

@ -70,32 +70,40 @@ class TimeMachine(_TaskManager):
"""get the next task if there's one that should be processed, """get the next task if there's one that should be processed,
and return how long it will be until the next one should be and return how long it will be until the next one should be
processed.""" processed."""
if _debug: TimeMachine._debug("get_next_task") if _debug: TimeMachine._debug("get_next_task @ %r", self.current_time)
if _debug: TimeMachine._debug(" - self: %r", self)
if _debug: TimeMachine._debug(" - self.tasks: %r", self.tasks) if _debug: TimeMachine._debug(" - self.tasks: %r", self.tasks)
task = None task = None
delta = None delta = None
if (self.time_limit is not None) and (self.current_time > self.time_limit): if (self.time_limit is not None) and (self.current_time >= self.time_limit):
if _debug: TimeMachine._debug(" - time limit reached") if _debug: TimeMachine._debug(" - time limit reached")
elif not self.tasks: elif not self.tasks:
if _debug: TimeMachine._debug(" - no more tasks") if _debug: TimeMachine._debug(" - no more tasks")
else: else:
# pull it off the list # peek at the next task and see when it is supposed to run
when, task = heappop(self.tasks) when, _ = self.tasks[0]
if _debug: TimeMachine._debug(" - when, task: %r, %r", when, task) if when >= self.time_limit:
if _debug: TimeMachine._debug(" - time limit reached")
# mark that it is no longer scheduled # bump up to the time limit
task.isScheduled = False self.current_time = self.time_limit
# advance the time else:
self.current_time = when # pull it off the list
when, task = heappop(self.tasks)
if _debug: TimeMachine._debug(" - when, task: %r, %s", when, task)
# do not wait, time has moved # mark that it is no longer scheduled
delta = 0.0 task.isScheduled = False
# advance the time
self.current_time = when
# do not wait, time has moved
delta = 0.0
# return the task to run and how long to wait for the next one # return the task to run and how long to wait for the next one
return (task, delta) return (task, delta)
@ -130,7 +138,7 @@ def reset_time_machine():
# #
@bacpypes_debugging @bacpypes_debugging
def run_time_machine(time_limit=None): def run_time_machine(time_limit):
"""This function is called after a set of tasks have been installed """This function is called after a set of tasks have been installed
and they should all run. and they should all run.
""" """
@ -140,8 +148,10 @@ def run_time_machine(time_limit=None):
# a little error checking # a little error checking
if not time_machine: if not time_machine:
raise RuntimeError("no time machine") raise RuntimeError("no time machine")
if time_limit <= 0.0:
raise ValueError("time limit required")
# let the task manager know there is a virtual time limit # pass the limit to the time machine
time_machine.time_limit = time_limit time_machine.time_limit = time_limit
# run until there is nothing left to do # run until there is nothing left to do