Source code for impy.utils.utilcls

import time
import sys
from importlib import import_module
import threading
from .._const import Const

__all__ = ["Timer", "ImportOnRequest", "Progress"]

[docs]class Timer: def __init__(self): self.tic()
[docs] def tic(self): self.t = time.time()
@property def now(self): return time.time() - self.t
[docs] def toc(self): self.t = self.now
def __str__(self): minute, sec = divmod(self.t, 60) sec = round(sec, 2) if minute == 0: out = f"{sec} sec" else: out = f"{int(minute)} min {sec} sec" return out
[docs]class ImportOnRequest: def __init__(self, name:str): self.name = name def __getattr__(self, name:str): try: mod = super().__getattribute__("mod") except AttributeError: self.mod = import_module(self.name) mod = super().__getattribute__("mod") return getattr(mod, name)
[docs]class Progress: n_ongoing = 0 def __init__(self, name, out=sys.stdout): self.name = name self.timer = None self.stop_event = None self.thread = None if out is None: self.out = DummyOut() elif out == "stdout": self.out = sys.stdout else: self.out = out
[docs] def update(self): event = self.stop_event chars = r"-\|/" i = 0 while not event.is_set(): event.wait(0.25) self.out.write(f"\b{chars[i]}") self.out.flush() i = i + 1 if i >= 4: i = 0
def __enter__(self): self.__class__.n_ongoing += 1 if Const["SHOW_PROGRESS"] and self.__class__.n_ongoing == 1: self.timer = Timer() self.stop_event = threading.Event() self.thread = threading.Thread(target=self.update) self.thread.start() self.out.write(f"{self.name} -") self.out.flush() def __exit__(self, exc_type, exc_value, traceback): # We must be careful not to make any error here. self.__class__.n_ongoing -= 1 if self.stop_event is not None: self.stop_event.set() if self.thread is not None: self.thread.join() if Const["SHOW_PROGRESS"] and self.__class__.n_ongoing == 0: if self. timer is not None: self.timer.toc() self.out.write(f"\r{self.name} finished ({self.timer})\n") self.out.flush()
class DummyOut: def write(self, a): pass def flush(self): pass