Python非循环函数的进度条

Question

Python非循环函数的进度条

4

给定一个函数 f，如何为其编写实时进度条以显示执行进度呢？请注意，我无法更改 f（它是来自另一个库的函数），因此无法在 f 中插入 pbar.update 调用（因此这是一篇有关非循环函数进度条的文章）。其他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, expected_time, 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"

- Ryan Park

你想让进度条在每次调用f时更新，还是在f内部的每次loop发生时更新？前者可以通过你所提到的“类似于某物”的注释实现，而后者则需要f支持一个回调函数或者像你提到的那样，能够访问f的内容。 - Geoffrey Garrett

你可能可以通过某种猴子补丁来实现后者。 - Acorn

这将取决于"f"是什么。函数调用将首先执行，经过一段时间后可能会返回。不能保证“进度条”的概念甚至对于一个函数都有意义，因此您的进度条必须根据该函数的工作方式进行特定设计。这个问题没有“通用”的解决方案。 - Hymns For Disco

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Acorn · Accepted Answer

如果该函数不允许您在工作单元之后采取行动，即通过暴露生成器接口或某种回调，那么唯一的解决方案将是使用修改后的函数版本或进行某种猴子补丁。

解决方案将针对具体的代码问题进行。

更新：

因此，如果您不介意进度条不能准确反映进度，并且只使用时间估计，您可以尝试以下操作。

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%等待，直到完成。