"""Read-only go/no-go audit for legacy caches before a clean KD experiment. This script never edits H5 data or checkpoints. It writes one diagnostic JSON report. """ from __future__ import annotations import hashlib import json import math from pathlib import Path import sys ROOT = Path(r"E:\Gaze_estimation") sys.path.insert(0, str(ROOT / ".codex_deps")) import h5py import numpy as np SUBJECTS = ("p01", "p08", "p11") OUT_DIR = ROOT / "artifacts" / "kd-teacher-trap-diagnostic" LOADER_AUDIT = OUT_DIR / "teacher_loader_key_audit.json" OUTPUT = OUT_DIR / "clean_kd_gate_audit.json" IDENTITY_FIELDS = ( "sample_id", "relative_frame_path", "participant", "day", "frame_id", "annotation_row", "raw_image_sha256", ) MODEL_FIELDS = ( "left_patches", "right_patches", "landmarks", "left_gaze", "right_gaze", "teacher_pitch_logits", "teacher_yaw_logits", ) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest().upper() def expectation_deg(logits: np.ndarray) -> np.ndarray: shifted = logits.astype(np.float64) - logits.max(axis=1, keepdims=True) probability = np.exp(shifted) probability /= probability.sum(axis=1, keepdims=True) return (probability * np.arange(90, dtype=np.float64)).sum(axis=1) * 2.0 - 90.0 def gaze_vectors(pitch_deg: np.ndarray, yaw_deg: np.ndarray) -> np.ndarray: pitch = np.deg2rad(pitch_deg) yaw = np.deg2rad(yaw_deg) return np.column_stack( (-np.cos(pitch) * np.sin(yaw), -np.sin(pitch), -np.cos(pitch) * np.cos(yaw)) ) def angular_error_deg(prediction: np.ndarray, target: np.ndarray) -> np.ndarray: dots = np.sum(prediction * target, axis=1) return np.rad2deg(np.arccos(np.clip(dots, -1.0, 1.0))) def audit_h5(subject: str) -> dict: path = ROOT / "data" / "processed" / f"{subject}_v16.h5" result = { "subject": subject, "path": str(path), "sha256": sha256(path), } with h5py.File(path, "r") as handle: keys = sorted(handle.keys()) result["datasets"] = keys result["missing_identity_fields"] = [name for name in IDENTITY_FIELDS if name not in handle] result["missing_model_fields"] = [name for name in MODEL_FIELDS if name not in handle] counts = {name: int(handle[name].shape[0]) for name in MODEL_FIELDS if name in handle} result["row_counts"] = counts result["row_counts_equal"] = len(set(counts.values())) == 1 result["teacher_logit_shapes_valid"] = all( name in handle and handle[name].ndim == 2 and handle[name].shape[1] == 90 for name in ("teacher_pitch_logits", "teacher_yaw_logits") ) finite = {} for name in MODEL_FIELDS: if name in handle: finite[name] = bool(np.isfinite(handle[name][:]).all()) result["all_numeric_values_finite"] = bool(finite) and all(finite.values()) result["finite_by_dataset"] = finite if not result["missing_model_fields"] and result["teacher_logit_shapes_valid"]: pitch = expectation_deg(handle["teacher_pitch_logits"][:]) yaw = expectation_deg(handle["teacher_yaw_logits"][:]) target_deg = np.rad2deg(handle["left_gaze"][:].astype(np.float64)) teacher_vec = gaze_vectors(pitch, yaw) target_vec = gaze_vectors(target_deg[:, 0], target_deg[:, 1]) error_3d = angular_error_deg(teacher_vec, target_vec) axis_error = np.abs(np.column_stack((pitch, yaw)) - target_deg).mean(axis=1) result["samples"] = int(len(error_3d)) result["teacher_axis_mae_deg"] = float(axis_error.mean()) result["teacher_3d_angular_error_deg"] = float(error_3d.mean()) result["teacher_3d_error_median_deg"] = float(np.median(error_3d)) result["teacher_3d_error_p90_deg"] = float(np.quantile(error_3d, 0.90)) result["cache_integrity_pass"] = bool( not result["missing_identity_fields"] and not result["missing_model_fields"] and result["row_counts_equal"] and result["teacher_logit_shapes_valid"] and result["all_numeric_values_finite"] ) return result def main() -> None: loader = json.loads(LOADER_AUDIT.read_text(encoding="utf-8")) loader_pass = bool( loader["historical_loader_loaded_key_count"] == loader["checkpoint_tensor_keys"] and loader["historical_loader_unexpected_key_count"] == 0 and set(loader["historical_loader_missing_keys"]) <= {"idx_tensor"} ) caches = [audit_h5(subject) for subject in SUBJECTS] cache_pass = all(item["cache_integrity_pass"] for item in caches) teacher_errors = [item.get("teacher_3d_angular_error_deg", math.nan) for item in caches] report = { "audit_type": "read-only clean-KD execution gate", "legacy_files_modified": False, "gate_a_loader": { "pass": loader_pass, "loaded_checkpoint_tensors": loader["historical_loader_loaded_key_count"], "checkpoint_tensors": loader["checkpoint_tensor_keys"], "unexpected_checkpoint_tensors": loader["historical_loader_unexpected_key_count"], "reason": "Recovered historical loader does not strictly load the trained backbone." if not loader_pass else "Strict loading requirements satisfied.", }, "gate_b_cache_integrity": { "pass": cache_pass, "required_identity_fields": list(IDENTITY_FIELDS), "subjects": caches, "reason": "Legacy caches have no row-level source identity/provenance fields." if not cache_pass else "Row-level provenance and numeric integrity requirements satisfied.", }, "gate_c_teacher_quality": { "pass": None, "status": "DESCRIPTIVE_ONLY", "mean_of_subjects_teacher_3d_angular_error_deg": float(np.nanmean(teacher_errors)), "reason": "A matched clean control is required to apply the preregistered relative-quality gate.", }, "decision": "STOP_BEFORE_TRAINING" if not (loader_pass and cache_pass) else "READY_FOR_TEACHER_QUALITY_GATE", "next_action": "Create a new strict-loader, manifest-backed cache; do not repair or overwrite legacy H5 files.", } OUTPUT.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print(json.dumps(report, indent=2)) if __name__ == "__main__": main()