| """Parse GROMACS mdrun log lines into structured progress events. |
| |
| mdrun emits step lines like: |
| step 12345000, t = 24690.000 ps, kT = 308.4 |
| and at the end: |
| Performance: 47.823 0.502 |
| """ |
| from __future__ import annotations |
| import re |
| from dataclasses import dataclass |
|
|
|
|
| @dataclass(frozen=True) |
| class ProgressEvent: |
| """One snapshot of mdrun progress.""" |
| current_step: int | None = None |
| current_time_ps: float | None = None |
| total_steps: int | None = None |
| total_time_ps: float | None = None |
| ns_per_day: float | None = None |
|
|
|
|
| |
| _STEP_RE = re.compile( |
| r"step\s+(\d+)\s*,\s*t\s*=\s*([\d.]+)\s*ps", |
| re.IGNORECASE, |
| ) |
|
|
| |
| _PERF_RE = re.compile( |
| r"Performance:\s+([\d.]+)\s+([\d.]+)", |
| ) |
|
|
|
|
| def parse_progress_line(line: str) -> ProgressEvent | None: |
| """Return a ProgressEvent if the line carries any known metric, else None.""" |
| if not line or line.lstrip().startswith("#"): |
| return None |
|
|
| perf = _PERF_RE.search(line) |
| if perf: |
| return ProgressEvent(ns_per_day=float(perf.group(1))) |
|
|
| step = _STEP_RE.search(line) |
| if step: |
| return ProgressEvent( |
| current_step=int(step.group(1)), |
| current_time_ps=float(step.group(2)), |
| ) |
|
|
| return None |
|
|
|
|
| def percent_complete(current_step: int, total_steps: int) -> float: |
| """Return percent in [0, 100]. 0 when total_steps is unknown.""" |
| if total_steps <= 0: |
| return 0.0 |
| return min(100.0, 100.0 * current_step / total_steps) |
|
|
