Tqdm-一个用于Python和CLI的快速、可扩展的进度条

from tqdm import tqdm for i in tqdm(range(10000)):
    ...

from tqdm import tqdm from time import sleep text = "" for char in tqdm(["a", "b", "c", "d"]):
    sleep(0.25)
    text = text + char

from tqdm import trange for i in trange(100):
    sleep(0.01)

pbar = tqdm(["a", "b", "c", "d"])
for char in pbar:
    sleep(0.25)
    pbar.set_description("Processing %s" % char)

with tqdm(total=100) as pbar:
    for i in range(10):
        sleep(0.1)
        pbar.update(10)

pbar = tqdm(total=100)
for i in range(10):
    sleep(0.1)
    pbar.update(10)
pbar.close()

class tqdm():
  """   Decorate an iterable object, returning an iterator which acts exactly   like the original iterable, but prints a dynamically updating   progressbar every time a value is requested.   """ def __init__(self, iterable=None, desc=None, total=None, leave=True,
               file=None, ncols=None, mininterval=0.1,
               maxinterval=10.0, miniters=None, ascii=None, disable=False,
               unit='it', unit_scale=False, dynamic_ncols=False,
               smoothing=0.3, bar_format=None, initial=0, position=None,
               postfix=None, unit_divisor=1000):

class tqdm():
  def update(self, n=1):
      """       Manually update the progress bar, useful for streams       such as reading files.       E.g.:       >>> t = tqdm(total=filesize) # Initialise       >>> for current_buffer in stream:       ...    ...       ...    t.update(len(current_buffer))       >>> t.close()       The last line is highly recommended, but possibly not necessary if       ``t.update()`` will be called in such a way that ``filesize`` will be       exactly reached and printed.       Parameters       ----------       n  : int or float, optional           Increment to add to the internal counter of iterations           [default: 1]. If using float, consider specifying ``{n:.3f}``           or similar in ``bar_format``, or specifying ``unit_scale``.       Returns       -------       out  : bool or None           True if a ``display()`` was triggered.       """ def close(self):
      """Cleanup and (if leave=False) close the progressbar.""" def clear(self, nomove=False):
      """Clear current bar display.""" def refresh(self):
      """       Force refresh the display of this bar.       Parameters       ----------       nolock  : bool, optional           If ``True``, does not lock.           If [default: ``False``]: calls ``acquire()`` on internal lock.       lock_args  : tuple, optional           Passed to internal lock's ``acquire()``.           If specified, will only ``display()`` if ``acquire()`` returns ``True``.       """ def unpause(self):
      """Restart tqdm timer from last print time.""" def reset(self, total=None):
      """       Resets to 0 iterations for repeated use.       Consider combining with ``leave=True``.       Parameters       ----------       total  : int or float, optional. Total to use for the new bar.       """ def set_description(self, desc=None, refresh=True):
      """       Set/modify description of the progress bar.       Parameters       ----------       desc  : str, optional       refresh  : bool, optional           Forces refresh [default: True].       """ def set_postfix(self, ordered_dict=None, refresh=True, **tqdm_kwargs):
      """       Set/modify postfix (additional stats)       with automatic formatting based on datatype.       Parameters       ----------       ordered_dict  : dict or OrderedDict, optional       refresh  : bool, optional           Forces refresh [default: True].       kwargs  : dict, optional       """ @classmethod def write(cls, s, file=sys.stdout, end="\n"):
      """Print a message via tqdm (without overlap with bars).""" @property def format_dict(self):
      """Public API for read-only member access.""" def display(self, msg=None, pos=None):
      """       Use ``self.sp`` to display ``msg`` in the specified ``pos``.       Consider overloading this function when inheriting to use e.g.:       ``self.some_frontend(**self.format_dict)`` instead of ``self.sp``.       Parameters       ----------       msg  : str, optional. What to display (default: ``repr(self)``).       pos  : int, optional. Position to ``moveto``         (default: ``abs(self.pos)``).       """ @classmethod @contextmanager def wrapattr(cls, stream, method, total=None, bytes=True, **tqdm_kwargs):
      """       stream  : file-like object.       method  : str, "read" or "write". The result of ``read()`` and           the first argument of ``write()`` should have a ``len()``.       >>> with tqdm.wrapattr(file_obj, "read", total=file_obj.size) as fobj:       ...     while True:       ...         chunk = fobj.read(chunk_size)       ...         if not chunk:       ...             break       """ @classmethod def pandas(cls, *targs, **tqdm_kwargs):
      """Registers the current `tqdm` class with `pandas`.""" def trange(*args, **tqdm_kwargs):
    """     A shortcut for `tqdm(xrange(*args), **tqdm_kwargs)`.     On Python3+, `range` is used instead of `xrange`.     """

