3 # © Copyright 2021-2022, Scott Gasch
5 """Utilities for dealing with threads + threading."""
11 from typing import Any, Callable, Optional, Tuple
13 # This module is commonly used by others in here and should avoid
14 # taking any unnecessary dependencies back on them.
16 logger = logging.getLogger(__name__)
19 def current_thread_id() -> str:
22 a string composed of the parent process' id, the current
23 process' id and the current thread identifier. The former two are
24 numbers (pids) whereas the latter is a thread id passed during thread
27 >>> ret = current_thread_id()
28 >>> (ppid, pid, tid) = ret.split('/')
37 tid = threading.current_thread().name
38 return f'{ppid}/{pid}/{tid}:'
41 def is_current_thread_main_thread() -> bool:
44 True is the current (calling) thread is the process' main
45 thread and False otherwise.
47 >>> is_current_thread_main_thread()
53 ... result = is_current_thread_main_thread()
60 >>> thread = threading.Thread(target=thunk)
67 return threading.current_thread() is threading.main_thread()
70 def background_thread(
71 _funct: Optional[Callable[..., Any]],
72 ) -> Callable[..., Tuple[threading.Thread, threading.Event]]:
73 """A function decorator to create a background thread.
78 def random(a: int, b: str, stop_event: threading.Event) -> None:
80 print(f"Hi there {b}: {a}!")
82 if stop_event.is_set():
86 (thread, event) = random(22, "dude")
94 In addition to any other arguments the function has, it must
95 take a stop_event as the last unnamed argument which it should
96 periodically check. If the event is set, it means the thread has
97 been requested to terminate ASAP.
100 def wrapper(funct: Callable):
101 @functools.wraps(funct)
102 def inner_wrapper(*a, **kwa) -> Tuple[threading.Thread, threading.Event]:
103 should_terminate = threading.Event()
104 should_terminate.clear()
105 newargs = (*a, should_terminate)
106 thread = threading.Thread(
112 logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
113 return (thread, should_terminate)
118 return wrapper # type: ignore
120 return wrapper(_funct)
123 def periodically_invoke(
125 stop_after: Optional[int],
128 Periodically invoke the decorated function.
131 period_sec: the delay period in seconds between invocations
132 stop_after: total number of invocations to make or, if None,
136 a :class:`Thread` object and an :class:`Event` that, when
137 signaled, will stop the invocations.
140 It is possible to be invoked one time after the :class:`Event`
141 is set. This event can be used to stop infinite
142 invocation style or finite invocation style decorations.
146 @periodically_invoke(period_sec=0.5, stop_after=None)
147 def there(name: str, age: int) -> None:
148 print(f" ...there {name}, {age}")
150 @periodically_invoke(period_sec=1.0, stop_after=3)
151 def hello(name: str) -> None:
152 print(f"Hello, {name}")
155 def decorator_repeat(func):
156 def helper_thread(should_terminate, *args, **kwargs) -> None:
157 if stop_after is None:
159 func(*args, **kwargs)
160 res = should_terminate.wait(period_sec)
164 for _ in range(stop_after):
165 func(*args, **kwargs)
166 res = should_terminate.wait(period_sec)
171 @functools.wraps(func)
172 def wrapper_repeat(*args, **kwargs):
173 should_terminate = threading.Event()
174 should_terminate.clear()
175 newargs = (should_terminate, *args)
176 thread = threading.Thread(target=helper_thread, args=newargs, kwargs=kwargs)
178 logger.debug('Started thread "%s" tid=%d', thread.name, thread.ident)
179 return (thread, should_terminate)
181 return wrapper_repeat
183 return decorator_repeat
186 if __name__ == '__main__':