"""A refresh thread keeps elapsed time moving even during native processing.""" import time from rich.console import Console from rich.progress import (BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn, TimeElapsedColumn, TimeRemainingColumn) class Display: def __init__(self): self.console = Console(stderr=True) self.progress = Progress(SpinnerColumn(), TextColumn("{task.description}"), BarColumn(), TaskProgressColumn(), TimeElapsedColumn(), TimeRemainingColumn(), console=self.console, refresh_per_second=5, disable=not self.console.is_terminal) self.task = None self.stage = "" self.started = time.monotonic() self.last_print = 0.0 def __enter__(self): self.progress.start() return self def __exit__(self, *args): self.progress.stop() def update(self, stage, completed=None, total=None): changed = stage != self.stage if changed: if self.task is not None: self.progress.remove_task(self.task) self.task = self.progress.add_task(stage, total=total) self.stage = stage self.progress.update(self.task, completed=completed if total else 0, total=total) now = time.monotonic() if not self.console.is_terminal and (changed or now - self.last_print >= 5): fraction = f" {min(100, 100 * (completed or 0) / total):.0f}%" if total else "" self.console.print(f"[{now - self.started:7.1f}s] {stage}{fraction}", markup=False) self.last_print = now