classtqdm():
""" 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):
classtqdm():
defupdate(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. """defclose(self):
"""Cleanup and (if leave=False) close the progressbar."""defclear(self, nomove=False):
"""Clear current bar display."""defrefresh(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``. """defunpause(self):
"""Restart tqdm timer from last print time."""defreset(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. """defset_description(self, desc=None, refresh=True):
""" Set/modify description of the progress bar. Parameters ---------- desc : str, optional refresh : bool, optional Forces refresh [default: True]. """defset_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 """@classmethoddefwrite(cls, s, file=sys.stdout, end="\n"):
"""Print a message via tqdm (without overlap with bars)."""@propertydefformat_dict(self):
"""Public API for read-only member access."""defdisplay(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@contextmanagerdefwrapattr(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 """@classmethoddefpandas(cls, *targs, **tqdm_kwargs):
"""Registers the current `tqdm` class with `pandas`."""deftrange(*args, **tqdm_kwargs):
""" A shortcut for `tqdm(xrange(*args), **tqdm_kwargs)`. On Python3+, `range` is used instead of `xrange`. """
fromtqdmimporttqdm, trangefromrandomimportrandom, randintfromtimeimportsleepwithtrange(10) ast:
foriint:
# Description will be displayed on the leftt.set_description('GEN %i'%i)
# Postfix will be displayed on the right,# formatted automatically based on argument's datatypet.set_postfix(loss=random(), gen=randint(1,999), str='h',
lst=[1, 2])
sleep(0.1)
withtqdm(total=10, bar_format="{postfix[0]} {postfix[1][value]:>8.2g}",
postfix=["Batch", dict(value=0)]) ast:
foriinrange(10):
sleep(0.1)
t.postfix[1]["value"] =i/2t.update()
importpandasaspdimportnumpyasnpfromtqdmimporttqdmdf=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(lambdax: x**2)
# can also groupby:# df.groupby(0).progress_apply(lambda x: x**2)
fromtqdm.daskimportTqdmCallbackwithTqdmCallback(desc="compute"):
...
arr.compute()
# or use callback globallycb=TqdmCallback(desc="global")
cb.register()
arr.compute()
# different celliterable=range(100)
pbar.reset(total=len(iterable)) # initialise with new `total`foriiniterable:
pbar.update()
pbar.refresh() # force print final status but don't `close()`
fromtqdmimporttqdmimportos.pathdeffind_files_recursively(path, show_progress=True):
files= []
# total=1 assumes `path` is a filet=tqdm(total=1, unit="file", disable=notshow_progress)
ifnotos.path.exists(path):
raiseIOError("Cannot find:"+path)
defappend_found_file(f):
files.append(f)
t.update()
deflist_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 directoryt.total+=len(listing) -1# fancy way to give info without forcing a refresht.set_postfix(dir=path[-10:], refresh=False)
t.update(0) # may trigger a refreshreturnlistingdefrecursively_search(path):
ifos.path.isdir(path):
forfinlist_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()
returnfiles
fromtqdm.autoimporttqdm, trangefromtimeimportsleepbar=trange(10)
foriinbar:
# Print using tqdm class method .write()sleep(0.1)
ifnot (i%3):
tqdm.write("Done task %i"%i)
# Can also use bar.write()
fromrich.consoleimportConsolefromrich.syntaximportSyntaxmy_code='''def iter_first_last(values: Iterable[T]) -> Iterable[Tuple[bool, bool, T]]: """Iterate and generate a tuple with a flag for first and last value.""" iter_values = iter(values) try: previous_value = next(iter_values) except StopIteration: return first = True for value in iter_values: yield first, False, previous_value first = False previous_value = value yield first, True, previous_value'''syntax=Syntax(my_code, "python", theme="monokai", line_numbers=True)
console=Console()
console.print(syntax)
