-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtimers.py
81 lines (67 loc) · 2.67 KB
/
timers.py
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import time
class Time:
_timers = dict()
_cancelled = list()
def __init__(self, seconds: int = None, function: object = None, mode: str = None, *args, **kw) -> None:
self.mode = mode
self.secs = seconds
self.function = function
self.args = args
self.kwargs = kw
self.time = time.time()
def setInterval(self, name: str, seconds: int, function: object, *args, **kw):
"""
set a never ending timer to run x seconds
@name: unique string to name timer object used to cancel
@seconds: a number of secounds before timer runs
@function: function to be called when seconds is hit
"""
interval = Time(seconds, function, "interval", *args, **kw)
if name not in self._timers:
self._timers[name] = interval
else:
raise Exception(f"INTERVAL: {name} exists already")
def setTimeout(self, name: str, seconds: int, function: object, *args, **kw):
"""
set an ending timer to run at x seconds then stops
@name: unique string to name timer object used to cancel
@seconds: a number of secounds before timer runs
@function: function to be called when seconds is hit
"""
timeout = Time(seconds, function, "timeout", *args, **kw)
if name not in self._timers:
self._timers[name] = timeout
else:
raise Exception(f"TIMEOUT: {name} exists already")
def cancel(self, name):
"""
cancel the timer
@name: the unique string you used to create the timer
"""
if name not in self._cancelled:
self._cancelled.append(name)
def run(self):
"""
call in a while loop runs timers and checks if cancelled
"""
for key, value in iter(self._timers.items()):
t = int((time.time() - value.time) % 60)
if t >= value.secs:
if value.mode == "timeout":
print(f"RUNNING: {key} MODE: {value.mode}")
value.function(*value.args, **value.kwargs)
self.cancel(key)
else:
print(f"RUNNING: {key} MODE: {value.mode}")
value.function(*value.args, **value.kwargs)
value.time = time.time()
for key in self._cancelled:
del self._timers[key]
print(f"CANCELLED: {key}")
self._cancelled = list()
if __name__=="__main__":
timer = Time()
timer.setTimeout("test", 3, print, "#" * 80)
timer.setInterval("test2", 2, print, "#" * 80)
while True:
timer.run()