File size: 5,572 Bytes
a16cd95
 
 
 
 
 
7d39c29
a16cd95
 
 
 
 
 
 
 
 
7d39c29
 
 
 
 
 
 
a16cd95
 
 
 
 
 
 
 
 
 
 
 
 
 
7d39c29
 
 
a16cd95
 
7d39c29
 
 
 
a16cd95
7d39c29
 
 
 
 
 
 
 
 
 
 
 
a16cd95
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7d39c29
 
 
a16cd95
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""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()