7 from typing import Callable, Optional, Tuple
9 # This module is commonly used by others in here and should avoid
10 # taking any unnecessary dependencies back on them.
12 logger = logging.getLogger(__name__)
15 def current_thread_id() -> str:
18 tid = threading.current_thread().name
19 return f'{ppid}/{pid}/{tid}:'
22 def is_current_thread_main_thread() -> bool:
23 """Returns True is the current (calling) thread is the process' main
24 thread and False otherwise.
26 return threading.current_thread() is threading.main_thread()
29 def background_thread(
30 _funct: Optional[Callable]
31 ) -> Tuple[threading.Thread, threading.Event]:
32 """A function decorator to create a background thread.
34 *** Please note: the decorated function must take an shutdown ***
35 *** event as an input parameter and should periodically check ***
36 *** it and stop if the event is set. ***
41 def random(a: int, b: str, stop_event: threading.Event) -> None:
43 print(f"Hi there {b}: {a}!")
45 if stop_event.is_set():
50 (thread, event) = random(22, "dude")
56 Note: in addition to any other arguments the function has, it must
57 take a stop_event as the last unnamed argument which it should
58 periodically check. If the event is set, it means the thread has
59 been requested to terminate ASAP.
61 def wrapper(funct: Callable):
62 @functools.wraps(funct)
65 ) -> Tuple[threading.Thread, threading.Event]:
66 should_terminate = threading.Event()
67 should_terminate.clear()
68 newargs = (*a, should_terminate)
69 thread = threading.Thread(
76 f'Started thread {thread.name} tid={thread.ident}'
78 return (thread, should_terminate)
84 return wrapper(_funct)
87 def periodically_invoke(
89 stop_after: Optional[int],
92 Periodically invoke a decorated function. Stop after N invocations
93 (or, if stop_after is None, call forever). Delay period_sec between
96 Returns a Thread object and an Event that, when signaled, will stop
97 the invocations. Note that it is possible to be invoked one time
98 after the Event is set. This event can be used to stop infinite
99 invocation style or finite invocation style decorations.
101 @periodically_invoke(period_sec=0.5, stop_after=None)
102 def there(name: str, age: int) -> None:
103 print(f" ...there {name}, {age}")
106 @periodically_invoke(period_sec=1.0, stop_after=3)
107 def hello(name: str) -> None:
108 print(f"Hello, {name}")
111 def decorator_repeat(func):
112 def helper_thread(should_terminate, *args, **kwargs) -> None:
113 if stop_after is None:
115 func(*args, **kwargs)
116 res = should_terminate.wait(period_sec)
120 for _ in range(stop_after):
121 func(*args, **kwargs)
122 res = should_terminate.wait(period_sec)
127 @functools.wraps(func)
128 def wrapper_repeat(*args, **kwargs):
129 should_terminate = threading.Event()
130 should_terminate.clear()
131 newargs = (should_terminate, *args)
132 thread = threading.Thread(
133 target=helper_thread,
139 f'Started thread {thread.name} tid={thread.ident}'
141 return (thread, should_terminate)
142 return wrapper_repeat
143 return decorator_repeat