"""Generate a new manifest-backed, strictly loaded, roll-aligned KD cache. Legacy H5 files are never opened by this script. Existing outputs are never overwritten. Use bounded ``--max-*`` arguments for a smoke test before a full participant run. """ from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json from pathlib import Path import subprocess 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 audit_to_dict, file_sha256, load_teacher_model_strict from src.utils.preprocess import GazePreprocessor DATASET_ROOT = ROOT / "data" / "MPIIGaze" / "MPIIGaze" / "MPIIGaze" ORIGINAL_ROOT = DATASET_ROOT / "Data" / "Original" MANIFEST_ROOT = ROOT / "data" / "processed_kd_clean_v1" / "manifests" OUTPUT_ROOT = ROOT / "data" / "processed_kd_clean_v1" / "cache" LANDMARK_MODEL = ROOT / "src" / "utils" / "face_landmarker.task" CHECKPOINT = ROOT / "checkpoints" / "resnet50.pt" BIN_CENTERS_DEG = np.arange(90, dtype=np.float64) * 4.0 - 180.0 def text_sha256(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest().upper() def softmax(values: np.ndarray) -> np.ndarray: shifted = values.astype(np.float64) - np.max(values) exp = np.exp(shifted) return exp / exp.sum() def gaze_vector(pitch_rad: float, yaw_rad: float) -> np.ndarray: return np.array( [ -np.cos(pitch_rad) * np.sin(yaw_rad), -np.sin(pitch_rad), -np.cos(pitch_rad) * np.cos(yaw_rad), ], dtype=np.float64, ) def angular_error_deg(first: np.ndarray, second: np.ndarray) -> float: first = first / np.linalg.norm(first) second = second / np.linalg.norm(second) return float(np.rad2deg(np.arccos(np.clip(np.dot(first, second), -1.0, 1.0)))) def rotation_matrix_z(angle_deg: float) -> np.ndarray: angle = np.deg2rad(angle_deg) cosine, sine = np.cos(angle), np.sin(angle) return np.array(((cosine, -sine, 0.0), (sine, cosine, 0.0), (0.0, 0.0, 1.0))) def angles_from_vectors(vectors: np.ndarray) -> tuple[np.ndarray, np.ndarray]: pitch = np.arcsin(np.clip(-vectors[:, 1], -1.0, 1.0)) yaw = np.arctan2(-vectors[:, 0], -vectors[:, 2]) return np.rad2deg(pitch), np.rad2deg(yaw) def bin_indices(angle_deg: np.ndarray) -> np.ndarray: return np.clip(np.rint((angle_deg + 90.0) / 2.0), 0, 89).astype(np.int64) def make_teacher_grid() -> np.ndarray: pitch, yaw = np.meshgrid(BIN_CENTERS_DEG, BIN_CENTERS_DEG, indexing="ij") pitch_rad, yaw_rad = np.deg2rad(pitch.ravel()), np.deg2rad(yaw.ravel()) return np.column_stack( ( -np.cos(pitch_rad) * np.sin(yaw_rad), -np.sin(pitch_rad), -np.cos(pitch_rad) * np.cos(yaw_rad), ) ) TEACHER_GRID = make_teacher_grid() def roll_align_teacher_distribution( pitch_logits: np.ndarray, yaw_logits: np.ndarray, roll_deg: float ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: """Rotate the independent pitch/yaw joint distribution into the eye-normalized frame.""" pitch_probability = softmax(pitch_logits) yaw_probability = softmax(yaw_logits) joint = np.outer(pitch_probability, yaw_probability).ravel() rotated = TEACHER_GRID @ rotation_matrix_z(roll_deg).T rotated_pitch, rotated_yaw = angles_from_vectors(rotated) pitch_marginal = np.bincount(bin_indices(rotated_pitch), weights=joint, minlength=90) yaw_marginal = np.bincount(bin_indices(rotated_yaw), weights=joint, minlength=90) pitch_marginal /= pitch_marginal.sum() yaw_marginal /= yaw_marginal.sum() aligned_logits = ( np.log(np.maximum(pitch_marginal, 1e-30)).astype(np.float32), np.log(np.maximum(yaw_marginal, 1e-30)).astype(np.float32), ) expected_vector = (rotated * joint[:, None]).sum(axis=0) expected_vector /= np.linalg.norm(expected_vector) return aligned_logits[0], aligned_logits[1], expected_vector def face_crop(frame: np.ndarray, landmarks, target_size: tuple[int, int] = (448, 448)) -> np.ndarray | None: height, width = frame.shape[:2] coordinates = np.array([[point.x * width, point.y * height] for point in landmarks]) minimum = coordinates.min(axis=0) maximum = coordinates.max(axis=0) center = (minimum + maximum) / 2.0 size = float(np.max(maximum - minimum) * 1.5) x1, y1 = np.maximum(0, (center - size / 2.0).astype(int)) x2 = min(width, int(center[0] + size / 2.0)) y2 = min(height, int(center[1] + size / 2.0)) crop = frame[y1:y2, x1:x2] if crop.size == 0: return None crop = cv2.resize(crop, target_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 normalize_eye_with_matrix( frame: np.ndarray, landmarks, indices: list[int], preprocessor: GazePreprocessor, target_size: tuple[int, int] = (64, 32) ) -> tuple[np.ndarray, float, np.ndarray]: height, width = frame.shape[:2] first = np.array([landmarks[indices[0]].x * width, landmarks[indices[0]].y * height]) second = np.array([landmarks[indices[1]].x * width, landmarks[indices[1]].y * height]) center = (first + second) / 2.0 delta = second - first angle = float(np.degrees(np.arctan2(delta[1], delta[0]))) scale = (target_size[0] * 0.7) / (np.linalg.norm(delta) + 1e-6) matrix = cv2.getRotationMatrix2D(tuple(center), angle, scale) matrix[0, 2] += target_size[0] / 2.0 - center[0] matrix[1, 2] += target_size[1] / 2.0 - center[1] normalized = cv2.warpAffine(frame, matrix, target_size, flags=cv2.INTER_CUBIC) normalized = cv2.cvtColor(normalized, cv2.COLOR_BGR2GRAY) normalized = cv2.medianBlur(normalized, 3) normalized = preprocessor.clahe.apply(normalized) return normalized, angle, matrix.astype(np.float32) def git_commit() -> str: try: return subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, stderr=subprocess.DEVNULL ).strip() except Exception: return "UNAVAILABLE" def append_or_reject(records: list[dict], row: dict, accepted: bool, reason: str, output_index: int | None) -> None: records.append( { "source_index": row["source_index"], "sample_id": row["sample_id"], "relative_frame_path": row["relative_frame_path"], "accepted": accepted, "rejection_reason": reason, "output_index": output_index, } ) def create_h5(path: Path, accepted: list[dict], attributes: dict) -> None: string = h5py.string_dtype(encoding="utf-8") with h5py.File(path, "x") as handle: for key, value in attributes.items(): handle.attrs[key] = value if isinstance(value, (str, int, float, bool)) else json.dumps(value, sort_keys=True) text_fields = ("sample_id", "relative_frame_path", "participant", "day", "frame_id", "raw_image_sha256") for field in text_fields: handle.create_dataset(field, data=np.array([row[field] for row in accepted], dtype=object), dtype=string) handle.create_dataset("source_index", data=np.array([row["source_index"] for row in accepted], dtype=np.int64)) handle.create_dataset("annotation_row", data=np.array([row["annotation_row"] for row in accepted], dtype=np.int32)) numeric_fields = ( "left_patches", "right_patches", "landmarks", "left_gaze", "right_gaze", "left_affine_matrix", "right_affine_matrix", "left_roll_deg", "right_roll_deg", "teacher_pitch_logits_raw", "teacher_yaw_logits_raw", "teacher_pitch_logits", "teacher_yaw_logits", "teacher_aligned_vector", "teacher_pitch_deg", "teacher_yaw_deg", "teacher_error_deg", ) for field in numeric_fields: handle.create_dataset(field, data=np.asarray([row[field] for row in accepted]), compression="gzip") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--participant", required=True, choices=[f"p{i:02d}" for i in range(15)]) parser.add_argument("--max-source-rows", type=int) parser.add_argument("--max-accepted", type=int) parser.add_argument("--tag", default="full") parser.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") args = parser.parse_args() manifest_path = MANIFEST_ROOT / f"{args.participant}.source.jsonl" summary_path = MANIFEST_ROOT / f"{args.participant}.source.summary.json" validation_path = MANIFEST_ROOT / f"{args.participant}.source.validation.json" summary = json.loads(summary_path.read_text(encoding="utf-8")) validation = json.loads(validation_path.read_text(encoding="utf-8")) if not validation.get("pass") or file_sha256(manifest_path) != summary["manifest_sha256"]: raise RuntimeError("source manifest is not validated or its hash changed") OUTPUT_ROOT.mkdir(parents=True, exist_ok=True) output_path = OUTPUT_ROOT / f"{args.participant}.{args.tag}.h5" decision_path = OUTPUT_ROOT / f"{args.participant}.{args.tag}.processing.jsonl" summary_output = OUTPUT_ROOT / f"{args.participant}.{args.tag}.summary.json" for path in (output_path, decision_path, summary_output): if path.exists(): raise FileExistsError(f"refusing to overwrite clean artifact: {path}") teacher, teacher_audit = load_teacher_model_strict(CHECKPOINT, device=args.device) preprocessor = GazePreprocessor(model_path=str(LANDMARK_MODEL)) manifest_rows = manifest_path.read_text(encoding="utf-8").splitlines() accepted: list[dict] = [] decisions: list[dict] = [] examined = 0 for encoded in manifest_rows: if args.max_source_rows is not None and examined >= args.max_source_rows: break if args.max_accepted is not None and len(accepted) >= args.max_accepted: break row = json.loads(encoded) examined += 1 image_path = ORIGINAL_ROOT / Path(row["relative_frame_path"]) if file_sha256(image_path) != row["raw_image_sha256"]: append_or_reject(decisions, row, False, "raw_image_hash_mismatch", None) continue frame = cv2.imread(str(image_path)) if frame is None: append_or_reject(decisions, row, False, "opencv_decode_failed", None) continue landmarks = preprocessor.get_landmarks(frame) if landmarks is None: append_or_reject(decisions, row, False, "face_landmarks_not_found", None) continue crop = face_crop(frame, landmarks) if crop is None: append_or_reject(decisions, row, False, "teacher_face_crop_empty", None) continue left_eye, left_angle, left_matrix = normalize_eye_with_matrix( frame, landmarks, preprocessor.LEFT_CORNERS, preprocessor ) right_eye, right_angle, right_matrix = normalize_eye_with_matrix( frame, landmarks, preprocessor.RIGHT_CORNERS, preprocessor ) left_patches = preprocessor.extract_patches(left_eye, patch_size=16) right_patches = preprocessor.extract_patches(right_eye, patch_size=16) landmark_array = np.array([[point.x, point.y] for point in landmarks], dtype=np.float32) left_center = landmark_array[preprocessor.LEFT_CORNERS].mean(axis=0) right_center = landmark_array[preprocessor.RIGHT_CORNERS].mean(axis=0) centered_landmarks = landmark_array - (left_center + right_center) / 2.0 target = np.asarray(row["target_ccs"], dtype=np.float64) left_vector = target - np.asarray(row["left_eye_ccs"], dtype=np.float64) right_vector = target - np.asarray(row["right_eye_ccs"], dtype=np.float64) left_vector /= np.linalg.norm(left_vector) right_vector /= np.linalg.norm(right_vector) left_aligned_vector = rotation_matrix_z(left_angle) @ left_vector right_aligned_vector = rotation_matrix_z(right_angle) @ right_vector left_gaze = np.asarray(preprocessor.gaze_3d_to_mag(left_aligned_vector), dtype=np.float32) right_gaze = np.asarray(preprocessor.gaze_3d_to_mag(right_aligned_vector), dtype=np.float32) with torch.inference_mode(): tensor = torch.from_numpy(crop).unsqueeze(0).to(args.device) raw_pitch, raw_yaw = teacher(tensor) raw_pitch_np = raw_pitch[0].detach().cpu().numpy().astype(np.float32) raw_yaw_np = raw_yaw[0].detach().cpu().numpy().astype(np.float32) aligned_pitch, aligned_yaw, teacher_vector = roll_align_teacher_distribution( raw_pitch_np, raw_yaw_np, left_angle ) teacher_pitch_deg, teacher_yaw_deg = angles_from_vectors(teacher_vector[None, :]) accepted_row = dict(row) accepted_row.update( { "left_patches": left_patches, "right_patches": right_patches, "landmarks": centered_landmarks, "left_gaze": left_gaze, "right_gaze": right_gaze, "left_affine_matrix": left_matrix, "right_affine_matrix": right_matrix, "left_roll_deg": np.float32(left_angle), "right_roll_deg": np.float32(right_angle), "teacher_pitch_logits_raw": raw_pitch_np, "teacher_yaw_logits_raw": raw_yaw_np, "teacher_pitch_logits": aligned_pitch, "teacher_yaw_logits": aligned_yaw, "teacher_aligned_vector": teacher_vector.astype(np.float32), "teacher_pitch_deg": np.float32(teacher_pitch_deg[0]), "teacher_yaw_deg": np.float32(teacher_yaw_deg[0]), "teacher_error_deg": np.float32(angular_error_deg(teacher_vector, left_aligned_vector)), } ) output_index = len(accepted) accepted.append(accepted_row) append_or_reject(decisions, row, True, "", output_index) if not accepted: raise RuntimeError("no rows were accepted; no cache written") attributes = { "schema": "mpiigaze-kd-clean-cache-v2-official448", "created_utc": datetime.now(timezone.utc).isoformat(), "git_commit": git_commit(), "participant": args.participant, "partial": args.max_source_rows is not None or args.max_accepted is not None, "source_manifest_sha256": summary["manifest_sha256"], "teacher_loader_audit": audit_to_dict(teacher_audit), "teacher_checkpoint_sha256": teacher_audit.checkpoint_sha256, "generator_script_sha256": file_sha256(Path(__file__)), "teacher_strict_script_sha256": file_sha256(ROOT / "src" / "models" / "teacher_strict.py"), "preprocess_script_sha256": file_sha256(ROOT / "src" / "utils" / "preprocess.py"), "landmark_model_sha256": file_sha256(LANDMARK_MODEL), "patch_size": 16, "teacher_input_size": 448, "teacher_bin_centers_deg": "index * 4 - 180", "teacher_protocol_basis": "recovered checkpoint key layout plus documented Gaze360 ResNet configuration", "training_target": "left-eye gaze rotated by left eye affine roll angle", "teacher_alignment": "independent pitch/yaw joint distribution rotated by same left-eye Z roll; marginals rebinned to 90 bins", } create_h5(output_path, accepted, attributes) with decision_path.open("x", encoding="utf-8", newline="\n") as stream: for decision in decisions: stream.write(json.dumps(decision, sort_keys=True, separators=(",", ":")) + "\n") run_summary = { "schema": "mpiigaze-kd-clean-cache-summary-v1", "participant": args.participant, "source_rows_examined": examined, "accepted_rows": len(accepted), "rejected_rows": examined - len(accepted), "cache_path": str(output_path.resolve()), "cache_sha256": file_sha256(output_path), "processing_manifest_path": str(decision_path.resolve()), "processing_manifest_sha256": file_sha256(decision_path), "source_manifest_sha256": summary["manifest_sha256"], "teacher_checkpoint_sha256": teacher_audit.checkpoint_sha256, "strict_loader_inference_sha256": teacher_audit.inference_sha256, "teacher_error_mean_deg": float(np.mean([row["teacher_error_deg"] for row in accepted])), "teacher_error_median_deg": float(np.median([row["teacher_error_deg"] for row in accepted])), } summary_output.write_text(json.dumps(run_summary, indent=2) + "\n", encoding="utf-8", newline="\n") print(json.dumps(run_summary, indent=2)) if __name__ == "__main__": main()