Progress printer

This commit is contained in:
Maximilian Schmeller 2022-05-31 16:49:47 +02:00
parent 813d571089
commit ebd9d68a40
2 changed files with 153 additions and 334 deletions

File diff suppressed because one or more lines are too long

View file

@ -1,16 +1,32 @@
from typing import Callable, Iterable, Optional, TypeVar import math
T = TypeVar("T") def left_abbreviate(string, limit=120):
def filter_none(ls: Iterable[Optional[T]]) -> filter[T]: return string if len(string) <= limit else f"...{string[:limit-3]}"
return filter(lambda x: x is not None, ls)
S = TypeVar("S")
def safe_map(func: Callable[[T], S], ls: Iterable[T]) -> map[Optional[S]]:
def safe_func(arg):
try:
return func(arg)
except:
return None
return map(safe_func, ls) class ProgressPrinter:
def __init__(self, verb, n) -> None:
self.verb = verb
self.n = n
self.i = 0
self.fmt_len = math.ceil(math.log10(n if n > 0 else 1))
def step(self, msg):
self.i += 1
print(f"({self.i:>{self.fmt_len}d}/{self.n}) {self.verb} {left_abbreviate(msg):<120}", end="\r")
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, exc_traceback):
self.i -= 1
if exc_value:
self.step("error.")
print()
print(exc_value)
return
self.step("done.")
print()