#!/usr/bin/env python3 """plan_house_crops.py — the house corpus's crop plan, in the pinned instrument's own format. `PREREG-HOUSE-ARMS.md` §8: *"Amendment A1 of PREREG-COHERENCE.md pins this instrument, and it is reused, not rebuilt."* §8.3: **re-cutting the mnml crops is forbidden** — it re-opens the one thing that makes 0.4489 comparable. So this script does the ONE thing the pinned bench cannot do for a corpus that did not exist when it was sealed: it emits `crops-plan-.csv` and `crops-plan-.tsv` for the house corpus. It changes nothing else. Every constant it uses — the window percentages, the window length, the crop naming, the window-start formula, the hash function — is **imported from the sealed `run_bench.py`**, not copied, so a crop this plan names is byte-identical to a crop `cmd_plan` would have named. The `crop` and `embed` steps then run from the sealed file verbatim. The corpus key is ``house-wide``, and each track contributes: * three ``norm`` crops (the §4.2 primary policy: mono 48 kHz, −16 LUFS / −1 dBTP, 16-bit PCM) — for every track in the frozen corpus, training side and holdout; * three ``raw`` crops (level untouched — arm S1's sensitivity) — for training-set tracks only, exactly as ``cmd_plan`` does it for the five mnml-arc corpora. The split side rides the plan as its own column so the metrics step can read "training set", "holdout" and "full pool" — §9.2's three registered P-A sets — off one file. With ``--render-dir`` it plans the RENDER clips instead, under the corpus key ``render-house`` — the same three windows, the same ``norm`` policy, the same naming — which is what H3/H3b/H5 (§8.2) need: "for each base/adapter pair at the same caption and seed, is the adapter's render closer to its own training corpus's centroid?". A render is cropped exactly as a corpus track is, or the distance is measuring the crop policy rather than the audio. """ from __future__ import annotations import argparse import csv import importlib.util import json import sys from pathlib import Path CORPUS_KEY = "house-wide" def load_sealed_bench(path: Path): """Import the pinned run_bench.py as a module, by path, without running it.""" spec = importlib.util.spec_from_file_location("run_bench_pinned", path) mod = importlib.util.module_from_spec(spec) sys.modules["run_bench_pinned"] = mod spec.loader.exec_module(mod) return mod def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--bench", type=Path, default=Path.home() / "music/out/coherence-bench/run_bench.py", help="the SEALED run_bench.py; its helpers are imported, never copied") ap.add_argument("--bench-sha", default=None, help="assert the sealed file's sha256 before importing it") ap.add_argument("--audio-dir", type=Path, default=Path.home() / "music/corpus/house-wide/audio") ap.add_argument("--render-dir", type=Path, default=None, help="plan the RENDER wavs in this directory instead of the corpus") ap.add_argument("--split", type=Path, default=None, help="the freeze's dataset/split.json") ap.add_argument("--manifest", type=Path, default=None, help="the freeze's corpus/manifest.jsonl (carries every local_sha256)") ap.add_argument("--workdir", type=Path, required=True) ap.add_argument("--box", default="the training box") args = ap.parse_args() if args.bench_sha: import hashlib got = hashlib.sha256(args.bench.read_bytes()).hexdigest() if got != args.bench_sha: print(f"ABORT: {args.bench} hashes to {got}, not {args.bench_sha}", file=sys.stderr) return 2 print(f"PLAN sealed instrument {args.bench.name} sha256 = {got} OK") rb = load_sealed_bench(args.bench) print(f"PLAN imported the sealed bench: WINDOW_PCTS={rb.WINDOW_PCTS} " f"WINDOW_SECONDS={rb.WINDOW_SECONDS} SAMPLE_RATE={rb.SAMPLE_RATE}") if args.render_dir: rows: list[dict] = [] seen: set[str] = set() wavs = sorted(p for p in args.render_dir.glob("*.wav")) if not wavs: print(f"ABORT: no wavs in {args.render_dir}", file=sys.stderr) return 2 for path in wavs: src_sha = rb.sha256_file(path) dur = rb.ffprobe_duration(path) for pct in rb.WINDOW_PCTS: name = rb.crop_name("render-house", src_sha, pct, "norm") if name in seen: continue seen.add(name) rows.append(dict( corpus="render-house", in_train=0, source_path=str(path), source_sha256=src_sha, source_duration_s=f"{dur:.6f}", window_pct=pct, mode="norm", start_s=f"{rb.window_start(pct, dur):.6f}", out_name=name, track_id=path.stem, artist_id="", split_side="render", )) (args.workdir / "crops").mkdir(parents=True, exist_ok=True) plan_csv = args.workdir / f"crops-plan-{args.box}.csv" with open(plan_csv, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) w.writeheader(); w.writerows(rows) plan_tsv = args.workdir / f"crops-plan-{args.box}.tsv" with open(plan_tsv, "w") as fh: for r in rows: fh.write(f"{args.workdir / 'crops' / r['out_name']}\t{r['source_path']}\t" f"{r['start_s']}\t{r['mode']}\n") print(f"PLAN {len(rows)} render crops planned over {len(wavs)} clips " f"({len(seen)} distinct); {plan_csv}") return 0 if not (args.split and args.manifest): print("ABORT: --split and --manifest are required unless --render-dir is given", file=sys.stderr) return 2 split = json.loads(args.split.read_text("utf-8")) manifest = [json.loads(l) for l in args.manifest.read_text("utf-8").splitlines() if l.strip()] held = {t for a in split["held_out"]["artists"] for t in a["track_ids"]} trained = {t for a in split["train"]["artists"] for t in a["track_ids"]} if held & trained: print("ABORT: split.json puts a track on both sides", file=sys.stderr) return 2 print(f"PLAN split.json: {len(trained)} training, {len(held)} held out, " f"{len(trained) + len(held)} total") (args.workdir / "crops").mkdir(parents=True, exist_ok=True) rows: list[dict] = [] seen: set[str] = set() for rec in sorted(manifest, key=lambda r: r["track_id"]): path = args.audio_dir / rec["filename"] if not path.exists(): print(f"ABORT: missing source {path}", file=sys.stderr) return 2 # The manifest's digest was computed on these very bytes at extraction; # re-hashing 5 GB here would only re-measure what MANIFEST.sha256 covers, # so the frozen digest is what names the crop — and it is asserted to be # the file's own by the corpus verify that precedes this step. src_sha = rec["local_sha256"] dur = rb.ffprobe_duration(path) in_train = rec["track_id"] in trained side = "train" if in_train else "held_out" for mode in ["norm"] + (["raw"] if in_train else []): for pct in rb.WINDOW_PCTS: name = rb.crop_name(CORPUS_KEY, src_sha, pct, mode) if name in seen: continue seen.add(name) rows.append(dict( corpus=CORPUS_KEY, in_train=int(in_train), source_path=str(path), source_sha256=src_sha, source_duration_s=f"{dur:.6f}", window_pct=pct, mode=mode, start_s=f"{rb.window_start(pct, dur):.6f}", out_name=name, track_id=rec["track_id"], artist_id=rec["artist_id"], split_side=side, )) plan_csv = args.workdir / f"crops-plan-{args.box}.csv" with open(plan_csv, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) plan_tsv = args.workdir / f"crops-plan-{args.box}.tsv" with open(plan_tsv, "w") as fh: for r in rows: fh.write(f"{args.workdir / 'crops' / r['out_name']}\t{r['source_path']}\t" f"{r['start_s']}\t{r['mode']}\n") short = [r for r in rows if float(r["source_duration_s"]) < rb.WINDOW_SECONDS] n_norm = sum(1 for r in rows if r["mode"] == "norm") print(f"PLAN {len(rows)} crops planned ({n_norm} norm + {len(rows) - n_norm} raw) " f"over {len(manifest)} tracks on {args.box}") print(f"PLAN sources shorter than {rb.WINDOW_SECONDS}s: {len(short)}") print(f"PLAN {plan_csv}") return 0 if __name__ == "__main__": raise SystemExit(main())