scripts/summarize.py

#!/usr/bin/env python3
"""Summarize actual benchmark JSONL without inventing results or pooling configs."""

import argparse
from collections import defaultdict
import json
import math
from pathlib import Path
import statistics


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("files", nargs="+", type=Path)
    args = parser.parse_args()
    groups = defaultdict(list)
    failed = []
    for path in args.files:
        for line in path.read_text().splitlines():
            row = json.loads(line)
            if row.get("status") != "ok":
                failed.append({"file": str(path), "trial": row.get("trial"), "error_type": row.get("error_type")})
                continue
            recipe = row["recipe"]
            identity = {
                "model_revision": recipe["model_revision"],
                "diffusers_revision": recipe["diffusers_revision"],
                "versions": recipe["versions"], "parameters": recipe["parameters"],
                "reference_pixel_sha256": recipe["reference_pixel_sha256"],
                "effective_prompt": recipe["effective_prompt"],
                "gpu_name": recipe["metrics"]["gpu"]["name"],
                "gpu_size": recipe["metrics"]["gpu"]["size"],
                "classification": "reused_gpu_worker" if recipe["metrics"].get("gpu_worker_call_index", 1) > 1
                    else row["classification"],
                "source_sha256": recipe.get("source_sha256"),
                "hub_commit": json.loads((path.parent / "manifest.json").read_text()).get("hub_commit"),
            }
            groups[json.dumps(identity, sort_keys=True)].append(row)
    summaries = []
    for key, rows in groups.items():
        summary = {"configuration": json.loads(key), "n": len(rows),
                   "distinct_pixel_hashes": len({r["pixel_sha256"] for r in rows})}
        for field in ["inference_seconds", "gpu_call_wall_seconds", "client_wall_seconds"]:
            values = [r[field] if field == "client_wall_seconds" else r["recipe"]["metrics"][field] for r in rows]
            summary[field] = {"values": values, "median": statistics.median(values),
                              "min": min(values), "max": max(values)}
            if len(values) >= 20:
                summary[field]["p95_nearest_rank"] = sorted(values)[math.ceil(.95 * len(values)) - 1]
        summaries.append(summary)
    print(json.dumps({"successful_groups": summaries, "failures": failed,
                      "quality_scores": None,
                      "note": "Descriptive observations only. No automatic aesthetic quality score."}, indent=2))


if __name__ == "__main__":
    main()