scripts/verify_dl3dv_download.py
"""Independently verify the approved DL3DV download selection.
The verifier rebuilds the selection from the pinned inventory and protocol,
checks every local file against its Hub Git/LFS identity, and records local
SHA-256 values as ``local_only`` evidence. A local SHA-256 is never treated
as an official source checksum.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any
from mcss.data.dl3dv_download import (
DEFAULT_REPRESENTATION,
METADATA_PATH,
REPO_ID,
REPO_TYPE,
REVISION,
atomic_write_json,
load_inventory_artifacts,
)
PROJECT = Path(__file__).resolve().parents[1]
DEFAULT_OUTPUT = PROJECT / "outputs/dl3dv_download_20260920"
DEFAULT_DATA_ROOT = PROJECT / "data/dl3dv_benchmark"
DEFAULT_PROTOCOL = (
PROJECT
/ "outputs/benchmark_protocol_20260920/sources/Long-LRM/data/"
/ "dl3dv_fold_8_kmeans_input_idx.json"
)
EXPECTED_PROTOCOL_SHA256 = "4b052aef605fda183683175de66250aaaf337e50003131dbf0113ea781a3145a"
EXPECTED_FILE_COUNT = 12_777
EXPECTED_TOTAL_BYTES = 11_313_259_491
EXPECTED_RETAINED_EXTRA_FILES = 281
EXPECTED_RETAINED_EXTRA_BYTES = 246_096_151
EXPECTED_PHYSICAL_FILE_COUNT = 13_058
EXPECTED_PHYSICAL_BYTES = 11_559_355_642
PROTOCOL_INPUT_KEYS = (
"fold_8_kmeans_16_input",
"fold_8_kmeans_32_input",
)
PROTOCOL_NAMES = ("full16", "ar32")
FRAME_NAME_RE = re.compile(r"(?:^|/)frame_(\d+)(?:\.[^/]*)?$")
HASH_CHUNK_SIZE = 4 * 1024 * 1024
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--inventory-dir", type=Path, default=DEFAULT_OUTPUT)
parser.add_argument(
"--output", type=Path, default=DEFAULT_OUTPUT / "completion_verification.json"
)
parser.add_argument("--data-root", type=Path, default=DEFAULT_DATA_ROOT)
parser.add_argument("--protocol-json", type=Path, default=DEFAULT_PROTOCOL)
parser.add_argument("--workers", type=int, default=8)
parser.add_argument("--target-every", type=int, default=8)
return parser
def _source_identity(remote: Any) -> dict[str, str | None]:
if remote.lfs_sha256 is not None:
return {"kind": "hf_lfs_sha256", "value": remote.lfs_sha256}
if remote.git_blob_id is not None:
return {"kind": "git_blob_sha1", "value": remote.git_blob_id}
return {"kind": None, "value": None}
def _hash_local_file(path: Path) -> tuple[int, str, str]:
size = path.stat().st_size
sha256 = hashlib.sha256()
git_sha1 = hashlib.sha1()
git_sha1.update(f"blob {size}\0".encode("ascii"))
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(HASH_CHUNK_SIZE), b""):
sha256.update(chunk)
git_sha1.update(chunk)
return size, sha256.hexdigest(), git_sha1.hexdigest()
def _verify_remote(remote: Any, path: Path) -> dict[str, Any]:
if not path.is_file():
raise ValueError(f"missing local file: {path}")
actual_size, local_sha256, local_git_sha1 = _hash_local_file(path)
if actual_size != remote.size:
raise ValueError(
f"size mismatch for {path}: expected {remote.size}, got {actual_size}"
)
source = _source_identity(remote)
if source["value"] is None:
raise ValueError(f"remote entry has no source identity: {remote.path}")
if source["kind"] == "hf_lfs_sha256":
source_hash_verified = local_sha256 == source["value"]
else:
source_hash_verified = local_git_sha1 == source["value"]
if not source_hash_verified:
raise ValueError(f"source identity mismatch for {remote.path}")
return {
"repo_path": remote.path,
"relative_path": path.as_posix(),
"bytes": actual_size,
"source_hash_kind": source["kind"],
"source_hash": source["value"],
"source_hash_verified": True,
"local_sha256": local_sha256,
"local_sha256_role": "local_only",
}
def _verify_selected(item: Any) -> dict[str, Any]:
"""Keep the small item-shaped helper for callers of the old verifier API."""
return _verify_remote(item.remote, item.destination)
def _relative_to_root(root: Path, path: Path) -> str:
try:
return path.resolve().relative_to(root.resolve()).as_posix()
except ValueError as error:
raise ValueError(f"local path escapes data root: {path}") from error
def _destination(root: Path, remote: Any) -> Path:
if remote.scene_hash is None:
raise ValueError(f"scene file has no scene hash: {remote.path}")
path = root / remote.scene_hash / remote.relative_path
_relative_to_root(root, path)
return path
def _frame_basename(raw_path: str, *, scene: str, index: int) -> str:
normalized = raw_path.replace("\\", "/")
candidate = Path(normalized)
parts = tuple(part for part in normalized.split("/") if part not in ("", "."))
if (
not parts
or candidate.is_absolute()
or normalized.startswith("/")
or any(part == ".." for part in parts)
or ":" in parts[0]
or len(parts) < 2
):
raise ValueError(f"unsafe transforms frame file_path at index {index}: {scene}")
basename = parts[-1]
if FRAME_NAME_RE.fullmatch(basename) is None:
raise ValueError(f"transforms frame path has no frame number: {scene}/{basename}")
return basename
def _load_verified_frame_paths(
scene: str,
inventory: Any,
data_root: Path,
remote_by_path: dict[str, Any],
verified_by_repo: dict[str, dict[str, Any]],
*,
representation: str = DEFAULT_REPRESENTATION,
) -> tuple[str, ...]:
transform_path = f"{scene}/{representation}/transforms.json"
transform = remote_by_path.get(transform_path)
if transform is None or transform.kind != "transforms":
raise ValueError(f"inventory has no canonical transforms target: {scene}")
local_path = _destination(data_root, transform)
transform_entry = _verify_remote(transform, local_path)
transform_entry["relative_path"] = _relative_to_root(data_root, local_path)
verified_by_repo[transform.path] = transform_entry
try:
transforms = json.loads(local_path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
raise ValueError(f"cannot read transforms metadata for scene: {scene}") from error
if not isinstance(transforms, dict):
raise ValueError(f"transforms metadata must contain an object: {scene}")
raw_frames = transforms.get("frames")
if not isinstance(raw_frames, list) or not raw_frames:
raise ValueError(f"transforms metadata has no non-empty frames list: {scene}")
frame_paths: list[str] = []
seen: set[str] = set()
for index, raw_frame in enumerate(raw_frames):
if not isinstance(raw_frame, dict):
raise ValueError(f"transforms frame {index} is not an object: {scene}")
raw_path = raw_frame.get("file_path")
if not isinstance(raw_path, str) or not raw_path:
raise ValueError(f"transforms frame {index} has no file_path: {scene}")
basename = _frame_basename(raw_path, scene=scene, index=index)
relative = f"{representation}/images_4/{basename}"
if relative in seen:
raise ValueError(f"duplicate transforms frame path: {scene}/{relative}")
seen.add(relative)
remote_path = f"{scene}/{relative}"
image = remote_by_path.get(remote_path)
if image is None or image.kind != "images_4":
raise ValueError(f"missing inventory target for transforms frame: {remote_path}")
frame_paths.append(relative)
return tuple(frame_paths)
def _build_required_paths(
protocol_path: Path,
inventory: Any,
data_root: Path,
verified_by_repo: dict[str, dict[str, Any]],
*,
target_every: int,
representation: str = DEFAULT_REPRESENTATION,
) -> tuple[set[str], dict[str, Any]]:
"""Build the required set directly from verified transforms frame order."""
if target_every < 1:
raise ValueError("target_every must be positive")
try:
rows = json.loads(protocol_path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
raise ValueError(f"cannot read protocol frame JSON: {protocol_path}") from error
if not isinstance(rows, list):
raise ValueError("protocol frame JSON must contain a list")
remote_by_path: dict[str, Any] = {}
for remote in inventory.files:
if remote.path in remote_by_path:
raise ValueError(f"inventory contains duplicate repository path: {remote.path}")
remote_by_path[remote.path] = remote
inventory_scenes = set(inventory.scene_hashes)
selected_images: set[str] = set()
seen_scenes: set[str] = set()
for row in rows:
if not isinstance(row, dict) or not isinstance(row.get("scene_name"), str):
raise ValueError("protocol frame row has no scene_name")
scene = row["scene_name"]
if scene not in inventory_scenes:
raise ValueError(f"protocol frame row has unknown scene: {scene}")
if scene in seen_scenes:
raise ValueError(f"duplicate protocol row for scene: {scene}")
seen_scenes.add(scene)
frame_paths = _load_verified_frame_paths(
scene,
inventory,
data_root,
remote_by_path,
verified_by_repo,
representation=representation,
)
scene_images: set[str] = set()
for key in PROTOCOL_INPUT_KEYS:
raw_values = row.get(key)
if not isinstance(raw_values, list):
raise ValueError(f"protocol frame row has no list for {key}: {scene}")
for value in raw_values:
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError(f"invalid zero-based frame index in {key}: {scene}")
if value >= len(frame_paths):
raise ValueError(
f"zero-based frame index out of range in {key}: {scene} index={value}"
)
scene_images.add(f"{scene}/{frame_paths[value]}")
# Query sampling is positional in transforms.json, just like the split inputs.
scene_images.update(
f"{scene}/{frame_paths[index]}"
for index in range(0, len(frame_paths), target_every)
)
selected_images.update(scene_images)
if seen_scenes != inventory_scenes:
missing = sorted(inventory_scenes - seen_scenes)
extra = sorted(seen_scenes - inventory_scenes)
raise ValueError(
f"protocol scene coverage mismatch; missing={missing[:3]}, extra={extra[:3]}"
)
transform_paths = {
f"{scene}/{representation}/transforms.json" for scene in inventory.scene_hashes
}
required = transform_paths | selected_images
for path in required:
remote = remote_by_path.get(path)
if remote is None:
raise ValueError(f"required protocol path is absent from pinned inventory: {path}")
if remote.kind not in {"transforms", "images_4"}:
raise ValueError(f"required protocol path has unexpected kind: {path}")
return required, {
"status": "complete",
"mapping": "transforms_frame_order",
"protocols": list(PROTOCOL_NAMES),
"required_files": len(required),
"missing_files": 0,
}
def _entry_list_sha256(entries: list[dict[str, Any]]) -> str:
payload = "".join(
f"{entry['repo_path']}\t{entry['bytes']}\n"
for entry in sorted(entries, key=lambda value: value["repo_path"])
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _inventory_remote_map(inventory: Any) -> dict[str, Any]:
remote_by_path: dict[str, Any] = {}
for remote in inventory.files:
if remote.path in remote_by_path:
raise ValueError(f"inventory contains duplicate repository path: {remote.path}")
remote_by_path[remote.path] = remote
return remote_by_path
def _verify_pair(pair: tuple[Any, Path]) -> tuple[str, dict[str, Any]]:
remote, path = pair
return remote.path, _verify_remote(remote, path)
def run(args: argparse.Namespace) -> int:
if args.workers < 1 or args.workers > 16:
raise ValueError("workers must be between 1 and 16")
data_root = args.data_root.resolve()
inventory_dir = args.inventory_dir.resolve()
protocol_sha256 = hashlib.sha256(args.protocol_json.read_bytes()).hexdigest()
if protocol_sha256 != EXPECTED_PROTOCOL_SHA256:
raise ValueError(
"protocol JSON SHA256 does not match the approved pinned protocol: "
f"{protocol_sha256}"
)
inventory = load_inventory_artifacts(inventory_dir)
remote_by_path = _inventory_remote_map(inventory)
verified_by_repo: dict[str, dict[str, Any]] = {}
required_repo_paths, protocol_coverage = _build_required_paths(
args.protocol_json,
inventory,
data_root,
verified_by_repo,
target_every=args.target_every,
representation=DEFAULT_REPRESENTATION,
)
required_remotes = [remote_by_path[path] for path in sorted(required_repo_paths)]
required_destinations = {
remote.path: _destination(data_root, remote) for remote in required_remotes
}
expected_required_paths = {
path.resolve() for path in required_destinations.values()
}
actual_paths = {
path.resolve()
for path in data_root.rglob("*")
if path.is_file()
}
missing_paths = expected_required_paths - actual_paths
if missing_paths:
raise ValueError(
"required protocol files are missing from the data root: "
f"missing={len(missing_paths)}"
)
remaining_required = [
(remote, required_destinations[remote.path])
for remote in required_remotes
if remote.path not in verified_by_repo
]
if remaining_required:
with ThreadPoolExecutor(max_workers=min(args.workers, len(remaining_required))) as pool:
for repo_path, entry in pool.map(_verify_pair, remaining_required):
verified_by_repo[repo_path] = entry
for remote in required_remotes:
entry = verified_by_repo[remote.path]
entry["relative_path"] = _relative_to_root(data_root, required_destinations[remote.path])
entries = [verified_by_repo[remote.path] for remote in required_remotes]
if len(entries) != EXPECTED_FILE_COUNT:
raise ValueError(f"approved selection count changed: {len(entries)}")
total_bytes = sum(entry["bytes"] for entry in entries)
if total_bytes != EXPECTED_TOTAL_BYTES:
raise ValueError(f"approved selection byte total changed: {total_bytes}")
local_to_remote: dict[Path, Any] = {}
for remote in inventory.files:
if remote.scene_hash is None:
continue
local_path = _destination(data_root, remote).resolve()
if local_path in local_to_remote and local_to_remote[local_path].path != remote.path:
raise ValueError(f"inventory has duplicate local destinations: {local_path}")
local_to_remote[local_path] = remote
extra_paths = actual_paths - expected_required_paths
unknown_extra = sorted(str(path) for path in extra_paths if path not in local_to_remote)
if unknown_extra:
raise ValueError(
"DL3DV data root contains files outside the pinned inventory: "
f"unexpected={len(unknown_extra)}"
)
extra_pairs = [
(local_to_remote[path], path)
for path in sorted(extra_paths, key=str)
]
extra_by_repo: dict[str, dict[str, Any]] = {}
if extra_pairs:
with ThreadPoolExecutor(max_workers=min(args.workers, len(extra_pairs))) as pool:
for repo_path, entry in pool.map(_verify_pair, extra_pairs):
entry["relative_path"] = _relative_to_root(data_root, Path(entry["relative_path"]))
extra_by_repo[repo_path] = entry
extra_entries = [extra_by_repo[path] for path in sorted(extra_by_repo)]
retained_extra_bytes = sum(entry["bytes"] for entry in extra_entries)
if len(extra_entries) != EXPECTED_RETAINED_EXTRA_FILES:
raise ValueError(f"retained extra file count changed: {len(extra_entries)}")
if retained_extra_bytes != EXPECTED_RETAINED_EXTRA_BYTES:
raise ValueError(f"retained extra byte total changed: {retained_extra_bytes}")
physical_count = len(actual_paths)
physical_bytes = total_bytes + retained_extra_bytes
if physical_count != EXPECTED_PHYSICAL_FILE_COUNT:
raise ValueError(f"physical file count changed: {physical_count}")
if physical_bytes != EXPECTED_PHYSICAL_BYTES:
raise ValueError(f"physical byte total changed: {physical_bytes}")
metadata_path = inventory_dir / METADATA_PATH
if not metadata_path.is_file():
raise ValueError(f"missing local metadata file: {metadata_path}")
metadata_size, metadata_sha256, metadata_git_sha1 = _hash_local_file(metadata_path)
if metadata_size != inventory.metadata_file.size:
raise ValueError("local benchmark metadata size differs from pinned metadata")
if inventory.metadata_file.git_blob_id is None:
raise ValueError("pinned benchmark metadata has no Git blob identity")
if metadata_git_sha1 != inventory.metadata_file.git_blob_id:
raise ValueError("local benchmark metadata Git blob identity mismatch")
manifest = json.loads((inventory_dir / "manifest.json").read_text(encoding="utf-8"))
if metadata_sha256 != manifest["metadata_sha256"]:
raise ValueError("local benchmark metadata differs from the pinned inventory snapshot")
status_path = inventory_dir / "status.json"
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
if status_payload.get("status") != "complete":
raise ValueError(f"download status is not complete: {status_payload.get('status')!r}")
status_entries = status_payload.get("entries")
if not isinstance(status_entries, list):
raise ValueError("download status has no entries list")
status_by_repo: dict[str, dict[str, Any]] = {}
for status_entry in status_entries:
repo_path = status_entry.get("repo_path")
if not isinstance(repo_path, str) or repo_path in status_by_repo:
raise ValueError("download status entries are missing or duplicate repo paths")
status_by_repo[repo_path] = status_entry
independent_by_repo = {entry["repo_path"]: entry for entry in entries}
status_repo_paths = set(status_by_repo)
if status_repo_paths != required_repo_paths:
raise ValueError(
"download status entries must cover every required file: "
f"expected={len(required_repo_paths)}, actual={len(status_repo_paths)}"
)
for repo_path, status_entry in status_by_repo.items():
entry = independent_by_repo[repo_path]
if status_entry.get("status") not in {"downloaded", "reused"}:
raise ValueError(f"download status is not successful: {repo_path}")
if status_entry.get("sha256") != entry["local_sha256"]:
raise ValueError(f"status SHA256 differs from independent hash: {repo_path}")
if status_entry.get("expected_bytes", entry["bytes"]) != entry["bytes"]:
raise ValueError(f"status size differs from independent size: {repo_path}")
counts: dict[str, int] = {}
for remote in required_remotes:
counts[remote.kind] = counts.get(remote.kind, 0) + 1
source_kinds: dict[str, int] = {}
all_entries = entries + extra_entries
for entry in all_entries:
kind = str(entry["source_hash_kind"])
source_kinds[kind] = source_kinds.get(kind, 0) + 1
protocol_coverage["required_files"] = len(entries)
protocol_coverage["missing_files"] = 0
payload = {
"schema": "mcss.dl3dv.completion_verification.v1",
"status": "verified",
"verified_unix": time.time(),
"repo_id": REPO_ID,
"repo_type": REPO_TYPE,
"revision": REVISION,
"inventory_dir": str(inventory_dir),
"protocol_json": str(args.protocol_json.resolve()),
"protocol_sha256": protocol_sha256,
"data_root": str(data_root),
"representation": DEFAULT_REPRESENTATION,
"scene_count": len(inventory.scenes),
"file_count": len(entries),
"total_bytes": total_bytes,
"retained_extra_files": len(extra_entries),
"retained_extra_bytes": retained_extra_bytes,
"physical_count": physical_count,
"physical_bytes": physical_bytes,
"entry_list_sha256": _entry_list_sha256(entries),
"entry_count_by_kind": counts,
"protocol_coverage": protocol_coverage,
"download_status": {
"path": str(status_path.resolve()),
"status": status_payload["status"],
"entry_count": len(status_entries),
"entry_sha256_matches": True,
"required_entry_count": len(entries),
},
"source_hash_verified": {
"all_entries": all(entry["source_hash_verified"] for entry in all_entries),
"entry_count": len(entries),
"extra_entry_count": len(extra_entries),
"physical_file_count": physical_count,
"source_hash_kind_counts": source_kinds,
"meaning": "Each local file matched the pinned Hub LFS SHA256 or Git blob SHA1.",
},
"local_only": {
"field": "entries[*].local_sha256",
"algorithm": "sha256",
"meaning": "Local content digest for reproducibility; not an official source checksum.",
},
"metadata": {
"repo_path": METADATA_PATH,
"bytes": metadata_size,
"source_hash_kind": "git_blob_sha1",
"source_hash": inventory.metadata_file.git_blob_id,
"source_hash_verified": True,
"local_sha256": metadata_sha256,
"local_sha256_role": "local_only",
"manifest_metadata_sha256": manifest["metadata_sha256"],
"manifest_metadata_sha256_match": True,
},
"entries": entries,
"extra_entries": extra_entries,
}
atomic_write_json(args.output, payload)
print(
json.dumps(
{
"status": payload["status"],
"revision": payload["revision"],
"file_count": payload["file_count"],
"total_bytes": payload["total_bytes"],
"retained_extra_files": payload["retained_extra_files"],
"retained_extra_bytes": payload["retained_extra_bytes"],
"physical_count": payload["physical_count"],
"physical_bytes": payload["physical_bytes"],
"entry_list_sha256": payload["entry_list_sha256"],
"data_root": payload["data_root"],
"protocol_sha256": payload["protocol_sha256"],
},
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(run(_parser().parse_args()))