1"""A generally useful event scheduler class. 2 3Each instance of this class manages its own queue. 4No multi-threading is implied; you are supposed to hack that 5yourself, or use a single instance per application. 6 7Each instance is parametrized with two functions, one that is 8supposed to return the current time, one that is supposed to 9implement a delay. You can implement real-time scheduling by 10substituting time and sleep from built-in module time, or you can 11implement simulated time by writing your own functions. This can 12also be used to integrate scheduling with STDWIN events; the delay 13function is allowed to modify the queue. Time can be expressed as 14integers or floating point numbers, as long as it is consistent. 15 16Events are specified by tuples (time, priority, action, argument, kwargs). 17As in UNIX, lower priority numbers mean higher priority; in this 18way the queue can be maintained as a priority queue. Execution of the 19event means calling the action function, passing it the argument 20sequence in "argument" (remember that in Python, multiple function 21arguments are be packed in a sequence) and keyword parameters in "kwargs". 22The action function may be an instance method so it 23has another way to reference private data (besides global variables). 24""" 25 26import time 27import heapq 28from collections import namedtuple 29try: 30 import threading 31except ImportError: 32 import dummy_threading as threading 33from time import monotonic as _time 34 35__all__ = ["scheduler"] 36 37class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')): 38 __slots__ = [] 39 def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority) 40 def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority) 41 def __le__(s, o): return (s.time, s.priority) <= (o.time, o.priority) 42 def __gt__(s, o): return (s.time, s.priority) > (o.time, o.priority) 43 def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority) 44 45Event.time.__doc__ = ('''Numeric type compatible with the return value of the 46timefunc function passed to the constructor.''') 47Event.priority.__doc__ = ('''Events scheduled for the same time will be executed 48in the order of their priority.''') 49Event.action.__doc__ = ('''Executing the event means executing 50action(*argument, **kwargs)''') 51Event.argument.__doc__ = ('''argument is a sequence holding the positional 52arguments for the action.''') 53Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword 54arguments for the action.''') 55 56_sentinel = object() 57 58class scheduler: 59 60 def __init__(self, timefunc=_time, delayfunc=time.sleep): 61 """Initialize a new instance, passing the time and delay 62 functions""" 63 self._queue = [] 64 self._lock = threading.RLock() 65 self.timefunc = timefunc 66 self.delayfunc = delayfunc 67 68 def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel): 69 """Enter a new event in the queue at an absolute time. 70 71 Returns an ID for the event which can be used to remove it, 72 if necessary. 73 74 """ 75 if kwargs is _sentinel: 76 kwargs = {} 77 event = Event(time, priority, action, argument, kwargs) 78 with self._lock: 79 heapq.heappush(self._queue, event) 80 return event # The ID 81 82 def enter(self, delay, priority, action, argument=(), kwargs=_sentinel): 83 """A variant that specifies the time as a relative time. 84 85 This is actually the more commonly used interface. 86 87 """ 88 time = self.timefunc() + delay 89 return self.enterabs(time, priority, action, argument, kwargs) 90 91 def cancel(self, event): 92 """Remove an event from the queue. 93 94 This must be presented the ID as returned by enter(). 95 If the event is not in the queue, this raises ValueError. 96 97 """ 98 with self._lock: 99 self._queue.remove(event) 100 heapq.heapify(self._queue) 101 102 def empty(self): 103 """Check whether the queue is empty.""" 104 with self._lock: 105 return not self._queue 106 107 def run(self, blocking=True): 108 """Execute events until the queue is empty. 109 If blocking is False executes the scheduled events due to 110 expire soonest (if any) and then return the deadline of the 111 next scheduled call in the scheduler. 112 113 When there is a positive delay until the first event, the 114 delay function is called and the event is left in the queue; 115 otherwise, the event is removed from the queue and executed 116 (its action function is called, passing it the argument). If 117 the delay function returns prematurely, it is simply 118 restarted. 119 120 It is legal for both the delay function and the action 121 function to modify the queue or to raise an exception; 122 exceptions are not caught but the scheduler's state remains 123 well-defined so run() may be called again. 124 125 A questionable hack is added to allow other threads to run: 126 just after an event is executed, a delay of 0 is executed, to 127 avoid monopolizing the CPU when other threads are also 128 runnable. 129 130 """ 131 # localize variable access to minimize overhead 132 # and to improve thread safety 133 lock = self._lock 134 q = self._queue 135 delayfunc = self.delayfunc 136 timefunc = self.timefunc 137 pop = heapq.heappop 138 while True: 139 with lock: 140 if not q: 141 break 142 time, priority, action, argument, kwargs = q[0] 143 now = timefunc() 144 if time > now: 145 delay = True 146 else: 147 delay = False 148 pop(q) 149 if delay: 150 if not blocking: 151 return time - now 152 delayfunc(time - now) 153 else: 154 action(*argument, **kwargs) 155 delayfunc(0) # Let other threads run 156 157 @property 158 def queue(self): 159 """An ordered list of upcoming events. 160 161 Events are named tuples with fields for: 162 time, priority, action, arguments, kwargs 163 164 """ 165 # Use heapq to sort the queue rather than using 'sorted(self._queue)'. 166 # With heapq, two events scheduled at the same time will show in 167 # the actual order they would be retrieved. 168 with self._lock: 169 events = self._queue[:] 170 return list(map(heapq.heappop, [events]*len(events))) 171