File size: 3,101 Bytes
ed3aeeb | 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 | #!/usr/bin/env python3
"""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())
|