def tqdm.contrib.tenumerate(iterable, start=0, total=None,
                            tqdm_class=tqdm.auto.tqdm, **tqdm_kwargs):
    """Equivalent of `numpy.ndenumerate` or builtin `enumerate`.""" def tqdm.contrib.tzip(iter1, *iter2plus, **tqdm_kwargs):
    """Equivalent of builtin `zip`.""" def tqdm.contrib.tmap(function, *sequences, **tqdm_kwargs):
    """Equivalent of builtin `map`."""

class tqdm.notebook.tqdm(tqdm.tqdm):
    """IPython/Jupyter Notebook widget.""" class tqdm.auto.tqdm(tqdm.tqdm):
    """Automatically chooses beween `tqdm.notebook` and `tqdm.tqdm`.""" class tqdm.asyncio.tqdm(tqdm.tqdm):
  """Asynchronous version.""" @classmethod def as_completed(cls, fs, *, loop=None, timeout=None, total=None,
                   **tqdm_kwargs):
      """Wrapper for `asyncio.as_completed`.""" class tqdm.gui.tqdm(tqdm.tqdm):
    """Matplotlib GUI version.""" class tqdm.tk.tqdm(tqdm.tqdm):
    """Tkinter GUI version.""" class tqdm.rich.tqdm(tqdm.tqdm):
    """`rich.progress` version.""" class tqdm.keras.TqdmCallback(keras.callbacks.Callback):
    """Keras callback for epoch and batch progress.""" class tqdm.dask.TqdmCallback(dask.callbacks.Callback):
    """Dask callback for task progress."""

from tqdm import tqdm, trange from random import random, randint from time import sleep with trange(10) as t:
    for i in t:
        # Description will be displayed on the left t.set_description('GEN %i' % i)
        # Postfix will be displayed on the right, # formatted automatically based on argument's datatype t.set_postfix(loss=random(), gen=randint(1,999), str='h',
                      lst=[1, 2])
        sleep(0.1)

with tqdm(total=10, bar_format="{postfix[0]} {postfix[1][value]:>8.2g}",
          postfix=["Batch", dict(value=0)]) as t:
    for i in range(10):
        sleep(0.1)
        t.postfix[1]["value"] = i / 2 t.update()

from tqdm import tqdm class TqdmExtraFormat(tqdm):
    """Provides a `total_time` format parameter""" @property def format_dict(self):
        d = super(TqdmExtraFormat, self).format_dict total_time = d["elapsed"] * (d["total"] or 0) / max(d["n"], 1)
        d.update(total_time=self.format_interval(total_time) + " in total")
        return d for i in TqdmExtraFormat(
      range(9), ascii=" .oO0",
      bar_format="{total_time}: {percentage:.0f}%|{bar}{r_bar}"):
    if i == 4:
        break

from tqdm.auto import trange from time import sleep for i in trange(4, desc='1st loop'):
    for j in trange(5, desc='2nd loop'):
        for k in trange(50, desc='3rd loop', leave=False):
            sleep(0.01)

from time import sleep from tqdm import trange, tqdm from multiprocessing import Pool, RLock, freeze_support L = list(range(9))

def progresser(n):
    interval = 0.001 / (n + 2)
    total = 5000 text = "#{}, est. {:<04.2}s".format(n, interval * total)
    for _ in trange(total, desc=text, position=n):
        sleep(interval)

