Skip to main content
  1. Maoli's Digital Garden/

Multi-threading and Multi-processing Progress Visualization with Python's `rich` Library

The Python package rich is for rich text and beautiful formatting in the terminal.

Here are some notes about how to use rich for visualizing the progress of long-running tasks, printing rich text… and some tips for displaying progress bars in multi-threading and multi-processing scenarios.1

Visualizing the Progress of Long-running Tasks #

When running long tasks in Python, it’s helpful to have a visual progress indicator. rich provides several methods to achieve this.

Using track #

For for loops with a known number of iterations, track is a useful method.

  • Example

    from rich.progress import track
    from time import sleep
    for step in track(range(10), description="[green]Processing..."):

    There will be a progress bar with a green “Processing…” message, increasing by 10% every 1 second.

Using status #

The status method can be used to indicate ongoing tasks when the exact progress percentage is not crucial or known, displaying a spinner with a temporary status message.

  • Example

    from time import sleep
    from rich.console import Console
    console = Console()
    tasks = [f"task {n}" for n in range(1, 11)]
    with console.status("[bold green]Working on tasks...") as status:
      while tasks:
          task = tasks.pop(0)
          console.log(f"{task} complete")

    There will be a spinner with the message “Working on tasks…” and logs each completed task.

  • Some arguments:

    • status: specifying the message to be displayed.
    • spinner: specifying the spinner to be used.
      • common spinners: dots, star, growHorizontal, bouncingBar, bouncingBall, …
      • use python -m rich.spinner to check all spinners.

Using Progress #

The Progress class provides a more customizable way and can be used for more complex scenarios.

For example, when there are multiple tasks to be monitored, we can use Progress to display multiple progress bars simultaneously.

  • Example

    import time
    from concurrent.futures import ThreadPoolExecutor
    from rich import print
    from rich.progress import Progress
    progress = Progress()
    def working(worker, amount, efficiency):
        progress.update(worker, total=amount)
        while not progress.finished:
            progress.update(worker, advance=efficiency)
    if __name__ == '__main__':
        task_load = {"maoli": 100, "zhuohua": 80}
        efficiency = {"maoli": 2, "zhuohua": 1}
        with progress:
            maoli = progress.add_task("[cyan]Maoli's working...", start=False)
            zhuohua = progress.add_task("[magenta]Zhuohua's working...", start=False)
            with ThreadPoolExecutor() as pool:
                pool.submit(working, maoli, task_load["maoli"], efficiency["maoli"])
                pool.submit(working, zhuohua, task_load["zhuohua"], efficiency["zhuohua"])
        print("[green]Tasks complete!")

    Maoli and Zhuohua are working. 😫

Handling Multi-threading and Multi-processing Scenarios #

  • When there are multiple threads in a single process (e.g., ThreadPoolExecutor, multiprocessing.pool.ThreadPool), the progress bars will be updated correctly.

  • However, in scenarios involving multiple processes (e.g., multiprocessing.Pool), Progress may not work as expected since the progress bars are designed to be updated in the same process. Comnumication between processes is required if we want to use Progress in this scenario.

  • A toy example: using multiprocessing.Pipe to communicate between processes.

    import time
    from multiprocessing import Pipe, Pool
    from rich import print
    from import Live
    from rich.progress import Progress
    def working(worker, total_amount, efficiency, pipe):
        completed_amount = 0
        while completed_amount < total_amount:
            amount = efficiency
            completed_amount += amount
            completed_amount = min(completed_amount, total_amount)
            pipe.send((worker, completed_amount))
        return f"[bold red]{worker} completed!"
    if __name__ == '__main__':
        task_amount = {"maoli": 20, "zhuohua": 20}
        efficiency = {"maoli": 2, "zhuohua": 1}
        maoli_conn, maoli_child_conn = Pipe()
        zhuohua_conn, zhuohua_child_conn = Pipe()
        progress = Progress()
        with progress:
            main_progress = progress.add_task("[green]Main progress...", total=2)
            maoli = progress.add_task("[cyan]Maoli's Working...", total=task_amount["maoli"])
            zhuohua = progress.add_task("[magenta]Zhuohua's Working...", total=task_amount["zhuohua"])
            pool = Pool(processes=2)
            maoli_process = pool.apply_async(working, args=("maoli", task_amount["maoli"], efficiency["maoli"], maoli_child_conn))
            zhuohua_process = pool.apply_async(working, args=("zhuohua", task_amount["zhuohua"], efficiency["zhuohua"], zhuohua_child_conn))
            completed_workers = {'maoli': 0, 'zhuohua': 0}
            while sum(completed_workers.values()) < 2:
                #check maoli
                if not maoli_conn.closed and maoli_conn.poll():
                    task_id, completed = maoli_conn.recv()
                    progress.update(maoli, description=f"[cyan]Maoli completed: {100*completed/task_amount['maoli']:.2f}%", completed=completed)
                    if completed >= task_amount["maoli"]:
                        completed_workers["maoli"] = 1
                #check zhuohua
                if not zhuohua_conn.closed and zhuohua_conn.poll():
                    task_id, completed = zhuohua_conn.recv()
                                    description=f"[magenta]Zhuohua completed: {100*completed/task_amount['zhuohua']:.2f}%",
                    if completed >= task_amount["zhuohua"]:
                        completed_workers["zhuohua"] = 1
            # for res in results:
            #     print(res.get())
        print("[bold yellow]Tasks completed, exiting...")

Printing Rich Text in Python #

We can import the print function from rich to replace the built-in print function.

  • Example
    from rich import print
    print("Hello, [bold magenta]World[/bold magenta]!")

Working with the Console #

Basic Printing #

from rich.console import Console

console = Console()
console.print("Hello, Maoli!")

Styling Text #

console.print("Hello", style="bold red")

Logging with Timestamps and File Information #

  • Console.log()

Creating Tables with Table #

  • Example

    from rich.table import Table
    from rich.console import Console
    table = Table(title="A Table")
    table.add_column("Column 1")
    table.add_row("Row 1")
    console = Console()

  1. This note is with the help of GitHub Copilot. ↩︎