Refactoring. Add ImmutablePoint.
import time
class TimerEvent:
__slots__ = ('time', 'callback', 'args')
def __init__(self, delay, callback, args):
self.time = time.time() + delay
self.callback = callback
self.args = args
def execute(self):
self.callback(*self.args)
def __lt__(self, other):
return self.time < other.time
def __str__(self):
return '{:.2f}s {} {}'.format(
self.time - time.time(), self.callback, self.args)
class Timer:
def __init__(self):
#: Each timeout
self.events = []
def add_timeout(self, delay, callback, *args):
"""Register `callback` to be called after `delay` seconds."""
self.events.append(TimerEvent(delay, callback, args))
self.events.sort(reverse=True)
def remove_timeout(self, callback, *args):
"""Unregister callback previously registered with add_timeout.
Removes all timeouts with the `callback` and `args`.
"""
for event in self.events[:]:
if event.callback == callback and event.args == args:
self.events.remove(event)
def nearest_timeout(self):
"""Get interval in seconds when next timeout expires.
Returns None when no timeout is registered.
"""
if not len(self.events):
return None
timeout = self.events[-1].time - time.time()
return max(0, timeout)
def process_timeouts(self):
"""Execute all expired timeouts."""
now = time.time()
while len(self.events) and self.events[-1].time <= now:
self.events[-1].execute()
self.events.pop()
def __str__(self):
return '\n'.join(str(event) for event in self.events)
if __name__ == '__main__':
# Self Test
t = Timer()
t.process_timeouts()
t.add_timeout(1, print, 'timeout 1')
t.add_timeout(3, print, 'timeout 3')
t.add_timeout(2, print, 'timeout 2')
print(t)
print('Nearest: %.2f' % t.nearest_timeout())
t.process_timeouts()
time.sleep(1)
t.process_timeouts()
print(t)
time.sleep(2)
t.process_timeouts()