バッチスクリプトの進捗をコンソールに出力する方法

README.md

progress

progress.py

import sys
import time


def process(wait):
    time.sleep(wait / 1000)


def main():

    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])

    for i in range(repeat):
        process(wait)


if __name__ == "__main__":
    main()

progress1.py

import sys
import time


def process(wait):
    time.sleep(wait / 1000)


def main():

    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])

    for i in range(1, repeat+1):
        print(i)
        process(wait)


if __name__ == "__main__":
    main()

progress2.py

import sys
import time


def process(wait):
    time.sleep(wait / 1000)


def main():

    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])
    interval = int(sys.argv[3])

    for i in range(1, repeat+1):
        if i % interval == 0:
            print(i)
        process(wait)

if __name__ == "__main__":
    main()

progress3.py

import sys
import time


def process(wait):
    time.sleep(wait / 1000)


def main():

    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])
    interval = int(sys.argv[3])

    for i in range(1, repeat+1):
        if i % interval == 0:
            print("\r", i, sep="", end="", flush=True)
        process(wait)
    print("\r", i, sep="", flush=True)


if __name__ == "__main__":
    main()

progress4.py

import sys
import time
from contextlib import contextmanager
from threading import Thread, Event


@contextmanager
def progress(report, interval=30):

    def run():
        while not event.wait(interval / 1000):
            print("\r", report(), sep="", end="", flush=True)
        print("\r", report(), sep="", flush=True)

    event = Event()
    Thread(target=run).start()

    yield

    event.set()


def process(wait):
    time.sleep(wait / 1000)


def main():
    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])
    interval = int(sys.argv[3])

    i = 0
    with progress(lambda: i, interval):
        for i in range(1, repeat+1):
            process(wait)


if __name__ == "__main__":
    main()

progress5.py

import sys
import time
from contextlib import contextmanager
from threading import Thread, Event


@contextmanager
def progress(report, interval=30):

    def run():
        while not event.wait(interval / 1000):
            print("\r", report(), sep="", end="", flush=True)
        print("\r", report(), sep="", flush=True)

    event = Event()
    Thread(target=run).start()

    yield

    event.set()


def process(wait):
    time.sleep(wait / 1000)


def main():
    wait = int(sys.argv[1])
    repeat = int(sys.argv[2])
    interval = int(sys.argv[3])

    i = 0

    def report():
        bar = "{: <{}s}".format(("=" * int(i / repeat * 40) + ">")[:40], 40)
        return "[{}] {}/{}".format(bar, i, repeat)

    with progress(report, interval):
        for i in range(1, repeat+1):
            process(wait)


if __name__ == "__main__":
    main()

progress6.py

from concurrent.futures.thread import ThreadPoolExecutor
from contextlib import contextmanager
from threading import Thread, Event
import time

import sys


@contextmanager
def progress(report, interval=0.03):

    def run():
        while not event.wait(interval):
            print("\r", report(), sep="", end="", flush=True)
        print("\r", report(), sep="", flush=True)

    event = Event()
    Thread(target=run).start()

    yield

    event.set()


class Task:

    def __init__(self, repeat):
        self.repeat = repeat
        self.current = 0

    def report(self):
        bar = "{: <{}s}".format(("=" * int(self.current / self.repeat * 40) + ">")[:40], 40)
        return "[{}] {}/{}".format(bar, self.current, self.repeat)

    def run(self):
        for self.current in range(1, self.repeat + 1):
            time.sleep(0.001)


def up(rows=1):
    return "\x1b[{}A".format(rows)


def down(rows=1):
    return "\x1b[{}B".format(rows)


def main():

    max_workers = int(sys.argv[1])
    tasks = [Task(int(repeat)) for repeat in sys.argv[2:]]

    # タスク数分改行する
    print("\n" * len(tasks), end="")

    def report():
        # タスク数分カーソルを上に移動 + 各タスクの進捗
        return up(len(tasks)) + "".join(task.report() + "\r" + down() for task in tasks)

    with progress(report):
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            executor.map(lambda task: task.run(), tasks)


if __name__ == "__main__":
    main()