I needed to display a waiting message for the user while a function is doing its work (in this case specifically, waiting for a response from Google's speech-to-text service). I therefore used threading for this task, and it works well. Later it occured to me this could be useful in the future, so decided to turn this into a decorator, which is something I haven't done before. After many tries (including an ill-fated attempt at using a class for it), this is what I came up with:
def loading_decorator(function):
def loading_wrapper(*function_args, active=True):
def loading_animation(speed=4, loading_text='Processing... '):
for symbol in cycle(r'-\|/-\|/'):
if not active:
print('\b' * (len(loading_text) + 1))
break
print('\r', f'{loading_text}', symbol, sep='', end='')
sleep(1 / speed)
loading_thread = threading.Thread(target=loading_animation)
loading_thread.start()
result = function(*function_args)
active = False
loading_thread.join()
return result
return loading_wrapper
... and this is the function I've decorated:
@loading_decorator
def process_speech(audio):
return r.recognize_google(audio, language=LANGUAGE)
I was happy to see it worked as intended. I have two questions however:
time.sleep(0.5)
in the code in place of loading_thread.join()
. I put it in because without sleeping, the result would be returned before the cleanup activated by active = False
happened, and the loading text remained in the window. I found this surprising, can this behaviour be explained easily? I guess I assumed (wrongly) the change to active
would somehow nudge the loading thread to stop looping. Is there a reason the main thread is given priority here? Is this something obvious to more knowledgeable people?Only later I thought using loading_thread.join()
instead of sleeping might be more proper. I'm not feeling confident here, since this is my first foray into both threading and decorators (except for some very basic ones). Which brings me to my second, more general question:
Your code seems allright + you have some suggestions already in comments.
Here's my take on the problem:
I'd "reverse" the roles and let the function run inside ThreadPoolExecutor
and the printing code in the main thread. That way I'd get rid-off the variable active
.
Also I'm using functools.wraps
+ I use decorator parameters to make it more nice.
import concurrent.futures
from functools import wraps
from itertools import cycle
from time import sleep
def loading_decorator(speed=0.2, loading_text="Processing... "):
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(f, *args, **kwargs)
for ch in cycle(r"-\|/-\|/"):
print("\r" + loading_text + ch, end="", flush=True)
if future.done():
print()
return future.result()
sleep(speed)
return wrapper
return decorator
@loading_decorator(speed=0.1)
def some_long_task():
sleep(5)
return "result"
x = some_long_task()
print(x)