38 lines
1.0 KiB
Python
38 lines
1.0 KiB
Python
"""
|
|
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
|