if __name__ == '__main__':
    freeze_support()  # for Windows support tqdm.set_lock(RLock())  # for managing output contention p = Pool(initializer=tqdm.set_lock, initargs=(tqdm.get_lock(),))
    p.map(progresser, L)

from time import sleep from tqdm import tqdm, trange from concurrent.futures import ThreadPoolExecutor L = list(range(9))

def progresser(n):
    interval = 0.001 / (n + 2)
    total = 5000 text = "#{}, est. {:<04.2}s".format(n, interval * total)
    for _ in trange(total, desc=text):
        sleep(interval)
    if n == 6:
        tqdm.write("n == 6 completed.")
        tqdm.write("`tqdm.write()` is thread-safe in py3!")

if __name__ == '__main__':
    with ThreadPoolExecutor() as p:
        p.map(progresser, L)

import urllib, os from tqdm import tqdm urllib = getattr(urllib, 'request', urllib)

class TqdmUpTo(tqdm):
    """Provides `update_to(n)` which uses `tqdm.update(delta_n)`.""" def update_to(self, b=1, bsize=1, tsize=None):
        """         b  : int, optional             Number of blocks transferred so far [default: 1].         bsize  : int, optional             Size of each block (in tqdm units) [default: 1].         tsize  : int, optional             Total size (in tqdm units). If [default: None] remains unchanged.         """ if tsize is not None:
            self.total = tsize return self.update(b * bsize - self.n)  # also sets self.n = b * bsize eg_link = "https://caspersci.uk.to/matryoshka.zip" with TqdmUpTo(unit='B', unit_scale=True, unit_divisor=1024, miniters=1,
              desc=eg_link.split('/')[-1]) as t:  # all optional kwargs urllib.urlretrieve(eg_link, filename=os.devnull,
                       reporthook=t.update_to, data=None)
    t.total = t.n

from tqdm.auto import tqdm from tqdm.utils import CallbackIOWrapper with tqdm(total=file_obj.size,
          unit='B', unit_scale=True, unit_divisor=1024) as t:
    fobj = CallbackIOWrapper(t.update, file_obj, "read")
    while True:
        chunk = fobj.read(chunk_size)
        if not chunk:
            break t.reset()
    # ... continue to use `t` for something else

import urllib, os from tqdm import tqdm eg_link = "https://caspersci.uk.to/matryoshka.zip" response = getattr(urllib, 'request', urllib).urlopen(eg_link)
with tqdm.wrapattr(open(os.devnull, "wb"), "write",
                   miniters=1, desc=eg_link.split('/')[-1],
                   total=getattr(response, 'length', None)) as fout:
    for chunk in response:
        fout.write(chunk)

import requests, os from tqdm import tqdm eg_link = "https://caspersci.uk.to/matryoshka.zip" response = requests.get(eg_link, stream=True)
with tqdm.wrapattr(open(os.devnull, "wb"), "write",
                   miniters=1, desc=eg_link.split('/')[-1],
                   total=int(response.headers.get('content-length', 0))) as fout:
    for chunk in response.iter_content(chunk_size=4096):
        fout.write(chunk)

from tqdm.auto import tqdm as std_tqdm def external_callback(*args, **kwargs):
    ...

class TqdmExt(std_tqdm):
    def update(self, n=1):
        displayed = super(TqdmExt, self).update(n):
        if displayed:
            external_callback(**self.format_dict)
        return displayed

from tqdm.asyncio import tqdm async for i in tqdm(range(9)):
    if i == 2:
        break

from tqdm.asyncio import tqdm with tqdm(range(9)) as pbar:
    async for i in pbar:
        if i == 2:
            break

import pandas as pd import numpy as np from tqdm import tqdm df = pd.DataFrame(np.random.randint(0, 100, (100000, 6)))

# Register `pandas.progress_apply` and `pandas.Series.map_apply` with `tqdm` # (can use `tqdm.gui.tqdm`, `tqdm.notebook.tqdm`, optional kwargs, etc.) tqdm.pandas(desc="my bar!")

