| """Read-only structural and provenance validation for a clean KD cache.""" |
| from __future__ import annotations |
|
|
| import argparse |
| 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 |
|
|
|
|
| MANIFEST_ROOT = ROOT / "data" / "processed_kd_clean_v1" / "manifests" |
| 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 decode(values: np.ndarray) -> list[str]: |
| return [value.decode("utf-8") if isinstance(value, bytes) else str(value) for value in values] |
|
|
|
|
| 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 vector_from_angles(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 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("--tag", required=True) |
| args = parser.parse_args() |
|
|
| cache_path = CACHE_ROOT / f"{args.participant}.{args.tag}.h5" |
| processing_path = CACHE_ROOT / f"{args.participant}.{args.tag}.processing.jsonl" |
| summary_path = CACHE_ROOT / f"{args.participant}.{args.tag}.summary.json" |
| source_path = MANIFEST_ROOT / f"{args.participant}.source.jsonl" |
| output_path = CACHE_ROOT / f"{args.participant}.{args.tag}.validation.json" |
| if output_path.exists(): |
| raise FileExistsError(f"refusing to overwrite validation artifact: {output_path}") |
|
|
| summary = json.loads(summary_path.read_text(encoding="utf-8")) |
| source_rows = { |
| row["sample_id"]: row |
| for row in (json.loads(line) for line in source_path.read_text(encoding="utf-8").splitlines()) |
| } |
| decisions = [json.loads(line) for line in processing_path.read_text(encoding="utf-8").splitlines()] |
| errors: list[str] = [] |
| if file_sha256(cache_path) != summary["cache_sha256"]: |
| errors.append("cache SHA-256 differs from summary") |
| if file_sha256(processing_path) != summary["processing_manifest_sha256"]: |
| errors.append("processing manifest SHA-256 differs from summary") |
| if len(decisions) != summary["source_rows_examined"]: |
| errors.append("processing-decision count differs from examined-row summary") |
| if sum(bool(row["accepted"]) for row in decisions) != summary["accepted_rows"]: |
| errors.append("accepted decision count differs from summary") |
|
|
| required = { |
| "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", "teacher_pitch_logits", |
| "teacher_yaw_logits", "teacher_aligned_vector", "teacher_pitch_deg", |
| "teacher_yaw_deg", "teacher_error_deg", |
| } |
| metrics: dict[str, float | int | bool] = {} |
| with h5py.File(cache_path, "r") as handle: |
| missing = sorted(required - set(handle.keys())) |
| if missing: |
| errors.append(f"missing datasets: {missing}") |
| row_counts = {name: int(handle[name].shape[0]) for name in required if name in handle} |
| if len(set(row_counts.values())) != 1: |
| errors.append(f"dataset row counts differ: {row_counts}") |
| count = min(row_counts.values()) if row_counts else 0 |
| if count != summary["accepted_rows"]: |
| errors.append("H5 row count differs from accepted-row summary") |
|
|
| sample_ids = decode(handle["sample_id"][:]) |
| paths = decode(handle["relative_frame_path"][:]) |
| hashes = decode(handle["raw_image_sha256"][:]) |
| source_indices = handle["source_index"][:] |
| if len(sample_ids) != len(set(sample_ids)): |
| errors.append("H5 sample IDs are not unique") |
| for output_index, sample_id in enumerate(sample_ids): |
| source = source_rows.get(sample_id) |
| if source is None: |
| errors.append(f"H5 sample_id not found in source manifest: {sample_id}") |
| continue |
| if source["relative_frame_path"] != paths[output_index]: |
| errors.append(f"path mismatch at H5 row {output_index}") |
| if source["raw_image_sha256"] != hashes[output_index]: |
| errors.append(f"image hash mismatch at H5 row {output_index}") |
| if source["source_index"] != int(source_indices[output_index]): |
| errors.append(f"source index mismatch at H5 row {output_index}") |
|
|
| numeric_names = [name for name in required if name in handle and handle[name].dtype.kind not in "OSU"] |
| if any(not np.isfinite(handle[name][:]).all() for name in numeric_names): |
| errors.append("one or more numeric datasets contain non-finite values") |
| if handle["teacher_pitch_logits"].shape[1:] != (90,) or handle["teacher_yaw_logits"].shape[1:] != (90,): |
| errors.append("aligned teacher logits do not have 90 bins") |
| if handle["left_patches"].shape[1:] != (4, 16, 16): |
| errors.append("left patches do not use the declared V16 shape") |
| if handle["landmarks"].shape[1:] != (478, 2): |
| errors.append("landmarks do not have shape [N,478,2]") |
|
|
| aligned_pitch_probability = softmax(handle["teacher_pitch_logits"][:]) |
| aligned_yaw_probability = softmax(handle["teacher_yaw_logits"][:]) |
| metrics["max_aligned_pitch_probability_sum_error"] = float( |
| np.max(np.abs(aligned_pitch_probability.sum(axis=1) - 1.0)) |
| ) |
| metrics["max_aligned_yaw_probability_sum_error"] = float( |
| np.max(np.abs(aligned_yaw_probability.sum(axis=1) - 1.0)) |
| ) |
| target_deg = np.rad2deg(handle["left_gaze"][:].astype(np.float64)) |
| target_vector = vector_from_angles(target_deg[:, 0], target_deg[:, 1]) |
| aligned_vector = handle["teacher_aligned_vector"][:].astype(np.float64) |
| recomputed_error = angular_error(aligned_vector, target_vector) |
| stored_error = handle["teacher_error_deg"][:].astype(np.float64) |
| metrics["max_teacher_error_recompute_difference_deg"] = float( |
| np.max(np.abs(recomputed_error - stored_error)) |
| ) |
| metrics["teacher_aligned_error_mean_deg"] = float(recomputed_error.mean()) |
|
|
| bin_definition = str(handle.attrs["teacher_bin_centers_deg"]) |
| if bin_definition == "index * 4 - 180": |
| binwidth, offset = 4.0, 180.0 |
| elif bin_definition == "index * 2 - 90": |
| binwidth, offset = 2.0, 90.0 |
| else: |
| errors.append(f"unknown teacher bin definition: {bin_definition}") |
| binwidth, offset = np.nan, np.nan |
| raw_pitch = softmax(handle["teacher_pitch_logits_raw"][:]) @ np.arange(90) * binwidth - offset |
| raw_yaw = softmax(handle["teacher_yaw_logits_raw"][:]) @ np.arange(90) * binwidth - offset |
| raw_vector = vector_from_angles(raw_pitch, raw_yaw) |
| metrics["teacher_raw_vs_roll_corrected_target_error_mean_deg"] = float( |
| angular_error(raw_vector, target_vector).mean() |
| ) |
| metrics["teacher_alignment_error_change_deg"] = float( |
| metrics["teacher_aligned_error_mean_deg"] |
| - metrics["teacher_raw_vs_roll_corrected_target_error_mean_deg"] |
| ) |
|
|
| teacher_audit = json.loads(handle.attrs["teacher_loader_audit"]) |
| if teacher_audit["mapped_tensor_count"] != teacher_audit["checkpoint_tensor_count"]: |
| errors.append("embedded strict-loader audit does not map every checkpoint tensor") |
|
|
| report = { |
| "schema": "mpiigaze-kd-clean-cache-validation-v1", |
| "participant": args.participant, |
| "tag": args.tag, |
| "cache_path": str(cache_path.resolve()), |
| "cache_sha256": file_sha256(cache_path), |
| "rows": summary["accepted_rows"], |
| "metrics": metrics, |
| "errors": errors, |
| "pass": not errors, |
| } |
| output_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8", newline="\n") |
| print(json.dumps(report, indent=2)) |
| if errors: |
| raise SystemExit(1) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
