33 lines
1.1 KiB
Python
33 lines
1.1 KiB
Python
"""
|
|
This file contains common functions, used across the project.
|
|
"""
|
|
|
|
import re
|
|
import time
|
|
from typing import Any, Callable
|
|
|
|
|
|
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]]]) -> 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:
|
|
try:
|
|
progress = extractor(match)
|
|
current_time = time.time()
|
|
if progress > last_progress and (current_time - last_emit_time) > min_emit_interval:
|
|
last_progress = progress
|
|
last_emit_time = current_time
|
|
print(f"\nPROGRESS: {progress:.0f}", flush=True)
|
|
break
|
|
except:
|
|
pass
|
|
return last_progress
|