提供多样式的显示
python -m rich.progress
[17:27:42] Text may be printed while the progress bars are rendering. progress.py:1095 [17:27:45] ┌────────────────────────────────────────────────────────────────────────────────────┐ progress.py:1095
│ In fact, any renderable will work │
└────────────────────────────────────────────────────────────────────────────────────┘
[17:27:46] Such as tables... progress.py:1095 [17:27:47] ┌─────┬─────┬─────┐ progress.py:1095
│ foo │ bar │ baz │
├─────┼─────┼─────┤
│ 1 │ 2 │ 3 │
└─────┴─────┴─────┘
[17:27:50] Pretty printed structures... progress.py:1095 [17:27:51] {'type': 'example', 'text': 'Pretty printed'} progress.py:1095 [17:27:53] Syntax... progress.py:1095
1 def loop_last(values: Iterable[T]) -> Iterable[Tuple[bool, T]]: progress.py:1095
2 """Iterate and generate a tuple with a flag for last value."""
3 iter_values = iter(values)
4 try:
5 previous_value = next(iter_values)
6 except StopIteration:
7 return
8 for value in iter_values:
9 yield False, previous_value
10 previous_value = value
11 yield True, previous_value
[17:27:54] ─────────────────────────────────── Give it a try! ─────────────────────────────────── progress.py:1095
进度条
from rich.progress import track
import time
def do_step(step):
time.sleep(step)
for step in track(range(100)):
do_step(step)
# python demo.py
# Working... ---------- ----------------------------- 26% 0:01:14
多个任务进度条
from concurrent.futures import ThreadPoolExecutor
from functools import partial
import os.path
import sys
from typing import Iterable
from urllib.request import urlopen
from rich.progress import (
BarColumn,
DownloadColumn,
TextColumn,
TransferSpeedColumn,
TimeRemainingColumn,
Progress,
TaskID,
)
progress = Progress(
TextColumn("[bold blue]{task.fields[filename]}", justify="right"),
BarColumn(bar_width=None),
"[progress.percentage]{task.percentage:>3.1f}%",
"•",
DownloadColumn(),
"•",
TransferSpeedColumn(),
"•",
TimeRemainingColumn(),
)
def copy_url(task_id: TaskID, url: str, path: str) -> None:
"""Copy data from a url to a local file."""
response = urlopen(url)
# This will break if the response doesn't contain content length
progress.update(task_id, total=int(response.info()["Content-length"]))
with open(path, "wb") as dest_file:
progress.start_task(task_id)
for data in iter(partial(response.read, 32768), b""):
dest_file.write(data)
progress.update(task_id, advance=len(data))
def download(urls: Iterable[str], dest_dir: str):
"""Download multuple files to the given directory."""
with progress:
with ThreadPoolExecutor(max_workers=4) as pool:
for url in urls:
filename = url.split("/")[-1]
dest_path = os.path.join(dest_dir, filename)
task_id = progress.add_task("download", filename=filename, start=False)
pool.submit(copy_url, task_id, url, dest_path)
if __name__ == "__main__":
# Try with https://releases.ubuntu.com/20.04/ubuntu-20.04.1-desktop-amd64.iso
if sys.argv[1:]:
download(sys.argv[1:], "./")
else:
print("Usage:\n\tpython downloader.py URL1 URL2 URL3 (etc)")
