SACFlow / sacflow /utils /eta.py
sathiiii's picture
Add SACFlow source code
d65ae7d verified
Raw
History Blame Contribute Delete
1.04 kB
from __future__ import annotations
import time
class ETA:
def __init__(self, total: int, smoothing: float = 0.95):
self.total = max(1, total)
self.smoothing = smoothing
self.start = time.time()
self.last = self.start
self.avg = None
self.n = 0
def step(self, count: int = 1):
now = time.time()
dt = (now - self.last) / max(1, count)
self.last = now
self.n += count
if self.avg is None:
self.avg = dt
else:
self.avg = self.smoothing * self.avg + (1 - self.smoothing) * dt
def eta_seconds(self) -> float:
rem = max(0, self.total - self.n)
return rem * (self.avg or 0.0)
@staticmethod
def fmt(seconds: float) -> str:
seconds = int(seconds)
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h:d}h {m:02d}m {s:02d}s"
if m > 0:
return f"{m:d}m {s:02d}s"
return f"{s:d}s"