#!/usr/bin/env python3 """Export pgbench summary.csv into an HTML report with TPS and latency charts.""" import argparse import csv import json from pathlib import Path from typing import Dict, List, Optional def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Create an HTML chart report from pgbench summary.csv") parser.add_argument("summary_csv", help="Path to summary.csv generated by pgbench benchmark script") parser.add_argument( "-o", "--output", default=None, help="Output HTML path (default: /pgbench_report.html)", ) return parser.parse_args() def load_rows(path: Path) -> List[Dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) expected = {"mode", "clients", "tps", "latency_ms"} if not expected.issubset(set(reader.fieldnames or [])): missing = sorted(expected - set(reader.fieldnames or [])) raise ValueError(f"summary.csv missing required columns: {', '.join(missing)}") return list(reader) def parse_optional_float(raw: str) -> Optional[float]: value = (raw or "").strip() if not value: return None try: return float(value) except ValueError: return None def build_series(rows: List[Dict[str, str]]) -> Dict[str, List[Dict[str, float]]]: series: Dict[str, List[Dict[str, float]]] = {} for row in rows: mode = row["mode"].strip() try: clients = int(row["clients"]) tps = float(row["tps"]) latency = float(row["latency_ms"]) except ValueError: continue point: Dict[str, float] = { "clients": clients, "tps": tps, "latency_ms": latency, } p95 = parse_optional_float(row.get("p95_ms", "")) p99 = parse_optional_float(row.get("p99_ms", "")) if p95 is not None: point["p95_ms"] = p95 if p99 is not None: point["p99_ms"] = p99 series.setdefault(mode, []).append(point) for mode in series: series[mode].sort(key=lambda item: item["clients"]) return series def html_report(title: str, data: Dict[str, List[Dict[str, float]]]) -> str: payload = json.dumps(data) return f""" {title}

{title}

Generated from summary.csv. X-axis = clients. Lines are grouped by mode.

TPS by Clients

Latency (ms) by Clients

p95 Latency (ms) by Clients

p99 Latency (ms) by Clients

""" def main() -> int: args = parse_args() summary_path = Path(args.summary_csv).resolve() if not summary_path.is_file(): raise FileNotFoundError(f"summary.csv not found: {summary_path}") rows = load_rows(summary_path) if not rows: raise ValueError("summary.csv has no data rows") series = build_series(rows) if not series: raise ValueError("No valid numeric rows found in summary.csv") output_path = ( Path(args.output).resolve() if args.output else summary_path.parent / "pgbench_report.html" ) title = f"pgbench report: {summary_path.parent.name}" output_path.write_text(html_report(title=title, data=series), encoding="utf-8") print(f"Report written: {output_path}") return 0 if __name__ == "__main__": raise SystemExit(main())