feat(emotion): overnight autonomous scale-up infra (#93)

Three tools for the unattended ~6h night run that scales the emotion GT corpus
and validates it across models:

- emotion_corpus_merge.py — fold model-authored GT batches into the corpus
  SAFELY: validate vs the fixed 12-emotion vocab + V/A range, dedup by slug AND
  by normalized artist|title (no track twice under two slugs), default the
  yt-dlp query. Idempotent.
- emotion_overnight.py — disk-safe (volume at 97%%), deadline-bounded grind:
  phase 1 fetch→whole-mix→delete mp3 over the whole corpus (bank the cheap
  signal first); phase 2 re-fetch→demucs stems→delete for as many as the
  deadline allows (grows the #92 stem evidence base). Per-track try/except,
  yt-dlp throttle + consecutive-failure backoff, audio purged after every track.
- emotion_drift.py — cross-model agreement on the GT itself: a second model
  (sonnet) labels the same tracks; high opus-vs-sonnet V/A drift or quadrant
  mismatch flags UNCERTAIN ground truth to down-weight, not trust. Single-model
  GT is a blind spot — a second pair of ears finds it.

Spirit of PLN's overnight brief: scale the dataset (the lever), and check
cross-model drift instead of trusting one model's priors.

armada/tide-table/emotion_corpus_merge.py

+#!/usr/bin/env python3
+"""emotion_corpus_merge — fold model-authored GT batches into emotion_corpus_gt.json.
+
+The GT-expansion workflow fans opus subagents over genre cells; each returns a batch
+of training-knowledge labels. This merges them SAFELY: validate against the fixed
+12-emotion vocabulary + V/A range, dedup by slug AND by normalized artist|title (so
+"Daft Punk – One More Time" can't sneak in twice under two slugs), default the yt-dlp
+query, and append. Idempotent — re-running with the same batch adds nothing.
+
+    python3 emotion_corpus_merge.py <batch1.json> [batch2.json …]   # merge & report
+    python3 emotion_corpus_merge.py --dry <batch.json>              # validate only
+
+Batch shape: {"tracks": [ {slug, artist, title, query?, emotions[], valence, arousal,
+genre[], structure?} ]}  (or a bare list). See [[project_hexa_emotion_convergence]]."""
+import json
+import re
+import sys
+import unicodedata
+from pathlib import Path
+
+import emotion_ontology as EMO
+
+HERE = Path(__file__).resolve().parent
+GT = HERE / "emotion_corpus_gt.json"
+VALID = set(EMO.NAMES)
+
+
+def _norm(s):
+    s = unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode().lower()
+    return re.sub(r"[^a-z0-9]+", "", s)
+
+
+def _slug(s):
+    s = unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode().lower()
+    return re.sub(r"[^a-z0-9]+", "-", s).strip("-")
+
+
+def _clean(t):
+    """Return a validated track dict, or (None, reason)."""
+    if not t.get("artist") or not t.get("title"):
+        return None, "missing artist/title"
+    emos = [e for e in (t.get("emotions") or []) if e in VALID]
+    if not emos:
+        return None, f"no valid emotion in {t.get('emotions')}"
+    try:
+        v, a = float(t["valence"]), float(t["arousal"])
+    except (KeyError, TypeError, ValueError):
+        return None, "bad/missing valence|arousal"
+    if not (-1.0 <= v <= 1.0 and -1.0 <= a <= 1.0):
+        return None, f"V/A out of range ({v},{a})"
+    slug = _slug(t.get("slug") or f"{t['artist']}-{t['title']}")
+    return {
+        "slug": slug,
+        "artist": t["artist"].strip(),
+        "title": t["title"].strip(),
+        "query": (t.get("query") or f"{t['artist']} {t['title']} official audio").strip(),
+        "emotions": emos[:3],
+        "valence": round(v, 2),
+        "arousal": round(a, 2),
+        "genre": [g for g in (t.get("genre") or []) if g][:4],
+        **({"structure": t["structure"]} if t.get("structure") else {}),
+    }, None
+
+
+def _load_batch(path):
+    d = json.loads(Path(path).read_text())
+    return d["tracks"] if isinstance(d, dict) and "tracks" in d else d
+
+
+def main():
+    args = sys.argv[1:]
+    dry = "--dry" in args
+    paths = [a for a in args if not a.startswith("--")]
+    if not paths:
+        sys.exit("usage: emotion_corpus_merge.py [--dry] <batch.json> …")
+
+    gt = json.loads(GT.read_text())
+    have_slug = {t["slug"] for t in gt["tracks"]}
+    have_name = {_norm(t["artist"]) + "|" + _norm(t["title"]) for t in gt["tracks"]}
+    added, dup, bad = [], 0, []
+
+    for path in paths:
+        for raw in _load_batch(path):
+            t, reason = _clean(raw)
+            if not t:
+                bad.append((raw.get("title", "?"), reason))
+                continue
+            key = _norm(t["artist"]) + "|" + _norm(t["title"])
+            if t["slug"] in have_slug or key in have_name:
+                dup += 1
+                continue
+            have_slug.add(t["slug"]); have_name.add(key)
+            added.append(t)
+
+    print(f"⛵ merge: +{len(added)} new · {dup} dups skipped · {len(bad)} invalid")
+    for title, reason in bad[:20]:
+        print(f"    ✗ {title}: {reason}")
+    if dry:
+        print("  (--dry: not written)")
+        return
+    gt["tracks"].extend(added)
+    GT.write_text(json.dumps(gt, ensure_ascii=False, indent=1))
+    print(f"  ✓ {GT.name} now {len(gt['tracks'])} tracks")
+
+
+if __name__ == "__main__":
+    main()

armada/tide-table/emotion_drift.py

+#!/usr/bin/env python3
+"""emotion_drift — cross-model agreement on the ground truth itself.
+
+The corpus GT is one model's training-knowledge read of each track's feeling. That's a
+prior, not gospel. So we ask a DIFFERENT model (Sonnet) to label the same tracks
+independently, and measure where the two disagree. High inter-model drift on a track =
+its GT is UNCERTAIN — either the track is genuinely ambiguous, or our 12-emotion
+ontology can't pin it. Those tracks should be down-weighted in calibration and surfaced
+for review, instead of trusted like the rest. (Spirit of PLN's "prompt sonnet, check
+drift" — single-model GT is a blind spot; a second pair of ears finds it.)
+
+    python3 emotion_drift.py <sonnet_labels.json>   # report + write emotion_drift.json
+
+sonnet_labels.json: [{slug|artist|title, emotions[], valence, arousal}, …] — matched to
+the opus GT by normalized artist|title. See [[project_hexa_emotion_convergence]]."""
+import json
+import re
+import sys
+import unicodedata
+from pathlib import Path
+
+import numpy as np
+
+HERE = Path(__file__).resolve().parent
+GT = HERE / "emotion_corpus_gt.json"
+OUT = HERE / "emotion_drift.json"
+
+
+def _norm(s):
+    s = unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode().lower()
+    return re.sub(r"[^a-z0-9]+", "", s)
+
+
+def _key(t):
+    return _norm(t.get("artist")) + "|" + _norm(t.get("title"))
+
+
+def _quad(v, a):
+    return ("+" if v >= 0 else "-") + ("+" if a >= 0 else "-")
+
+
+def main():
+    args = [a for a in sys.argv[1:] if not a.startswith("--")]
+    if not args:
+        sys.exit("usage: emotion_drift.py <sonnet_labels.json>")
+    opus = {(_key(t)): t for t in json.loads(GT.read_text())["tracks"]}
+    sondata = json.loads(Path(args[0]).read_text())
+    if isinstance(sondata, dict) and "tracks" in sondata:
+        sondata = sondata["tracks"]
+    son = {_key(t): t for t in sondata}
+
+    rows = []
+    for k, o in opus.items():
+        s = son.get(k)
+        if not s:
+            continue
+        try:
+            sv, sa = float(s["valence"]), float(s["arousal"])
+        except (KeyError, TypeError, ValueError):
+            continue
+        va_drift = float(np.hypot(o["valence"] - sv, o["arousal"] - sa))
+        oe, se = set(o.get("emotions", [])), set(s.get("emotions", []))
+        jac = len(oe & se) / len(oe | se) if (oe | se) else 0.0
+        rows.append({
+            "slug": o["slug"],
+            "artist": o["artist"], "title": o["title"],
+            "va_drift": round(va_drift, 3),
+            "quad_agree": _quad(o["valence"], o["arousal"]) == _quad(sv, sa),
+            "emo_jaccard": round(jac, 2),
+            "opus": f"{'/'.join(o.get('emotions', []))} {o['valence']:+.2f}/{o['arousal']:+.2f}",
+            "sonnet": f"{'/'.join(s.get('emotions', []))} {sv:+.2f}/{sa:+.2f}",
+        })
+    if not rows:
+        sys.exit("no tracks matched between opus GT and sonnet labels (check keys)")
+
+    n = len(rows)
+    mean_drift = float(np.mean([r["va_drift"] for r in rows]))
+    quad_agree = float(np.mean([r["quad_agree"] for r in rows]))
+    mean_jac = float(np.mean([r["emo_jaccard"] for r in rows]))
+    # uncertain GT: top-quartile V/A drift OR quadrant disagreement
+    thresh = float(np.quantile([r["va_drift"] for r in rows], 0.75))
+    uncertain = sorted([r for r in rows if r["va_drift"] >= thresh or not r["quad_agree"]],
+                       key=lambda r: -r["va_drift"])
+
+    print(f"⛵ cross-model GT drift — {n} tracks matched (opus vs sonnet)\n")
+    print(f"  mean V/A drift     : {mean_drift:.3f}")
+    print(f"  quadrant agreement : {quad_agree*100:.0f}%")
+    print(f"  emotion Jaccard    : {mean_jac:.2f}")
+    print(f"  uncertain-GT (drift ≥ p75={thresh:.2f} or quad mismatch): {len(uncertain)}\n")
+    for r in uncertain[:25]:
+        flag = "" if r["quad_agree"] else "  ⚠ quad"
+        print(f"    {r['va_drift']:.2f}  {r['artist'][:18]:<18} {r['title'][:22]:<22} "
+              f"O:{r['opus']:<24} S:{r['sonnet']}{flag}")
+
+    OUT.write_text(json.dumps({
+        "schema": "cross-model GT drift (opus vs sonnet) — uncertain-GT detector",
+        "n_matched": n,
+        "mean_va_drift": mean_drift,
+        "quad_agreement": quad_agree,
+        "mean_emo_jaccard": mean_jac,
+        "uncertain_threshold_p75": thresh,
+        "uncertain_slugs": [r["slug"] for r in uncertain],
+        "rows": rows,
+    }, ensure_ascii=False, indent=1))
+    print(f"\n  ✓ {OUT.name}")
+
+
+if __name__ == "__main__":
+    main()

armada/tide-table/emotion_overnight.py

+#!/usr/bin/env python3
+"""emotion_overnight — disk-safe, deadline-bounded corpus grind for the ~6h night run.
+
+The volume is at 97%%, so we NEVER hoard audio: every track is fetched, analyzed, and
+its mp3 + demucs stems deleted immediately — only the tiny analysis JSON survives. Two
+phases so the cheap signal is banked first:
+
+  phase 1  fetch → WHOLE-MIX analyze → delete mp3        (fast; covers the whole corpus)
+  phase 2  re-fetch → demucs STEMS → stem-analyze → delete mp3+stems  (slow; as many as
+           the deadline allows — grows the #92 whole-mix-vs-stem evidence base)
+
+Robust by design: per-track try/except, yt-dlp throttle + consecutive-failure backoff
+(YouTube rate-limits a residential IP at volume), and a hard wall-clock deadline after
+which it stops cleanly (cleanup always runs). Re-runnable — skips anything already
+analyzed.
+
+    python3 emotion_overnight.py --hours 6 [--phase1-only] [--throttle 4]
+
+See [[project_hexa_emotion_convergence]], [[feedback_freebox_ssot]] (disk discipline)."""
+import shutil
+import sys
+import time
+from pathlib import Path
+
+import emotion_corpus as EC
+
+FOUNDRY = EC.CACHE / "foundry"
+
+
+def _purge(slug):
+    """Delete the mp3 and any demucs workspace for this slug. Keep analysis JSON."""
+    for p in EC.AUDIO.glob(f"{slug}.*"):
+        p.unlink(missing_ok=True)
+    ws = FOUNDRY / slug
+    if ws.exists():
+        shutil.rmtree(ws, ignore_errors=True)
+
+
+def _throttled_fetch(t, throttle, state):
+    """Fetch with throttle + exponential backoff on consecutive failures."""
+    time.sleep(throttle + min(state["fails"], 8) * throttle)  # backoff grows with fails
+    p = EC.fetch_one(t)
+    state["fails"] = 0 if p else state["fails"] + 1
+    return p
+
+
+def run(hours, phase1_only, throttle):
+    deadline = time.time() + hours * 3600
+    tracks = EC._tracks()
+    state = {"fails": 0}
+    n1 = n2 = 0
+
+    # ── phase 1: whole-mix over the whole corpus (cheap, bank it first) ──────────
+    print(f"⛵ overnight grind — {len(tracks)} tracks, deadline {hours}h, throttle {throttle}s",
+          flush=True)
+    for t in tracks:
+        if time.time() > deadline:
+            print("  ⏰ deadline — stopping phase 1", flush=True); break
+        slug = t["slug"]
+        if (EC.ANALYSIS / f"{slug}_mix.json").exists():
+            continue
+        try:
+            if not _throttled_fetch(t, throttle, state):
+                print(f"  ✗ fetch {slug}", flush=True); continue
+            r = EC.analyze_one(t, mix=True)
+            if r:
+                n1 += 1
+                s = r["summary"]
+                print(f"  mix ✓ {slug:<30} {s['dominant']:<11} {s['valence']:+.2f}/{s['arousal']:+.2f}",
+                      flush=True)
+        except Exception as e:
+            print(f"  ! {slug}: {type(e).__name__} {e}", flush=True)
+        finally:
+            _purge(slug)
+    print(f"  phase 1 done: +{n1} whole-mix\n", flush=True)
+
+    if phase1_only:
+        print(f"✓ overnight (phase1-only): +{n1} whole-mix", flush=True)
+        return
+
+    # ── phase 2: demucs stems for as many as the deadline allows ────────────────
+    state["fails"] = 0
+    for t in tracks:
+        if time.time() > deadline:
+            print("  ⏰ deadline — stopping phase 2", flush=True); break
+        slug = t["slug"]
+        if (EC.ANALYSIS / f"{slug}.json").exists():
+            continue
+        try:
+            if not _throttled_fetch(t, throttle, state):
+                print(f"  ✗ refetch {slug}", flush=True); continue
+            r = EC.analyze_one(t, mix=False)        # demucs → stem-aware
+            if r:
+                n2 += 1
+                s = r["summary"]
+                print(f"  stem ✓ {slug:<30} {s['dominant']:<11} {s['valence']:+.2f}/{s['arousal']:+.2f}",
+                      flush=True)
+        except Exception as e:
+            print(f"  ! stem {slug}: {type(e).__name__} {e}", flush=True)
+        finally:
+            _purge(slug)
+
+    # final disk hygiene — nothing should remain in audio/ or foundry/
+    for p in EC.AUDIO.glob("*"):
+        p.unlink(missing_ok=True)
+    if FOUNDRY.exists():
+        shutil.rmtree(FOUNDRY, ignore_errors=True)
+    print(f"\n✓ overnight done: +{n1} whole-mix, +{n2} stem. Audio purged.", flush=True)
+
+
+def main():
+    args = sys.argv[1:]
+    hours = float(args[args.index("--hours") + 1]) if "--hours" in args else 6.0
+    throttle = float(args[args.index("--throttle") + 1]) if "--throttle" in args else 4.0
+    run(hours, "--phase1-only" in args, throttle)
+
+
+if __name__ == "__main__":
+    main()