| |
| """Compare frozen/trainable transfer-forward artifacts.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
|
|
| import numpy as np |
|
|
|
|
| def compare(left: np.ndarray, right: np.ndarray) -> dict[str, object]: |
| delta = np.abs(left.astype(np.float64) - right.astype(np.float64)) |
| passed = np.isclose(left, right, atol=1e-5, rtol=1e-5) |
| return { |
| "status": "PASS" if bool(np.all(passed)) else "FAIL", |
| "shape": list(left.shape), |
| "max_abs": float(delta.max()), |
| "outside_tolerance": int((~passed).sum()), |
| } |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("legacy_frozen", type=Path) |
| parser.add_argument("rewrite_frozen", type=Path) |
| parser.add_argument("legacy_trainable", type=Path) |
| parser.add_argument("rewrite_trainable", type=Path) |
| parser.add_argument("output", type=Path) |
| args = parser.parse_args() |
| reports = {} |
| for name, left_path, right_path in ( |
| ("frozen_backbone", args.legacy_frozen, args.rewrite_frozen), |
| ("trainable_backbone", args.legacy_trainable, args.rewrite_trainable), |
| ): |
| with np.load(left_path) as left, np.load(right_path) as right: |
| reports[name] = compare(left["logits"], right["logits"]) |
| report = { |
| "schema_version": 1, |
| "stages": reports, |
| "overall_status": "PASS" |
| if all(item["status"] == "PASS" for item in reports.values()) |
| else "FAIL", |
| } |
| args.output.mkdir(parents=True, exist_ok=True) |
| (args.output / "transfer.json").write_text( |
| json.dumps(report, indent=2, sort_keys=True) + "\n" |
| ) |
| print(json.dumps({"overall_status": report["overall_status"]}, indent=2)) |
| return 0 if report["overall_status"] == "PASS" else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
