kwutil.util_progress module

POC for a better ProgIter with rich support

CommandLine

DEMO_PROGRESS=1 xdoctest -m kwutil.util_progress __doc__:0

Example

>>> # xdoctest: +REQUIRES(env:DEMO_PROGRESS)
>>> from kwutil.util_progress import ProgressManager
>>> import time
>>> delay = 0.05
>>> # Can use plain progiter or rich
>>> # The usecase for plain progiter is when threads / live output
>>> # is not desirable and you just want plain stdout progress
>>> for backend in ['rich', 'progiter']:
>>>     print(f'\n\n -- starting {backend} --\n\n')
>>>     pman = ProgressManager(backend=backend)
>>>     with pman:
>>>         pbar1 = pman.progiter(range(5), desc='outer loop', verbose=3)
>>>         for i in pbar1:
>>>             pbar1.set_postfix(f'\\[step {i}]', refresh=False)
>>>             for j1 in pman.progiter(range(100), desc=f'prepare inner loop {i}', transient=True):
>>>                 time.sleep(delay / 3)
>>>             for j2 in pman.progiter(range(100), desc=f'execute inner loop {i}'):
>>>                 time.sleep(delay)
>>>             for j3 in pman.progiter(range(100), desc=f'shutdown inner loop {i}', transient=True):
>>>                 time.sleep(delay / 3)

class kwutil.util_progress.ProgIter2(iterable=None, desc=None, total=None, freq=1, initial=0, eta_window=64, clearline=True, adjust=True, time_thresh=2.0, show_percent=True, show_times=True, show_rate=True, show_eta=True, show_total=True, show_wall=False, enabled=True, verbose=None, stream=None, chunksize=None, rel_adjust_limit=4.0, homogeneous='auto', timer=None, **kwargs)

Bases: ProgIter

Note

See attributes for arg information **kwargs accepts most of the tqdm api

_set_manager(manager)

update_info(text)

update(n=1)

display_message()

set_postfix(s='', refresh=True)

tqdm api compatibility. calls set_extra

class kwutil.util_progress.RichProgIter(iterable=None, desc=None, total=None, freq=1, initial=0, eta_window=64, clearline=True, adjust=True, time_thresh=2.0, show_times=True, show_wall=False, enabled=True, verbose=None, stream=None, chunksize=None, rel_adjust_limit=4.0, transient=False, manager=None, spinner=False, **kwargs)

Bases: object

Ducktypes ProgIter

TODO: enhance with the ability to have a update info panel that removes the circular reference

start()

stop()

begin()

end()

update(n=1)

step(n=1)

remove()

Remove this progress task from its rich manager

update_info(text)

ensure_newline()

set_postfix_str(text, refresh=True)

set_postfix(text, refresh=True)

set_extra(text, refresh=True)

class kwutil.util_progress.BaseProgIterManager

Bases: object

new(*args, **kw)

__call__(*args, **kw)

start()

begin()

stop(**kwargs)

class kwutil.util_progress._RichProgIterManager(**kwargs)

Bases: BaseProgIterManager

rich specific backend.

Example

>>> # Test verbose = 0
>>> # xdoctest: +REQUIRES(module:rich)
>>> from kwutil.util_progress import ProgressManager
>>> from kwutil import util_progress
>>> import time
>>> pman = ProgressManager(backend='rich', verbose=0)
>>> with pman:
>>>     for i in pman.progiter(range(10), desc='should not show1'):
>>>         ...
>>>     for i in pman.progiter(range(10), verbose=1, desc='should show2'):
>>>         ...
>>>     for i in pman.progiter(range(10), verbose=0, desc='should not show3'):
>>>         ...

progiter(iterable=None, total=None, desc=None, transient=False, spinner=False, verbose='auto', **kw)

ProgIter(iterable=None, total=None, desc=None, transient=False, spinner=False, verbose='auto', **kw)

setup_rich()

update_info(text)

start()

stop(**kw)

class kwutil.util_progress._ProgIterManager(**kwargs)

Bases: BaseProgIterManager

progiter specific backend

progiter(iterable=None, total=None, desc=None, transient=False, spinner=False, verbose='auto', **kw)

update_info(text)

class kwutil.util_progress.ProgressManager(backend='rich', **kwargs)

Bases: BaseProgIterManager

A progress manager.

Manage multiple progress bars, either with rich or ProgIter.

CommandLine

xdoctest -m kwutil.util_progress ProgressManager:0
xdoctest -m kwutil.util_progress ProgressManager:1
xdoctest -m kwutil.util_progress ProgressManager:2

Example

>>> from kwutil.util_progress import ProgressManager
>>> from progiter import progiter
>>> # Can use plain progiter or rich
>>> # The usecase for plain progiter is when threads / live output
>>> # is not desirable and you just want plain stdout progress
>>> pman = ProgressManager(backend='progiter')
>>> with pman:
>>>     oprog = pman.progiter(range(20), desc='outer loop', verbose=3)
>>>     for i in oprog:
>>>         oprog.set_postfix(f'Doing step {i}', refresh=False)
>>>         for i in pman.progiter(range(100), desc=f'inner loop {i}'):
>>>             pass
>>> # xdoctest: +REQUIRES(module:rich)
>>> self = pman = ProgressManager(backend='rich')
>>> pman = ProgressManager(backend='rich')
>>> with pman:
>>>     oprog = pman.progiter(range(20), desc='outer loop', verbose=3)
>>>     for i in oprog:
>>>         oprog.set_postfix(f'Doing step {i}', refresh=False)
>>>         for i in pman.progiter(range(100), desc=f'inner loop {i}'):
>>>             pass

