Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| """ | |
| Progress and timing helpers. | |
| This module provides small utilities to improve the user experience of long-running | |
| pipeline stages by emitting: | |
| - stage start/end messages | |
| - elapsed time | |
| - optional spinner UI while blocking library calls run | |
| We intentionally keep this lightweight and dependency-minimal (Rich only). | |
| """ | |
| from contextlib import contextmanager | |
| from time import perf_counter | |
| from typing import Iterator, Optional | |
| from rich.console import Console | |
| from rich.status import Status | |
| def stage_status( | |
| title: str, | |
| *, | |
| console: Optional[Console] = None, | |
| spinner: str = "dots", | |
| ) -> Iterator[None]: | |
| """ | |
| Context manager that shows a Rich spinner and prints elapsed time. | |
| Use this for stages where work happens inside a single blocking call | |
| (e.g., Docling loader.load()) and you cannot expose per-item progress. | |
| Parameters | |
| ---------- | |
| title: | |
| Human readable stage label shown in the console. | |
| console: | |
| Rich Console instance. If None, a default Console is created. | |
| spinner: | |
| Rich spinner name. Examples: "dots", "earth", "bouncingBall". | |
| Example | |
| ------- | |
| >>> with stage_status("🦆 Docling: PDF → paragraphs"): | |
| ... docs = loader.load() | |
| """ | |
| c = console or Console() | |
| start = perf_counter() | |
| with Status(title, console=c, spinner=spinner): | |
| yield | |
| elapsed = perf_counter() - start | |
| c.print(f"[green]✅ {title}[/green] [dim](took {elapsed:.2f}s)[/dim]") | |