1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
"""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
|
