| """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() |
|
|
