import argparse
import re
import subprocess
import sys
from pathlib import Path
from cases import CASE_NAMES, matpower_filename
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
SCRIPT_DIR = Path(__file__).resolve().parent
CONVERT_SCRIPT = SCRIPT_DIR / "matpower_to_pgm.py"
BENCH_GRIDOXIDE_SCRIPT = SCRIPT_DIR / "bench_gridoxide_native.py"
BENCH_PGM_SCRIPT = SCRIPT_DIR / "bench_pgm.py"
LS2G_SCRIPT = SCRIPT_DIR / "bench_lightsim2grid.py"
PYPOWSYBL_SCRIPT = SCRIPT_DIR / "bench_pypowsybl.py"
PANDAPOWER_SCRIPT = SCRIPT_DIR / "bench_pandapower.py"
VERAGRID_SCRIPT = SCRIPT_DIR / "bench_veragrid.py"
MATPOWER_DIR = REPO_ROOT / "tests" / "data" / "benchmark-grids" / "matpower"
MEAN_RE = re.compile(r"min=[\d.]+ms mean=([\d.]+)ms")
NODES_RE = re.compile(r"nodes=(\d+)")
def fetch_matpower_case(case_name: str) -> tuple[Path | None, str | None]:
filename = matpower_filename(case_name)
m_path = MATPOWER_DIR / filename
if not m_path.exists():
return None, (
f"{m_path} not found — run "
"`git submodule update --init tests/data/benchmark-grids`"
)
return m_path, None
def convert_case(python: str, case_name: str, cache_dir: Path) -> tuple[Path | None, Path | None, str | None]:
m_path, fetch_err = fetch_matpower_case(case_name)
if m_path is None:
return None, None, fetch_err
out_path = cache_dir / f"{case_name}.json"
if out_path.exists():
return m_path, out_path, None
proc = subprocess.run(
[python, str(CONVERT_SCRIPT), str(m_path), str(out_path)],
capture_output=True, text=True, timeout=300,
)
if proc.returncode != 0 or not out_path.exists():
return m_path, None, (extract_error(proc.stderr) if proc.stderr else "conversion failed")
return m_path, out_path, None
def run_gridoxide(python: str, json_path: Path, backend: str) -> tuple[float | None, int | None, str | None]:
proc = subprocess.run(
[python, str(BENCH_GRIDOXIDE_SCRIPT), str(json_path), backend],
capture_output=True, text=True, timeout=300,
)
if proc.returncode != 0:
if "ModuleNotFoundError" in proc.stderr and "gridoxide" in proc.stderr:
return None, None, "gridoxide extension module not built (see this directory's README)"
return None, None, (extract_error(proc.stderr) if proc.stderr else "failed")
mean_match = MEAN_RE.search(proc.stdout)
nodes_match = NODES_RE.search(proc.stdout)
if not mean_match:
return None, None, "could not parse output"
return float(mean_match.group(1)), (int(nodes_match.group(1)) if nodes_match else None), None
def extract_error(stderr: str) -> str:
if not stderr:
return "failed"
lines = [l.strip() for l in stderr.strip().splitlines() if l.strip()]
lines = [l for l in lines if "validate_input_data" not in l and "validate_batch_data" not in l]
for line in reversed(lines):
if re.match(r"^[\w.]+:\s", line):
return line
return lines[-1] if lines else "failed"
def run_subprocess_mean(python: str, script: Path, args: list[str], timeout: int = 600) -> tuple[float | None, str | None]:
proc = subprocess.run(
[python, str(script), *args],
capture_output=True, text=True, timeout=timeout,
)
if proc.returncode != 0:
return None, extract_error(proc.stderr)
mean_match = MEAN_RE.search(proc.stdout)
if not mean_match:
return None, "could not parse output"
return float(mean_match.group(1)), None
def fmt(value: float | None, err: str | None) -> str:
if value is not None:
return f"{value:.3f} ms"
if err and err.startswith("N/A"):
return err
return f"FAILED ({err})" if err else "FAILED"
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--python", default=sys.executable)
parser.add_argument("--cache-dir", type=Path, default=Path(__file__).resolve().parent / ".case-cache")
parser.add_argument("--out", type=Path, default=None, help="also write the table to this file")
args = parser.parse_args()
args.cache_dir.mkdir(exist_ok=True)
rows = []
for case_name in CASE_NAMES:
print(f"=== {case_name} ===", file=sys.stderr)
m_path, json_path, conv_err = convert_case(args.python, case_name, args.cache_dir)
if json_path is None:
rows.append((case_name, "-", f"FAILED ({conv_err})", "-", "-", "-", "-", "-", "-", "-", "-", "-"))
continue
n_nodes = None
backend_times = {}
for backend in ("scalar", "block", "klu", "klu_native", "pardiso"):
t, n, err = run_gridoxide(args.python, json_path, backend)
n_nodes = n_nodes or n
backend_times[backend] = fmt(t, err)
pgm_time, pgm_err = run_subprocess_mean(args.python, BENCH_PGM_SCRIPT, [str(json_path)])
ls2g_time, ls2g_err = run_subprocess_mean(args.python, LS2G_SCRIPT, [case_name])
pypowsybl_time, pypowsybl_err = run_subprocess_mean(args.python, PYPOWSYBL_SCRIPT, [case_name, str(m_path)])
pandapower_time, pandapower_err = run_subprocess_mean(args.python, PANDAPOWER_SCRIPT, [case_name])
veragrid_time, veragrid_err = run_subprocess_mean(args.python, VERAGRID_SCRIPT, [case_name, str(m_path)])
rows.append((
case_name,
str(n_nodes) if n_nodes else "?",
backend_times["scalar"],
backend_times["block"],
backend_times["klu"],
backend_times["klu_native"],
backend_times["pardiso"],
fmt(pgm_time, pgm_err),
fmt(ls2g_time, ls2g_err),
fmt(pypowsybl_time, pypowsybl_err),
fmt(pandapower_time, pandapower_err),
fmt(veragrid_time, veragrid_err),
))
header = ["case", "buses", "gridoxide scalar", "gridoxide block", "gridoxide klu", "gridoxide klu_native",
"gridoxide pardiso", "PGM", "lightsim2grid (KLU)", "pypowsybl", "pandapower", "VeraGrid"]
lines = [
"| " + " | ".join(header) + " |",
"|" + "|".join(["---"] * len(header)) + "|",
]
for row in rows:
lines.append("| " + " | ".join(row) + " |")
table = "\n".join(lines)
print()
print(table)
if args.out:
args.out.write_text(table + "\n")
print(f"\nwrote {args.out}", file=sys.stderr)
if __name__ == "__main__":
main()