"""Evaluate the recovered checkpoint under the documented 448px/4-degree protocol. This is a protocol audit, not a KD-ready cache generator. It evaluates all explicit head-order/sign candidates and reports the documented candidate separately. """ from __future__ import annotations import hashlib import json from pathlib import Path import sys ROOT = Path(r"E:\Gaze_estimation") sys.path.insert(0, str(ROOT / ".codex_deps")) sys.path.insert(0, str(ROOT)) import cv2 import h5py import numpy as np import torch from src.models.teacher_strict import file_sha256, load_teacher_model_strict from src.utils.preprocess import GazePreprocessor CLEAN = ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.full.h5" ORIGINAL = ROOT / "data" / "MPIIGaze" / "MPIIGaze" / "MPIIGaze" / "Data" / "Original" LANDMARK_MODEL = ROOT / "src" / "utils" / "face_landmarker.task" CHECKPOINT = ROOT / "checkpoints" / "resnet50.pt" OUT = ROOT / "artifacts" / "kd-teacher-trap-diagnostic" def decode(values: np.ndarray) -> list[str]: return [value.decode("utf-8") if isinstance(value, bytes) else str(value) for value in values] def face_crop(frame: np.ndarray, landmarks, size: int = 448) -> np.ndarray: height, width = frame.shape[:2] coordinates = np.array([[point.x * width, point.y * height] for point in landmarks]) minimum, maximum = coordinates.min(axis=0), coordinates.max(axis=0) center = (minimum + maximum) / 2.0 extent = float(np.max(maximum - minimum) * 1.5) x1, y1 = np.maximum(0, (center - extent / 2.0).astype(int)) x2, y2 = min(width, int(center[0] + extent / 2.0)), min(height, int(center[1] + extent / 2.0)) crop = frame[y1:y2, x1:x2] if crop.size == 0: raise RuntimeError("empty face crop") crop = cv2.resize(crop, (size, size)) crop = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0 crop = (crop - np.array([0.485, 0.456, 0.406], dtype=np.float32)) / np.array( [0.229, 0.224, 0.225], dtype=np.float32 ) return np.transpose(crop, (2, 0, 1)) def softmax(values: np.ndarray) -> np.ndarray: shifted = values.astype(np.float64) - values.max(axis=1, keepdims=True) exp = np.exp(shifted) return exp / exp.sum(axis=1, keepdims=True) def vectors(pitch_deg: np.ndarray, yaw_deg: np.ndarray) -> np.ndarray: pitch, yaw = np.deg2rad(pitch_deg), np.deg2rad(yaw_deg) return np.column_stack( (-np.cos(pitch) * np.sin(yaw), -np.sin(pitch), -np.cos(pitch) * np.cos(yaw)) ) def rotate_z(vectors_: np.ndarray, roll_deg: np.ndarray) -> np.ndarray: angle = np.deg2rad(roll_deg) cosine, sine = np.cos(angle), np.sin(angle) output = vectors_.copy() output[:, 0] = cosine * vectors_[:, 0] - sine * vectors_[:, 1] output[:, 1] = sine * vectors_[:, 0] + cosine * vectors_[:, 1] return output def angular_error(prediction: np.ndarray, target: np.ndarray) -> np.ndarray: prediction /= np.linalg.norm(prediction, axis=1, keepdims=True) target /= np.linalg.norm(target, axis=1, keepdims=True) return np.rad2deg(np.arccos(np.clip(np.sum(prediction * target, axis=1), -1.0, 1.0))) def main() -> None: with h5py.File(CLEAN, "r") as handle: paths = decode(handle["relative_frame_path"][:]) sample_ids = decode(handle["sample_id"][:]) left_roll = handle["left_roll_deg"][:].astype(np.float64) target_deg = np.rad2deg(handle["left_gaze"][:].astype(np.float64)) target_vector = vectors(target_deg[:, 0], target_deg[:, 1]) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model, loader_audit = load_teacher_model_strict(CHECKPOINT, device=device) preprocessor = GazePreprocessor(model_path=str(LANDMARK_MODEL)) fc_pitch_batches: list[np.ndarray] = [] fc_yaw_batches: list[np.ndarray] = [] batch: list[np.ndarray] = [] batch_size = 16 def infer_batch() -> None: if not batch: return tensor = torch.from_numpy(np.stack(batch)).to(device) with torch.inference_mode(): fc_pitch, fc_yaw = model(tensor) fc_pitch_batches.append(fc_pitch.detach().cpu().numpy().astype(np.float32)) fc_yaw_batches.append(fc_yaw.detach().cpu().numpy().astype(np.float32)) batch.clear() for index, relative in enumerate(paths): frame = cv2.imread(str(ORIGINAL / Path(relative))) if frame is None: raise RuntimeError(f"failed to decode mapped frame {relative}") landmarks = preprocessor.get_landmarks(frame) if landmarks is None: raise RuntimeError(f"mapped accepted frame no longer produces landmarks: {relative}") batch.append(face_crop(frame, landmarks, size=448)) if len(batch) == batch_size: infer_batch() if (index + 1) % 200 == 0: print(f"protocol448 {index + 1}/{len(paths)}", flush=True) infer_batch() fc_pitch = np.concatenate(fc_pitch_batches) fc_yaw = np.concatenate(fc_yaw_batches) npz_path = OUT / "p01_teacher_protocol_448_logits.npz" np.savez_compressed( npz_path, sample_id=np.asarray(sample_ids, dtype="S64"), relative_frame_path=np.asarray(paths, dtype="S128"), fc_pitch_logits=fc_pitch, fc_yaw_logits=fc_yaw, left_roll_deg=left_roll.astype(np.float32), target_pitch_yaw_rad=np.deg2rad(target_deg).astype(np.float32), ) probabilities = {"fc_pitch": softmax(fc_pitch), "fc_yaw": softmax(fc_yaw)} rows: list[dict] = [] for binwidth, offset in ((2.0, 90.0), (4.0, 180.0)): decoded = { name: probability @ np.arange(90, dtype=np.float64) * binwidth - offset for name, probability in probabilities.items() } for swap in (False, True): pitch_base = decoded["fc_yaw"] if swap else decoded["fc_pitch"] yaw_base = decoded["fc_pitch"] if swap else decoded["fc_yaw"] for pitch_sign in (-1, 1): for yaw_sign in (-1, 1): raw_vector = vectors(pitch_sign * pitch_base, yaw_sign * yaw_base) for apply_roll in (False, True): prediction = rotate_z(raw_vector, left_roll) if apply_roll else raw_vector.copy() error = angular_error(prediction, target_vector.copy()) rows.append( { "input_size": 448, "binwidth_deg": binwidth, "offset_deg": offset, "swap_fc_pitch_fc_yaw": swap, "pitch_sign": pitch_sign, "yaw_sign": yaw_sign, "apply_left_roll": apply_roll, "mean_3d_error_deg": float(error.mean()), "median_3d_error_deg": float(np.median(error)), } ) rows.sort(key=lambda row: row["mean_3d_error_deg"]) documented = next( row for row in rows if row["binwidth_deg"] == 4.0 and not row["swap_fc_pitch_fc_yaw"] and row["pitch_sign"] == 1 and row["yaw_sign"] == 1 and row["apply_left_roll"] ) report = { "status": "PROTOCOL_AUDIT_NOT_KD_CACHE", "participant": "p01", "samples": len(paths), "checkpoint_sha256": loader_audit.checkpoint_sha256, "strict_loader_inference_sha256": loader_audit.inference_sha256, "input_size": 448, "logits_npz": str(npz_path), "logits_npz_sha256": file_sha256(npz_path), "documented_named-head_4deg_roll_candidate": documented, "best_posthoc_candidate": rows[0], "top_ten_candidates": rows[:10], "warning": ( "Head order/sign search is post-hoc and cannot define a confirmatory protocol. " "The fc_yaw/fc_pitch checkpoint lineage and training-code semantics must be fixed before KD." ), } output = OUT / "p01_teacher_protocol_448_audit.json" output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8", newline="\n") print(json.dumps(report, indent=2)) if __name__ == "__main__": main()