Upload release_check.py with huggingface_hub

release_check.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Is this release deployable? Checks the checkpoint and the code set before anything is published.
+
+Two failures this exists to stop (both happened):
+
+* **A checkpoint the robot cannot use.** Without a condition head the safety gate
+  rejects every frame ("surface condition not assessed"); without a completion
+  threshold sign-off is disabled; without temperatures the confidence thresholds
+  mean nothing; without ``condition_operating_point`` damage flags run on
+  uncalibrated defaults that flag nearly every real photo.
+* **Code that does not import.** A hand-kept file list drifted behind ``predict.py``'s
+  imports. ``release_files()`` takes every tracked top-level module instead, and
+  ``missing_imports()`` proves each published module's local imports are published too.
+
+Usage::
+
+    python3 release_check.py --checkpoint artifacts/candidates/ft4-cal.pt     # exit 1 = not deployable
+"""
+from __future__ import annotations
+
+import argparse
+import ast
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+PROJECT = Path(__file__).resolve().parent
+# Data-collection and one-off scripts that are not part of the runtime kit.
+EXCLUDE = {"fetch_construction_data.py", "fetch_surface_eval.py"}
+EXTRA = ["requirements.txt", "metadata_template.csv", "console/index.html", "console/app.js", "console/style.css"]
+
+
+def tracked_files(project=PROJECT):
+    out = subprocess.run(["git", "ls-files"], cwd=project, capture_output=True, text=True, check=True).stdout
+    return set(out.split())
+
+
+def release_files(project=PROJECT):
+    """Every tracked top-level module (and its tests), minus data scripts, plus runtime assets."""
+    tracked = tracked_files(project)
+    modules = sorted(f for f in tracked if f.endswith(".py") and "/" not in f and f not in EXCLUDE)
+    return modules + [f for f in EXTRA if f in tracked]
+
+
+def missing_imports(files, project=PROJECT):
+    """{published module: [local modules it imports that are not published]}."""
+    local = {p.stem for p in project.glob("*.py")}
+    published = {Path(f).stem for f in files if f.endswith(".py")}
+    gaps = {}
+    for name in files:
+        if not name.endswith(".py"):
+            continue
+        tree = ast.parse((project / name).read_text(), filename=name)
+        imported = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                imported.update(alias.name.split(".")[0] for alias in node.names)
+            elif isinstance(node, ast.ImportFrom) and node.module and not node.level:
+                imported.add(node.module.split(".")[0])
+        missing = sorted((imported & local) - published)
+        if missing:
+            gaps[name] = missing
+    return gaps
+
+
+def check_checkpoint(checkpoint):
+    """``{"errors": [...], "warnings": [...]}`` for a loaded checkpoint dict."""
+    from label_schema import CLASSES
+    errors, warnings = [], []
+    if tuple(checkpoint.get("classes") or ()) != tuple(CLASSES):
+        errors.append("class schema does not match label_schema.CLASSES")
+    if not checkpoint.get("condition_classes"):
+        errors.append("no surface-condition head: the gate rejects every frame (condition not assessed)")
+    if checkpoint.get("temperature") is None:
+        errors.append("no temperature calibration: run train.py --calibrate-checkpoint")
+    if checkpoint.get("completion_threshold") is None:
+        errors.append("no completion threshold: sign-off is disabled (train.py --calibrate-checkpoint)")
+    point = checkpoint.get("condition_operating_point")
+    if checkpoint.get("condition_classes") and not point:
+        errors.append("no condition_operating_point: damage flags use uncalibrated defaults "
+                      "(calibrate_condition.py --write)")
+    elif point and point.get("false_alarm_set") != "real":
+        warnings.append("condition operating point was calibrated without real sound walls")
+    if "wallpaper" not in (checkpoint.get("material_classes") or []):
+        warnings.append("no wallpaper material class: wallpaper rests on the pattern cue alone")
+    return {"errors": errors, "warnings": warnings}
+
+
+def run(checkpoint_path=None, project=PROJECT):
+    report = {"files": release_files(project)}
+    report["missing_imports"] = missing_imports(report["files"], project)
+    report["errors"] = [f"{m} imports unpublished {', '.join(g)}" for m, g in report["missing_imports"].items()]
+    report["warnings"] = []
+    if checkpoint_path:
+        import torch
+        checkpoint = torch.load(checkpoint_path, map_location="cpu", weights_only=True)
+        result = check_checkpoint(checkpoint)
+        report["errors"] += result["errors"]
+        report["warnings"] += result["warnings"]
+        report["checkpoint"] = str(checkpoint_path)
+    report["deployable"] = not report["errors"]
+    return report
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--checkpoint", type=Path)
+    ap.add_argument("--json", action="store_true")
+    args = ap.parse_args()
+    report = run(args.checkpoint)
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print(f"{len(report['files'])} files in the release")
+        for e in report["errors"]:
+            print("ERROR  ", e)
+        for w in report["warnings"]:
+            print("warning", w)
+        print("deployable" if report["deployable"] else "NOT deployable")
+    sys.exit(0 if report["deployable"] else 1)
+
+
+if __name__ == "__main__":
+    main()