Recommended way to have multiple progress bars & multiple status spinners active at the same time

[akhileshraju]

My use case requires that I have a way to support multiple status spinners and progress bars active at the same time.
I followed the recipe from #1500 (comment) and came up with the below snippet.

import threading
from typing import Callable

import time

from rich.console import Group
from rich.live import Live
from rich.progress import (
    Progress as RichProgress,
    TaskID as RichProgressTaskId,
    TextColumn,
    BarColumn,
)
from rich.status import Status as RichStatus


class Spinner:
    def __init__(self, live_updater: Callable):
        self.spinners_by_task_id: dict[int, RichStatus] = dict()
        self._live: Live = None
        self.live_updater = live_updater

    @property
    def __rich_renderable__(self) -> list[RichStatus]:
        return list(self.spinners_by_task_id.values())

    def show(self, task_id: int, status: str) -> None:
        if task_id not in self.spinners_by_task_id:
            self.spinners_by_task_id[task_id] = RichStatus(status=status)
            self.spinners_by_task_id[task_id]._live = self._live
            self.live_updater()
        self.spinners_by_task_id[task_id].status = status
        self.spinners_by_task_id[task_id].update()

    def stop(self, task_id: int) -> None:
        if task_id not in self.spinners_by_task_id:
            raise KeyError(f"Spinner not found for task ID '{task_id}'")
        del self.spinners_by_task_id[task_id]
        self.live_updater()


class ProgressBar:
    def __init__(self, *columns, live_updater: Callable):
        self.progress_bars_by_task_id: dict[int, RichProgress] = dict()
        self.rich_task_ids: dict[int, RichProgressTaskId] = dict()
        self.columns = columns
        self.live_updater = live_updater

    @property
    def __rich_renderable__(self) -> list[RichProgress]:
        return list(self.progress_bars_by_task_id.values())

    def add(self, task_id: int, description: str) -> None:
        self.progress_bars_by_task_id[task_id] = RichProgress(*self.columns)
        rich_task_id = self.progress_bars_by_task_id[task_id].add_task(description)
        self.live_updater()
        self.rich_task_ids[task_id] = rich_task_id

    def update(self, task_id: int, status: str, completed: int):
        if task_id not in self.progress_bars_by_task_id:
            raise RuntimeError(f"Progress bar not found for task ID '{task_id}'")
        self.progress_bars_by_task_id[task_id].update(
            self.rich_task_ids[task_id], status=status, completed=completed
        )

    def remove(self, task_id: int) -> None:
        if task_id not in self.progress_bars_by_task_id:
            raise KeyError(f"Progress bar not found for task ID '{task_id}'")
        del self.progress_bars_by_task_id[task_id]
        del self.rich_task_ids[task_id]
        self.live_updater()


class Printer:
    def __init__(self):
        self.spinner = Spinner(live_updater=self._update_live_render_group)
        self.progress_bar = ProgressBar(
            TextColumn("{task.description}"),
            BarColumn(),
            TextColumn("{task.percentage:>3.0f}%"),
            TextColumn("{task.fields[status]}"),
            live_updater=self._update_live_render_group,
        )

        self.live = Live(
            Group(
                *self.spinner.__rich_renderable__,
                *self.progress_bar.__rich_renderable__,
            ),
            refresh_per_second=12,
        )
        self.spinner._live = self.live
        # self.progress_bar.live = self.live
        self.live.start()

    def _update_live_render_group(self) -> None:
        self.live.update(
            Group(
                *self.spinner.__rich_renderable__,
                *self.progress_bar.__rich_renderable__,
            )
        )


if __name__ == "__main__":
    printer = Printer()

    def update_status():
        time.sleep(5)
        printer.spinner.show(1, "1-1")
        time.sleep(2)
        printer.spinner.show(2, "2-1")
        time.sleep(1)
        printer.spinner.show(3, "3-1")
        printer.spinner.show(2, "2-2")
        time.sleep(2)
        printer.spinner.stop(2)
        printer.spinner.stop(3)
        time.sleep(2)
        printer.spinner.show(1, "1-2")
        printer.spinner.show(4, "4-1")
        time.sleep(2)
        printer.spinner.stop(4)
        printer.spinner.stop(1)

    def update_progress():
        printer.progress_bar.add(1, "Task 1")
        printer.progress_bar.add(2, "Task 2")
        printer.progress_bar.add(3, "Task 3")
        printer.progress_bar.add(4, "Task 4")
        printer.progress_bar.add(5, "Task 5")

        def p1_updater():
            for i in range(0, 100):
                printer.progress_bar.update(1, status="Working...", completed=i)
                time.sleep(0.1)
            printer.progress_bar.remove(1)

        def p2_updater():
            for i in range(0, 100, 5):
                printer.progress_bar.update(2, status="Working...", completed=i)
                time.sleep(0.5)
            printer.progress_bar.remove(2)

        def p3_updater():
            for i in range(0, 100, 10):
                printer.progress_bar.update(3, status="Working...", completed=i)
                time.sleep(0.25)
            printer.progress_bar.remove(3)

        def p4_updater():
            for i in range(0, 100, 15):
                printer.progress_bar.update(4, status="Working...", completed=i)
                time.sleep(1.25)
            printer.progress_bar.remove(4)

        def p5_updater():
            for i in range(0, 100, 20):
                printer.progress_bar.update(5, status="Working...", completed=i)
                time.sleep(3)
            printer.progress_bar.remove(5)

        tp1 = threading.Thread(target=p1_updater)
        tp2 = threading.Thread(target=p2_updater)
        tp3 = threading.Thread(target=p3_updater)
        tp4 = threading.Thread(target=p4_updater)
        tp5 = threading.Thread(target=p5_updater)

        tp1.start()
        tp2.start()
        tp3.start()
        tp4.start()
        tp5.start()
        tp1.join()
        tp2.join()
        tp3.join()
        tp4.join()
        tp5.join()

    ts = threading.Thread(target=update_status)
    tp = threading.Thread(target=update_progress)

    ts.start()
    tp.start()
    ts.join()
    tp.join()

    time.sleep(2)

This works as I expect to and shows up like so

MultipleProgressAndSpinner.mp4

Questions

1. Is this the recommended way to achieve multiple active progress & spinners?

2. Since there can be only one Live instance active at any given time, I had to find a way to get the Status and Progress classes to use the Live instance that I am creating.

   Progress seems to already have a way to get the current console like below.
   

   rich/rich/progress.py

   Line 1095 in e1e6d74

   console=console or get_console(),

   
   However, Status doesn't seem to have any such provision.
   

   rich/rich/status.py

   Line 39 in e1e6d74

   console=console,

   
   So, I had to resort to overriding the Status._live attribute to get it to use the Live instance I am creating. I don't feel very comfortable override a "protected" member variable. Is this a safe approach? Alternatively, would the maintainers be willing to entertain a PR for getting Status to call get_console() when creating its Live instance?

   NOTE - I see that Progress is creating its own Live instance different from the one I am creating, but that does seem to peacefully co-exist with the Live I am creating. I assume that this is has something to do with the get_console() call within the Live instance creation in Progress