| """Validation suite v4 (6.25 Hz, new intensity target). |
| |
| Checks: 6.25 Hz rate contract, interface contract, teacher fidelity on held-out |
| RAVDESS actors 19-24, intensity corr vs the NEW salience-gated absolute-energy |
| target, and — the acceptance gate — separation of RAVDESS ground-truth intensity |
| labels (normal 01 vs strong 02) over ALL 336 emotional held-out files. |
| Writes /tmp/work/surgical_out_v4/eval_report_v4.json + dynamics_plot_v4.png |
| """ |
| from __future__ import annotations |
| import glob, json, os, random, sys |
| import numpy as np |
| import torch |
| import torch.nn.functional as F |
| import torchaudio |
|
|
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| from surgical_emotion import SurgicalEmotionEncoder |
|
|
| FUNASR_DIR = "/tmp/work/emotion2vec_plus_base" |
| OUT_DIR = "/tmp/work/surgical_out_v4" |
| CKPT = os.path.join(OUT_DIR, "surgical_emotion_module.pt") |
| RAVDESS_DIR = "/tmp/work/ravdess_kag" |
| EMO_IDX = [0, 1, 2, 3, 6, 7] |
| GATE_FLOOR = 0.02 |
|
|
|
|
| def rms_raw(wav, T): |
| L = wav.numel(); n_full = (L + 319) // 320 |
| w = torch.zeros(T * 320); w[: min(L, T * 320)] = wav[: min(L, T * 320)] |
| ms = w.view(T, 320).pow(2).mean(1).sqrt() |
| if T > n_full: ms[n_full:] = ms[n_full - 1] |
| if T >= 9: |
| ms = F.conv1d(ms[None, None], torch.ones(9)[None, None] / 9.0, padding=4)[0, 0] |
| return ms |
|
|
|
|
| def js_div(p, q, eps=1e-8): |
| p = p.clamp(min=eps); q = q.clamp(min=eps) |
| m = 0.5 * (p + q) |
| kl = lambda a, b: float((a * (a.log() - b.log())).sum(-1)) |
| return 0.5 * kl(p, m) + 0.5 * kl(q, m) |
|
|
|
|
| def load_wav(path, max_s=12.0): |
| wav, sr = torchaudio.load(path) |
| wav = wav.mean(0) |
| if sr != 16000: |
| wav = torchaudio.functional.resample(wav, sr, 16000) |
| wav = wav[: int(max_s * 16000)] |
| return wav / wav.abs().max().clamp(min=1e-4) |
|
|
|
|
| def new_target(enc, wav, states): |
| probs = torch.softmax(enc.teacher.proj(states), -1)[0] |
| p6 = probs[:, EMO_IDX].max(-1).values.clamp(min=GATE_FLOOR).cpu() |
| env = (rms_raw(wav, states.size(1)) / META["global_ref"]).clamp(0, 1) |
| return p6 * env, probs |
|
|
|
|
| def main(): |
| device = "cuda:0" |
| enc = SurgicalEmotionEncoder.from_pretrained(FUNASR_DIR, CKPT, device=device) |
| enc.eval() |
| report = {"frame_rate_hz": enc.cfg.frame_rate_hz, "checks": {}, "intensity": {}} |
|
|
| |
| ok = True |
| for sec in [1.0, 2.5, 5.0, 10.0]: |
| n = int(sec * 16000) |
| wav = (torch.randn(1, n, device=device)); wav = wav / wav.abs().max().clamp(min=1e-3) |
| out = enc(wav, torch.tensor([n], device=device)) |
| expect = int(np.ceil(sec * enc.cfg.frame_rate_hz)) |
| got = int(out["frame_embs"].shape[1]) |
| good = abs(got - expect) <= 1 |
| ok &= good |
| report["checks"][f"{sec}s"] = {"expected~": expect, "got": got, "ok": good} |
| assert out["frame_embs"].shape[-1] == enc.cfg.out_dim |
| assert 0.0 <= float(out["frame_intensity"].min()) and float(out["frame_intensity"].max()) <= 1.0 |
| report["rate_contract_ok"] = bool(ok) |
| print("rate contract ok:", ok) |
|
|
| files = sorted(glob.glob(os.path.join(RAVDESS_DIR, "**", "*.wav"), recursive=True)) |
| speech = [f for f in files if os.path.basename(f).split("-")[1] == "01" |
| and 19 <= int(os.path.basename(f).split("-")[-1].split(".")[0]) <= 24] |
| emotional = [f for f in speech if os.path.basename(f).split("-")[2] != "01"] |
| random.seed(0) |
| sel = random.sample(speech, min(60, len(speech))) |
|
|
| agree, jss, corrs = [], [], [] |
| plot_sample = None |
| with torch.no_grad(): |
| for f in sel: |
| wav = load_wav(f) |
| if wav.numel() < 8000: continue |
| out = enc(wav[None].to(device), torch.tensor([wav.numel()], device=device)) |
| states = enc.teacher.extract_features( |
| F.layer_norm(wav[None].to(device), wav[None].to(device).shape), |
| padding_mask=None, mask=False)["x"] |
| tp = torch.softmax(enc.teacher.proj(states.mean(1)), -1)[0] |
| sp = out["utt_probs"][0] |
| agree.append(int(sp.argmax().item() == tp.argmax().item())) |
| jss.append(js_div(sp[None], tp[None])) |
| tgt50, _ = new_target(enc, wav, states) |
| si = out["frame_intensity"][0, :, 0].cpu() |
| tgt = F.interpolate(tgt50[None, None], size=si.size(0), mode="area")[0, 0] |
| if si.var() > 1e-8 and tgt.var() > 1e-8: |
| corrs.append(float(np.corrcoef(si.numpy(), tgt.numpy())[0, 1])) |
| if plot_sample is None and "-05-" in f: |
| plot_sample = (os.path.basename(f), out, states, tgt50) |
| report["intensity"]["fidelity_agreement"] = float(np.mean(agree)) |
| report["intensity"]["fidelity_mean_js"] = float(np.mean(jss)) |
| report["intensity"]["mean_corr_new_target"] = float(np.mean(corrs)) |
| print(f"agreement={report['intensity']['fidelity_agreement']:.3f} " |
| f"JS={report['intensity']['fidelity_mean_js']:.4f} " |
| f"int_corr={report['intensity']['mean_corr_new_target']:.3f}") |
|
|
| |
| groups = {"01": [], "02": []} |
| groups_tgt = {"01": [], "02": []} |
| with torch.no_grad(): |
| for f in emotional: |
| wav = load_wav(f) |
| if wav.numel() < 8000: continue |
| out = enc(wav[None].to(device), torch.tensor([wav.numel()], device=device)) |
| states = enc.teacher.extract_features( |
| F.layer_norm(wav[None].to(device), wav[None].to(device).shape), |
| padding_mask=None, mask=False)["x"] |
| tgt50, _ = new_target(enc, wav, states) |
| lbl = os.path.basename(f).split("-")[3] |
| groups[lbl].append(float(out["frame_intensity"][0, :, 0].mean())) |
| groups_tgt[lbl].append(float(tgt50.mean())) |
| a, b = np.array(groups["01"]), np.array(groups["02"]) |
| d = (b.mean() - a.mean()) / np.sqrt((a.var() + b.var()) / 2 + 1e-9) |
| report["intensity"]["sep_normal_mean"] = float(a.mean()) |
| report["intensity"]["sep_strong_mean"] = float(b.mean()) |
| report["intensity"]["sep_diff"] = float(b.mean() - a.mean()) |
| report["intensity"]["sep_cohen_d"] = float(d) |
| print(f"intensity separation: normal {a.mean():.3f} vs strong {b.mean():.3f} " |
| f"diff {b.mean()-a.mean():+.3f} Cohen-d {d:.2f} (n={len(a)}/{len(b)})") |
| ta, tb = np.array(groups_tgt["01"]), np.array(groups_tgt["02"]) |
| td = (tb.mean() - ta.mean()) / np.sqrt((ta.var() + tb.var()) / 2 + 1e-9) |
| report["intensity"]["target_sep_normal_mean"] = float(ta.mean()) |
| report["intensity"]["target_sep_strong_mean"] = float(tb.mean()) |
| report["intensity"]["target_sep_cohen_d"] = float(td) |
| print(f"target itself : normal {ta.mean():.3f} vs strong {tb.mean():.3f} " |
| f"Cohen-d {td:.2f} (ceiling diagnostic)") |
|
|
| |
| try: |
| import matplotlib; matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| name, out, states, tgt50 = plot_sample |
| fp = out["frame_probs"][0].cpu().numpy() |
| si = out["frame_intensity"][0, :, 0].cpu() |
| tgt = F.interpolate(tgt50[None, None], size=si.size(0), mode="area")[0, 0] |
| t = np.arange(si.size(0)) * 0.16 |
| fig, ax = plt.subplots(2, 1, figsize=(11, 6)) |
| for k, lab in enumerate(["angry", "disgusted", "fearful", "happy", |
| "neutral", "other", "sad", "surprised"]): |
| ax[0].plot(t, fp[:, k], label=lab, lw=1.2) |
| ax[0].set_title(f"v4 frame emotion probs @6.25Hz — {name}") |
| ax[0].legend(fontsize=7, ncol=4); ax[0].set_ylabel("p") |
| ax[1].plot(t, si, color="crimson", lw=1.8, label="v4 intensity") |
| ax[1].plot(t, tgt, color="steelblue", lw=1.4, ls="--", label="target") |
| ax[1].set_ylabel("intensity"); ax[1].set_ylim(0, 1.05); ax[1].legend(fontsize=8) |
| ax[1].set_xlabel("time (s) — frame = 160 ms | 6.25 Hz (1 frame / dots.tts token)") |
| fig.tight_layout() |
| fig.savefig(os.path.join(OUT_DIR, "dynamics_plot_v4.png"), dpi=110) |
| print("plot saved") |
| except Exception as e: |
| print("plot skipped:", e) |
|
|
| with open(os.path.join(OUT_DIR, "eval_report_v4.json"), "w") as f: |
| json.dump(report, f, indent=1) |
| print("EVAL_DONE") |
|
|
|
|
| if __name__ == "__main__": |
| with open(os.path.join(OUT_DIR, "target_meta.json")) as f: |
| META = json.load(f) |
| main() |
|
|
