-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathrepeated_timer.py
More file actions
33 lines (29 loc) · 1.03 KB
/
repeated_timer.py
File metadata and controls
33 lines (29 loc) · 1.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import time
from threading import Event, Thread
class RepeatedTimer:
"""Repeat `function` every `interval` seconds."""
def __init__(self, interval, function, *args, **kwargs):
""" Start the timer and repeat the
function every "interval" seconds. """
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.start = time.time()
self.event = Event()
self.thread = Thread(target=self._target)
self.thread.start()
def _target(self):
while not self.event.wait(self._time):
self.function(*self.args, **self.kwargs)
@property
def _time(self):
return self.interval - ((time.time() - self.start) % self.interval)
def stop(self):
""" Stop the timer. """
self.event.set()
try:
self.thread.join()
except RuntimeError:
# Just in case the timer thread is already stopped, let it go.
pass