| |
| """Build the checksum-guarded 42-model-variant inventory for Netron capture.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from collections import Counter |
| from pathlib import Path |
|
|
| from netron_capture_common import INVENTORY_FIELDS, REPO_ROOT, atomic_csv, atomic_json, discover_slots |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--repo-root", type=Path, default=REPO_ROOT) |
| parser.add_argument( |
| "--output", type=Path, default=Path("reports/graphs/netron/netron_input_inventory.csv") |
| ) |
| parser.add_argument( |
| "--summary", type=Path, default=Path("reports/graphs/netron/netron_input_summary.json") |
| ) |
| args = parser.parse_args() |
| root = args.repo_root.resolve() |
| output = args.output if args.output.is_absolute() else root / args.output |
| summary_path = args.summary if args.summary.is_absolute() else root / args.summary |
| rows = discover_slots(root) |
| counts = { |
| "slots": len(rows), |
| "models": len({row["model_id"] for row in rows}), |
| "available": sum(row["artifact_status"] == "AVAILABLE" for row in rows), |
| "not_available": sum(row["artifact_status"] == "NOT_AVAILABLE" for row in rows), |
| "checksum_mismatch": sum(row["artifact_status"] == "BLOCKED_CHECKSUM_MISMATCH" for row in rows), |
| "canonical_s7_selected": sum(row["canonical_s7_selected"] for row in rows), |
| "available_by_format": dict(Counter(row["format"] for row in rows if row["artifact_status"] == "AVAILABLE")), |
| "validation_status": dict(Counter(row["validation_stage_status"] for row in rows)), |
| } |
| scalar_ok = ( |
| counts["slots"] == 42 |
| and counts["models"] == 21 |
| and counts["available"] == 42 |
| and counts["not_available"] == 0 |
| and counts["checksum_mismatch"] == 0 |
| and counts["canonical_s7_selected"] == 42 |
| ) |
| status = "PASS" if scalar_ok else "FAIL" |
| atomic_csv(output, rows, INVENTORY_FIELDS) |
| atomic_json( |
| summary_path, |
| { |
| "schema_version": "1.0", |
| "stage": "T80_NETRON_INPUT_INVENTORY", |
| "status": status, |
| "failure_code": None if status == "PASS" else "FAIL_ANALYSIS", |
| "counts": counts, |
| "not_available": [ |
| {key: row[key] for key in ("model_id", "variant", "format", "pipeline_stage_status", "production_stage_status", "production_failure_code", "validation_stage_status", "validation_failure_code", "source_artifact")} |
| for row in rows |
| if row["artifact_status"] != "AVAILABLE" |
| ], |
| "policy": { |
| "tflite_to_onnx_for_netron": False, |
| "model_conversion_performed": False, |
| "model_weight_architecture_modified": False, |
| "allocator_work_performed": False, |
| }, |
| }, |
| ) |
| print(json.dumps({"status": status, **counts}, sort_keys=True)) |
| return 0 if status == "PASS" else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
