File size: 7,210 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 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 | #!/usr/bin/env python3
"""Validate the independent T90 official-quality audit for ELIGIBLE vision models."""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import shlex
import sys
from collections import Counter
from pathlib import Path
from typing import Any
VISION_TASKS = {"vision_classification", "object_detection", "semantic_segmentation"}
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def md5_file(path: Path) -> str:
digest = hashlib.md5(usedforsecurity=False)
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def atomic_json(path: Path, value: Any) -> None:
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
temporary.replace(path)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project-root", type=Path, default=Path.cwd())
parser.add_argument("--audit", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
root = args.project_root.absolute()
audit_path = args.audit if args.audit.is_absolute() else root / args.audit
audit = json.loads(audit_path.read_text())
checks: list[dict[str, Any]] = []
def check(name: str, condition: bool, detail: Any) -> None:
checks.append({"name": name, "pass": bool(condition), "detail": detail})
with (root / "model_registry.csv").open(newline="") as handle:
registry = list(csv.DictReader(handle))
expected = {
row["model_id"]
for row in registry
if row["eligibility"] == "ELIGIBLE" and row["task"] in VISION_TASKS
}
models = audit["models"]
by_id = {row["model_id"]: row for row in models}
check("audit_ids_unique", len(by_id) == len(models), len(by_id))
check("all_eligible_vision_models", set(by_id) == expected, {"expected": sorted(expected), "actual": sorted(by_id)})
decisions = Counter(row["decision"] for row in models)
check("decision_vocabulary", set(decisions) <= {"EXECUTED_PASS", "EXECUTABLE", "EXTERNAL_BLOCKED"}, decisions)
check("summary_counts", audit["summary"]["models"] == len(models) == 16 and audit["summary"]["executed_pass"] == decisions["EXECUTED_PASS"] and audit["summary"]["executable_or_running"] == decisions["EXECUTABLE"] and audit["summary"]["external_blocked"] == decisions["EXTERNAL_BLOCKED"], {"summary": audit["summary"], "decisions": decisions})
blocked_missing_detail = [row["model_id"] for row in models if row["decision"] == "EXTERNAL_BLOCKED" and (not row.get("blocker") or not row.get("resolution"))]
check("blocked_resolution_conditions", not blocked_missing_detail, blocked_missing_detail)
check("segmentation_threshold_policy", all(by_id[model_id]["official_threshold"] is None and by_id[model_id]["expected_acceptance_status"] == "MEASURED_NO_ACCEPTANCE_THRESHOLD" for model_id in ["SG06", "SG07"]), {model_id: by_id[model_id] for model_id in ["SG06", "SG07"]})
executed_details: dict[str, Any] = {}
for model_id, threshold, expected_count in [("VC01", 0.8, 1000), ("VC02", 0.85, 200)]:
quality_root = root / "models" / "vision_classification" / model_id / "quality_evaluation"
summary = json.loads((quality_root / "results" / "quality_summary.json").read_text())
validation = json.loads((quality_root / "results" / "validation_report.json").read_text())
manifest = json.loads((quality_root / "execution_manifest.json").read_text())
check(f"{model_id}_summary_pass", summary["status"] == "PASS" and summary["acceptance_status"] == "PASS", {"status": summary["status"], "acceptance": summary["acceptance_status"]})
check(f"{model_id}_threshold", summary["threshold"] == threshold and summary["quality"]["public_int8"]["accuracy"] >= threshold, {"threshold": summary["threshold"], "accuracy": summary["quality"]["public_int8"]["accuracy"]})
check(f"{model_id}_generic_quality_fields", all(summary["quality"][variant]["format"] == "tflite" and summary["quality"][variant]["sample_count"] == expected_count for variant in ["fp32", "public_int8"]), summary["quality"])
check(f"{model_id}_resume", summary["resume"]["rows_reused"] == expected_count and summary["resume"]["rows_evaluated"] == 0, summary["resume"])
check(f"{model_id}_validator", validation["status"] == "PASS" and all(item["pass"] for item in validation["checks"]), {"status": validation["status"], "checks": len(validation["checks"])})
check(f"{model_id}_execution_manifest", manifest["overall"]["status"] == "PASS" and not manifest["overall"]["forbidden_operations_performed"], manifest["overall"])
check(f"{model_id}_prohibited_operations", not any(summary["prohibited_operations"].values()), summary["prohibited_operations"])
executed_details[model_id] = {"summary": summary["quality"], "validation_checks": len(validation["checks"]), "resume": summary["resume"]}
cifar = root / "research" / "downloads" / "cifar10" / "cifar-10-python.tar.gz"
transport = by_id["VC02"]["transport_provenance"]
check("vc02_transport_sha256_identity", sha256_file(cifar) == transport["downloaded_sha256"] == transport["mirror_lfs_pointer_sha256"] == transport["keras_v3_12_canonical_url_sha256"], sha256_file(cifar))
check("vc02_transport_md5_identity", md5_file(cifar) == transport["downloaded_md5"] == transport["official_page_md5"], md5_file(cifar))
check("vc02_canonical_source_unchanged", transport["canonical_source_changed"] is False and transport["canonical_url"] == "https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz", transport)
check("vc02_canonical_failure_logged", transport["canonical_probe_exit_code"] == 28 and (root / transport["canonical_probe_stdout"]).is_file() and (root / transport["canonical_probe_stderr"]).is_file(), {"exit_code": transport["canonical_probe_exit_code"]})
passed = all(item["pass"] for item in checks)
report = {
"stage": "T90_VISION_OFFICIAL_QUALITY_AUDIT_VALIDATION",
"status": "PASS" if passed else "FAIL",
"failure_code": None if passed else "FAIL_ANALYSIS",
"command": shlex.join([sys.executable, *sys.argv]),
"tool_versions": {"python": sys.version.split()[0]},
"audit": str(audit_path),
"audit_sha256": sha256_file(audit_path),
"eligible_vision_model_count": len(expected),
"decision_counts": dict(decisions),
"checks": checks,
"executed_details": executed_details,
}
output = args.output if args.output.is_absolute() else root / args.output
atomic_json(output, report)
print(json.dumps({"status": report["status"], "checks": len(checks), "passed": sum(item["pass"] for item in checks)}, sort_keys=True))
return 0 if passed else 1
if __name__ == "__main__":
raise SystemExit(main())
|