feat(tide-table): emotion_timeline — slide CLAP emotion kernel over a track → arc

Windows a master (~12s/50% hop), CLAP-embeds each window (new sample_semantics.
embed_audio_arrays seam), runs the emotion_ontology kernel per window → (valence,
arousal, top-emotion) series; median-smooths, relabels from the smoothed circumplex
point, fuses short runs into emotion SECTIONS. Emits frames+sections+summary JSON —
the continuous-V/A + discrete-scene lane hexa wants. Validated on Bohemian Rhapsody:
caught the warm-intro → energetic-middle → tense rock-peak (A+0.77) → calm-outro arc.

armada/tide-table/_corpus_cache/.gitignore

+*
+!.gitignore

armada/tide-table/emotion_timeline.py

+#!/usr/bin/env python3
+"""emotion_timeline — slide the CLAP emotion kernel over a track → an emotion arc.
+
+A whole track has ONE dominant vibe but an ARC: a euphoric build, a tense breakdown,
+a dark drop. So we window the master (~12 s, 50 % hop — long enough that CLAP reads a
+mood, short enough to catch a section change), CLAP-embed each window, and run the
+emotion_ontology kernel per window → a time series of (valence, arousal, top-emotion).
+We smooth out single-frame flicker, then merge contiguous frames into emotion SECTIONS
+(short blips fused into neighbours). The result is the lane hydra-live-hexa wants:
+continuous V/A to drive palette/intensity + discrete sections to switch FX scenes.
+See [[project_hexa_emotion_convergence]].
+
+    python3 emotion_timeline.py track <file> [--win 12 --hop 6] [--json out.json]
+
+Window scale matters: emotion reads decisively at 10-30 s (CLAP's strength), weakly on
+one-shots — so this is where the engine is actually trustworthy. PLN's ear confirms feel
+([[feedback_verify_own_renders]]); the machine reports the arc."""
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import numpy as np
+
+import emotion_ontology as EMO
+import sample_classify as CLF
+import sample_semantics as SEM
+
+EMBED_BATCH = 24            # windows per CLAP forward (bounds memory on long tracks)
+
+
+def decode_full(path, sr=CLF.SR, max_min=25):
+    """Decode any file → mono float32 @ CLF.SR, capped at max_min. None on error."""
+    try:
+        p = subprocess.run(
+            ["ffmpeg", "-v", "quiet", "-i", str(path), "-ac", "1", "-ar", str(sr),
+             "-t", str(max_min * 60), "-f", "f32le", "-"],
+            capture_output=True, timeout=300)
+        a = np.frombuffer(p.stdout, dtype=np.float32)
+        return a if a.size >= sr else None
+    except Exception:
+        return None
+
+
+def _median_smooth(x, k=3):
+    """Median filter (odd k) — kills single-window flicker without lagging like an EMA."""
+    if k <= 1 or len(x) < k:
+        return np.asarray(x, dtype="float32")
+    r = k // 2
+    pad = np.pad(x, r, mode="edge")
+    return np.array([np.median(pad[i:i + k]) for i in range(len(x))], dtype="float32")
+
+
+def _sections(times, labels, min_sec=8.0):
+    """Contiguous equal-label runs → sections; runs shorter than min_sec are fused into
+    the longer neighbour so the arc reads as scenes, not strobing."""
+    if not len(labels):
+        return []
+    segs = []
+    s = 0
+    for i in range(1, len(labels) + 1):
+        if i == len(labels) or labels[i] != labels[s]:
+            segs.append([times[s], times[i - 1], labels[s], s, i - 1])
+            s = i
+    # fuse short segments into the neighbour with the longer duration
+    changed = True
+    while changed and len(segs) > 1:
+        changed = False
+        for j, seg in enumerate(segs):
+            if seg[1] - seg[0] >= min_sec:
+                continue
+            left = segs[j - 1] if j > 0 else None
+            right = segs[j + 1] if j < len(segs) - 1 else None
+            pick = left if (right is None or
+                            (left is not None and (left[1] - left[0]) >= (right[1] - right[0]))) else right
+            pick[0] = min(pick[0], seg[0]); pick[1] = max(pick[1], seg[1])
+            pick[3] = min(pick[3], seg[3]); pick[4] = max(pick[4], seg[4])
+            segs.pop(j); changed = True
+            break
+    return segs
+
+
+def analyze(path, win_s=12.0, hop_s=6.0):
+    y = decode_full(path)
+    if y is None:
+        return None
+    sr = CLF.SR
+    win, hop = int(win_s * sr), int(hop_s * sr)
+    starts = list(range(0, max(1, len(y) - win + 1), hop)) or [0]
+    if len(y) - starts[-1] > win * 0.4 and starts[-1] + win < len(y):
+        starts.append(len(y) - win)                    # trailing partial window
+    windows = [y[s:s + win] for s in starts]
+
+    rows = []
+    for b in range(0, len(windows), EMBED_BATCH):
+        rows.append(SEM.embed_audio_arrays(windows[b:b + EMBED_BATCH]))
+    ae = np.vstack(rows) if rows else np.zeros((0, 512), dtype="float32")
+
+    frames = []
+    for s, row in zip(starts, ae):
+        r = EMO.score(row)
+        frames.append({"t": round(s / sr, 2), "valence": r["valence"],
+                       "arousal": r["arousal"], "emotion": r["top"][0][0],
+                       "confidence": r["confidence"]})
+    if not frames:
+        return None
+
+    # smooth V/A and de-flicker the label track (argmax of the smoothed circumplex point
+    # against each emotion's V/A — keeps labels consistent with the smoothed trajectory).
+    val = _median_smooth([f["valence"] for f in frames], 3)
+    aro = _median_smooth([f["arousal"] for f in frames], 3)
+    va = EMO._matrix()["va"]; names = EMO._matrix()["names"]
+    for i, f in enumerate(frames):
+        f["valence"], f["arousal"] = round(float(val[i]), 3), round(float(aro[i]), 3)
+        d = (va[:, 0] - val[i]) ** 2 + (va[:, 1] - aro[i]) ** 2
+        f["emotion"] = names[int(np.argmin(d))]
+
+    times = [f["t"] for f in frames]
+    labels = [f["emotion"] for f in frames]
+    segs = _sections(times, labels, min_sec=max(win_s, 8.0))
+    dur = round(len(y) / sr, 2)
+    sections = []
+    for s0, s1, lab, i0, i1 in segs:
+        sl = frames[i0:i1 + 1]
+        sections.append({"start": round(s0, 2), "end": round(min(s1 + hop_s, dur), 2),
+                         "emotion": lab,
+                         "valence": round(float(np.mean([f["valence"] for f in sl])), 3),
+                         "arousal": round(float(np.mean([f["arousal"] for f in sl])), 3)})
+    from collections import Counter
+    dom = Counter(labels).most_common(1)[0][0]
+    arc = "→".join(dict.fromkeys(s["emotion"] for s in sections))   # de-duped order
+    return {"file": str(path), "duration": dur, "win_s": win_s, "hop_s": hop_s,
+            "n_frames": len(frames),
+            "summary": {"dominant": dom,
+                        "valence": round(float(np.mean(val)), 3),
+                        "arousal": round(float(np.mean(aro)), 3), "arc": arc},
+            "sections": sections, "frames": frames}
+
+
+# ── CLI ─────────────────────────────────────────────────────────────────────────
+def _arc_print(res):
+    print(f"🎬 {Path(res['file']).name}  ({res['duration']:.0f}s, {res['n_frames']} frames)")
+    s = res["summary"]
+    print(f"   dominant: {s['dominant']}   mean V/A {s['valence']:+.2f}/{s['arousal']:+.2f}")
+    print(f"   arc: {s['arc']}\n")
+    span = max(1.0, res["duration"])
+    for sec in res["sections"]:
+        a = int(sec["start"] / span * 40); b = max(a + 1, int(sec["end"] / span * 40))
+        bar = " " * a + "█" * (b - a)
+        print(f"   {sec['start']:>6.0f}-{sec['end']:<6.0f} {sec['emotion']:<12} "
+              f"V/A {sec['valence']:+.2f}/{sec['arousal']:+.2f}")
+        print(f"   {bar}")
+
+
+def main():
+    args = sys.argv[1:]
+    if not (args and args[0] == "track" and len(args) > 1):
+        sys.exit("usage: emotion_timeline.py track <file> [--win N] [--hop N] [--json out]")
+    win = float(args[args.index("--win") + 1]) if "--win" in args else 12.0
+    hop = float(args[args.index("--hop") + 1]) if "--hop" in args else 6.0
+    res = analyze(args[1], win_s=win, hop_s=hop)
+    if not res:
+        sys.exit("decode/analyze failed")
+    _arc_print(res)
+    if "--json" in args:
+        out = Path(args[args.index("--json") + 1])
+        out.write_text(json.dumps(res, ensure_ascii=False, indent=1))
+        print(f"\n✓ {out}")
+
+
+if __name__ == "__main__":
+    main()

