Module `saev.helpers`

Useful helpers for saev.

Functions

def flattened(dct: dict[str, object], *, sep: str = '.') ‑> dict[str, str | int | float | bool | None]: Flatten a potentially nested dict to a single-level dict with .-separated keys.
def get(dct: dict[str, object], key: str, *, sep: str = '.') ‑> object
def get_cache_dir() ‑> str: Get cache directory from environment variables, defaulting to the current working directory (.)

Returns

A path to a cache directory (might not exist yet).

Classes

class progress (it, *, every: int = 10, desc: str = 'progress', total: int = 0)

Wraps an iterable with a logger like tqdm but doesn't use any control codes to manipulate a progress bar, which doesn't work well when your output is redirected to a file. Instead, simple logging statements are used, but it includes quality-of-life features like iteration speed and predicted time to finish.

Args

it: Iterable to wrap.
every: How many iterations between logging progress.
desc: What to name the logger.
total: If non-zero, how long the iterable is.

Expand source code

@beartype.beartype
class progress:
    def __init__(self, it, *, every: int = 10, desc: str = "progress", total: int = 0):
        """
        Wraps an iterable with a logger like tqdm but doesn't use any control codes to manipulate a progress bar, which doesn't work well when your output is redirected to a file. Instead, simple logging statements are used, but it includes quality-of-life features like iteration speed and predicted time to finish.

        Args:
            it: Iterable to wrap.
            every: How many iterations between logging progress.
            desc: What to name the logger.
            total: If non-zero, how long the iterable is.
        """
        self.it = it
        self.every = every
        self.logger = logging.getLogger(desc)
        self.total = total

    def __iter__(self):
        start = time.time()

        try:
            total = len(self)
        except TypeError:
            total = None

        for i, obj in enumerate(self.it):
            yield obj

            if (i + 1) % self.every == 0:
                now = time.time()
                duration_s = now - start
                per_min = (i + 1) / (duration_s / 60)

                if total is not None:
                    pred_min = (total - (i + 1)) / per_min
                    self.logger.info(
                        "%d/%d (%.1f%%) | %.1f it/m (expected finish in %.1fm)",
                        i + 1,
                        total,
                        (i + 1) / total * 100,
                        per_min,
                        pred_min,
                    )
                else:
                    self.logger.info("%d/? | %.1f it/m", i + 1, per_min)

    def __len__(self) -> int:
        if self.total > 0:
            return self.total

        # Will throw exception.
        return len(self.it)

Functions

Returns

Classes

Args