Skip to content

Instantly share code, notes, and snippets.

@linuskmr
Last active July 15, 2021 16:10
Show Gist options
  • Save linuskmr/92e1f3438813c30006aa470e1b0e46ea to your computer and use it in GitHub Desktop.
Save linuskmr/92e1f3438813c30006aa470e1b0e46ea to your computer and use it in GitHub Desktop.
Simple ETA (estimated time of availability) calculation
import dataclasses
import json
import sys
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Tuple, TypeVar, Optional
T = TypeVar('T')
@dataclass
class AbsRelTime:
time: timedelta
absolute: int
percentage: float
class ETA:
@dataclass(frozen=True)
class Stats:
total: AbsRelTime
remaining: AbsRelTime
done: AbsRelTime
eta: datetime
def print_refresh(self, file=None):
print(f'\r{self}', end='', file=file, flush=True)
def dict(self) -> dict:
return dataclasses.asdict(self)
def json(self, **kwargs):
return json.dumps(self.dict(), default=lambda x: str(x), **kwargs)
def __str__(self):
return (
f'{self.done.absolute}/{self.total.absolute} = {self.done.percentage:.1f}%, '
f'remaining {self.remaining.time}'
)
def __init__(self, data: Any, auto_print: bool = True):
if not hasattr(data, '__len__'):
raise TypeError('Expected object with len property.')
self.data = iter(data)
self.total = len(data)
self.done = 0
self.start_time = datetime.now()
self.auto_print = auto_print
def __iter__(self):
return self
def __next__(self) -> Tuple[T, Stats]:
self.done += 1
next_data = next(self.data)
stats = self.stats()
if self.auto_print:
stats.print_refresh()
return next_data, stats
def stats(self) -> Stats:
duration_per_operation = (datetime.now() - self.start_time) / self.done
remaining_absolute = self.total - self.done
remaining_time = duration_per_operation * remaining_absolute
return self.Stats(
total=AbsRelTime(
time=duration_per_operation * self.total,
absolute=self.total,
percentage=100
),
remaining=AbsRelTime(
time=remaining_time,
absolute=remaining_absolute,
percentage=(remaining_absolute / self.total)*100
),
done=AbsRelTime(
time=duration_per_operation * self.done,
absolute=self.done,
percentage=(self.done / self.total)*100
),
eta=datetime.now() + remaining_time
)
import time
from eta import ETA
# Some data to be processed
data = list(range(42))
for num, _ in ETA(data):
# Do something useful here that takes a little time
time.sleep(0.1)
# Example output:
# 13/42 = 31.0%, ETA Delta 0:00:02.685864, ETA 2021-07-04 14:50:12.486002
# Or disable auto print and access all eta data
for num, eta in ETA(data, auto_print=False):
# Do something useful here that takes a little time
time.sleep(0.1)
print(eta.json(indent=' '))
# {
# "total": {
# "time": "0:00:03.762402",
# "absolute": 42,
# "percentage": 100
# },
# "remaining": {
# "time": "0:00:02.956173",
# "absolute": 33,
# "percentage": 78.57142857142857
# },
# "done": {
# "time": "0:00:00.806229",
# "absolute": 9,
# "percentage": 21.428571428571427
# },
# "eta": "2021-07-04 18:05:35.103308"
# }
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment