cars.core.progress.progress

Progress tree with Rich UI display.

One progress bar is shown per pipeline via Rich. Tasks are registered under a pipeline with weights, and task progress contributes to the pipeline bar accordingly.

Classes

TaskState	Internal state for one tracked task.
PipelineState	Internal state for one pipeline.
ProgressTree	Singleton registry with pipeline-level weighted progress bars.

Module Contents

class cars.core.progress.progress.TaskState

Internal state for one tracked task.

task_id: int

name: str

pipeline_id: int

weight: float

expected_runs: int = 1

started_runs: int = 0

nominal_started_runs: int = 0

progress_in_run: float = 0.0

total: int = 0

last_logged_percent: int = 0

progressed_tiles: int = 0

retries: int = 0

failed: int = 0

pending_retry_pass: bool = False

current_pass_is_retry: bool = False

completed_nominal_runs: int = 0

class cars.core.progress.progress.PipelineState

Internal state for one pipeline.

pipeline_id: int

name: str

position: int

total_weight: float = 0.0

weighted_progress: float = 0.0

retries: int = 0

failed: int = 0

class cars.core.progress.progress.ProgressTree

Singleton registry with pipeline-level weighted progress bars.

_instance: ProgressTree | None = None

_initialized: bool = False

_pipeline_id_gen

_task_id_gen

_pipeline_position_gen

_pipelines: dict[int, PipelineState]

_tasks: dict[int, TaskState]

_task_to_pipeline: dict[int, int]

_pipeline_parent: dict[int, int | None]

_pipeline_children: dict[int, list[int]]

_ui

_ui_enabled = True

_pipeline_order: list[int] = []

_finished_successfully = False

set_ui_enabled(enabled: bool) → None

Enable/disable Rich UI rendering for progress updates.

update_log_file_path(log_file_path: str | None) → None

Update log file path used by the footer log link button.

update_empty_status_text(status_text: str) → None

Update startup status, shown before any pipeline is registered.

begin_pipeline(pipeline_name: str, parent_id: int | None = None) → int

Declare that subsequent tasks belong to this pipeline.

register_task(pipeline_id: int, task_name: str, *, weight: float = 1.0, expected_runs: int = 1) → int

Register a task under a pipeline and return its unique ID.

_refresh_pipeline_bar(pipeline: PipelineState) → None

Update Rich UI display for one pipeline from weighted progress and retry metadata.

_aggregate_pipeline_progress(pipeline_id: int) → tuple[float, float]

Aggregate progress/weight recursively over a pipeline subtree.

Retries and failed counters are intentionally not aggregated.

_refresh_pipeline_and_ancestors(pipeline_id: int) → None

Refresh one pipeline and all its ancestors for progress roll-up.

_set_running_state_for_lineage(pipeline_id: int) → None

Mark a pipeline and all its ancestors as running in the UI.

_handle_started(task: TaskState, pipeline: PipelineState, total: int | None) → None

_handle_progressed(task: TaskState, pipeline: PipelineState, value: int) → None

_handle_retries(task: TaskState, pipeline: PipelineState, value: int) → None

_handle_failed(task: TaskState, pipeline: PipelineState, value: int) → None

_handle_completed(task: TaskState, pipeline: PipelineState) → None

notify(task_id: int, event: str, *, total: int | None = None, value: int = 1) → None

Handle task events from orchestrators and roll up to pipeline bars.

finalize() → None

Clean up UI resources (call after pipelines are done).

draw() → None

Render the current tree state immediately.

notify_crash(exception: BaseException) → None

Notify progress UI that a crash occurred and render details.

notify_success(output_dir: str | None) → None

Notify progress UI that pipeline completed successfully.

get_task_started_runs(task_id: int) → int | None

Return started runs for a task, or None if task is unknown.