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