File size: 4,875 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 | """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()
|