armada/tide-table/sample_semantics.py

@@ -96,6 +96,23 @@ def _axis_embeds():
     return _S["axes"]
 
 
+def embed_audio_arrays(arrays):
+    """Normalized CLAP audio embeddings for a list of decoded mono float32 arrays
+    (already at CLF.SR) → np.ndarray (n×512). The array-level seam: file decoding and
+    window slicing both feed through here (emotion_timeline slides windows in)."""
+    arrays = [a for a in arrays if a is not None and len(a)]
+    if not arrays:
+        return np.zeros((0, 512), dtype=np.float32)
+    C = CLF._clap()
+    torch = C["torch"]
+    with torch.no_grad():
+        ai = C["proc"](audio=arrays, sampling_rate=CLF.SR,
+                       return_tensors="pt", padding=True)
+        ae = C["model"].get_audio_features(input_features=ai["input_features"]).pooler_output
+        ae = ae / ae.norm(p=2, dim=-1, keepdim=True)
+    return ae.cpu().numpy()
+
+
 def embed_audios(paths):
     """Normalized CLAP audio embeddings for paths → (np.ndarray (n×512), kept_paths).
     Decodes via sample_classify.load_audio (48k mono, thread-pooled)."""
@@ -105,14 +122,7 @@ def embed_audios(paths):
     keep = [(p, a) for p, a in zip(paths, decoded) if a is not None]
     if not keep:
         return np.zeros((0, 512), dtype=np.float32), []
-    C = CLF._clap()
-    torch = C["torch"]
-    with torch.no_grad():
-        ai = C["proc"](audio=[a for _, a in keep], sampling_rate=CLF.SR,
-                       return_tensors="pt", padding=True)
-        ae = C["model"].get_audio_features(input_features=ai["input_features"]).pooler_output
-        ae = ae / ae.norm(p=2, dim=-1, keepdim=True)
-    return ae.cpu().numpy(), [p for p, _ in keep]
+    return embed_audio_arrays([a for _, a in keep]), [p for p, _ in keep]
 
 
 # ── tagging: multi-axis labels for one sound ──────────────────────────────────