File size: 6,802 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 | """Assemble the p01 official-448 KD cache from validated rows and sealed raw logits."""
from __future__ import annotations
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
ROOT = Path(r"E:\Gaze_estimation")
sys.path.insert(0, str(ROOT / ".codex_deps"))
sys.path.insert(0, str(ROOT))
import h5py
import numpy as np
from scripts.generate_kd_clean_cache import (
angles_from_vectors,
angular_error_deg,
create_h5,
file_sha256,
gaze_vector,
roll_align_teacher_distribution,
)
SOURCE_CACHE = ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.full.h5"
SOURCE_PROCESSING = ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.full.processing.jsonl"
LOGITS = ROOT / "artifacts" / "kd-teacher-trap-diagnostic" / "p01_teacher_protocol_448_logits.npz"
STRICT_AUDIT = ROOT / "artifacts" / "kd-teacher-trap-diagnostic" / "strict_teacher_loader_audit.json"
OUTPUT = ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.official448_full.h5"
OUTPUT_PROCESSING = (
ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.official448_full.processing.jsonl"
)
OUTPUT_SUMMARY = ROOT / "data" / "processed_kd_clean_v1" / "cache" / "p01.official448_full.summary.json"
def decode(values: np.ndarray) -> list[str]:
return [value.decode("utf-8") if isinstance(value, bytes) else str(value) for value in values]
def main() -> None:
for path in (OUTPUT, OUTPUT_PROCESSING, OUTPUT_SUMMARY):
if path.exists():
raise FileExistsError(f"refusing to overwrite clean artifact: {path}")
logits = np.load(LOGITS)
strict_audit = json.loads(STRICT_AUDIT.read_text(encoding="utf-8"))
if not strict_audit.get("gate_a_loader_pass"):
raise RuntimeError("current strict-loader audit does not pass")
with h5py.File(SOURCE_CACHE, "r") as source:
sample_ids = decode(source["sample_id"][:])
paths = decode(source["relative_frame_path"][:])
if sample_ids != decode(logits["sample_id"]):
raise RuntimeError("448 logits sample IDs do not exactly match validated cache rows")
if paths != decode(logits["relative_frame_path"]):
raise RuntimeError("448 logits paths do not exactly match validated cache rows")
row_count = len(sample_ids)
rows: list[dict] = []
text_fields = ("sample_id", "relative_frame_path", "participant", "day", "frame_id", "raw_image_sha256")
scalar_fields = ("source_index", "annotation_row")
copied_numeric = (
"left_patches", "right_patches", "landmarks", "left_gaze", "right_gaze",
"left_affine_matrix", "right_affine_matrix", "left_roll_deg", "right_roll_deg",
)
text_values = {field: decode(source[field][:]) for field in text_fields}
scalar_values = {field: source[field][:] for field in scalar_fields}
numeric_values = {field: source[field][:] for field in copied_numeric}
raw_pitch = logits["fc_pitch_logits"].astype(np.float32)
raw_yaw = logits["fc_yaw_logits"].astype(np.float32)
for index in range(row_count):
aligned_pitch, aligned_yaw, teacher_vector = roll_align_teacher_distribution(
raw_pitch[index], raw_yaw[index], float(numeric_values["left_roll_deg"][index])
)
pitch_deg, yaw_deg = angles_from_vectors(teacher_vector[None, :])
left_gaze = numeric_values["left_gaze"][index]
target_vector = gaze_vector(float(left_gaze[0]), float(left_gaze[1]))
row = {field: text_values[field][index] for field in text_fields}
row.update({field: scalar_values[field][index] for field in scalar_fields})
row.update({field: numeric_values[field][index] for field in copied_numeric})
row.update(
{
"teacher_pitch_logits_raw": raw_pitch[index],
"teacher_yaw_logits_raw": raw_yaw[index],
"teacher_pitch_logits": aligned_pitch,
"teacher_yaw_logits": aligned_yaw,
"teacher_aligned_vector": teacher_vector.astype(np.float32),
"teacher_pitch_deg": np.float32(pitch_deg[0]),
"teacher_yaw_deg": np.float32(yaw_deg[0]),
"teacher_error_deg": np.float32(angular_error_deg(teacher_vector, target_vector)),
}
)
rows.append(row)
attributes = {
"schema": "mpiigaze-kd-clean-cache-v2-official448",
"created_utc": datetime.now(timezone.utc).isoformat(),
"participant": "p01",
"partial": False,
"source_manifest_sha256": source.attrs["source_manifest_sha256"],
"teacher_loader_audit": strict_audit,
"teacher_checkpoint_sha256": strict_audit["checkpoint_sha256"],
"assembly_script_sha256": file_sha256(Path(__file__)),
"source_preprocessing_cache_sha256": file_sha256(SOURCE_CACHE),
"source_448_logits_sha256": file_sha256(LOGITS),
"patch_size": 16,
"teacher_input_size": 448,
"teacher_bin_centers_deg": "index * 4 - 180",
"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",
"status": "KD_READY_PENDING_VALIDATION",
}
create_h5(OUTPUT, rows, attributes)
OUTPUT_PROCESSING.write_bytes(SOURCE_PROCESSING.read_bytes())
summary = {
"schema": "mpiigaze-kd-clean-cache-summary-v2-official448",
"participant": "p01",
"source_rows_examined": sum(1 for _ in SOURCE_PROCESSING.open("r", encoding="utf-8")),
"accepted_rows": len(rows),
"rejected_rows": sum(1 for line in SOURCE_PROCESSING.open("r", encoding="utf-8") if not json.loads(line)["accepted"]),
"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_manifest_sha256": attributes["source_manifest_sha256"],
"teacher_checkpoint_sha256": attributes["teacher_checkpoint_sha256"],
"strict_loader_inference_sha256": strict_audit["inference_sha256"],
"teacher_error_mean_deg": float(np.mean([row["teacher_error_deg"] for row in rows])),
"teacher_error_median_deg": float(np.median([row["teacher_error_deg"] for row in rows])),
}
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()
|