Simple nested progress bars

[Sourish07]

I was wondering if there were any current efforts to support nested for loops like how tqdm does? Using track throws an error: rich.errors.LiveError: Only one live display may be active at once.

I quickly hacked together the following tqdm_track function to showcase the functionality that would be convenient for rich users to have.

from rich.progress import tqdm_track
import time

for i in tqdm_track(range(10), description="One"):
    time.sleep(0.1)
    for j in tqdm_track(range(100), description="Two"):
        time.sleep(0.01)
        print(f"Verbose info! {i, j}")

# In progress.py

_shared_progress = None

def tqdm_track(
    sequence: Union[Sequence[ProgressType], Iterable[ProgressType]],
    description: str = "Working...",
    total: Optional[float] = None,
    completed: int = 0,
    auto_refresh: bool = True,
    console: Optional[Console] = None,
    transient: bool = False,
    refresh_per_second: float = 10,
    style: StyleType = "bar.back",
    complete_style: StyleType = "bar.complete",
    finished_style: StyleType = "bar.finished",
    pulse_style: StyleType = "bar.pulse",
    disable: bool = False,
    show_speed: bool = True,
) -> Iterable[ProgressType]:
    """Track progress by iterating over a sequence. Supports nested loops.

    Args:
        sequence (Iterable[ProgressType]): A sequence (must support "len") you wish to iterate over.
        description (str, optional): Description of task show next to progress bar. Defaults to "Working".
        total: (float, optional): Total number of steps. Default is len(sequence).
        completed (int, optional): Number of steps completed so far. Defaults to 0.
        auto_refresh (bool, optional): Automatic refresh, disable to force a refresh after each iteration. Default is True.
        transient: (bool, optional): Clear the progress on exit. Defaults to False.
        console (Console, optional): Console to write to. Default creates internal Console instance.
        refresh_per_second (float): Number of times per second to refresh the progress information. Defaults to 10.
        style (StyleType, optional): Style for the bar background. Defaults to "bar.back".
        complete_style (StyleType, optional): Style for the completed bar. Defaults to "bar.complete".
        finished_style (StyleType, optional): Style for a finished bar. Defaults to "bar.finished".
        pulse_style (StyleType, optional): Style for pulsing bars. Defaults to "bar.pulse".
        disable (bool, optional): Disable display of progress.
        show_speed (bool, optional): Show speed if total isn't known. Defaults to True.

    Returns:
        Iterable[ProgressType]: An iterable of the values in the sequence.
    """
    if disable:
        yield from sequence
        return

    global _shared_progress
    
    # Create shared progress instance if it doesn't exist
    if _shared_progress is None:
        columns: List["ProgressColumn"] = (
            [TextColumn("[progress.description]{task.description}")] if description else []
        )
        columns.extend(
            (
                BarColumn(
                    style=style,
                    complete_style=complete_style,
                    finished_style=finished_style,
                    pulse_style=pulse_style,
                ),
                TaskProgressColumn(show_speed=show_speed),
                TimeRemainingColumn(),
            )
        )
        _shared_progress = Progress(
            *columns,
            auto_refresh=auto_refresh,
            console=console,
            transient=transient,
            refresh_per_second=refresh_per_second,
            disable=disable,
        )
        _shared_progress.start()

    # Get total from sequence length if not provided
    if total is None:
        try:
            total = len(sequence)  # type: ignore
        except (TypeError, AttributeError):
            total = None

    # Add task to progress
    task_id = _shared_progress.add_task(description, total=total, completed=completed)

    try:
        # Iterate and update progress
        for value in sequence:
            yield value
            _shared_progress.advance(task_id)
    finally:
        # Remove task when done
        _shared_progress.remove_task(task_id)
        # Stop progress if no more tasks
        if not _shared_progress.tasks:
            _shared_progress.stop()
            _shared_progress = None

---

Discussed in https://github.com/Textualize/rich/discussions/2272

^{Originally posted by nik-sm May 11, 2022} Hi - I'd love to switch from using tqdm for tracking progress to using rich.

Is there a simple recipe for tracking progress over several nested iterables?

---

I frequently have several nested loops going, and I'd like to keep their progress bars at the bottom while also printing verbose debug info above. It seems that this layout is definitely a good use case for rich.

Here's the typical code I use for tqdm:

from tqdm import tqdm
from time import sleep

for i in tqdm(range(100), "Outer"):
    for j in tqdm(range(10), "Inner", leave=False):
        # print(f"Verbose info! {i, j}")
        sleep(0.1)

If no printing occurs, this works great and makes it possible to track progress of nested tasks without any development overhead. However, if you uncomment the print statement, you can see that this makes the output very ugly.

As a newbie to rich, I would have naively loved to simply do:

from rich import print
from rich.progress import track
from time import sleep
for i in track(range(100), "Outer"):
    for j in track(range(10), "Inner", transient=True):
        print(f"Verbose info! {i, j}")
        sleep(0.1)

This or a similar recipe would totally serve my use case (tracking nested progress, code is still simple), but rich.progress.track currently cannot be nested.

I found this example from the docs that demonstrates having multiple bars, but I can't quite see how to adapt this to my situation without adding substantially more code, and I'm wondering if there is a simpler approach.

import time

from rich.progress import Progress

with Progress() as progress:

    task1 = progress.add_task("[red]Downloading...", total=1000)
    task2 = progress.add_task("[green]Processing...", total=1000)
    task3 = progress.add_task("[cyan]Cooking...", total=1000)

    while not progress.finished:
        progress.update(task1, advance=0.5)
        progress.update(task2, advance=0.3)
        progress.update(task3, advance=0.9)
        time.sleep(0.02)

Thanks!

[github-actions[bot]]

Thank you for your issue. Give us a little time to review it.

PS. You might want to check the FAQ if you haven't done so already.

This is an automated reply, generated by FAQtory

[mattseddon]

I think the examples that you want to look at are examples/dynamic_progress.py & examples/jobs.py.