codekingpro/portable-devtools

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 that5yourself, or use a single instance per application.6 7Each instance is parametrized with two functions, one that is8supposed to return the current time, one that is supposed to9implement a delay.  You can implement real-time scheduling by10substituting time and sleep from built-in module time, or you can11implement simulated time by writing your own functions.  This can12also be used to integrate scheduling with STDWIN events; the delay13function is allowed to modify the queue.  Time can be expressed as14integers 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 this18way the queue can be maintained as a priority queue.  Execution of the19event means calling the action function, passing it the argument20sequence in "argument" (remember that in Python, multiple function21arguments are be packed in a sequence) and keyword parameters in "kwargs".22The action function may be an instance method so it23has another way to reference private data (besides global variables).24"""25 26import time27import heapq28from collections import namedtuple29from itertools import count30import threading31from time import monotonic as _time32 33__all__ = ["scheduler"]34 35Event = namedtuple('Event', 'time, priority, sequence, action, argument, kwargs')36Event.time.__doc__ = ('''Numeric type compatible with the return value of the37timefunc function passed to the constructor.''')38Event.priority.__doc__ = ('''Events scheduled for the same time will be executed39in the order of their priority.''')40Event.sequence.__doc__ = ('''A continually increasing sequence number that41    separates events if time and priority are equal.''')42Event.action.__doc__ = ('''Executing the event means executing43action(*argument, **kwargs)''')44Event.argument.__doc__ = ('''argument is a sequence holding the positional45arguments for the action.''')46Event.kwargs.__doc__ = ('''kwargs is a dictionary holding the keyword47arguments for the action.''')48 49_sentinel = object()50 51class scheduler:52 53    def __init__(self, timefunc=_time, delayfunc=time.sleep):54        """Initialize a new instance, passing the time and delay55        functions"""56        self._queue = []57        self._lock = threading.RLock()58        self.timefunc = timefunc59        self.delayfunc = delayfunc60        self._sequence_generator = count()61 62    def enterabs(self, time, priority, action, argument=(), kwargs=_sentinel):63        """Enter a new event in the queue at an absolute time.64 65        Returns an ID for the event which can be used to remove it,66        if necessary.67 68        """69        if kwargs is _sentinel:70            kwargs = {}71 72        with self._lock:73            event = Event(time, priority, next(self._sequence_generator),74                          action, argument, kwargs)75            heapq.heappush(self._queue, event)76        return event # The ID77 78    def enter(self, delay, priority, action, argument=(), kwargs=_sentinel):79        """A variant that specifies the time as a relative time.80 81        This is actually the more commonly used interface.82 83        """84        time = self.timefunc() + delay85        return self.enterabs(time, priority, action, argument, kwargs)86 87    def cancel(self, event):88        """Remove an event from the queue.89 90        This must be presented the ID as returned by enter().91        If the event is not in the queue, this raises ValueError.92 93        """94        with self._lock:95            self._queue.remove(event)96            heapq.heapify(self._queue)97 98    def empty(self):99        """Check whether the queue is empty."""100        with self._lock:101            return not self._queue102 103    def run(self, blocking=True):104        """Execute events until the queue is empty.105        If blocking is False executes the scheduled events due to106        expire soonest (if any) and then return the deadline of the107        next scheduled call in the scheduler.108 109        When there is a positive delay until the first event, the110        delay function is called and the event is left in the queue;111        otherwise, the event is removed from the queue and executed112        (its action function is called, passing it the argument).  If113        the delay function returns prematurely, it is simply114        restarted.115 116        It is legal for both the delay function and the action117        function to modify the queue or to raise an exception;118        exceptions are not caught but the scheduler's state remains119        well-defined so run() may be called again.120 121        A questionable hack is added to allow other threads to run:122        just after an event is executed, a delay of 0 is executed, to123        avoid monopolizing the CPU when other threads are also124        runnable.125 126        """127        # localize variable access to minimize overhead128        # and to improve thread safety129        lock = self._lock130        q = self._queue131        delayfunc = self.delayfunc132        timefunc = self.timefunc133        pop = heapq.heappop134        while True:135            with lock:136                if not q:137                    break138                (time, priority, sequence, action,139                 argument, kwargs) = q[0]140                now = timefunc()141                if time > now:142                    delay = True143                else:144                    delay = False145                    pop(q)146            if delay:147                if not blocking:148                    return time - now149                delayfunc(time - now)150            else:151                action(*argument, **kwargs)152                delayfunc(0)   # Let other threads run153 154    @property155    def queue(self):156        """An ordered list of upcoming events.157 158        Events are named tuples with fields for:159            time, priority, action, arguments, kwargs160 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 in164        # 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