Last active
August 29, 2024 00:37
-
-
Save pirate/c89b7d42be148e9180d8c7cf81e734c8 to your computer and use it in GitHub Desktop.
Animated CLI progress bar that fills up over N seconds, gracefully handling delays by filling slower and slower if the task exceeds the expected time. Good for when you don't know how long a task is going to take, but you still need it to A. feel fast in the avg case, and B. not freeze in the worst case (unknown total time).
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
# asymptotic_progress_bar.py | |
# MIT License © 2021 | |
# | |
# A pretty non-blocking python progress bar timer that fills up asymptotically until you stop it. | |
# Good for when you don't know how long a task is going to take (up to some max timeout), | |
# but you want it to feel fast and accurate / not stuck the whole time. | |
# ████████████████████ 0.9% (1/60sec) | |
# useful for animating e.g. file copy progress, download progress, package install progress, etc. | |
# | |
# Also gracefully handles terminal resizing mid-task, early termination, | |
# non-UTF-8 stdout encoding, indented / same-line progress rendering, | |
# and it stores start and end timestamps in a readable stats dict. | |
# | |
# Usage: | |
# >>> timer = TimedProgress(60) # start progressbar (forks to separate process) | |
# >>> render_large_video() # call your long-running blocking task | |
# >>> timer.end() # stop the bar when finished | |
# >>> elapsed = timer.stats['end_ts'] - timer.stats['start_ts'] | |
# >>> print('Your task finished in', elapsed.seconds, 'seconds.') | |
# TimedProcess(seconds: int, visible: bool=True) | |
# seconds: int = 60 refers to your estimated completion time for the average case the user will encounter | |
# this calibrates the progress bar to hit ~ 50% at the mean average time of 60sec, leaving ~50% of asymptotic range to approach infinity above that time | |
# this means the 50% that are faster than the mean appear to jump to 100% complete before they fill up, which I've found desirable over lots of user testing | |
# users interpret that as "feeling fast" because "all the tasks are finishing earlier than expected", | |
# but also it doesn't feel as useless / or like the progress bar is lying to you by freezing or jumping around | |
import sys | |
import shutil | |
from datetime import datetime, timezone | |
from multiprocessing import Process | |
from collections import defaultdict | |
IS_TTY = sys.stdout.isatty() | |
USE_COLOR = IS_TTY | |
SHOW_PROGRESS = IS_TTY | |
SUPPORTS_UTF8 = sys.stdout.encoding.upper() == 'UTF-8' | |
GET_TERM_WIDTH = lambda: shutil.get_terminal_size((100, 10)).columns | |
COLORS = { | |
'reset': '\033[00;00m', | |
'yellow': '\033[01;33m', | |
'red': '\033[01;31m', | |
'green': '\033[01;32m', | |
} if USE_COLOR else defaultdict(str) | |
def progress_bar(seconds: int, prefix: str='') -> None: | |
"""show timer in the form of progress bar, with percentage and seconds remaining""" | |
chunk = '█' SUPPORTS_UTF8 else '#' | |
last_width = GET_TERM_WIDTH() | |
chunks = last_width - len(prefix) - 20 # number of progress chunks to show (aka max bar width) | |
try: | |
for s in range(seconds * chunks): | |
max_width = GET_TERM_WIDTH() | |
if max_width < last_width: | |
# when the terminal size is shrunk, we have to write a newline | |
# otherwise the progress bar will keep wrapping incorrectly | |
sys.stdout.write('\r\n') | |
sys.stdout.flush() | |
chunks = max_width - len(prefix) - 20 | |
pct_complete = s / chunks / seconds * 100 | |
# everyone likes faster progress bars ;) | |
# accelerate logarithmically instead of linearly. tune the constants | |
# here to match the percieved vs actual speed in your particular situation | |
log_pct = (log(pct_complete or 1, 10) / 2) * 100 | |
bar_width = round(log_pct/(100/chunks)) | |
last_width = max_width | |
# ████████████████████ 0.9% (1/60sec) | |
sys.stdout.write('\r{0}{1}{2}{3} {4}% ({5}/{6}sec)'.format( | |
prefix, | |
COLORS['green' if pct_complete < 80 else 'yellow'], | |
(chunk * bar_width).ljust(chunks), | |
COLORS['reset'], | |
round(pct_complete, 1), | |
round(s/chunks), | |
seconds, | |
)) | |
sys.stdout.flush() | |
time.sleep(1 / chunks) | |
# ██████████████████████████████████ 100.0% (60/60sec) | |
sys.stdout.write('\r{0}{1}{2}{3} {4}% ({5}/{6}sec)'.format( | |
prefix, | |
COLORS['red'], | |
chunk * chunks, | |
COLORS['reset'], | |
100.0, | |
seconds, | |
seconds, | |
)) | |
sys.stdout.flush() | |
# uncomment to have it disappear when it hits 100% instead of staying full red: | |
# time.sleep(0.5) | |
# sys.stdout.write('\r{}{}\r'.format((' ' * GET_TERM_WIDTH()), COLORS['reset'])) | |
# sys.stdout.flush() | |
except (KeyboardInterrupt, BrokenPipeError, IOError): | |
print() | |
class TimedProgress: | |
"""Show a progress bar and measure elapsed time until .end() is called""" | |
def __init__(self, seconds, prefix='', visible: bool=True): | |
self.SHOW_PROGRESS = SHOW_PROGRESS and visible | |
if self.SHOW_PROGRESS: | |
self.p = Process(target=progress_bar, args=(seconds, prefix)) | |
self.p.start() | |
self.stats = {'start_ts': datetime.now(timezone.utc), 'end_ts': None, 'duration': None} | |
def end(self): | |
"""immediately end progress, clear the progressbar line, and save end_ts""" | |
end_ts = datetime.now(timezone.utc) | |
self.stats['end_ts'] = end_ts | |
self.stats['duration'] = end_ts - start_ts | |
if self.SHOW_PROGRESS: | |
# terminate if we havent already terminated | |
try: | |
# kill the progress bar subprocess | |
try: | |
self.p.close() # must be closed *before* its terminnated | |
except (KeyboardInterrupt, SystemExit): | |
print() | |
raise | |
except BaseException: # lgtm [py/catch-base-exception] | |
# this can fail in a ton of different, nasty, platform-dependent ways, | |
# and we dont really care if it fails, we need to finish subprocess cleanup | |
pass | |
self.p.terminate() | |
self.p.join() | |
# clear whole terminal line | |
sys.stdout.write('\r{}{}\r'.format((' ' * GET_TERM_WIDTH()), COLORS['reset'])) | |
except (ValueError, IOError, BrokenPipeError): | |
# ignore when the parent proc rejects, or has stopped listening to our stdout | |
pass |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment