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