给定功能f
,为f
编写进度条的最佳方法是什么?即,具有一个实时进度栏,该进度栏在执行f
时进行更新。请注意,我无法更改f
(这是另一个库中的函数),因此无法在pbar.update
中插入f
调用(因此这是有关进度条的帖子非循环功能)。其他SO帖子在您可以更改f
中的代码的情况下解决了此问题,但是当我无法访问f
的内容时,我找不到/没有解决方案
我是否必须使用线程或多处理来实现类似的目的?
类似的东西:
@progress_bar
def func_wrapper(*args,**kwargs):
return f(*args,**kwargs)
或:
start_progress_bar()
f()
感谢您的帮助!
更新:我已经使用@Acorn答案中提供的代码,并以装饰器形式重写了它。
import concurrent.futures
import functools
import time
from tqdm import tqdm
def progress_bar(expected_time,increments=10):
def _progress_bar(func):
def timed_progress_bar(future,expected_time,increments=10):
"""
Display progress bar for expected_time seconds.
Complete early if future completes.
Wait for future if it doesn't complete in expected_time.
"""
interval = expected_time / increments
with tqdm(total=increments) as pbar:
for i in range(increments - 1):
if future.done():
# finish the progress bar
# not sure if there's a cleaner way to do this?
pbar.update(increments - i)
return
else:
time.sleep(interval)
pbar.update()
# if the future still hasn't completed,wait for it.
future.result()
pbar.update()
@functools.wraps(func)
def _func(*args,**kwargs):
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
future = pool.submit(func,*args,**kwargs)
timed_progress_bar(future,increments)
return future.result()
return _func
return _progress_bar
if __name__ == "__main__":
@progress_bar(expected_time=11)
def test_func():
time.sleep(10)
return "result"
print(test_func()) # prints "result"