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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
#!/usr/bin/env python3
import functools
import logging
import os
import threading
from typing import Callable, Optional, Tuple
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
logger = logging.getLogger(__name__)
def current_thread_id() -> str:
"""Returns a string composed of the parent process' id, the current
process' id and the current thread identifier. The former two are
numbers (pids) whereas the latter is a thread id passed during thread
creation time.
>>> ret = current_thread_id()
>>> (ppid, pid, tid) = ret.split('/')
>>> ppid.isnumeric()
True
>>> pid.isnumeric()
True
"""
ppid = os.getppid()
pid = os.getpid()
tid = threading.current_thread().name
return f'{ppid}/{pid}/{tid}:'
def is_current_thread_main_thread() -> bool:
"""Returns True is the current (calling) thread is the process' main
thread and False otherwise.
>>> is_current_thread_main_thread()
True
>>> result = None
>>> def thunk():
... global result
... result = is_current_thread_main_thread()
>>> thunk()
>>> result
True
>>> import threading
>>> thread = threading.Thread(target=thunk)
>>> thread.start()
>>> thread.join()
>>> result
False
"""
return threading.current_thread() is threading.main_thread()
def background_thread(
_funct: Optional[Callable],
) -> Tuple[threading.Thread, threading.Event]:
"""A function decorator to create a background thread.
*** Please note: the decorated function must take an shutdown ***
*** event as an input parameter and should periodically check ***
*** it and stop if the event is set. ***
Usage:
@background_thread
def random(a: int, b: str, stop_event: threading.Event) -> None:
while True:
print(f"Hi there {b}: {a}!")
time.sleep(10.0)
if stop_event.is_set():
return
def main() -> None:
(thread, event) = random(22, "dude")
print("back!")
time.sleep(30.0)
event.set()
thread.join()
Note: in addition to any other arguments the function has, it must
take a stop_event as the last unnamed argument which it should
periodically check. If the event is set, it means the thread has
been requested to terminate ASAP.
"""
def wrapper(funct: Callable):
@functools.wraps(funct)
def inner_wrapper(*a, **kwa) -> Tuple[threading.Thread, threading.Event]:
should_terminate = threading.Event()
should_terminate.clear()
newargs = (*a, should_terminate)
thread = threading.Thread(
target=funct,
args=newargs,
kwargs=kwa,
)
thread.start()
logger.debug(f'Started thread {thread.name} tid={thread.ident}')
return (thread, should_terminate)
return inner_wrapper
if _funct is None:
return wrapper
else:
return wrapper(_funct)
def periodically_invoke(
period_sec: float,
stop_after: Optional[int],
):
"""
Periodically invoke a decorated function. Stop after N invocations
(or, if stop_after is None, call forever). Delay period_sec between
invocations.
Returns a Thread object and an Event that, when signaled, will stop
the invocations. Note that it is possible to be invoked one time
after the Event is set. This event can be used to stop infinite
invocation style or finite invocation style decorations.
@periodically_invoke(period_sec=0.5, stop_after=None)
def there(name: str, age: int) -> None:
print(f" ...there {name}, {age}")
@periodically_invoke(period_sec=1.0, stop_after=3)
def hello(name: str) -> None:
print(f"Hello, {name}")
"""
def decorator_repeat(func):
def helper_thread(should_terminate, *args, **kwargs) -> None:
if stop_after is None:
while True:
func(*args, **kwargs)
res = should_terminate.wait(period_sec)
if res:
return
else:
for _ in range(stop_after):
func(*args, **kwargs)
res = should_terminate.wait(period_sec)
if res:
return
return
@functools.wraps(func)
def wrapper_repeat(*args, **kwargs):
should_terminate = threading.Event()
should_terminate.clear()
newargs = (should_terminate, *args)
thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
thread.start()
logger.debug(f'Started thread {thread.name} tid={thread.ident}')
return (thread, should_terminate)
return wrapper_repeat
return decorator_repeat
if __name__ == '__main__':
import doctest
doctest.testmod()
|
