from collections.abc import Generator, Iterable, Iterator
from itertools import takewhile
from typing import Callable, Optional
from tqdm import tqdm
from napari.utils.events.containers import EventedSet
from napari.utils.events.event import EmitterGroup, Event
from napari.utils.translations import trans
__all__ = ['cancelable_progress', 'progrange', 'progress']
[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.
While this interface is primarily designed to be displayed in
the viewer, it can also be used without a viewer open, in which
case it behaves identically to tqdm and produces the progress
bar in the terminal.
See tqdm.tqdm API for valid args and kwargs:
https://tqdm.github.io/docs/tqdm/
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
.. code-block:: python
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
# to give us a way to hook into the creation and update of progress objects
# without progress knowing anything about a Viewer, we track all instances in
# this evented *class* attribute, accessed through `progress._all_instances`
# this allows the ActivityDialog to find out about new progress objects and
# hook up GUI progress bars to its update events
_all_instances: EventedSet['progress'] = EventedSet()
def __init__(
self,
iterable: Optional[Iterable] = None,
desc: Optional[str] = None,
total: Optional[int] = None,
nest_under: Optional['progress'] = None,
*args,
**kwargs,
) -> None:
self.events = EmitterGroup(
value=Event,
description=Event,
overflow=Event,
eta=Event,
total=Event,
)
self.nest_under = nest_under
self.is_init = True
super().__init__(iterable, desc, total, *args, **kwargs)
# if the progress bar is set to disable the 'desc' member is not set by the
# tqdm super constructor, so we set it to a dummy value to avoid errors thrown below
if self.disable:
self.desc = ''
if not self.desc:
self.set_description(trans._('progress'))
progress._all_instances.add(self)
self.is_init = False
def __repr__(self) -> str:
return self.desc
@property
def total(self):
return self._total
@total.setter
def total(self, total):
self._total = total
self.events.total(value=self.total)
[docs]
def display(
self, msg: Optional[str] = None, pos: Optional[int] = None
) -> None:
"""Update the display and emit eta event."""
# just plain tqdm if we don't have gui
if not self.gui and not self.is_init:
super().display(msg, pos)
return
# TODO: This could break if user is formatting their own terminal tqdm
etas = str(self).split('|')[-1] if self.total != 0 else ''
self.events.eta(value=etas)
[docs]
def update(self, n=1):
"""Update progress value by n and emit value event"""
super().update(n)
self.events.value(value=self.n)
[docs]
def increment_with_overflow(self):
"""Update if not exceeding total, else set indeterminate range."""
if self.n == self.total:
self.total = 0
self.events.overflow()
else:
self.update(1)
[docs]
def set_description(self, desc):
"""Update progress description and emit description event."""
super().set_description(desc, refresh=True)
self.events.description(value=desc)
[docs]
def close(self):
"""Close progress object and emit event."""
if self.disable:
return
progress._all_instances.remove(self)
super().close()
[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)
[docs]
class cancelable_progress(progress):
"""This class inherits from progress, providing the additional
ability to cancel expensive executions. When progress is
canceled by the user in the napari UI, two things will happen:
Firstly, the is_canceled attribute will become True, and the
for loop will terminate after the current iteration, regardless
of whether or not the iterator had more items.
Secondly, cancel_callback will be called, allowing the computation
to close resources, repair state, etc.
See napari.utils.progress and tqdm.tqdm API for valid args and kwargs:
https://tqdm.github.io/docs/tqdm/
Examples
--------
>>> def long_running(steps=10, delay=0.1):
... def on_cancel():
... print("Canceled operation")
... for i in cancelable_progress(range(steps), cancel_callback=on_cancel):
... sleep(delay)
"""
def __init__(
self,
iterable: Optional[Iterable] = None,
desc: Optional[str] = None,
total: Optional[int] = None,
nest_under: Optional['progress'] = None,
cancel_callback: Optional[Callable] = None,
*args,
**kwargs,
) -> None:
self.cancel_callback = cancel_callback
self.is_canceled = False
super().__init__(iterable, desc, total, nest_under, *args, **kwargs)
def __iter__(self) -> Iterator:
itr = super().__iter__()
def is_canceled(_):
if self.is_canceled:
# If we've canceled, run the callback and then notify takewhile
if self.cancel_callback:
self.cancel_callback()
# Perform additional cleanup for generators
if isinstance(self.iterable, Generator):
self.iterable.close()
return False
# Otherwise, continue
return True
return takewhile(is_canceled, itr)
[docs]
def cancel(self):
"""Cancels the execution of the underlying computation.
Note that the current iteration will be allowed to complete, however
future iterations will not be run.
"""
self.is_canceled = True
