| |
| """Build the compact 21-model accuracy CSV from retained audited values.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import json |
| import os |
| import tempfile |
| from pathlib import Path |
|
|
|
|
| FIELDS = [ |
| "model_id", "model_name", "metric", "fp32", "quantized", "delta", |
| "published_comparison", |
| ] |
|
|
|
|
| def load_rows(root: Path) -> list[dict[str, str]]: |
| source = root / "reports/accuracy/model_accuracy_values.json" |
| payload = json.loads(source.read_text(encoding="utf-8")) |
| if payload.get("schema_version") != "1.0": |
| raise ValueError("unsupported model accuracy value schema") |
| records = payload.get("models") |
| if not isinstance(records, list) or len(records) != 21: |
| raise ValueError("model accuracy source must contain 21 records") |
| rows = [{field: str(record.get(field, "")) for field in FIELDS} for record in records] |
| for row in rows: |
| if not row["published_comparison"].strip(): |
| row["published_comparison"] = "공개 수치 없음." |
| registry = { |
| row["model_id"]: row |
| for row in csv.DictReader((root / "model_registry.csv").open(newline="", encoding="utf-8")) |
| if row["eligibility"] == "ELIGIBLE" |
| } |
| if {row["model_id"] for row in rows} != set(registry): |
| raise ValueError("accuracy source model IDs differ from active registry") |
| for row in rows: |
| if row["model_name"] != registry[row["model_id"]]["model_name"]: |
| raise ValueError(f"model name differs from registry: {row['model_id']}") |
| return rows |
|
|
|
|
| def atomic_write_csv(path: Path, rows: list[dict[str, str]]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| fd, temporary = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=path.parent) |
| try: |
| with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle: |
| writer = csv.DictWriter(handle, fieldnames=FIELDS, lineterminator="\n") |
| writer.writeheader() |
| writer.writerows(rows) |
| os.replace(temporary, path) |
| except BaseException: |
| Path(temporary).unlink(missing_ok=True) |
| raise |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--repo-root", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| args = parser.parse_args() |
| root = args.repo_root.resolve() |
| output = args.output if args.output.is_absolute() else root / args.output |
| records = load_rows(root) |
| atomic_write_csv(output, records) |
| print(json.dumps({ |
| "status": "PASS", |
| "output": str(output), |
| "row_count": len(records), |
| "source": "reports/accuracy/model_accuracy_values.json", |
| "model_runtime_executed": False, |
| "dataset_evaluation_executed": False, |
| }, sort_keys=True)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
