scripts/build_netron_report.py

#!/usr/bin/env python3
"""Build the ONNX Netron report, model matrix, and gallery."""

from __future__ import annotations

import argparse
import csv
import html
import json
import os
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

from netron_capture_common import REPO_ROOT, atomic_csv, atomic_json, load_csv, relative, resolve, sha256


MODEL_FIELDS = [
    "model_id",
    "task",
    "task_group",
    "architecture_family",
    "format",
    "pair_netron_status",
    "fp32_onnx_status",
    "fp32_onnx_png",
    "public_quantized_onnx_status",
    "public_quantized_onnx_png",
]


def markdown_link(report_dir: Path, root: Path, row: dict[str, str]) -> str:
    if row["capture_status"] != "PASS":
        return f"`{row['capture_status']}`"
    path = resolve(root, row["output_png"])
    target = Path(os.path.relpath(path, report_dir)).as_posix()
    return f"[PNG]({target}) {row['output_png_width']}×{row['output_png_height']}"


def _task_summary(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
    grouped: dict[str, list[dict[str, str]]] = defaultdict(list)
    for row in rows:
        grouped[row["task_group"]].append(row)
    result = []
    for task_group, values in sorted(grouped.items()):
        result.append(
            {
                "task_group": task_group,
                "models": len({row["model_id"] for row in values}),
                "slots": len(values),
                "pass": sum(row["capture_status"] == "PASS" for row in values),
                "onnx_pass": sum(row["capture_status"] == "PASS" and row["format"] == "onnx" for row in values),
            }
        )
    return result


def _model_matrix(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
    by_key = {(row["model_id"], row["variant"], row["format"]): row for row in rows}
    result = []
    for model_id in sorted({row["model_id"] for row in rows}):
        sample = next(row for row in rows if row["model_id"] == model_id)
        slots = {
            variant: by_key[(model_id, variant, "onnx")]
            for variant in ("fp32", "public_quantized")
        }
        pair_pass = all(slots[variant]["capture_status"] == "PASS" for variant in slots)
        result.append(
            {
                "model_id": model_id,
                "task": sample["task"],
                "task_group": sample["task_group"],
                "architecture_family": sample["architecture_family"],
                "format": "onnx",
                "pair_netron_status": "PASS" if pair_pass else "FAIL",
                "fp32_onnx_status": slots["fp32"]["capture_status"],
                "fp32_onnx_png": slots["fp32"]["output_png"],
                "public_quantized_onnx_status": slots["public_quantized"]["capture_status"],
                "public_quantized_onnx_png": slots["public_quantized"]["output_png"],
            }
        )
    return result


def build_report(root: Path, report_dir: Path) -> dict[str, Any]:
    rows = load_csv(report_dir / "netron_capture_inventory.csv")
    input_rows = load_csv(report_dir / "netron_input_inventory.csv")
    model_rows = _model_matrix(rows)
    atomic_csv(report_dir / "netron_model_matrix.csv", model_rows, MODEL_FIELDS)
    task_rows = _task_summary(rows)
    pass_rows = [row for row in rows if row["capture_status"] == "PASS"]
    missing_rows = [row for row in rows if row["capture_status"] != "PASS"]
    canonical_rows = [row for row in rows if row["canonical_s7_selected"].lower() == "true"]
    dimensions = [(int(row["output_png_width"]), int(row["output_png_height"])) for row in pass_rows]
    summary = {
        "schema_version": "1.0",
        "stage": "T80_NETRON_VISUALIZATION",
        "status": "PASS",
        "failure_code": None,
        "result_interpretation": "all 42 FP32/quantized ONNX variants exported",
        "counts": {
            "models": len(model_rows),
            "theoretical_slots": len(rows),
            "source_artifacts_available": sum(row["artifact_status"] == "AVAILABLE" for row in input_rows),
            "netron_exports_pass": len(pass_rows),
            "onnx_exports_pass": sum(row["format"] == "onnx" for row in pass_rows),
            "not_available": len(missing_rows),
            "canonical_pair_exports_pass": sum(row["capture_status"] == "PASS" for row in canonical_rows),
            "canonical_pair_exports_expected": len(canonical_rows),
            "ui_proof_images": len(pass_rows),
            "metadata_records": len(rows),
            "total_netron_export_bytes": sum(int(row["output_png_bytes"]) for row in pass_rows),
            "minimum_export_width": min(width for width, _ in dimensions),
            "maximum_export_width": max(width for width, _ in dimensions),
            "minimum_export_height": min(height for _, height in dimensions),
            "maximum_export_height": max(height for _, height in dimensions),
        },
        "tool_versions": {
            "netron": sorted({row["netron_version"] for row in rows}),
            "playwright": sorted({row["playwright_version"] for row in rows}),
            "chromium": sorted({row["chromium_version"] for row in rows}),
        },
        "task_coverage": task_rows,
        "not_available": [
            {
                "model_id": row["model_id"],
                "variant": row["variant"],
                "format": row["format"],
                "failure_code": row["failure_code"],
                "failure_detail": row["failure_detail"],
                "production_stage_status": row["production_stage_status"],
                "production_failure_code": row["production_failure_code"],
                "production_command": row["production_command"],
                "production_stdout_log": row["production_stdout_log"],
                "production_stderr_log": row["production_stderr_log"],
            }
            for row in missing_rows
        ],
        "policy": {
            "netron_layout_used_as_execution_order": False,
            "conversion_or_converter_retry_performed": False,
            "model_weight_architecture_modified": False,
            "allocator_work_performed": False,
            "prohibited_operations_performed": [],
        },
    }
    atomic_json(report_dir / "netron_capture_summary.json", summary)

    by_key = {(row["model_id"], row["variant"], row["format"]): row for row in rows}
    lines = [
        "# Netron ONNX 그래프",
        "",
        "## 결론",
        "",
        "21개 모델의 FP32·공개 양자화 ONNX 42개를 Netron PNG로 생성했다.",
        "",
        "## 캡처 방식과 범위",
        "",
        "- 변환이 끝난 `.onnx` 파일을 Netron에 직접 열었다.",
        "- 전체 graph 그림은 Netron 브라우저의 `Export as PNG` (`Control+Shift+E`)를 사용했다. `*_netron_ui.png`는 실제 UI load 증빙용 viewport screenshot이다.",
        "- ONNX FP32와 공개 양자화 모델을 공통 비교 pair로 사용한다.",
        "",
        "## Coverage",
        "",
        "| Task | 모델 | ONNX variant | PASS |",
        "|---|---:|---:|---:|",
    ]
    for row in task_rows:
        lines.append(
            f"| {row['task_group']} | {row['models']} | {row['slots']} | {row['pass']} |"
        )
    lines.extend(
        [
            "",
            "## 모델별 Netron export",
            "",
            "| 모델 | Task | FP32 ONNX | Q ONNX |",
            "|---|---|---|---|",
        ]
    )
    for model in model_rows:
        model_id = model["model_id"]
        cells = [
            markdown_link(report_dir, root, by_key[(model_id, "fp32", "onnx")]),
            markdown_link(report_dir, root, by_key[(model_id, "public_quantized", "onnx")]),
        ]
        lines.append(
            f"| {model_id} | {model['task_group']} | {' | '.join(cells)} |"
        )
    lines.extend(
        [
            "",
            "## 결과 파일",
            "",
            "- [Netron 전체 gallery](netron_gallery.html): UI 증빙 thumbnail과 full graph export 링크",
            "- [Netron model matrix](netron_model_matrix.csv): 모델별 FP32·양자화 ONNX",
            "- [Netron capture inventory](netron_capture_inventory.csv): source/result/log/checksum/dimension 전체",
            "- [독립 검증](validation.json), [artifact manifest](artifact_manifest.json), [checksum 목록](artifacts.sha256)",
            "",
            "Netron 그림과 capture inventory가 이 단계의 결과다.",
            "",
            "## 재현 명령",
            "",
            "```bash",
            "bash environment/visualization/netron/bootstrap.sh",
            "PLAYWRIGHT_BROWSERS_PATH=environment/visualization/netron/browsers \\",
            "  environment/visualization/netron/.venv/bin/python scripts/capture_netron_graphs.py \\",
            "  --run-dir logs/graphs/netron/full_20260807_attempt3_resume_provenance --scope full",
            ".venv/bin/python scripts/run_netron_checks.py \\",
            "  --run-dir logs/graphs/netron/final_20260807",
            "```",
        ]
    )
    (report_dir / "netron_capture_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")

    cards = []
    for row in sorted(rows, key=lambda value: (value["model_id"], value["variant"], value["format"])):
        title = f"{row['model_id']} · {row['variant']} · {row['format'].upper()}"
        if row["capture_status"] == "PASS":
            full_path = Path(os.path.relpath(resolve(root, row["output_png"]), report_dir)).as_posix()
            ui_path = Path(os.path.relpath(resolve(root, row["ui_proof_png"]), report_dir)).as_posix()
            metadata_path = Path(os.path.relpath(resolve(root, row["metadata_json"]), report_dir)).as_posix()
            cards.append(
                f'<article class="card"><h2>{html.escape(title)}</h2>'
                f'<a href="{html.escape(full_path)}"><img loading="lazy" src="{html.escape(ui_path)}" alt="{html.escape(title)} Netron UI"></a>'
                f'<p><strong>PASS</strong> · Netron export {row["output_png_width"]}×{row["output_png_height"]} · nodes {row["graph_node_count"]}</p>'
                f'<p><a href="{html.escape(full_path)}">full Netron PNG</a> · <a href="{html.escape(ui_path)}">UI proof</a> · <a href="{html.escape(metadata_path)}">metadata</a></p></article>'
            )
        else:
            cards.append(
                f'<article class="card missing"><h2>{html.escape(title)}</h2><p><strong>{html.escape(row["capture_status"])}</strong></p>'
                f'<p>{html.escape(row["failure_code"])}: {html.escape(row["failure_detail"])}</p></article>'
            )
    gallery = """<!doctype html>
<html lang="ko"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Netron graph gallery</title><style>
body{font-family:system-ui,sans-serif;margin:24px;background:#f5f6f8;color:#202124}header{max-width:1100px;margin:auto auto 24px}.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(360px,1fr));gap:18px}.card{background:white;border:1px solid #dfe1e5;border-radius:10px;padding:14px;box-shadow:0 1px 3px #0001}.card h2{font-size:17px;margin:0 0 12px}.card img{display:block;width:100%;height:280px;object-fit:contain;background:#fff;border:1px solid #eee}.card p{font-size:13px;overflow-wrap:anywhere}.missing{border-color:#c5221f;background:#fff8f7}a{color:#185abc}</style></head><body>
<header><h1>실제 Netron graph gallery</h1><p>각 thumbnail은 Netron UI 증빙 screenshot이며, <b>full Netron PNG</b> 링크가 Netron 자체 전체 graph export다. 배치는 실행 순서를 의미하지 않는다.</p>
<p><a href="netron_capture_report.md">보고서</a> · <a href="netron_capture_inventory.csv">machine-readable inventory</a></p></header><main class="grid">
""" + "\n".join(cards) + "\n</main></body></html>\n"
    (report_dir / "netron_gallery.html").write_text(gallery, encoding="utf-8")

    readme = """# Netron ONNX graphs

[netron_capture_report.md](netron_capture_report.md)에서 모델별 FP32·양자화 그래프를 확인한다.

- `*_netron.png`: Netron 9.2.0 자체 전체 graph PNG export
- `*_netron_ui.png`: Netron UI load 증빙 screenshot
- `netron_capture_inventory.csv`: source/checksum/command/log/load/PNG 증빙
"""
    (report_dir / "README.md").write_text(readme, encoding="utf-8")
    return summary


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=REPO_ROOT)
    parser.add_argument("--report-dir", type=Path, default=Path("reports/graphs/netron"))
    args = parser.parse_args()
    root = args.repo_root.resolve()
    report_dir = args.report_dir if args.report_dir.is_absolute() else root / args.report_dir
    report_dir.mkdir(parents=True, exist_ok=True)
    summary = build_report(root, report_dir)
    print(json.dumps({"status": summary["status"], **summary["counts"]}, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())