from __future__ import annotations import time def human_bytes(n: float) -> str: for unit in ["B", "KB", "MB", "GB", "TB"]: if n < 1024: return f"{n:.1f}{unit}" n /= 1024 return f"{n:.1f}PB" def fmt_eta(sec: float) -> str: sec = max(0, int(sec)) if sec < 60: return f"{sec}s" m = sec // 60 s = sec % 60 if m < 60: return f"{m}m {s}s" h = m // 60 m = m % 60 return f"{h}h {m}m" class RateProgress: def __init__(self, total: int, edit_every: float = 2.5): self.total = max(1, total) self.edit_every = edit_every self.t0 = time.time() self.last_edit = 0.0 self.last_bytes = 0 self.last_t = self.t0 def snapshot(self, done: int) -> dict: now = time.time() done = max(0, min(done, self.total)) dt = max(1e-6, now - self.last_t) dbytes = done - self.last_bytes speed = dbytes / dt # B/s elapsed = now - self.t0 pct = (done / self.total) * 100.0 remaining = self.total - done eta = remaining / speed if speed > 1 else float("inf") return { "pct": pct, "done": done, "total": self.total, "speed": speed, "eta": eta, "elapsed": elapsed, "now": now } def should_edit(self, now: float) -> bool: return (now - self.last_edit) >= self.edit_every def mark_edit(self, done: int, now: float): self.last_edit = now self.last_bytes = done self.last_t = now