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 29import threading 30from time import monotonic as _time 31 32__all__ = ["scheduler"] 33 34class Event(namedtuple('Event', 'time, priority, action, argument, kwargs')): 35 __slots__ = [] 36 def __eq__(s, o): return (s.time, s.priority) == (o.time, o.priority) 37 def __lt__(s, o): return (s.time, s.priority) < (o.time, o.priority) 38 def __le__(s, o): return (s.time, s.priority) <= (o.time, o.priority) 39 def __gt__(s, o): return (s.time, s.priority) > (o.time, o.priority) 40 def __ge__(s, o): return (s.time, s.priority) >= (o.time, o.priority) 41 42Event.time.__doc__ = ('''Numeric type compatible with the return value of the 43timefunc function passed to the constructor.''') 44Event.priority.__doc__ = ('''Events scheduled for the same time will be executed 45in the order of their priority.''') 46Event.action.__doc__ = ('''Executing the event means executing 47action(*argument, **kwargs)''') 48Event.argument.__doc__ = ('''argument is a sequence holding the positional 49arguments for the action.''') 50Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword 51arguments for the action.''') 52 53_sentinel = object() 54 55class scheduler: 56 57 def __init__(self, timefunc=_time, delayfunc=time.sleep): 58 """Initialize a new instance, passing the time and delay 59 functions""" 60 self._queue = [] 61 self._lock = threading.RLock() 62 self.timefunc = timefunc 63 self.delayfunc = delayfunc 64 65 def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel): 66 """Enter a new event in the queue at an absolute time. 67 68 Returns an ID for the event which can be used to remove it, 69 if necessary. 70 71 """ 72 if kwargs is _sentinel: 73 kwargs = {} 74 event = Event(time, priority, action, argument, kwargs) 75 with self._lock: 76 heapq.heappush(self._queue, event) 77 return event # The ID 78 79 def enter(self, delay, priority, action, argument=(), kwargs=_sentinel): 80 """A variant that specifies the time as a relative time. 81 82 This is actually the more commonly used interface. 83 84 """ 85 time = self.timefunc() + delay 86 return self.enterabs(time, priority, action, argument, kwargs) 87 88 def cancel(self, event): 89 """Remove an event from the queue. 90 91 This must be presented the ID as returned by enter(). 92 If the event is not in the queue, this raises ValueError. 93 94 """ 95 with self._lock: 96 self._queue.remove(event) 97 heapq.heapify(self._queue) 98 99 def empty(self): 100 """Check whether the queue is empty.""" 101 with self._lock: 102 return not self._queue 103 104 def run(self, blocking=True): 105 """Execute events until the queue is empty. 106 If blocking is False executes the scheduled events due to 107 expire soonest (if any) and then return the deadline of the 108 next scheduled call in the scheduler. 109 110 When there is a positive delay until the first event, the 111 delay function is called and the event is left in the queue; 112 otherwise, the event is removed from the queue and executed 113 (its action function is called, passing it the argument). If 114 the delay function returns prematurely, it is simply 115 restarted. 116 117 It is legal for both the delay function and the action 118 function to modify the queue or to raise an exception; 119 exceptions are not caught but the scheduler's state remains 120 well-defined so run() may be called again. 121 122 A questionable hack is added to allow other threads to run: 123 just after an event is executed, a delay of 0 is executed, to 124 avoid monopolizing the CPU when other threads are also 125 runnable. 126 127 """ 128 # localize variable access to minimize overhead 129 # and to improve thread safety 130 lock = self._lock 131 q = self._queue 132 delayfunc = self.delayfunc 133 timefunc = self.timefunc 134 pop = heapq.heappop 135 while True: 136 with lock: 137 if not q: 138 break 139 time, priority, action, argument, kwargs = q[0] 140 now = timefunc() 141 if time > now: 142 delay = True 143 else: 144 delay = False 145 pop(q) 146 if delay: 147 if not blocking: 148 return time - now 149 delayfunc(time - now) 150 else: 151 action(*argument, **kwargs) 152 delayfunc(0) # Let other threads run 153 154 @property 155 def queue(self): 156 """An ordered list of upcoming events. 157 158 Events are named tuples with fields for: 159 time, priority, action, arguments, kwargs 160 161 """ 162 # Use heapq to sort the queue rather than using 'sorted(self._queue)'. 163 # With heapq, two events scheduled at the same time will show in 164 # the actual order they would be retrieved. 165 with self._lock: 166 events = self._queue[:] 167 return list(map(heapq.heappop, [events]*len(events))) 168