scripts/summarize_math_ink_06_android_benchmarks.py

"""AIFlow Math Ink 0.6의 Android low/mid/high 실기기 benchmark를 AND gate로 요약한다."""

from __future__ import annotations

import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
from typing import Any, Sequence


REQUIRED_TIERS06 = frozenset({"low", "mid", "high"})


def model_bundle_sha25606(online_sha256: str, raster_sha256: str) -> str:
    """필요 변수: 두 flatbuffer hash. 작동 원리: Android와 같은 순서·UTF-8 계약으로 bundle 지문을 만든다."""

    payload = f"online:{online_sha256}\nraster:{raster_sha256}\n".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def summarize_android_benchmarks06(
    reports: Sequence[dict[str, Any]],
) -> dict[str, Any]:
    """필요 변수: 세 기기 benchmark report. 작동 원리: 동일 모델·고유 tier·전 metric gate를 AND로 검증한다."""

    tiers = [str(report.get("device_tier") or "") for report in reports]
    if len(reports) != 3 or set(tiers) != REQUIRED_TIERS06 or len(set(tiers)) != 3:
        raise ValueError("Android benchmark는 low·mid·high report가 정확히 하나씩 필요합니다.")
    if any(
        report.get("schema") != "aiflow-math-ink-06-android-benchmark-v1"
        for report in reports
    ):
        raise ValueError("지원하지 않는 Android benchmark schema입니다.")
    versions = {str(report.get("model_version") or "") for report in reports}
    online_hashes = {str(report.get("online_model_sha256") or "") for report in reports}
    raster_hashes = {str(report.get("raster_model_sha256") or "") for report in reports}
    bundle_hashes = {str(report.get("model_bundle_sha256") or "") for report in reports}
    if len(versions) != 1 or "" in versions:
        raise ValueError("세 기기의 model_version이 동일하고 비어 있지 않아야 합니다.")
    for name, hashes in (
        ("online", online_hashes),
        ("raster", raster_hashes),
        ("bundle", bundle_hashes),
    ):
        if (
            len(hashes) != 1
            or "" in hashes
            or re.fullmatch(r"[0-9a-f]{64}", next(iter(hashes))) is None
        ):
            raise ValueError(f"세 기기의 {name} SHA-256이 동일하고 유효해야 합니다.")
    expected_bundle = model_bundle_sha25606(
        next(iter(online_hashes)),
        next(iter(raster_hashes)),
    )
    if next(iter(bundle_hashes)) != expected_bundle:
        raise ValueError("model_bundle_sha256이 online/raster ordered pair와 일치하지 않습니다.")
    ordered = sorted(reports, key=lambda report: ("low", "mid", "high").index(report["device_tier"]))
    tier_rows: dict[str, Any] = {}
    for report in ordered:
        checks = report.get("checks") or {}
        required_checks = {
            "online_p95",
            "raster_p95",
            "peak_memory",
            "battery_measurement",
        }
        if set(checks) != required_checks:
            raise ValueError(f"{report['device_tier']} report의 metric check 계약이 다릅니다.")
        if report.get("product_validation") is not False:
            raise ValueError("개별 benchmark가 임의로 product_validation을 true로 만들 수 없습니다.")
        tier_rows[report["device_tier"]] = {
            "device": report["device"],
            "online_p95_ms": float(report["online"]["p95_ms"]),
            "raster_p95_ms": float(report["raster"]["p95_ms"]),
            "peak_pss_bytes": int(report["peak_pss_bytes"]),
            "battery_charge_delta_micro_ah": report["battery_charge_delta_micro_ah"],
            "checks": checks,
            "passed": bool(report.get("gate_passed") and all(checks.values())),
        }
    all_passed = all(row["passed"] for row in tier_rows.values())
    return {
        "schema": "aiflow-math-ink-06-android-3tier-summary-v1",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "model_version": next(iter(versions)),
        "online_model_sha256": next(iter(online_hashes)),
        "raster_model_sha256": next(iter(raster_hashes)),
        "model_bundle_sha256": next(iter(bundle_hashes)),
        "tiers": tier_rows,
        "android_hardware_validation": all_passed,
        "android_release_gate_passed": all_passed,
        "product_validation": False,
        "next_gate": (
            "P writer/device-disjoint model gate와 Android gate를 함께 release manifest에서 결합"
            if all_passed
            else "실패 tier의 latency·memory·battery를 개선한 뒤 같은 model SHA로 재측정"
        ),
    }


def main() -> None:
    """필요 변수: tier별 UTF-8 JSON report·출력. 작동 원리: 검증된 3-tier summary를 원자적으로 기록한다."""

    parser = argparse.ArgumentParser(description="Summarize Math Ink 0.6 Android benchmarks")
    parser.add_argument("--report", type=Path, action="append", required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    reports = [
        json.loads(path.read_text(encoding="utf-8"))
        for path in args.report
    ]
    summary = summarize_android_benchmarks06(reports)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    temporary = args.output.with_suffix(args.output.suffix + ".part")
    temporary.write_text(
        json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    temporary.replace(args.output)
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()