# Now you can use `progress_apply` instead of `apply` # and `progress_map` instead of `map` df.progress_apply(lambda x: x**2)
# can also groupby: # df.groupby(0).progress_apply(lambda x: x**2)

from tqdm.keras import TqdmCallback

...

model.fit(..., verbose=0, callbacks=[TqdmCallback()])

from tqdm.dask import TqdmCallback with TqdmCallback(desc="compute"):
    ...
    arr.compute()

# or use callback globally cb = TqdmCallback(desc="global")
cb.register()
arr.compute()

from tqdm.notebook import trange, tqdm from time import sleep for i in trange(3, desc='1st loop'):
    for j in tqdm(range(100), desc='2nd loop'):
        sleep(0.01)

from tqdm.autonotebook import tqdm tqdm.pandas()

from tqdm.notebook import tqdm pbar = tqdm(..., display=False)

# different cell display(pbar.container)

from tqdm.keras import TqdmCallback cbk = TqdmCallback(display=False)

# different cell cbk.display()
model.fit(..., verbose=0, callbacks=[cbk])

from tqdm.notebook import tqdm pbar = tqdm()

# different cell iterable = range(100)
pbar.reset(total=len(iterable))  # initialise with new `total` for i in iterable:
    pbar.update()
pbar.refresh()  # force print final status but don't `close()`

from functools import partial from tqdm import tqdm as std_tqdm tqdm = partial(std_tqdm, dynamic_ncols=True)

from tqdm import tqdm import os.path def find_files_recursively(path, show_progress=True):
    files = []
    # total=1 assumes `path` is a file t = tqdm(total=1, unit="file", disable=not show_progress)
    if not os.path.exists(path):
        raise IOError("Cannot find:" + path)

    def append_found_file(f):
        files.append(f)
        t.update()

    def list_found_dir(path):
        """returns os.listdir(path) assuming os.path.isdir(path)""" listing = os.listdir(path)
        # subtract 1 since a "file" we found was actually this directory t.total += len(listing) - 1 # fancy way to give info without forcing a refresh t.set_postfix(dir=path[-10:], refresh=False)
        t.update(0)  # may trigger a refresh return listing def recursively_search(path):
        if os.path.isdir(path):
            for f in list_found_dir(path):
                recursively_search(os.path.join(path, f))
        else:
            append_found_file(path)

    recursively_search(path)
    t.set_postfix(dir=path)
    t.close()
    return files

from tqdm.auto import tqdm, trange from time import sleep bar = trange(10)
for i in bar:
    # Print using tqdm class method .write() sleep(0.1)
    if not (i % 3):
        tqdm.write("Done task %i" % i)
    # Can also use bar.write()

from time import sleep import contextlib import sys from tqdm import tqdm from tqdm.contrib import DummyTqdmFile @contextlib.contextmanager def std_out_err_redirect_tqdm():
    orig_out_err = sys.stdout, sys.stderr try:
        sys.stdout, sys.stderr = map(DummyTqdmFile, orig_out_err)
        yield orig_out_err[0]
    # Relay exceptions except Exception as exc:
        raise exc # Always restore sys.stdout/err if necessary finally:
        sys.stdout, sys.stderr = orig_out_err def some_fun(i):
    print("Fee, fi, fo,".split()[i])

# Redirect stdout to tqdm.write() (don't forget the `as save_stdout`) with std_out_err_redirect_tqdm() as orig_stdout:
    # tqdm needs the original stdout # and dynamic_ncols=True to autodetect console width for i in tqdm(range(3), file=orig_stdout, dynamic_ncols=True):
        sleep(.5)
        some_fun(i)

# After the `with`, printing is restored print("Done!")

import logging from tqdm import trange from tqdm.contrib.logging import logging_redirect_tqdm LOG = logging.getLogger(__name__)

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    with logging_redirect_tqdm():
        for i in trange(9):
            if i == 4:
                LOG.info("console logging redirected to `tqdm.write()`")
    # logging restored