File size: 6,472 Bytes
178f61f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """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()
|