""" This file contains common functions, used across the project. """ import re import time from collections.abc import Callable from typing import Any def print_progress( last_emit_time: float | int, last_progress: int | float | Any, line: str, min_emit_interval: float, phase_patterns: list[tuple[str, Callable[[Any], int | None]]], ) -> Any: """ If we're at a good place, print a progress message. """ # Try to infer progress from patterns for pattern, extractor in phase_patterns: match = re.search(pattern, line, re.IGNORECASE) if match: progress = extractor(match) if progress is None: continue current_time = time.time() if ( progress > last_progress and (current_time - last_emit_time) > min_emit_interval ): last_progress = progress print(f"\nPROGRESS: {progress:.0f}", flush=True) break return last_progress