#!/usr/bin/env python3 """run_bench.py — THE CORPUS-COHERENCE BENCH. Executes PREREG-COHERENCE.md (sealed sha256 4bbe9548945d1bd849b355a080b49c481c067e21fd105e98d8d6d30c9f2fe5f2), ratified §13 + AMENDMENT A-R4, under ruling D-20260903-02. MEASURED, NEVER GATING (D-20260831-17). CPU ONLY. The pinned instrument (`music/venvs/eval/.venv`, Amendment A1 of PREREG-ACE-LORA-FIRST-TRAIN-2026-08-27.md) carries a torch 2.8.0+cpu build with no CUDA runtime linked, so "never on a live seat's card" holds by construction, not by scheduling. Subcommands, each a registered STEP of §10.2: step0 instrument check — reproduce A1's receipt or abort plan build the crop plan — ffprobe durations + sha256 every source crop STEP 2 — hand the plan to crop_windows.sh, sha256 every crop embed STEP 4 — 10 s crops -> unit vectors metrics STEPS 5+6 (+ Set B) — METRIC 1, METRIC 2, RESULTS.json Corpora live on two boxes; `--box` selects this box's share. Crops travel, sources do not (§10.4: 20x less bytes). """ from __future__ import annotations import argparse import csv import hashlib import json import math import os import subprocess import sys import time import wave from pathlib import Path # --------------------------------------------------------------------------- # REGISTERED CONSTANTS — none of these may be changed without an amendment. # --------------------------------------------------------------------------- SEAL = "4bbe9548945d1bd849b355a080b49c481c067e21fd105e98d8d6d30c9f2fe5f2" WINDOW_PCTS = (25, 50, 75) # §4.2 WINDOW_SECONDS = 10 # §4.1 (CLAP's native window) SAMPLE_RATE = 48_000 # §4.2 N_EQUAL = 19 # §5.2, RATIFY-4 N_DRAWS = 200 # §5.3 N_PERM = 10_000 # §5.4 SEED_SUBSAMPLE = 20260903 # §5.3 SEED_BOOTSTRAP = 20260904 # §5.3 SEED_PERMUTE = 20260905 # §5.4 CKPT_SHA = "fae3e9c087f2909c28a09dc31c8dfcdacbc42ba44c70e972b58c1bd1caf6dedd" CKPT_BYTES = 2_352_471_003 FREEZE_SHA = "258a10ee9966ed60eb7839ce9ae9695c431277e4e324564c9fe2bf4e2550ebb1" A1_RECEIPT = "CLAP-laion-music embedding: shape (1, 512), L2 norm 1.0000" HOME = Path.home() EVAL = HOME / "music" / "venvs" / "eval" CKPT = EVAL / "weights" / "music_audioset_epoch_15_esc_90.14.pt" AUDIO_EXT = {".wav", ".mp3", ".flac", ".m4a", ".aif", ".aiff", ".ogg", ".opus", ".alac"} # §2.2 + §2.3. `dataset` is the frozen training set (RATIFY-1: the unit of analysis); # `corpus_dir` is the superset that arm S3 (§4.4) measures. Hashes are §2.3 verbatim. CORPORA = { "chopin": dict( box="the evaluation box", label="Chopin (pd-shakedown)", dataset=HOME / "music/datasets/pd-shakedown/20260829T162506Z-stage0/dataset.json", dataset_sha="baf091ea1d251a84c6b96652874d706d78d0da54edc63d14357f76633b112d7d", corpus_dir=HOME / "music/corpus/pd-shakedown", corpus_recursive=False, train_n=19, corpus_n=24, ), "fm-control": dict( box="the training box", label="one-artist control (fm-control)", dataset=HOME / "music/datasets/fm-control/20260831T193904Z-fmctl0/dataset.json", dataset_sha="13b16b5314807d0c4d45d61e37ded51e5cd4e176a761e2e043f1303d58d358ac", corpus_dir=HOME / "music/corpus/fm-control/audio", corpus_recursive=False, train_n=24, corpus_n=30, manifest=HOME / "music/corpus/fm-control/manifest.json", manifest_sha="11a12a746a929b8046d09905bf46ed98abefa3ae160859ee334760769563754c", ), "bach": dict( box="the evaluation box", label="Bach (bach-shakedown)", dataset=HOME / "music/datasets/bach-shakedown/20260830T104030Z-bach0/dataset.json", dataset_sha="5460a13eaf487e0bd2689cec495e0883735b598adddc7e3b86cb265ebd37cbcf", corpus_dir=HOME / "music/corpus/bach-shakedown/audio", corpus_recursive=False, train_n=63, corpus_n=79, ), "sousa": dict( box="the evaluation box", label="Sousa (sousa-shakedown)", dataset=HOME / "music/datasets/sousa-shakedown/20260830T004943Z-sousa0/dataset.json", dataset_sha="5de3991c18e356be0cb82a9b08c0f8699b24306b3acea0a5285664245d077054", corpus_dir=HOME / "music/corpus/sousa-shakedown/audio", corpus_recursive=False, train_n=84, corpus_n=106, ), "mnml": dict( box="the training box", label="minimal techno (mnml-shakedown)", dataset=HOME / "music/datasets/mnml-shakedown/20260831T023450Z-mnml0/dataset.json", dataset_sha="d26114c0d3828f1b2aaab947db6e676fc5a617cfbe3def62508c1924b8ae8333", corpus_dir=HOME / "music/corpus/mnml-shakedown/audio", corpus_recursive=False, train_n=159, corpus_n=208, manifest=HOME / "music/corpus/mnml-shakedown/manifest.json", manifest_sha="73040ef2e45694efef61e308ecb333c10f7aa21580fc086826a42245bd64e41f", ), } # The order in which corpora consume the registered RNG streams. §5.3 fixes the seeds but # not the order; this is the §2.2 table order, recorded here so the draws are reproducible. CORPUS_ORDER = ("chopin", "fm-control", "bach", "sousa", "mnml") # §6.2 — Set A, the six primary pairs, keyed on output_sha256 (NEVER clip_id: the base arm of # pairs 3/5 and 4/6 is the same audio and the clip_ids collide too). SET_A_PAIRS = [ dict(n=1, adapter="mnml", scale=1.0, caption="A", seed=42, base="d6b62483885e", adapt="b1627a2f0972"), dict(n=2, adapter="mnml", scale=1.0, caption="B", seed=4242, base="b3f55e8eb66c", adapt="2cef5f056c7b"), dict(n=3, adapter="mnml", scale=1.0, caption="C", seed=777, base="e71b8c65156c", adapt="0327d0699971"), dict(n=4, adapter="mnml", scale=1.0, caption="D", seed=7777, base="5889ca518f8b", adapt="0cecc9a342b5"), dict(n=5, adapter="fm-control", scale=1.0, caption="C", seed=777, base="e71b8c65156c", adapt="5db77e0345b0"), dict(n=6, adapter="fm-control", scale=1.0, caption="D", seed=7777, base="5889ca518f8b", adapt="f0ebb8d25803"), ] # §6.2 secondary — the adapter-strength ladder. Reported as a curve, never pooled into the # sign test (the renders share seeds and are not independent). SET_A_LADDER = [ dict(rung="mnml 0.35", caption="C", seed=777, sha="32fed7511416"), dict(rung="mnml 0.5", caption="C", seed=777, sha="bf3bb89431fa"), dict(rung="mnml 0.7", caption="A", seed=42, sha="3ce6690db1ce"), dict(rung="mnml 0.7", caption="B", seed=4242, sha="f931809a479e"), dict(rung="mnml 1.0 ep5", caption="C", seed=777, sha="d69a459f9d49"), ] SET_A_DIRS = [ HOME / "music/out/mnml-shakedown/listen-1", HOME / "music/out/mnml-shakedown/listen-2", HOME / "music/out/fm-control/round-3", ] # A-R4 — Set B. Round 4 is 12 captions x 4 arms at 30 s, seed 4000+index. SET_B_DIRS = [ HOME / "music/out/mnml-shakedown/round-4", HOME / "music/out/mnml-round4/round-4", ] SET_B_ARMS = ("base", "mnml-0.5", "mnml-1.0", "fm-control-1.0") CENTROID_OF_ARM = {"mnml": "mnml", "mnml-0.5": "mnml", "mnml-1.0": "mnml", "fm-control": "fm-control", "fm-control-1.0": "fm-control"} # --------------------------------------------------------------------------- # small helpers # --------------------------------------------------------------------------- def log(msg: str) -> None: print(f"[{time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}] {msg}", flush=True) def sha256_file(path: Path, chunk: int = 1 << 20) -> str: h = hashlib.sha256() with open(path, "rb") as fh: for blk in iter(lambda: fh.read(chunk), b""): h.update(blk) return h.hexdigest() def wav_info(path: Path) -> tuple[int, int, int]: """(frames, samplerate, channels) of a PCM WAV, via the stdlib — so the cropping box needs nothing installed (the training box carries no `soundfile`).""" with wave.open(str(path), "rb") as w: return w.getnframes(), w.getframerate(), w.getnchannels() def read_wav_mono_f32(path: Path): """16-bit PCM WAV -> float32 mono in [-1, 1), stdlib + numpy only.""" import numpy as np with wave.open(str(path), "rb") as w: assert w.getsampwidth() == 2, f"{path}: expected 16-bit PCM" frames, sr, ch = w.getnframes(), w.getframerate(), w.getnchannels() raw = w.readframes(frames) a = np.frombuffer(raw, dtype=" 1: a = a.reshape(-1, ch).mean(axis=1) return a, sr def ffprobe_duration(path: Path) -> float: """True duration from the SOURCE file — never dataset.json, whose `duration` is clamped at the 240 s preprocessing cap (§4.2).""" out = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(path)], capture_output=True, text=True, check=True).stdout.strip() return float(out) def window_start(pct: int, duration: float) -> float: """§10.3: START = max(0, pct * true_duration - 5), clamped so the window stays inside.""" start = (pct / 100.0) * duration - (WINDOW_SECONDS / 2.0) start = min(start, duration - WINDOW_SECONDS) return round(max(0.0, start), 6) def load_training_paths(spec: dict) -> list[Path]: """RATIFY-1: the unit of analysis is the `samples` array of the run's dataset.json.""" doc = json.loads(spec["dataset"].read_text()) samples = doc["samples"] if isinstance(doc, dict) else doc return [Path(s["audio_path"]) for s in samples] def list_corpus_paths(spec: dict) -> list[Path]: """Arm S3 (§4.4): the full corpus directory, not the training set.""" d: Path = spec["corpus_dir"] it = d.rglob("*") if spec.get("corpus_recursive") else d.iterdir() return sorted(p for p in it if p.is_file() and p.suffix.lower() in AUDIO_EXT) def crop_name(corpus: str, src_sha: str, pct: int, mode: str) -> str: return f"{corpus}__{src_sha[:16]}__w{pct:02d}__{mode}.wav" def render_index(dirs: list[Path]) -> dict[str, Path]: """sha256(12) -> path for every render WAV, top level only (never `_engine/`).""" idx: dict[str, Path] = {} for d in dirs: if not d.is_dir(): continue for p in sorted(d.glob("*.wav")): idx.setdefault(sha256_file(p)[:12], p) return idx # --------------------------------------------------------------------------- # STEP 0 — instrument check (§10.2). No corpus is touched. # --------------------------------------------------------------------------- def cmd_step0(args) -> int: log("STEP 0 — instrument check (no corpus touched)") freeze = EVAL / "freeze-A1.txt" fs = sha256_file(freeze) log(f"STEP 0 freeze-A1.txt sha256 = {fs}") if fs != FREEZE_SHA: log(f"STEP 0 ABORT: freeze sha mismatch, expected {FREEZE_SHA}") return 2 log(f"STEP 0 freeze-A1.txt lines = {len(freeze.read_text().splitlines())} MATCHES A1") size = CKPT.stat().st_size cs = sha256_file(CKPT) log(f"STEP 0 {CKPT.name} bytes = {size} sha256 = {cs}") if cs != CKPT_SHA or size != CKPT_BYTES: log("STEP 0 ABORT: checkpoint hash/size mismatch") return 2 log("STEP 0 checkpoint MATCHES A1 pin") # weights.sha256 as shipped, re-verified rather than trusted ws = (EVAL / "weights" / "weights.sha256").read_text() log("STEP 0 weights.sha256 on disk:\n" + ws.strip()) import numpy as np import torch import laion_clap import soundfile as sf log(f"STEP 0 python {sys.version.split()[0]} torch {torch.__version__} " f"numpy {np.__version__} laion_clap {getattr(laion_clap, '__version__', 'n/a')}") if torch.cuda.is_available(): log("STEP 0 ABORT: this torch reports CUDA available — the pinned instrument is CPU-only") return 2 log("STEP 0 torch.cuda.is_available() = False (CPU-only by construction, §3.1)") # A1's live verification: a 25 s clip cut from the pd-shakedown corpus, loudness-matched # to the §1 operating point. A1 does not name the clip; this run names its own, cut with # the registered conditioning, and reproduces A1's shape/norm receipt. verify_src = Path(args.verify_src) tmp = Path(os.environ["TMPDIR"]) / "step0-verify.wav" dur = ffprobe_duration(verify_src) start = round(max(0.0, min(dur / 2.0 - 12.5, dur - 25.0)), 6) subprocess.run( ["ffmpeg", "-nostdin", "-v", "error", "-ss", str(start), "-t", "25", "-i", str(verify_src), "-ar", "48000", "-af", "loudnorm=I=-16:TP=-1.0:LRA=11:print_format=summary", "-c:a", "pcm_s16le", "-y", str(tmp)], check=True) log(f"STEP 0 verification clip: {verify_src.name} dur={dur:.3f}s start={start}s 25 s " f"@ -16 LUFS / -1 dBTP / 48 kHz stereo sha256={sha256_file(tmp)}") t0 = time.time() model = laion_clap.CLAP_Module(enable_fusion=False, amodel="HTSAT-base") model.load_ckpt(ckpt=str(CKPT)) # explicit: never the 630k default (§3.2a) log(f"STEP 0 CLAP_Module(enable_fusion=False, amodel='HTSAT-base') + explicit ckpt " f"loaded in {time.time() - t0:.1f}s") audio, sr = sf.read(str(tmp), dtype="float32", always_2d=True) mono = audio.mean(axis=1) emb = model.get_audio_embedding_from_data(x=mono[None, :], use_tensor=False) emb = np.asarray(emb, dtype=np.float64) norm = float(np.linalg.norm(emb[0])) receipt = f"CLAP-laion-music embedding: shape {tuple(emb.shape)}, L2 norm {norm:.4f}" log("STEP 0 RECEIPT: " + receipt) log("STEP 0 A1 REQUIRES: " + A1_RECEIPT) if receipt != A1_RECEIPT: log("STEP 0 ABORT: A1's receipt was not reproduced") return 2 log("STEP 0 PASS — A1's receipt reproduced exactly; the corpora may now be read.") return 0 # --------------------------------------------------------------------------- # STEP 1 — cross-box crop equivalence (§4.3) # --------------------------------------------------------------------------- def cmd_equiv(args) -> int: """Cut the three registered crops of one nominated track and print their sha256s. Run on both boxes on the same track; the caller compares.""" src = Path(args.src) outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) dur = ffprobe_duration(src) log(f"STEP 1 {src} ffprobe duration = {dur:.6f}s src sha256 = {sha256_file(src)}") for pct in WINDOW_PCTS: start = window_start(pct, dur) out = outdir / f"equiv__w{pct:02d}__norm.wav" subprocess.run([str(Path(__file__).parent / "crop_windows.sh"), "__one", str(out), str(src), str(start), "norm"], check=True) log(f"STEP 1 w{pct:02d} start={start} sha256={sha256_file(out)} bytes={out.stat().st_size}") return 0 # --------------------------------------------------------------------------- # STEP 2a — the crop plan # --------------------------------------------------------------------------- def cmd_plan(args) -> int: box = args.box work = Path(args.workdir) (work / "crops").mkdir(parents=True, exist_ok=True) rows: list[dict] = [] seen_out: set[str] = set() t0 = time.time() for corpus in CORPUS_ORDER: spec = CORPORA[corpus] if spec["box"] != box: continue ds_sha = sha256_file(spec["dataset"]) log(f"PLAN {corpus}: dataset.json sha256 = {ds_sha}") if ds_sha != spec["dataset_sha"]: log(f"PLAN ABORT (§2.3): {corpus} dataset.json hash mismatch, " f"expected {spec['dataset_sha']}") return 2 if "manifest" in spec: m_sha = sha256_file(spec["manifest"]) log(f"PLAN {corpus}: manifest.json sha256 = {m_sha}") if m_sha != spec["manifest_sha"]: log(f"PLAN ABORT (§2.3): {corpus} manifest.json hash mismatch") return 2 train = load_training_paths(spec) full = list_corpus_paths(spec) log(f"PLAN {corpus}: train n={len(train)} (expect {spec['train_n']}), " f"corpus n={len(full)} (expect {spec['corpus_n']})") if len(train) != spec["train_n"] or len(full) != spec["corpus_n"]: log(f"PLAN ABORT (§2.2): {corpus} counts differ from the registered table") return 2 train_set = {p.resolve() for p in train} for path in full: rp = path.resolve() if not path.exists(): log(f"PLAN ABORT: missing source {path}") return 2 src_sha = sha256_file(path) dur = ffprobe_duration(path) in_train = rp in train_set # norm crops cover primary + S2 (a subset) + S3; raw crops cover S1 and are only # needed for the training sets, which are what S1 recomputes. modes = ["norm"] + (["raw"] if in_train else []) for mode in modes: for pct in WINDOW_PCTS: name = crop_name(corpus, src_sha, pct, mode) if name in seen_out: continue # identical bytes listed twice inside one corpus seen_out.add(name) rows.append(dict( corpus=corpus, 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"{window_start(pct, dur):.6f}", out_name=name, )) if args.renders: which = args.renders dirs = SET_A_DIRS if which == "A" else SET_B_DIRS want = None if which == "A": want = {p[k] for p in SET_A_PAIRS for k in ("base", "adapt")} want |= {r["sha"] for r in SET_A_LADDER} idx = render_index(dirs) log(f"PLAN renders set {which}: {len(idx)} distinct sha256 across {len(dirs)} dirs") if want is not None: missing = want - set(idx) if missing: log(f"PLAN ABORT (§6.2): render sha256 prefixes not found: {sorted(missing)}") return 2 log(f"PLAN renders set A: all {len(want)} registered sha256 prefixes located") for sha12, path in sorted(idx.items()): src_sha = sha256_file(path) dur = ffprobe_duration(path) for pct in WINDOW_PCTS: name = crop_name(f"render{which}", src_sha, pct, "norm") if name in seen_out: continue seen_out.add(name) rows.append(dict( corpus=f"render{which}", 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"{window_start(pct, dur):.6f}", out_name=name, )) plan_csv = work / f"crops-plan-{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 = work / f"crops-plan-{box}.tsv" with open(plan_tsv, "w") as fh: for r in rows: fh.write(f"{work / '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"]) < WINDOW_SECONDS] log(f"PLAN {len(rows)} crops planned on {box} ({plan_csv}) " f"sources shorter than {WINDOW_SECONDS}s: {len(short)} " f"[{time.time() - t0:.1f}s]") return 0 # --------------------------------------------------------------------------- # STEP 2b — cut, then sha256 every crop # --------------------------------------------------------------------------- def cmd_crop(args) -> int: work = Path(args.workdir) plan_csv = work / f"crops-plan-{args.box}.csv" plan_tsv = work / f"crops-plan-{args.box}.tsv" t0 = time.time() rc = subprocess.call([str(Path(__file__).parent / "crop_windows.sh"), str(plan_tsv), str(args.par)]) log(f"CROP crop_windows.sh rc={rc} in {time.time() - t0:.1f}s") rows = list(csv.DictReader(open(plan_csv))) out_rows = [] bad = 0 for r in rows: p = work / "crops" / r["out_name"] if not p.exists() or p.stat().st_size == 0: log(f"CROP MISSING/EMPTY {p}") bad += 1 continue frames, sr, ch = wav_info(p) r = dict(r) r["crop_sha256"] = sha256_file(p) r["crop_bytes"] = p.stat().st_size r["crop_duration_s"] = f"{frames / sr:.6f}" r["crop_samplerate"] = sr r["crop_channels"] = ch out_rows.append(r) with open(work / f"crops-{args.box}.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(out_rows[0].keys())) w.writeheader() w.writerows(out_rows) log(f"CROP {len(out_rows)} crops written and hashed, {bad} bad, " f"total {time.time() - t0:.1f}s") return 2 if bad else 0 def cmd_verify_crops(args) -> int: """STEP 3's other half: re-verify every crop sha AFTER transfer.""" work = Path(args.workdir) n = ok = 0 for csv_path in sorted(work.glob("crops-*.csv")): if csv_path.name.startswith("crops-plan"): continue for r in csv.DictReader(open(csv_path)): p = work / "crops" / r["out_name"] n += 1 if p.exists() and sha256_file(p) == r["crop_sha256"]: ok += 1 else: log(f"VERIFY MISMATCH/MISSING {p}") log(f"VERIFY {ok}/{n} crops re-verified by sha256 after transfer") return 0 if ok == n else 2 # --------------------------------------------------------------------------- # STEP 4 — embed # --------------------------------------------------------------------------- def cmd_embed(args) -> int: import numpy as np import laion_clap work = Path(args.workdir) rows: list[dict] = [] for csv_path in sorted(work.glob("crops-*.csv")): if csv_path.name.startswith("crops-plan"): continue rows.extend(csv.DictReader(open(csv_path))) rows.sort(key=lambda r: r["out_name"]) log(f"EMBED {len(rows)} crops to embed") t0 = time.time() model = laion_clap.CLAP_Module(enable_fusion=False, amodel="HTSAT-base") model.load_ckpt(ckpt=str(CKPT)) log(f"EMBED instrument loaded in {time.time() - t0:.1f}s") vecs = np.zeros((len(rows), 512), dtype=np.float32) bs = args.batch t1 = time.time() for i in range(0, len(rows), bs): chunk = rows[i:i + bs] wavs = [] for r in chunk: a, sr = read_wav_mono_f32(work / "crops" / r["out_name"]) assert sr == SAMPLE_RATE, f"{r['out_name']} sr={sr}" need = SAMPLE_RATE * WINDOW_SECONDS if len(a) < need: a = np.pad(a, (0, need - len(a))) wavs.append(a[:need]) e = model.get_audio_embedding_from_data(x=np.stack(wavs), use_tensor=False) vecs[i:i + len(chunk)] = np.asarray(e, dtype=np.float32) if (i // bs) % 20 == 0: done = i + len(chunk) rate = done / max(time.time() - t1, 1e-9) log(f"EMBED {done}/{len(rows)} {rate:.2f} crops/s " f"eta {(len(rows) - done) / max(rate, 1e-9) / 60:.1f} min") log(f"EMBED {len(rows)} forwards in {time.time() - t1:.1f}s") norms = np.linalg.norm(vecs, axis=1) log(f"EMBED crop-vector L2 norms: min={norms.min():.4f} max={norms.max():.4f} " f"mean={norms.mean():.4f}") vecs = vecs / np.clip(norms, 1e-12, None)[:, None] np.savez_compressed(work / "embeddings.npz", crops=vecs) with open(work / "embeddings-index.csv", "w", newline="") as fh: w = csv.writer(fh) w.writerow(["row", "set", "corpus", "in_train", "source_sha256", "window_pct", "mode", "crop_sha256", "out_name", "source_path"]) for i, r in enumerate(rows): kind = "render" if r["corpus"].startswith("render") else "corpus" w.writerow([i, kind, r["corpus"], r["in_train"], r["source_sha256"], r["window_pct"], r["mode"], r["crop_sha256"], r["out_name"], r["source_path"]]) log(f"EMBED wrote embeddings.npz {(work / 'embeddings.npz').stat().st_size} bytes") return 0 # --------------------------------------------------------------------------- # STEPS 5 + 6 — the statistics # --------------------------------------------------------------------------- def d_pair(E): import numpy as np G = E @ E.T iu = np.triu_indices(len(E), k=1) return float((1.0 - G[iu]).mean()) def d_cent(E): import numpy as np c = E.mean(axis=0) c = c / np.linalg.norm(c) return float((1.0 - E @ c).mean()) def centroid(E): import numpy as np c = E.mean(axis=0) return c / np.linalg.norm(c) def equal_n_draws(E, rng, n=N_EQUAL, draws=N_DRAWS): """§5.3 point estimate: `draws` draws of n WITHOUT replacement.""" import numpy as np idx = np.arange(len(E)) # NOTE: the draw is taken even when len(E) == n — rng.choice then returns a permutation of # the same set, which is §5.3's registered degeneracy ("all 200 draws are the same set"). # Short-circuiting it would desynchronise the registered RNG stream, so it is not done. dp, dc = [], [] for _ in range(draws): s = rng.choice(idx, n, replace=False) dp.append(d_pair(E[s])) dc.append(d_cent(E[s])) return np.asarray(dp), np.asarray(dc) def bootstrap_draws(E, rng, n=N_EQUAL, draws=N_DRAWS): """§5.3 interval: `draws` draws of n WITH replacement, DISTINCT ids only — the duplicate-pair rule, registered because it is the easy way to get this wrong.""" import numpy as np dp, dc = [], [] for _ in range(draws): pick = np.unique(rng.choice(len(E), n, replace=True)) dp.append(d_pair(E[pick])) dc.append(d_cent(E[pick])) return np.asarray(dp), np.asarray(dc) def sign_test_one_sided(diffs): """Exact one-sided sign test (§6.3): P(X >= k | n, 0.5), zeros dropped.""" nz = [d for d in diffs if d != 0] n = len(nz) k = sum(1 for d in nz if d > 0) p = sum(math.comb(n, i) for i in range(k, n + 1)) / (2 ** n) if n else float("nan") return n, k, p def cmd_metrics(args) -> int: import numpy as np work = Path(args.workdir) out = Path(args.outdir) out.mkdir(parents=True, exist_ok=True) V = np.load(work / "embeddings.npz")["crops"] idx = list(csv.DictReader(open(work / "embeddings-index.csv"))) log(f"METRICS {V.shape[0]} crop vectors, {len(idx)} index rows") # ---- track vectors: L2-normalised mean of a track's three crops (§4.2) -------------- def track_vectors(pred): """pred(row) -> group key or None. Returns {key: {src_sha: vec}}.""" buckets: dict[str, dict[str, list[int]]] = {} for r in idx: key = pred(r) if key is None: continue buckets.setdefault(key, {}).setdefault(r["source_sha256"], []).append(int(r["row"])) outd = {} for key, tracks in buckets.items(): m = {} for sha, rowsi in tracks.items(): v = V[sorted(rowsi)].mean(axis=0) m[sha] = v / np.linalg.norm(v) outd[key] = m return outd ARMS = { # arm -> (predicate, human description) "primary": (lambda r: r["corpus"] if (r["set"] == "corpus" and r["mode"] == "norm" and r["in_train"] == "1") else None, "training sets, 3 windows, -16 LUFS (§4.2)"), "S1": (lambda r: r["corpus"] if (r["set"] == "corpus" and r["mode"] == "raw" and r["in_train"] == "1") else None, "training sets, 3 windows, NO loudness normalisation (§4.4 S1)"), "S2": (lambda r: r["corpus"] if (r["set"] == "corpus" and r["mode"] == "norm" and r["in_train"] == "1" and r["window_pct"] == "50") else None, "training sets, single centre window, -16 LUFS (§4.4 S2)"), "S3": (lambda r: r["corpus"] if (r["set"] == "corpus" and r["mode"] == "norm") else None, "FULL corpus directories, 3 windows, -16 LUFS (§4.4 S3)"), } results = {"seal": SEAL, "generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "seeds": dict(subsample=SEED_SUBSAMPLE, bootstrap=SEED_BOOTSTRAP, permute=SEED_PERMUTE), "n_equal": N_EQUAL, "n_draws": N_DRAWS, "n_perm": N_PERM, "corpus_rng_order": list(CORPUS_ORDER), "metric1": {}, "metric2": {}} m1_rows, draw_rows = [], [] arm_tracks = {} for arm, (pred, desc) in ARMS.items(): tv = track_vectors(pred) arm_tracks[arm] = tv # ONE rng per stream per arm, consumed in the registered CORPUS_ORDER (§5.3). rng_sub = np.random.default_rng(SEED_SUBSAMPLE) rng_boot = np.random.default_rng(SEED_BOOTSTRAP) stats = {} for corpus in CORPUS_ORDER: if corpus not in tv: continue shas = sorted(tv[corpus]) E = np.stack([tv[corpus][s] for s in shas]) dp_full, dc_full = d_pair(E), d_cent(E) dp_eq, dc_eq = equal_n_draws(E, rng_sub) dp_bs, dc_bs = bootstrap_draws(E, rng_boot) rec = dict( arm=arm, corpus=corpus, label=CORPORA[corpus]["label"], n=len(E), d_pair_full=round(dp_full, 6), d_cent_full=round(dc_full, 6), d_pair_equaln=round(float(dp_eq.mean()), 6), d_pair_equaln_sd=round(float(dp_eq.std(ddof=0)), 6), d_cent_equaln=round(float(dc_eq.mean()), 6), d_cent_equaln_sd=round(float(dc_eq.std(ddof=0)), 6), d_pair_ci_lo=round(float(np.percentile(dp_bs, 2.5)), 6), d_pair_ci_hi=round(float(np.percentile(dp_bs, 97.5)), 6), d_cent_ci_lo=round(float(np.percentile(dc_bs, 2.5)), 6), d_cent_ci_hi=round(float(np.percentile(dc_bs, 97.5)), 6), ) stats[corpus] = rec m1_rows.append(rec) for i in range(N_DRAWS): draw_rows.append(dict(arm=arm, corpus=corpus, draw=i, d_pair_subsample=f"{dp_eq[i]:.6f}", d_cent_subsample=f"{dc_eq[i]:.6f}", d_pair_bootstrap=f"{dp_bs[i]:.6f}", d_cent_bootstrap=f"{dc_bs[i]:.6f}")) # rank by the headline statistic order = sorted(stats, key=lambda c: stats[c]["d_pair_equaln"]) for rank, c in enumerate(order, 1): stats[c]["rank_d_pair"] = rank results["metric1"][arm] = dict(description=desc, corpora=stats) with open(out / "metric1-corpus-spread.csv", "w", newline="") as fh: cols = ["arm", "corpus", "label", "n", "d_pair_full", "d_cent_full", "d_pair_equaln", "d_pair_equaln_sd", "d_pair_ci_lo", "d_pair_ci_hi", "d_cent_equaln", "d_cent_equaln_sd", "d_cent_ci_lo", "d_cent_ci_hi", "rank_d_pair"] w = csv.DictWriter(fh, fieldnames=cols) w.writeheader() for r in m1_rows: w.writerow({k: r.get(k, "") for k in cols}) with open(out / "metric1-draws.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(draw_rows[0].keys())) w.writeheader() w.writerows(draw_rows) # ---- H1 permutation test (§5.4), primary arm --------------------------------------- tv = arm_tracks["primary"] E_w = np.stack([tv["mnml"][s] for s in sorted(tv["mnml"])]) E_c = np.stack([tv["fm-control"][s] for s in sorted(tv["fm-control"])]) n_w, n_c = len(E_w), len(E_c) pooled = np.concatenate([E_w, E_c]) G = pooled @ pooled.T def eq_dpair_from_G(members, rng, draws=N_DRAWS, n=N_EQUAL): members = np.asarray(members) if len(members) <= n: sub = G[np.ix_(members, members)] iu = np.triu_indices(len(members), k=1) return float((1.0 - sub[iu]).mean()) picks = np.array([rng.choice(members, n, replace=False) for _ in range(draws)]) sub = G[picks[:, :, None], picks[:, None, :]] iu = np.triu_indices(n, k=1) return float((1.0 - sub[:, iu[0], iu[1]]).mean()) rng_p = np.random.default_rng(SEED_PERMUTE) obs = (results["metric1"]["primary"]["corpora"]["mnml"]["d_pair_equaln"] - results["metric1"]["primary"]["corpora"]["fm-control"]["d_pair_equaln"]) all_idx = np.arange(n_w + n_c) t0 = time.time() ge = 0 perm_stats = np.zeros(N_PERM) for i in range(N_PERM): perm = rng_p.permutation(all_idx) stat = (eq_dpair_from_G(perm[:n_w], rng_p) - eq_dpair_from_G(perm[n_w:], rng_p)) perm_stats[i] = stat if stat >= obs: ge += 1 p_perm = (ge + 1) / (N_PERM + 1) log(f"METRICS permutation test: obs={obs:.6f} p_one_sided={p_perm:.6f} " f"({time.time() - t0:.1f}s)") results["metric1"]["permutation_H1"] = dict( statistic="equal-N (19) D_pair(mnml) - D_pair(fm-control), primary arm", observed=round(obs, 6), n_perm=N_PERM, n_ge_observed=ge, p_one_sided=round(p_perm, 6), seed=SEED_PERMUTE, perm_mean=round(float(perm_stats.mean()), 6), perm_p975=round(float(np.percentile(perm_stats, 97.5)), 6)) # H1 verdict (§5.4): primary rule = non-overlapping 95 % CIs on D_pair. w_ = results["metric1"]["primary"]["corpora"]["mnml"] c_ = results["metric1"]["primary"]["corpora"]["fm-control"] ci_disjoint_above = w_["d_pair_ci_lo"] > c_["d_pair_ci_hi"] results["metric1"]["H1"] = dict( rule_primary="wide corpus D_pair 95% CI entirely above the control's", ci_wide=[w_["d_pair_ci_lo"], w_["d_pair_ci_hi"]], ci_control=[c_["d_pair_ci_lo"], c_["d_pair_ci_hi"]], ci_supported=bool(ci_disjoint_above), permutation_supported=bool(p_perm < 0.05), agree=bool(ci_disjoint_above == (p_perm < 0.05))) # ---- A-C3 POST-HOC, NOT PRE-REGISTERED ------------------------------------------------ # §2.4 registered the two training sets as disjoint; they are not (21 of the control's 24 # tracks are byte-identical to mnml training tracks). This descriptive strips the shared # bytes so a reader can see how much of H1 rides on them. It is NOT a registered arm and # never replaces the headline. tvp = arm_tracks["primary"] shared = set(tvp["mnml"]) & set(tvp["fm-control"]) only_mnml = sorted(set(tvp["mnml"]) - shared) if only_mnml: E_om = np.stack([tvp["mnml"][s_] for s_ in only_mnml]) rng_a = np.random.default_rng(SEED_SUBSAMPLE) rng_b = np.random.default_rng(SEED_BOOTSTRAP) dp_e, _ = equal_n_draws(E_om, rng_a) dp_b, _ = bootstrap_draws(E_om, rng_b) results["metric1"]["POSTHOC_A_C3"] = dict( warning="POST-HOC, NOT PRE-REGISTERED (A-C3). Descriptive only.", shared_tracks=len(shared), mnml_train_minus_shared_n=len(only_mnml), d_pair_full=round(d_pair(E_om), 6), d_cent_full=round(d_cent(E_om), 6), d_pair_equaln=round(float(dp_e.mean()), 6), d_pair_ci_lo=round(float(np.percentile(dp_b, 2.5)), 6), d_pair_ci_hi=round(float(np.percentile(dp_b, 97.5)), 6)) # ---- METRIC 2 ---------------------------------------------------------------------- cents = {"mnml": centroid(np.stack([tv["mnml"][s] for s in sorted(tv["mnml"])])), "fm-control": centroid(np.stack([tv["fm-control"][s] for s in sorted(tv["fm-control"])]))} results["metric2"]["centroid_note"] = ( "centroids from the FULL training set of each adapter's corpus, primary arm " "(the centroid is a property of the corpus, not of a draw) — §6.1") rend = track_vectors(lambda r: r["corpus"] if r["set"] == "render" else None) rvec: dict[str, "np.ndarray"] = {} rpath: dict[str, str] = {} for setname, tracks in rend.items(): for sha, v in tracks.items(): rvec[sha[:12]] = v for r in idx: if r["set"] == "render": rpath[r["source_sha256"][:12]] = r["source_path"] def dists(sha12): v = rvec[sha12] return {c: float(1.0 - v @ cents[c]) for c in cents} rc_rows = [] for sha12 in sorted(rvec): d = dists(sha12) rc_rows.append(dict(sha256_12=sha12, path=rpath.get(sha12, ""), d_to_C_mnml=round(d["mnml"], 6), d_to_C_fm_control=round(d["fm-control"], 6))) with open(out / "metric2-render-centroid.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rc_rows[0].keys())) w.writeheader() w.writerows(rc_rows) pair_rows = [] def add_pairs(setlabel, pairs): diffs_by_adapter: dict[str, list[float]] = {} for p in pairs: own = CENTROID_OF_ARM[p["adapter"]] other = "fm-control" if own == "mnml" else "mnml" db, da = dists(p["base"]), dists(p["adapt"]) diff = db[own] - da[own] diff_other = db[other] - da[other] row = dict(set=setlabel, pair=p["n"], adapter=p["adapter"], scale=p["scale"], caption=p["caption"], seed=p["seed"], base_sha12=p["base"], adapter_sha12=p["adapt"], own_centroid=own, d_base_own=round(db[own], 6), d_adapter_own=round(da[own], 6), diff_own=round(diff, 6), direction="toward" if diff > 0 else ("away" if diff < 0 else "flat"), d_base_other=round(db[other], 6), d_adapter_other=round(da[other], 6), diff_other=round(diff_other, 6)) pair_rows.append(row) diffs_by_adapter.setdefault(p["adapter"], []).append(diff) return diffs_by_adapter a_diffs = add_pairs("A", SET_A_PAIRS) all_a = [d for p in SET_A_PAIRS for d in [next(r["diff_own"] for r in pair_rows if r["set"] == "A" and r["pair"] == p["n"])]] n6, k6, p6 = sign_test_one_sided(all_a) nm, km, pm = sign_test_one_sided(a_diffs.get("mnml", [])) nf, kf, pf = sign_test_one_sided(a_diffs.get("fm-control", [])) results["metric2"]["setA"] = dict( pooled=dict(n=n6, toward=k6, p_one_sided=round(p6, 6), note="the single registered inferential test of Set A (§6.3)"), mnml_only=dict(n=nm, toward=km, p_one_sided=round(pm, 6), note="DESCRIPTIVE by construction: min attainable one-sided p = " "1/2^4 = 0.0625 (§6.3)"), fm_control_only=dict(n=nf, toward=kf, p_one_sided=round(pf, 6), note="descriptive, n=2"), caveat="captions are identical modulo the adapter's trigger token, so the " "base/adapter difference is 'adapter + trigger', not 'adapter alone' (§6.2)") ladder = [] for r in SET_A_LADDER: d = dists(r["sha"]) ladder.append(dict(set="A-ladder", rung=r["rung"], caption=r["caption"], seed=r["seed"], sha256_12=r["sha"], d_to_C_mnml=round(d["mnml"], 6), d_to_C_fm_control=round(d["fm-control"], 6))) for p in SET_A_PAIRS: if p["adapter"] == "mnml": for tag, s in (("base", p["base"]), ("mnml 1.0", p["adapt"])): d = dists(s) ladder.append(dict(set="A-ladder", rung=tag, caption=p["caption"], seed=p["seed"], sha256_12=s, d_to_C_mnml=round(d["mnml"], 6), d_to_C_fm_control=round(d["fm-control"], 6))) results["metric2"]["setA_ladder"] = ladder # ---- Set B (A-R4) ------------------------------------------------------------------- setb = None b_records = None for d in SET_B_DIRS: if (d / "records.json").exists(): b_records = d break if b_records is not None: recs = json.loads((b_records / "records.json").read_text()) recs = recs["clips"] if isinstance(recs, dict) and "clips" in recs else recs by = {} for rec in (recs if isinstance(recs, list) else recs.get("records", [])): sha = (rec.get("output_sha256") or rec.get("sha256") or "")[:12] arm = rec.get("arm") pid = rec.get("prompt_id") or rec.get("prompt") or rec.get("id") if sha and arm and pid: by[(str(pid), arm)] = dict(sha=sha, seed=rec.get("seed")) prompts = sorted({k[0] for k in by}) log(f"METRICS Set B: {len(by)} records over {len(prompts)} prompts") bpairs = [] for i, pid in enumerate(prompts, 1): for arm, label in (("mnml-1.0", "B-mnml"), ("fm-control-1.0", "B-fm"), ("mnml-0.5", "B-half")): if (pid, "base") in by and (pid, arm) in by: bpairs.append(dict(n=i, adapter=arm, scale=float(arm.split("-")[-1]), caption=pid, seed=by[(pid, "base")]["seed"], base=by[(pid, "base")]["sha"], adapt=by[(pid, arm)]["sha"], _label=label)) missing = [p for p in bpairs if p["base"] not in rvec or p["adapt"] not in rvec] if missing: log(f"METRICS Set B: {len(missing)} pairs have unembedded renders — skipping") else: setb = {} for label in ("B-mnml", "B-fm", "B-half"): sel = [p for p in bpairs if p["_label"] == label] add_pairs(label, sel) diffs = [r["diff_own"] for r in pair_rows if r["set"] == label] n, k, pv = sign_test_one_sided(diffs) setb[label] = dict(n=n, toward=k, p_one_sided=round(pv, 6)) # H5 dose monotonicity per prompt: d(., C_mnml) non-increasing base -> 0.5 -> 1.0 mono, triples = 0, [] for pid in prompts: try: b = dists(by[(pid, "base")]["sha"])["mnml"] h = dists(by[(pid, "mnml-0.5")]["sha"])["mnml"] f = dists(by[(pid, "mnml-1.0")]["sha"])["mnml"] except KeyError: continue ok = (h <= b) and (f <= h) mono += int(ok) triples.append(dict(prompt=pid, d_base=round(b, 6), d_half=round(h, 6), d_full=round(f, 6), monotone=bool(ok))) setb["H5_dose_monotonicity"] = dict(monotone=mono, of=len(triples), per_prompt=triples) # H4: P01-P06 vs P07-P12 group medians of |effect| on d(., C_mnml) at 1.0 def med(vals): v = sorted(vals) m = len(v) return float("nan") if not m else (v[m // 2] if m % 2 else (v[m // 2 - 1] + v[m // 2]) / 2) near = [r["diff_own"] for r in pair_rows if r["set"] == "B-mnml" and str(r["caption"]) in {"P01", "P02", "P03", "P04", "P05", "P06"}] far = [r["diff_own"] for r in pair_rows if r["set"] == "B-mnml" and str(r["caption"]) in {"P07", "P08", "P09", "P10", "P11", "P12"}] setb["H4"] = dict(median_P01_P06=round(med(near), 6) if near else None, median_P07_P12=round(med(far), 6) if far else None, n_near=len(near), n_far=len(far), note="two group medians, no test (exploratory, A-R4)") results["metric2"]["setB"] = setb if setb is None: results["metric2"]["setB"] = "OWED — Round 4 had not landed when the bench ran (A-R4)" with open(out / "metric2-pairs.csv", "w", newline="") as fh: cols = ["set", "pair", "adapter", "scale", "caption", "seed", "base_sha12", "adapter_sha12", "own_centroid", "d_base_own", "d_adapter_own", "diff_own", "direction", "d_base_other", "d_adapter_other", "diff_other"] w = csv.DictWriter(fh, fieldnames=cols) w.writeheader() for r in pair_rows: w.writerow({k: r.get(k, "") for k in cols}) tvp2 = arm_tracks["primary"] results["corpus_overlap_A_C3"] = dict( fm_train_n=len(tvp2["fm-control"]), mnml_train_n=len(tvp2["mnml"]), shared_by_sha256=len(set(tvp2["mnml"]) & set(tvp2["fm-control"])), note="§2.4 registered these training sets as disjoint; measured, they are not (A-C3)") results["instrument"] = dict(laion_clap="1.1.7", torch="2.8.0+cpu", numpy="1.26.4", python="3.12.14", ckpt_sha256=CKPT_SHA, freeze_sha256=FREEZE_SHA, amodel="HTSAT-base", enable_fusion=False) (out / "RESULTS.json").write_text(json.dumps(results, indent=2, sort_keys=False) + "\n") log(f"METRICS wrote {out / 'RESULTS.json'}") return 0 def main() -> int: ap = argparse.ArgumentParser(description=__doc__) sub = ap.add_subparsers(dest="cmd", required=True) p = sub.add_parser("step0"); p.add_argument("--verify-src", required=True); p.set_defaults(fn=cmd_step0) p = sub.add_parser("equiv"); p.add_argument("--src", required=True) p.add_argument("--outdir", required=True); p.set_defaults(fn=cmd_equiv) p = sub.add_parser("plan"); p.add_argument("--box", required=True) p.add_argument("--workdir", required=True); p.add_argument("--renders", default=None) p.set_defaults(fn=cmd_plan) p = sub.add_parser("crop"); p.add_argument("--box", required=True) p.add_argument("--workdir", required=True); p.add_argument("--par", type=int, default=8) p.set_defaults(fn=cmd_crop) p = sub.add_parser("verify-crops"); p.add_argument("--workdir", required=True) p.set_defaults(fn=cmd_verify_crops) p = sub.add_parser("embed"); p.add_argument("--workdir", required=True) p.add_argument("--batch", type=int, default=8); p.set_defaults(fn=cmd_embed) p = sub.add_parser("metrics"); p.add_argument("--workdir", required=True) p.add_argument("--outdir", required=True); p.set_defaults(fn=cmd_metrics) args = ap.parse_args() return args.fn(args) if __name__ == "__main__": sys.exit(main())