""" make_multicommodity_manifests.py — multi-commodity, multi-source benchmark manifests for ARC-V Paper 2. Generalizes make_loso_manifests.py / make_source_specific_manifests.py from one-commodity/two-sources to C >= 2 commodities x S >= 3 acquisition sources. Spec: paper/BENCHMARK_PROTOCOL.md. This script is the reference implementation. It consumes a REGISTRY of the form {commodity: {"sources": {source: path_glob, ...}, "synonyms": {raw_label: canonical_name, ...}}} # synonyms optional and, for each commodity independently, emits in the existing manifest schema (v2, additive over unified_benchmark.json): * manifest__src_.json — per-source (renumbered classes) * manifest__overlap.json — classes present in >=2 sources * manifest__overlap_.json — overlap classes, one source (asym. probe) * manifest__loso_.json — train on others, test on held-out source Source identity is carried EXPLICITLY in a per-sample `meta` block (1:1 with `samples`), so v2-aware loaders never parse paths to recover the source. The `samples` lists keep the Paper-1 [path, label] shape, so existing consumers (src.dataset.load_manifest_splits / get_aifnet_dataloaders) still work. Real images are NOT required to develop/test this: `--smoke` builds a synthetic registry of dummy file lists and runs the whole pipeline on CPU. # smoke (no data needed, pure CPU): python make_multicommodity_manifests.py --smoke # real run from a registry file: python make_multicommodity_manifests.py --registry registry.json --val-frac 0.15 NOTE: this scaffold performs manifest construction only. The mandatory pixel-verified de-duplication (BENCHMARK_PROTOCOL.md sec.5) reuses audit_dedup.py and make_dedup_manifests.py unchanged and is invoked as a separate step; here we only emit the `dedup` placeholder block and wire the hook (see run_dedup()). """ import argparse import hashlib import json import re from collections import defaultdict from pathlib import Path import numpy as np # ── reuse existing helpers where they exist; fall back gracefully ─────────────── try: # src.dataset.source_of is the Paper-1 v1 path->source fallback only from src.dataset import source_of as _v1_source_of # noqa: F401 except Exception: # pragma: no cover - keeps the smoke path importable anywhere _v1_source_of = None # ── canonicalization ─────────────────────────────────────────────────────────── def slugify(label: str) -> str: """Canonical class name: lower, spaces/hyphens/dots -> underscore, collapse.""" s = str(label).strip().lower() s = re.sub(r"[\s\-.]+", "_", s) s = re.sub(r"[^a-z0-9_]+", "", s) s = re.sub(r"_+", "_", s).strip("_") return s or "unknown" def canonical(label: str, synonyms: dict) -> str: """Map a raw folder label to a canonical class name via synonyms then slugify.""" if synonyms and label in synonyms: return synonyms[label] # also allow a synonym keyed by the slug sl = slugify(label) if synonyms and sl in synonyms: return synonyms[sl] return sl # ── registry resolution: {commodity: [(path, source, canonical_class, raw), ...]} ─ def _class_label_from_path(path: str) -> str: """Raw class label = immediate parent directory name of the file.""" return Path(str(path).replace("\\", "/")).parent.name def resolve_registry(registry: dict, file_lists: dict | None = None) -> dict: """Expand a registry into per-commodity sample records. registry: {commodity: {"sources": {source: glob}, "synonyms": {...}}} file_lists: optional {commodity: {source: [paths...]}} that BYPASSES globbing (used by --smoke and by tests). When given, the registry glob is ignored for that (commodity, source) and these paths are used. Returns {commodity: list[(path, source, canonical_class, raw_label)]}. Raises if a real (non-overridden) source resolves to zero files. """ out: dict[str, list] = {} for commodity, spec in registry.items(): sources = spec.get("sources", {}) synonyms = spec.get("synonyms", {}) recs: list = [] for source, glob in sources.items(): paths = None if file_lists and commodity in file_lists and source in file_lists[commodity]: paths = list(file_lists[commodity][source]) else: # real path: glob is relative to CWD (matches existing scripts) paths = [str(p) for p in sorted(Path().glob(glob))] if not paths: raise SystemExit( f"[registry] commodity={commodity!r} source={source!r} glob={glob!r} " f"matched 0 files; a silently-empty source corrupts LOSO." ) for p in paths: raw = _class_label_from_path(p) recs.append((p, source, canonical(raw, synonyms), raw)) out[commodity] = recs return out def _registry_hash(registry: dict) -> str: blob = json.dumps(registry, sort_keys=True, default=str).encode() return "sha1:" + hashlib.sha1(blob).hexdigest() # ── manifest assembly helpers ────────────────────────────────────────────────── def _split_records(records, classes, val_frac, seed): """Carve train/val from a list of (path, source, class) over the given classes. Returns (train, val) where each is a list of (path, source, class). A per-(source,class) seeded shuffle puts `val_frac` into val. (Used for the training sources of a LOSO/per-source manifest; held-out test is added by caller.) """ rng = np.random.default_rng(seed) by_sc = defaultdict(list) for p, s, c in records: if c in classes: by_sc[(s, c)].append(p) train, val = [], [] for (s, c), paths in sorted(by_sc.items()): paths = list(paths) rng.shuffle(paths) n_val = max(1, int(round(len(paths) * val_frac))) if paths else 0 for p in paths[:n_val]: val.append((p, s, c)) for p in paths[n_val:]: train.append((p, s, c)) return train, val def _build_manifest(commodity, protocol, cls2idx, split_recs, *, seed, val_frac, registry_hash, extra=None): """Assemble a v2 manifest dict from per-split (path, source, class) records. split_recs: {"train":[...], "val":[...], "test":[...]} of (path, source, class). cls2idx: canonical class name -> contiguous index for THIS manifest. """ samples = {sp: [] for sp in ("train", "val", "test")} meta = {sp: [] for sp in ("train", "val", "test")} # per-class, per-split counts and the set of (source:raw-ish) provenance counts = {c: {"train": 0, "val": 0, "test": 0} for c in cls2idx} src_provenance = defaultdict(set) # class -> {"source:class"} sources_seen = set() for sp in ("train", "val", "test"): for p, s, c in split_recs[sp]: idx = cls2idx[c] samples[sp].append([p, idx]) meta[sp].append({"commodity": commodity, "source": s, "class": c}) counts[c][sp] += 1 src_provenance[c].add(f"{s}:{c}") sources_seen.add(s) classes = [] for c in sorted(cls2idx, key=lambda x: cls2idx[x]): cnt = counts[c] classes.append({ "name": c, "index": cls2idx[c], "commodity": commodity, "sources": sorted(src_provenance[c]), "total": cnt["train"] + cnt["val"] + cnt["test"], "train": cnt["train"], "val": cnt["val"], "test": cnt["test"], }) source_map = {s: i for i, s in enumerate(sorted(sources_seen))} manifest = { "version": 2, "seed": seed, "protocol": protocol, "commodity": commodity, "splits": {"train": 1 - val_frac, "val": val_frac, "test": "held-out source" if protocol == "leave-one-source-out" else val_frac}, "classes": classes, "samples": samples, "meta": meta, "source_map": source_map, "registry_hash": registry_hash, # Placeholder; populated by the dedup step (audit_dedup / make_dedup_manifests). "dedup": {"method": "PENDING — run audit_dedup.py + make_dedup_manifests.py", "cross_source_verified_dups": None, "removed_from_test": 0, "test_before": len(samples["test"]), "test_after": len(samples["test"])}, } if extra: manifest.update(extra) return manifest # ── the three manifest families, per commodity ───────────────────────────────── def emit_for_commodity(commodity, records, *, outdir, val_frac, seed, registry_hash): """Emit per-source, overlap, and LOSO manifests for one commodity. records: list of (path, source, canonical_class, raw_label). Returns list of (Path, manifest_dict) written. """ recs3 = [(p, s, c) for (p, s, c, _raw) in records] # drop raw for assembly by_src_cls = defaultdict(lambda: defaultdict(list)) for p, s, c in recs3: by_src_cls[s][c].append(p) sources = sorted(by_src_cls) written = [] def _dump(name, manifest): out = Path(outdir) / name out.parent.mkdir(parents=True, exist_ok=True) json.dump(manifest, open(out, "w"), indent=2) written.append((out, manifest)) nc = len(manifest["classes"]) s = manifest["samples"] print(f" {out.name}: {nc} classes | " f"train {len(s['train'])} val {len(s['val'])} test {len(s['test'])}") # classes present per source, and overlap (>=2 sources) classes_by_src = {s: set(by_src_cls[s]) for s in sources} all_classes = set().union(*classes_by_src.values()) if classes_by_src else set() overlap = sorted(c for c in all_classes if sum(c in classes_by_src[s] for s in sources) >= 2) # (1) per-source manifests — renumber classes contiguously within the source for s in sources: scls = sorted(classes_by_src[s]) cls2idx = {c: i for i, c in enumerate(scls)} tr, va = _split_records([(p, s, c) for (p, ss, c) in recs3 if ss == s], set(scls), val_frac, seed) mf = _build_manifest(commodity, "source-specific", cls2idx, {"train": tr, "val": va, "test": []}, seed=seed, val_frac=val_frac, registry_hash=registry_hash, extra={"single_source": s}) _dump(f"manifest_{commodity}_src_{s}.json", mf) # (2) overlap-only manifests (pooled + per-source asymmetric probes) if overlap: cls2idx = {c: i for i, c in enumerate(overlap)} tr, va = _split_records([r for r in recs3 if r[2] in overlap], set(overlap), val_frac, seed) mf = _build_manifest(commodity, "overlap-only", cls2idx, {"train": tr, "val": va, "test": []}, seed=seed, val_frac=val_frac, registry_hash=registry_hash) _dump(f"manifest_{commodity}_overlap.json", mf) for s in sources: srecs = [r for r in recs3 if r[1] == s and r[2] in overlap] if not srecs: continue tr, va = _split_records(srecs, set(overlap), val_frac, seed) mf = _build_manifest(commodity, "overlap-only", cls2idx, {"train": tr, "val": va, "test": []}, seed=seed, val_frac=val_frac, registry_hash=registry_hash, extra={"single_source": s}) _dump(f"manifest_{commodity}_overlap_{s}.json", mf) else: print(f" [overlap] commodity={commodity!r}: no class shared by >=2 sources") # (3) Leave-One-Source-Out — train on others, test on the entire held-out source if len(sources) < 2: print(f" [loso] commodity={commodity!r}: need >=2 sources, have {len(sources)}") return written for held in sources: train_srcs = [s for s in sources if s != held] train_classes = set().union(*[classes_by_src[s] for s in train_srcs]) common = sorted(set(classes_by_src[held]) & train_classes) if not common: print(f" [loso/skip] held={held!r}: no class shared with training sources") continue cls2idx = {c: i for i, c in enumerate(common)} tr, va = _split_records([(p, s, c) for (p, s, c) in recs3 if s in train_srcs and c in common], set(common), val_frac, seed) test = [(p, held, c) for c in common for p in by_src_cls[held][c]] mf = _build_manifest(commodity, "leave-one-source-out", cls2idx, {"train": tr, "val": va, "test": test}, seed=seed, val_frac=val_frac, registry_hash=registry_hash, extra={"held_out_source": held, "train_sources": train_srcs, "degenerate_loso": len(sources) < 3}) _dump(f"manifest_{commodity}_loso_{held}.json", mf) if len(sources) < 3: print(f" [warn] commodity={commodity!r}: only {len(sources)} sources; LOSO " f"manifests flagged degenerate_loso=true (protocol requires >=3).") return written # ── dedup hook (reuses the existing, validated tooling — NOT reimplemented) ────── def run_dedup(written): """Wire the mandatory pixel-verified dedup step (BENCHMARK_PROTOCOL.md sec.5). This intentionally does not duplicate audit_dedup.py. For each emitted manifest it shells out conceptually to make_dedup_manifests.py (train/test leakage) and, per commodity, to audit_dedup.py (cross-source hard gate). Kept as a documented stub so the smoke path stays pure/dependency-light (no PIL, no real images). """ print("\n[dedup] mandatory pixel-verified de-duplication is a separate step:") print(" per manifest : python make_dedup_manifests.py --manifest # (L) train/test leakage") print(" per commodity: python audit_dedup.py --base # (X) cross-source HARD gate (must be 0)") print(f" ({len(written)} manifests emitted; dedup blocks are PENDING until run.)") # ── synthetic registry for the CPU smoke path (no images needed) ──────────────── def synthetic_registry_and_files(seed=0): """Build a fake multi-commodity, multi-source registry + dummy file lists. 2 commodities x 3 sources, with deliberate overlap and source-specific classes, so every manifest family (per-source, overlap, LOSO) is exercised. Paths are synthetic strings; no file ever needs to exist. """ rng = np.random.default_rng(seed) def files(commodity, source, classes, n_per): fl = [] for c in classes: for i in range(n_per + int(rng.integers(0, 3))): fl.append(f"_smoke/{commodity}/{source}/{c}/img_{i:03d}.jpg") return fl # rice: 3 sources, shared {basmati, jasmine}, plus per-source extras + a synonym rice_files = { "koklu": files("rice", "koklu", ["Basmati", "Jasmine", "Arborio"], 12), "bangladeshi": files("rice", "bangladeshi", ["basmati_long", "Jasmine", "Miniket"], 10), "bina": files("rice", "bina", ["Basmati", "Jasmine", "BR29"], 8), } # spices: 3 sources, shared {black_pepper, cumin}, plus extras spice_files = { "indian": files("spices", "indian", ["Black Pepper", "Cumin", "Clove"], 11), "spice_spectrum": files("spices", "spice_spectrum", ["black pepper", "cumin", "Cardamom"], 9), "market": files("spices", "market", ["Black Pepper", "Cumin", "Fenugreek"], 7), } registry = { "rice": { "sources": {k: f"_smoke/rice/{k}/*/*.jpg" for k in rice_files}, "synonyms": {"basmati_long": "basmati"}, # canonicalize to 'basmati' }, "spices": { "sources": {k: f"_smoke/spices/{k}/*/*.jpg" for k in spice_files}, "synonyms": {}, # slugify handles 'Black Pepper' }, } file_lists = {"rice": rice_files, "spices": spice_files} return registry, file_lists # ── self-check for the smoke path ─────────────────────────────────────────────── def _validate(manifest): """Assert the v2 invariants from BENCHMARK_PROTOCOL.md sec.2.""" for sp in ("train", "val", "test"): assert len(manifest["meta"][sp]) == len(manifest["samples"][sp]), \ f"meta/samples length mismatch in {sp}" idx2name = {c["index"]: c["name"] for c in manifest["classes"]} # contiguous 0..k-1 idxs = sorted(c["index"] for c in manifest["classes"]) assert idxs == list(range(len(idxs))), f"non-contiguous class indices: {idxs}" for sp in ("train", "val", "test"): for (p, lbl), mt in zip(manifest["samples"][sp], manifest["meta"][sp]): assert idx2name[lbl] == mt["class"], "label<->class mismatch" assert mt["commodity"] == manifest["commodity"] or manifest["commodity"] == "*" def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--registry", help="path to registry JSON " "{commodity:{sources:{src:glob},synonyms:{...}}}") ap.add_argument("--smoke", action="store_true", help="run the CPU smoke path on a synthetic registry (no images)") ap.add_argument("--val-frac", type=float, default=0.15) ap.add_argument("--seed", type=int, default=42) ap.add_argument("--outdir", default="outputs") ap.add_argument("--no-dedup-hint", action="store_true", help="suppress the dedup next-step reminder") args = ap.parse_args() file_lists = None if args.smoke: registry, file_lists = synthetic_registry_and_files(seed=args.seed) if not args.outdir or args.outdir == "outputs": args.outdir = "outputs/_smoke_manifests" print(f"[smoke] synthetic registry: commodities={list(registry)}; " f"outdir={args.outdir}") elif args.registry: registry = json.load(open(args.registry)) else: ap.error("provide --registry or --smoke") registry_hash = _registry_hash(registry) resolved = resolve_registry(registry, file_lists=file_lists) all_written = [] for commodity, records in resolved.items(): srcs = sorted({s for _, s, _, _ in records}) ncls = len({c for _, _, c, _ in records}) print(f"\ncommodity={commodity!r}: {len(records)} samples, " f"sources={srcs}, classes={ncls}") written = emit_for_commodity( commodity, records, outdir=args.outdir, val_frac=args.val_frac, seed=args.seed, registry_hash=registry_hash) all_written += written if args.smoke: for _path, mf in all_written: _validate(mf) print(f"\n[smoke] OK — {len(all_written)} manifests written & validated " f"(v2 invariants hold).") if not args.no_dedup_hint: run_dedup(all_written) print(f"\nwrote {len(all_written)} manifests to {args.outdir}.") if __name__ == "__main__": main()