Example

>>> # xdoctest: +REQUIRES(module:rich)
>>> # A fairly complex example
>>> from kwutil.util_progress import ProgressManager
>>> import time
>>> delay = 0.00005
>>> N_inner = 300
>>> N_outer = 11
>>> self = pman = ProgressManager(backend='rich')
>>> with pman:
>>>     oprog = pman(range(N_outer), desc='outer loop')
>>>     for i in oprog:
>>>         if i > 7:
>>>             self.update_info(f'The info panel gives detailed updates\nWe are now at step {i}\nWe are just about done now')
>>>         elif i > 5:
>>>             self.update_info(f'The info panel gives detailed updates\nWe are now at step {i}')
>>>         oprog.set_postfix(f'Doing step {i}')
>>>         N = 1000
>>>         for j in pman(iter(range(N_inner)), total=None if i % 2 == 0 else N_inner, desc=f'inner loop {i}', transient=i < 4):
>>>             time.sleep(delay)

Example

>>> # xdoctest: +REQUIRES(module:rich)
>>> # Test complex example over a grid of parameters
>>> from kwutil.util_progress import ProgressManager, ProgIter2
>>> import time
>>> delay = 0.000001
>>> N_inner = 100
>>> N_outer = 11
>>> basis = {
>>>     'with_info': [0, 1],
>>>     'backend': ['progiter', 'rich'],
>>>     'enabled': [0, 1],
>>>     #'with_info': [1],
>>> }
>>> grid = list(ub.named_product(basis))
>>> grid_prog = ProgIter2(grid, desc='Test cases over grid', verbose=3)
>>> grid_prog.update_info('Here we go')
>>> for item in grid:
>>>     grid_prog.ensure_newline()
>>>     grid_prog.update_info(f'Running grid test {ub.urepr(item, nl=1)}')
>>>     print('\n\n')
>>>     self = ProgressManager(backend=item['backend'], enabled=item['enabled'])
>>>     with self:
>>>         outer_prog = self.progiter(range(N_outer), desc='outer loop')
>>>         for i in outer_prog:
>>>             if item['with_info']:
>>>                 if i > 7:
>>>                     outer_prog.update_info(f'The info panel gives detailed updates\nWe are now at step {i}\nWe are just about done now')
>>>                 elif i > 5:
>>>                     outer_prog.update_info(f'The info panel gives detailed updates\nWe are now at step {i}')
>>>             outer_prog.set_postfix(f'Doing step {i}')
>>>             inner_kwargs = dict(
>>>                 total=None if i % 2 == 0 else N_inner,
>>>                 transient=i < 4,
>>>                 time_thresh=delay * 2.3,
>>>                 desc=f'inner loop {i}',
>>>             )
>>>             for j in self.progiter(iter(range(N_inner)), **inner_kwargs):
>>>                 time.sleep(delay)
>>>     grid_prog.update_info(f'Finished test item')

Example

>>> # xdoctest: +REQUIRES(module:rich)
>>> # Demo manual usage
>>> from kwutil.util_progress import ProgressManager
>>> from kwutil import util_progress
>>> import time
>>> pman = ProgressManager()
>>> pman.start()
>>> task1 = pman.progiter(desc='task1', total=100)
>>> task2 = pman.progiter(desc='task2')
>>> for i in range(100):
>>>     task1.update()
>>>     task2.update(2)
>>>     time.sleep(0.001)
>>> ProgressManager.stopall()

Example

>>> # Demo manual usage (progiter backend)
>>> from kwutil.util_progress import ProgressManager
>>> from kwutil import util_progress
>>> import time
>>> pman = ProgressManager(backend='progiter', adjust=0, freq=1)
>>> pman.start()
>>> task1 = pman.progiter(desc='task1', total=12)
>>> task2 = pman.progiter(desc='task2')
>>> task1.update()
>>> task2.update()
>>> for i in range(10):
>>>     time.sleep(0.001)
>>>     task1.update()
>>>     time.sleep(0.001)
>>>     task2.update(2)
>>> ProgressManager.stopall()

progiter(*args, **kw)

ProgIter(*args, **kw)

update_info(text)

start()

stop(*args, **kwargs)

property _is_main_manager

classmethod stopall()

Stop all background progress threads (likely only 1 exists)

kwutil.util_progress._progman_test_multiple_managers()

Note

We want the user to be able let the user create multiple ProgressManagers, but we are only allowed one live display, therefore the first ProgressManager needs to becomes the “main” manager and all others will be secondary. Getting this exactly right may require a refactor and locks, but this tests that our simple implementation works well enough.

CommandLine

xdoctest -m kwutil.util_progress _progman_test_multiple_managers

Example

>>> # xdoctest: +REQUIRES(module:rich)
>>> _progman_test_multiple_managers()

kwutil.util_progress._progman_test_multiple_managers_tree()

” .. rubric:: CommandLine

xdoctest -m kwutil.util_progress _progman_test_multiple_managers_tree

Example

>>> # xdoctest: +REQUIRES(module:rich)
>>> _progman_test_multiple_managers_tree()