Skip to content

pipeloom.progress

pipeloom.progress

progress.py

Factory helpers for Rich progress managers.

We intentionally separate overall progress (sticky, remains visible at 100%) from per-task progress (transient, disappears once all tasks complete). Both share the same Console (see rlog.console).

make_overall_progress 

make_overall_progress() -> Progress

Create the overall persistent progress manager.

This remains visible after completion and is ideal for a single “All tasks: 100%” summary line once per-task bars disappear.

Returns:

Name Type Description
Progress Progress

The Rich Progress instance managing overall progress.
Source code in pipeloom/progress.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def make_overall_progress() -> Progress:
    """
    Create the overall persistent progress manager.

    This remains visible after completion and is ideal for a single “All tasks: 100%”
    summary line once per-task bars disappear.

    Returns:
        Progress: The Rich Progress instance managing overall progress.
    """
    return Progress(
        SpinnerColumn(),
        "[progress.description]{task.description}",
        BarColumn(),
        TaskProgressColumn(),
        MofNCompleteColumn(),
        "•",
        TimeElapsedColumn(),
        "•",
        TimeRemainingColumn(),
        transient=False,  # keep final frame visible
        console=console,
        refresh_per_second=8,  # throttle repaints to avoid log spam jitter
        disable=not console.is_terminal,
    )

make_task_progress 

make_task_progress() -> Progress

Create the transient per-task progress manager.

This hides when the context exits so your terminal remains tidy.

Returns:

Name Type Description
Progress Progress

The Rich Progress instance managing task bars.
Source code in pipeloom/progress.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def make_task_progress() -> Progress:
    """
    Create the transient per-task progress manager.

    This hides when the context exits so your terminal remains tidy.

    Returns:
        Progress: The Rich Progress instance managing task bars.
    """
    return Progress(
        SpinnerColumn(),
        "[progress.description]{task.description}",
        BarColumn(),
        TaskProgressColumn(),
        TimeElapsedColumn(),
        transient=True,  # auto-hide on exit
        console=console,
        refresh_per_second=8,
        disable=not console.is_terminal,
    )

preregister_task_bars 

preregister_task_bars(task_progress: Progress, num_tasks: int) -> dict[int, TaskID]

Pre-create one Rich bar per task and return a mapping {task_id -> TaskID}.

Parameters:

Name Type Description Default
task_progress Progress

The Rich Progress instance managing task bars.

 required
num_tasks int

The total number of tasks to create bars for.

 required

Returns:

Type Description
dict[int, TaskID]

dict[int, TaskID]: A mapping of task IDs to their Rich TaskIDs.
Why up-front?

If a worker publishes a MsgTaskProgress before the writer has created the bar, updates can be lost. Pre-registering eliminates this race completely.

Source code in pipeloom/progress.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def preregister_task_bars(task_progress: Progress, num_tasks: int) -> dict[int, TaskID]:
    """
    Pre-create one Rich bar per task and return a mapping {task_id -> TaskID}.

    Args:
        task_progress (Progress): The Rich Progress instance managing task bars.
        num_tasks (int): The total number of tasks to create bars for.

    Returns:
        dict[int, TaskID]: A mapping of task IDs to their Rich TaskIDs.

    Why up-front?
    -------------
    If a worker publishes a MsgTaskProgress before the writer has created the bar,
    updates can be lost. Pre-registering eliminates this race completely.
    """
    mapping: dict[int, TaskID] = {}
    for i in range(1, num_tasks + 1):
        mapping[i] = task_progress.add_task(f"[bold]task-{i}", total=100)
    return mapping