import inspect
from typing import Iterable, Optional
from tqdm import tqdm
from ..utils.translations import trans
_tqdm_kwargs = {
p.name
for p in inspect.signature(tqdm.__init__).parameters.values()
if p.kind is not inspect.Parameter.VAR_KEYWORD and p.name != "self"
}
[docs]class progress(tqdm):
"""This class inherits from tqdm and provides an interface for
progress bars in the napari viewer. Progress bars can be created
directly by wrapping an iterable or by providing a total number
of expected updates.
See tqdm.tqdm API for valid args and kwargs:
https://tqdm.github.io/docs/tqdm/
Also, any keyword arguments to the :class:`ProgressBar` `QWidget`
are also accepted and will be passed to the ``ProgressBar``.
Examples
--------
>>> def long_running(steps=10, delay=0.1):
... for i in progress(range(steps)):
... sleep(delay)
it can also be used as a context manager:
>>> def long_running(steps=10, repeats=4, delay=0.1):
... with progress(range(steps)) as pbr:
... for i in pbr:
... sleep(delay)
or equivalently, using the `progrange` shorthand
... with progrange(steps) as pbr:
... for i in pbr:
... sleep(delay)
For manual updates:
>>> def manual_updates(total):
... pbr = progress(total=total)
... sleep(10)
... pbr.set_description("Step 1 Complete")
... pbr.update(1)
... # must call pbr.close() when using outside for loop
... # or context manager
... pbr.close()
"""
monitor_interval = 0 # set to 0 to disable the thread
def __init__(
self,
iterable: Optional[Iterable] = None,
desc: Optional[str] = None,
total: Optional[int] = None,
nest_under: Optional['progress'] = None,
*args,
**kwargs,
) -> None:
kwargs = kwargs.copy()
pbar_kwargs = {k: kwargs.pop(k) for k in set(kwargs) - _tqdm_kwargs}
self._group_token = None
# get progress bar added to viewer
try:
from .._qt.dialogs.activity_dialog import get_pbar
pbar = get_pbar(self, nest_under=nest_under, **pbar_kwargs)
except ImportError:
pbar = None
if pbar is not None:
kwargs['gui'] = True
self._pbar = pbar
super().__init__(iterable, desc, total, *args, **kwargs)
if not self._pbar:
return
if self.total is not None:
self._pbar.setRange(self.n, self.total)
self._pbar._set_value(self.n)
else:
self._pbar.setRange(0, 0)
self.total = 0
if desc:
self.set_description(desc)
else:
self.set_description(trans._("progress"))
[docs] def display(self, msg: str = None, pos: int = None) -> None:
"""Update the display."""
if not self._pbar:
return super().display(msg=msg, pos=pos)
if self.total != 0:
etas = str(self).split('|')[-1]
try:
self._pbar._set_value(self.n)
self._pbar._set_eta(etas)
except AttributeError:
pass
[docs] def increment_with_overflow(self):
"""Update if not exceeding total, else set indeterminate range."""
if self.n == self.total:
self.total = 0
if self._pbar:
self._pbar.setRange(0, 0)
else:
self.update(1)
[docs] def set_description(self, desc):
"""Update progress bar description"""
super().set_description(desc, refresh=True)
if self._pbar:
self._pbar._set_description(self.desc)
[docs] def close(self):
"""Closes and deletes the progress bar widget"""
if self.disable:
return
if self._pbar:
self.close_pbar()
super().close()
def close_pbar(self):
if self.disable or not self._pbar:
return
from napari._qt.widgets.qt_progress_bar import (
ProgressBar,
ProgressBarGroup,
)
parent_widget = self._pbar.parent()
self._pbar.close()
self._pbar.deleteLater()
if isinstance(parent_widget, ProgressBarGroup):
pbar_children = [
child
for child in parent_widget.children()
if isinstance(child, ProgressBar)
]
if not any(child.isVisible() for child in pbar_children):
parent_widget.close()
self._pbar = None
[docs]def progrange(*args, **kwargs):
"""Shorthand for `progress(range(*args), **kwargs)`.
Adds tqdm based progress bar to napari viewer, if it
exists, and returns the wrapped range object.
Returns
-------
progress
wrapped range object
"""
return progress(range(*args), **kwargs)