Create app/progress.py

app/progress.py

@@ -0,0 +1,58 @@
+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