| """Validate identity, order, and a deterministic content-hash sample of a clean manifest.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| from pathlib import Path |
| import random |
|
|
|
|
| ROOT = Path(r"E:\Gaze_estimation") |
| DATASET_ROOT = ROOT / "data" / "MPIIGaze" / "MPIIGaze" / "MPIIGaze" |
| ORIGINAL_ROOT = DATASET_ROOT / "Data" / "Original" |
| MANIFEST_ROOT = ROOT / "data" / "processed_kd_clean_v1" / "manifests" |
|
|
|
|
| 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 text_sha256(value: str) -> str: |
| return hashlib.sha256(value.encode("utf-8")).hexdigest().upper() |
|
|
|
|
| def expected_sample_id(row: dict) -> str: |
| payload = "\0".join( |
| ( |
| row["dataset"], |
| row["participant"], |
| row["relative_frame_path"], |
| str(row["annotation_row"]), |
| "binocular-left-training-target", |
| ) |
| ) |
| return text_sha256(payload) |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--participant", required=True, choices=[f"p{i:02d}" for i in range(15)]) |
| parser.add_argument("--sample-size", type=int, default=100) |
| parser.add_argument("--seed", type=int, default=20260809) |
| args = parser.parse_args() |
|
|
| manifest_path = MANIFEST_ROOT / f"{args.participant}.source.jsonl" |
| summary_path = MANIFEST_ROOT / f"{args.participant}.source.summary.json" |
| rows = [json.loads(line) for line in manifest_path.read_text(encoding="utf-8").splitlines()] |
| summary = json.loads(summary_path.read_text(encoding="utf-8")) |
|
|
| ids = [row["sample_id"] for row in rows] |
| structural_errors: list[str] = [] |
| if len(ids) != len(set(ids)): |
| structural_errors.append("sample_id values are not unique") |
| if [row["source_index"] for row in rows] != list(range(len(rows))): |
| structural_errors.append("source_index is not consecutive in file order") |
| if any(row["sample_id"] != expected_sample_id(row) for row in rows): |
| structural_errors.append("one or more sample_id values do not match the identity formula") |
| if len(rows) != summary["rows"]: |
| structural_errors.append("manifest row count differs from summary") |
| if file_sha256(manifest_path) != summary["manifest_sha256"]: |
| structural_errors.append("manifest SHA-256 differs from summary") |
|
|
| sample_count = min(args.sample_size, len(rows)) |
| sampled_indices = sorted(random.Random(args.seed).sample(range(len(rows)), sample_count)) |
| sampled_errors: list[str] = [] |
| annotation_cache: dict[Path, tuple[str, list[str]]] = {} |
| for index in sampled_indices: |
| row = rows[index] |
| image_path = ORIGINAL_ROOT / Path(row["relative_frame_path"]) |
| annotation_path = ORIGINAL_ROOT / Path(row["annotation_file_relative_path"]) |
| if not image_path.is_file(): |
| sampled_errors.append(f"missing image for source_index={index}: {image_path}") |
| continue |
| if file_sha256(image_path) != row["raw_image_sha256"]: |
| sampled_errors.append(f"image hash mismatch for source_index={index}") |
| if annotation_path not in annotation_cache: |
| annotation_cache[annotation_path] = ( |
| file_sha256(annotation_path), |
| annotation_path.read_text(encoding="utf-8").splitlines(), |
| ) |
| annotation_hash, annotation_lines = annotation_cache[annotation_path] |
| if annotation_hash != row["annotation_file_sha256"]: |
| sampled_errors.append(f"annotation file hash mismatch for source_index={index}") |
| annotation_line = annotation_lines[row["annotation_row"] - 1] |
| if text_sha256(annotation_line) != row["annotation_line_sha256"]: |
| sampled_errors.append(f"annotation line hash mismatch for source_index={index}") |
|
|
| report = { |
| "schema": "mpiigaze-kd-clean-manifest-validation-v1", |
| "participant": args.participant, |
| "manifest_path": str(manifest_path.resolve()), |
| "manifest_sha256": file_sha256(manifest_path), |
| "rows": len(rows), |
| "unique_sample_ids": len(set(ids)), |
| "sample_validation_seed": args.seed, |
| "sampled_rows_rehashed": sample_count, |
| "sampled_annotation_files_rehashed": len(annotation_cache), |
| "structural_errors": structural_errors, |
| "sampled_content_errors": sampled_errors, |
| "pass": not structural_errors and not sampled_errors, |
| } |
| output = MANIFEST_ROOT / f"{args.participant}.source.validation.json" |
| output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8", newline="\n") |
| print(json.dumps(report, indent=2)) |
| if not report["pass"]: |
| raise SystemExit(1) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
