scripts/audit_static_unseen_independence.py
#!/usr/bin/env python3
"""Independently cross-check frozen unseen roles, identities, and allowed media."""
import argparse
import csv
import json
import re
from pathlib import Path
from mcss.mechanism_pilot.small_training import sha, write_json
from mcss.mechanism_pilot.unseen_data import asset_identity, source_assets
def audit(root):
root = Path(root)
data = root / "unseen_data"
candidate = json.loads((data / "candidate_lock.json").read_text())
member_lock = json.loads((data / "data_lock.json").read_text())
role = json.loads((data / "role_lock.json").read_text())
manifest = json.loads((data / "manifest.json").read_text())
assets = source_assets(root / "metadata/upstream_dataset_config.py")
train = candidate["train_scene_ids"]
selected = candidate["candidate_scene_ids"]
qualified = [r["scene_id"] for r in manifest["scenes"]]
with Path("configs/hypersim_er_partitions.csv").open() as f:
partitions = {r["scene_name"]: r for r in csv.DictReader(f)}
with (root / "rgb_audit/official_metadata/metadata_images_split_scene_v1.csv").open() as f:
official = {
(r["scene_name"], r["camera_name"], int(r["frame_id"])): r for r in csv.DictReader(f)
}
with (root / "rgb_audit/official_metadata/metadata_camera_trajectories.csv").open() as f:
trajectories = {r["Animation"]: r for r in csv.DictReader(f)}
source_hash_mismatch = [p for p, h in candidate["source_sha256"].items() if sha(Path(p)) != h]
accesses = [json.loads(x) for x in (data / "access.jsonl").read_text().splitlines()]
downloads = [r for r in accesses if r["operation"] == "download"]
bad_media, bad_roles = [], []
for plan in member_lock["scenes"]:
sid = plan["scene_id"]
for member in plan["members"]:
match = re.search(r"/frame\.(\d{4})\.", member)
if match:
row = official.get((sid, "cam_00", int(match[1])))
if (
not row
or row["included_in_public_release"] != "True"
or row["split_partition_name"] != "train"
):
bad_media.append(member)
for s in manifest["scenes"]:
r = s["roles"]
if r != role["roles"][s["scene_id"]] or set(r["primary_query"]) & (
set(r["context_a"]) | set(r["context_b"])
):
bad_roles.append(s["scene_id"])
source_rows = {
s: {
"archive_file": assets[s]["archive_file"],
"asset_file": assets[s]["asset_file"],
"volume": s.split("_")[1],
}
for s in train + selected
}
identities = [asset_identity(assets[s]) for s in train + selected]
first_download = min(r["time"] for r in downloads)
static_lock = root / "unseen_static/lock.json"
static_lock_mtime = static_lock.stat().st_mtime if static_lock.exists() else None
checks = {
"train_scene_overlap_absent": not set(train) & set(selected),
"source_asset_pairs_all_distinct": len(identities) == len(set(identities)),
"training_volumes_disjoint": not {s.split("_")[1] for s in train}
& {s.split("_")[1] for s in selected},
"source_hashes_unchanged": not source_hash_mismatch,
"candidate_lock_hash_matches_data_lock": sha(data / "candidate_lock.json")
== member_lock["candidate_lock_sha256"],
"data_lock_hash_matches_manifest": sha(data / "data_lock.json")
== manifest["data_lock_sha256"],
"manifest_hash_matches_role_lock": sha(data / "manifest.json") == role["manifest_sha256"],
"only_locked_candidates_downloaded": {r["scene_id"] for r in downloads} == set(selected),
"all_candidate_protocol_partitions_train": all(
partitions[s]["protocol_partition"] == "train" for s in selected
),
"all_candidate_trajectories_non_BAD": all(
"BAD" not in trajectories[s + "_cam_00"]["Scene type"].upper() for s in selected
),
"all_media_official_included_train": not bad_media,
"candidate_then_members_then_download_order": candidate["created_unix"]
< member_lock["created_unix"]
< first_download,
"locked_roles_identical_and_queries_disjoint": not bad_roles,
"role_lock_before_static_lock_local_mtime": static_lock_mtime is not None
and role["created_unix"] < static_lock_mtime,
"qualified_scenes_subset_frozen_candidates": set(qualified) <= set(selected),
}
report = {
"status": "PASS_WITH_PHYSICAL_IDENTITY_LIMITATION" if all(checks.values()) else "FAIL",
"checks": checks,
"train_scene_ids": train,
"candidate_scene_ids": selected,
"qualified_scene_ids": qualified,
"source_asset_mapping": source_rows,
"previously_observed": candidate["previously_observed"],
"independence_scope": (
"Unseen to this frozen B-final checkpoint; ai_007_002 has prior "
"project exposure and is not claimed globally virgin."
),
"physical_identity_evidence": (
"Distinct official archive/asset pairs and volumes; source mappings "
"immutable by SHA256."
),
"physical_identity_limit": (
"Vendor content duplication across different source assets cannot be"
" ruled out; no claim of externally certified physical-location "
"uniqueness."
),
"source_url": (
"https://github.com/apple/ml-hypersim/blob/main/evermotion_dataset/_dataset_config.py"
),
"candidate_lock_url_erratum": str(data / "preparation_notes.json"),
"protected_test_media_opened": False,
"bad_media": bad_media,
"bad_roles": bad_roles,
"source_hash_mismatch": source_hash_mismatch,
"order_evidence": {
"candidate_unix": candidate["created_unix"],
"members_unix": member_lock["created_unix"],
"first_media_download_unix": first_download,
"role_lock_unix": role["created_unix"],
"static_lock_mtime_unix": static_lock_mtime,
"limitation": (
"Local operation timestamps and file mtime; not an external "
"cryptographic timestamp service."
),
},
"artifact_sha256": {
str(p): sha(p)
for p in (
data / "candidate_lock.json",
data / "data_lock.json",
data / "role_lock.json",
data / "manifest.json",
data / "access.jsonl",
root / "metadata/upstream_dataset_config.py",
)
},
}
write_json(root / "unseen_independence_audit.json", report)
return report
if __name__ == "__main__":
p = argparse.ArgumentParser()
p.add_argument("--experiment", required=True)
a = p.parse_args()
print(json.dumps(audit(a.experiment), indent=2))