基于已用时间的进度条 Python

Progress bar based on elapsed time in Python

我想在 Python2.7 中添加一个基于经过时间的进度条。基本上,我从 python 执行一个 bash 脚本,大约需要 90 分钟才能完成,但被调用的底层应用程序并不经常提供信息。大约有 45 分钟,用户看不到任何事情发生,我不希望他们认为进程已冻结而终止脚本。因为我知道可执行文件大约需要多长时间才能完成,所以我想根据经过的时间添加一个进度条。这是我目前所拥有的:

    def sys_call(self, cmd, log_file="prog.log"):
        # setup toolbar and start timer
        start = datetime.datetime.now()
        stdo = sys.stdout
        msg = "elapsed {}".format(datetime.timedelta())
        stdo.write(msg)
        stdo.write("\b"*len(msg)) # go to beginning of line
        stdo.flush() # flush buffer

        # initiate command and monitor
        log = open(log_file, 'w')
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
        for line in iter(process.stdout.readline, ''): 
            log.write(line.decode('utf-8'))

            # display toolbar
            msg = "elapsed {}".format(datetime.datetime.now() - start)
            stdo.write(msg)
            stdo.write("\b"*len(msg)) # go to beginning of line
            stdo.flush() # flush buffer

        log.close()

我正在寻找一种方法来根据经过的时间更新终端中的进度。上面的 line 循环在脚本执行后长时间停止循环,因为应用程序停止向 stdout 发送任何内容,因此我的运行时间停止更新,这是误导性的。我可以计算和打印一个百分比,让用户看到进程没有被冻结,但我不确定如何 运行 在单独的线程中显示经过的时间代码。

我调查了 progressbartqdm,但不幸的是,由于我无法控制的要求,我无法使用第三方包。

每次打印条后,使用'\r'将光标移回行首,然后循环条直到任务完成,确保在每个末尾都有'\r'迭代。 这是一个如何使用的例子

import time
for x in range(101):
  print(f'{x}% done',end='\r')
  time.sleep(1)

我跟着打击片段from this link

# Python program killing
# threads using stop
# flag
  
import threading
import time
  
def run(stop):
    while True:
        print('thread running')
        if stop():
                break
                  
def main():
        stop_threads = False
        t1 = threading.Thread(target = run, args =(lambda : stop_threads, ))
        t1.start()
        time.sleep(1)
        stop_threads = True
        t1.join()
        print('thread killed')
main()

最终,这是我想出的:

import datetime
import time

    def thread_progressbar(self, stop):
        """Progress Bar Thread

        This function is to be used with `threading`. Reference example.
        Note, ensure the thread is a daemon otherwise CTRL+C will not
        work and if this occurs, you will have to kill the thread from
        the command line even though `main()` has already terminated.

        Example
        ```
        thread_stop = False
        t = threading.Thread(target=thread_progressbar, args=(lambda:thread_stop,))
        t.daemon = True # make daemon to kill thread if main terminates
        t.start() # start executing thread
        <do something else>
        thread_stop = True # break out of inf loop
        t.join() # kill thread
        ```
        """
        start = datetime.datetime.now()
        while True:
            # get elapsed time and stay at 99 if we exceed expected time
            elapsed = datetime.datetime.now() - start
            percent = int(100.0*(elapsed.total_seconds()/self.expect_time.total_seconds()))
            percent = 99 if percent >= 100 else percent

            # display toolbar
            sys.stdout.write("[{}%](elapsed {})".format(percent, elapsed))
            sys.stdout.write("\b"*len(msg)) # go to beginning of line
            sys.stdout.flush() # flush buffer

            # use stop to determine when we are done
            if stop():
                elapsed = datetime.datetime.now() - start
                sys.stdout.write("[100%](elapsed {})\n".format(elapsed))
                break
            time.sleep(1)