rpki/scripts/local_repo_replay/summarize_replay.py

48 lines
1.5 KiB
Python
Executable File

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
def count_lines(path: Path) -> int:
if not path.is_file():
return 0
return sum(1 for line in path.read_text(encoding="utf-8", errors="replace").splitlines() if line.strip())
def parse_time(path: Path) -> dict[str, object]:
if not path.is_file():
return {}
text = path.read_text(encoding="utf-8", errors="replace")
elapsed = re.search(r"Elapsed \(wall clock\) time .*: (.+)", text)
rss = re.search(r"Maximum resident set size \(kbytes\): (\d+)", text)
return {
"elapsed": elapsed.group(1).strip() if elapsed else "",
"maxRssKb": int(rss.group(1)) if rss else 0,
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--rp", required=True)
parser.add_argument("--out-dir", type=Path, required=True)
parser.add_argument("--summary", type=Path, required=True)
args = parser.parse_args()
summary = {
"rp": args.rp,
"vrps": count_lines(args.out_dir / "vrps.normalized.txt"),
"vaps": count_lines(args.out_dir / "vaps.normalized.txt"),
"time": parse_time(args.out_dir / "process-time.txt"),
"artifacts": sorted(path.name for path in args.out_dir.iterdir() if path.is_file()),
}
args.summary.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(args.summary)
return 0
if __name__ == "__main__":
raise SystemExit(main())