mirror of
https://github.com/TheAlgorithms/Python.git
synced 2026-09-28 21:45:27 +08:00
* added Token Bucket rate limiter algorithm * Update comments for clarity in token_bucket.py * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Implement progress iterator for item processing Added a progress iterator function that displays progress in stderr while processing items from an iterable. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Rename cheap_progress to cheap_progress.py --------- Co-authored-by: Christian Clauss <cclauss@me.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
from collections.abc import Iterable, Iterator
|
|
from sys import stderr
|
|
|
|
|
|
def progress[T](items: Iterable[T], desc: str = "", total: int = 0) -> Iterator[T]:
|
|
"""
|
|
A simple progress iterator that yields items from the given iterable while
|
|
displaying a progress indicator in place on a single line of stderr. The output is
|
|
not written to stdout, so the output of the program remains clean (see doctests).
|
|
|
|
for item in progress(range(1_000), desc="Processing"):
|
|
process(item)
|
|
|
|
Args:
|
|
items: The iterable of items to process.
|
|
desc: A description to display alongside the progress. Defaults to "".
|
|
total: The total number of items, defaults to 0. If 0, it will be inferred from
|
|
the iterable if possible.
|
|
|
|
Yields:
|
|
Iterator[T]: The items from the iterable, one by one.
|
|
|
|
>>> tuple(progress(range(5)))
|
|
(0, 1, 2, 3, 4)
|
|
>>> tuple(progress(range(5), desc="Processing", total=3))
|
|
(0, 1, 2, 3, 4)
|
|
>>> tuple(progress(range(5), desc="Processing", total=10))
|
|
(0, 1, 2, 3, 4)
|
|
>>> tuple(progress(range(5), desc="Processing", total=-5))
|
|
(0, 1, 2, 3, 4)
|
|
>>> from string import printable
|
|
>>> tuple(progress(printable, desc="Printable")) # doctest: +ELLIPSIS
|
|
('0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f',...
|
|
"""
|
|
total = max(total, 0)
|
|
if not total and hasattr(items, "__len__"):
|
|
total = len(items) # type: ignore[invalid-argument-type]
|
|
|
|
for i, item in enumerate(items, 1):
|
|
suffix = f"{i:,}/{total:,}" if total else f"{i:,}"
|
|
print(f"\r\033[K{desc}: {suffix}", end="", file=stderr, flush=True)
|
|
yield item
|
|
|
|
print("\r\033[K", end="", file=stderr, flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import time
|
|
|
|
print("start")
|
|
for _item in progress(range(1_000), desc="Processing"):
|
|
time.sleep(0.02)
|
|
|
|
print("stop")
|