非循环功能的Python进度栏

给定功能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"

ether1984 回答:非循环功能的Python进度栏

如果函数不允许您在工作单元后采取措施,即通过公开某种生成器接口或某种回调,则唯一的解决方案是使用函数的修改版本或进行某种操作猴子修补的。

解决方案将特定于所讨论的代码。


更新:

因此,如果您不介意进度栏不能准确反映进度,而仅使用时间估算,就可以执行以下操作。

import concurrent.futures
import time

from tqdm import tqdm


def timed_future_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()


def blocking_job():
    time.sleep(2)
    return 'result'


def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(blocking_job)
        timed_future_progress_bar(future,5)
    print(f'Work done: {future.result()}')

main()

无论工作花费的时间比预期的长还是短,这应该明智地表现。如果作业的运行时间比预期的长,那么进度将等待90%,直到完成。

本文链接:https://www.f2er.com/3043638.html

大家都在问