"""Create an unambiguous continuous-vector KD cache from official 448 raw logits.""" from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json 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 CACHE_ROOT = ROOT / "data" / "processed_kd_clean_v1" / "cache" def file_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 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(values: np.ndarray, roll_deg: np.ndarray) -> np.ndarray: angle = np.deg2rad(roll_deg) cosine, sine = np.cos(angle), np.sin(angle) result = values.copy() result[:, 0] = cosine * values[:, 0] - sine * values[:, 1] result[:, 1] = sine * values[:, 0] + cosine * values[:, 1] return result def angles(values: np.ndarray) -> tuple[np.ndarray, np.ndarray]: pitch = np.arcsin(np.clip(-values[:, 1], -1.0, 1.0)) yaw = np.arctan2(-values[:, 0], -values[:, 2]) return np.rad2deg(pitch), np.rad2deg(yaw) def angular_error(first: np.ndarray, second: np.ndarray) -> np.ndarray: first = first / np.linalg.norm(first, axis=1, keepdims=True) second = second / np.linalg.norm(second, axis=1, keepdims=True) return np.rad2deg(np.arccos(np.clip(np.sum(first * second, axis=1), -1.0, 1.0))) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--participant", required=True) parser.add_argument("--source-tag", default="official448_full") parser.add_argument("--output-tag", default="official448_pointkd") args = parser.parse_args() source = CACHE_ROOT / f"{args.participant}.{args.source_tag}.h5" source_processing = CACHE_ROOT / f"{args.participant}.{args.source_tag}.processing.jsonl" output = CACHE_ROOT / f"{args.participant}.{args.output_tag}.h5" output_processing = CACHE_ROOT / f"{args.participant}.{args.output_tag}.processing.jsonl" output_summary = CACHE_ROOT / f"{args.participant}.{args.output_tag}.summary.json" for path in (output, output_processing, output_summary): if path.exists(): raise FileExistsError(f"refusing to overwrite point-KD artifact: {path}") copied_fields = ( "sample_id", "relative_frame_path", "participant", "day", "frame_id", "raw_image_sha256", "source_index", "annotation_row", "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", ) with h5py.File(source, "r") as source_h5, h5py.File(output, "x") as output_h5: for field in copied_fields: source_h5.copy(field, output_h5) raw_pitch = source_h5["teacher_pitch_logits_raw"][:] raw_yaw = source_h5["teacher_yaw_logits_raw"][:] pitch = softmax(raw_pitch) @ np.arange(90) * 4.0 - 180.0 yaw = softmax(raw_yaw) @ np.arange(90) * 4.0 - 180.0 raw_vector = vectors(pitch, yaw) target_vector = rotate_z(raw_vector, source_h5["left_roll_deg"][:]) target_pitch, target_yaw = angles(target_vector) gaze_deg = np.rad2deg(source_h5["left_gaze"][:].astype(np.float64)) label_vector = vectors(gaze_deg[:, 0], gaze_deg[:, 1]) error = angular_error(target_vector, label_vector) output_h5.create_dataset("teacher_target_vector", data=target_vector.astype(np.float32), compression="gzip") output_h5.create_dataset("teacher_target_pitch_deg", data=target_pitch.astype(np.float32), compression="gzip") output_h5.create_dataset("teacher_target_yaw_deg", data=target_yaw.astype(np.float32), compression="gzip") output_h5.create_dataset("teacher_target_error_deg", data=error.astype(np.float32), compression="gzip") for key, value in source_h5.attrs.items(): output_h5.attrs[key] = value output_h5.attrs["schema"] = "mpiigaze-point-kd-cache-v3-official448" output_h5.attrs["created_utc"] = datetime.now(timezone.utc).isoformat() output_h5.attrs["source_official448_cache_sha256"] = file_sha256(source) output_h5.attrs["pointkd_assembly_script_sha256"] = file_sha256(Path(__file__)) output_h5.attrs["teacher_target_definition"] = ( "expectation of named 4-degree fc_pitch/fc_yaw logits converted to 3D, then rotated " "by the same left-eye Z roll used for the training label" ) output_h5.attrs["approved_distillation"] = "continuous 3D vector loss only" output_h5.attrs["prohibited_distillation"] = "KL on roll-rebinned marginal logits" output_h5.attrs["status"] = "POINT_KD_READY_PENDING_VALIDATION" output_processing.write_bytes(source_processing.read_bytes()) decisions = [json.loads(line) for line in output_processing.read_text(encoding="utf-8").splitlines()] summary = { "schema": "mpiigaze-point-kd-cache-summary-v3", "participant": args.participant, "source_rows_examined": len(decisions), "accepted_rows": int(sum(bool(row["accepted"]) for row in decisions)), "rejected_rows": int(sum(not bool(row["accepted"]) for row in decisions)), "cache_path": str(output.resolve()), "cache_sha256": file_sha256(output), "processing_manifest_path": str(output_processing.resolve()), "processing_manifest_sha256": file_sha256(output_processing), "source_official448_cache_sha256": file_sha256(source), "teacher_target_error_mean_deg": float(error.mean()), "teacher_target_error_median_deg": float(np.median(error)), } output_summary.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8", newline="\n") print(json.dumps(summary, indent=2)) if __name__ == "__main__": main()