"""Run V3 face inference on a MULTI-TURN arbitrary-audio scenario. `scripts/infer_song.py` handles one audio file with ONE emotion held for the whole clip (it calls build_cond_smoothed with a single turn and vad_smooth_sigma=0). That is the right tool for a song, but it flattens any scenario whose whole point is that the emotion CHANGES: the 2026-07 English clips came out with browInnerUp peaking at 0.003, versus ~1.0 on the Korean scenarios baked through data_pipeline. This script takes N per-turn audio files plus the per-turn (emotion, VAD) and rebuilds the same time-varying (T, 19) cond the model was trained on — hard one-hot + VAD per turn, then cross-turn Gaussian smoothing at σ=30 frames, exactly as data_pipeline.py does when baking .npz. Turn boundaries are derived from the CONCATENATED waveform rather than from per-turn mel lengths, so the frame grid the cond is cut on is the same grid the merged mp3 plays back on — no accumulating drift between animation and audio across turns. Usage: PYTHONPATH=. python3 scripts/infer_multiturn_audio.py \\ --spec data/audio_eng/en_long_046.json \\ --variants v14 v18m Spec format (JSON): { "sid": "en_long_046", "source": "long_046", "turns": [ {"audio": "data/audio_eng/en_long_046_t0.mp3", "emotion": "surprise", "vad": [0.062, 0.737, -0.315], "text": "The second I opened that text, ..."}, ... ] } Output (per variant): data/viewer/_pred__dataset.json data/viewer/_dataset.mp3 # merged audio, real file (not a symlink) """ from __future__ import annotations import argparse import json import subprocess import sys import tempfile from pathlib import Path from typing import List import librosa import numpy as np import soundfile as sf import torch PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) from scripts.compiler.constants import ARKIT_52_NAMES # noqa: E402 from scripts.compiler.data_pipeline import FPS, mel_features # noqa: E402 from models.v3_face.infer import ( # noqa: E402 crisp_mouth, smooth_brows, inject_blinks, load_model, ) from models.v3_face.infer_e2e import build_cond_smoothed # noqa: E402 CKPT_DIR = PROJECT_ROOT / "models" / "v3_face" / "checkpoints" DEFAULT_VIEWER_DIR = PROJECT_ROOT / "data" / "viewer" MEL_SR = 16000 OUT_SR = 44100 def load_and_trim(path: Path, sr: int, top_db: float): """Load one turn at `sr` mono and strip leading/trailing near-silence. ElevenLabs v3 prompts carry a leading ". [tag]" buffer (the tag is not voiced, but the period can leave a beat of room tone) and the renders tend to end with a tail of silence. Concatenating those untrimmed puts dead air at every turn boundary, which reads as the avatar freezing mid-scenario. top_db is deliberately conservative — it only removes what is genuinely near the noise floor. """ wav, _ = librosa.load(str(path), sr=sr, mono=True) if wav.size == 0: raise SystemExit(f"empty audio: {path}") trimmed, _ = librosa.effects.trim(wav, top_db=top_db) if trimmed.size < sr * 0.05: # trim ate the whole clip — keep original return wav, wav.size return trimmed, wav.size def encode_mp3(wav: np.ndarray, sr: int, out_path: Path) -> None: """Write `wav` to `out_path` as mp3 via ffmpeg (no mp3 encoder in soundfile).""" with tempfile.NamedTemporaryFile(suffix=".wav", delete=True) as tmp: sf.write(tmp.name, wav, sr, subtype="PCM_16") out_path.parent.mkdir(parents=True, exist_ok=True) if out_path.exists() or out_path.is_symlink(): out_path.unlink() # may be a stale/broken symlink subprocess.run( ["ffmpeg", "-y", "-loglevel", "error", "-i", tmp.name, "-codec:a", "libmp3lame", "-b:a", "128k", str(out_path)], check=True, ) def frame_edges(sample_edges: List[int], hop: int, total_frames: int) -> List[int]: """Turn cumulative sample offsets into per-turn frame counts. Cutting the cond on the concatenated timeline (rather than summing independently-computed per-turn mel lengths) is what keeps audio and animation locked together: each turn's mel length rounds up independently, so summing them drifts ~1 frame per turn against the merged file the viewer actually plays. """ edges = [min(total_frames, int(round(s / hop))) for s in sample_edges] edges[0], edges[-1] = 0, total_frames for i in range(1, len(edges)): # keep monotonic edges[i] = max(edges[i], edges[i - 1]) return [edges[i + 1] - edges[i] for i in range(len(edges) - 1)] def main() -> None: ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--spec", type=Path, required=True) ap.add_argument("--variants", nargs="+", default=["v14", "v18m"], help="Checkpoint tags; resolved to best_expression_.pt") ap.add_argument("--viewer_dir", type=Path, default=DEFAULT_VIEWER_DIR) ap.add_argument("--device", default="cuda:0") ap.add_argument("--vad-smooth-sigma", type=float, default=30.0, help="Cross-turn Gaussian σ in frames. 30 matches " "data_pipeline.py's bake default — the shape of cond " "V3 was trained on. Lower = sharper emotion switches.") ap.add_argument("--trim-top-db", type=float, default=40.0) ap.add_argument("--turn-gap", type=float, default=0.45, help="Seconds of silence inserted between turns. Each gap " "is charged to the turn BEFORE it, so the expression " "holds through the pause rather than snapping to the " "next emotion while the mouth is still closed.") ap.add_argument("--no-crisp", action="store_true") ap.add_argument("--crisp-threshold", type=float, default=0.3) ap.add_argument("--crisp-scale", type=float, default=1.0) ap.add_argument("--crisp-sigma", type=float, default=1.0) ap.add_argument("--crisp-mouthclose-sigma", type=float, default=1.0) ap.add_argument("--no-brow-smooth", action="store_true") ap.add_argument("--brow-min-cutoff", type=float, default=1.5) ap.add_argument("--brow-beta", type=float, default=0.01) ap.add_argument("--brow-d-cutoff", type=float, default=1.0) ap.add_argument("--no-blinks", action="store_true") ap.add_argument("--blink-interval", type=float, default=4.0) ap.add_argument("--blink-expressive-cap", type=float, default=0.5) args = ap.parse_args() spec = json.loads(args.spec.read_text()) sid, turns = spec["sid"], spec["turns"] print(f"{sid}: {len(turns)} turns") # ── Audio: trim each turn, concatenate once at 44.1k ────────────── gap_n = int(round(args.turn_gap * OUT_SR)) fade_n = int(0.012 * OUT_SR) # 12ms de-click ramp either side of a gap hi_chunks, edges_hi = [], [0] for ti, t in enumerate(turns): p = Path(t["audio"]) if not p.is_absolute(): p = PROJECT_ROOT / p wav, orig_n = load_and_trim(p, OUT_SR, args.trim_top_db) if gap_n and wav.size > 2 * fade_n: # Trimming cuts at a -40dB threshold, not at true zero, so butting # a trimmed tail against digital silence can tick. Ramp both ends. ramp = np.linspace(0.0, 1.0, fade_n, dtype=np.float32) wav = wav.copy() wav[:fade_n] *= ramp wav[-fade_n:] *= ramp[::-1] hi_chunks.append(wav) seg_n = wav.size if gap_n and ti < len(turns) - 1: hi_chunks.append(np.zeros(gap_n, dtype=np.float32)) seg_n += gap_n # gap belongs to the turn it follows edges_hi.append(edges_hi[-1] + seg_n) print(f" t{ti} {t['emotion']:<11} {wav.size / OUT_SR:6.2f}s" f"{f' +{args.turn_gap:.2f}s gap' if gap_n and ti < len(turns) - 1 else '':>12}" f" (silence trimmed {(orig_n - wav.size) / OUT_SR:.2f}s)") full_hi = np.concatenate(hi_chunks).astype(np.float32) out_mp3 = args.viewer_dir / f"{sid}_dataset.mp3" encode_mp3(full_hi, OUT_SR, out_mp3) print(f" merged audio {full_hi.size / OUT_SR:.2f}s -> {out_mp3}") # Resample the SAME concatenation for mel, so the frame grid the cond is # cut on is the grid the merged mp3 plays on. full_lo = librosa.resample(full_hi, orig_sr=OUT_SR, target_sr=MEL_SR) mel = mel_features(full_lo, sr=MEL_SR, fps=FPS) T = mel.shape[0] hop = MEL_SR // FPS scale = MEL_SR / OUT_SR turn_Ts = frame_edges([int(e * scale) for e in edges_hi], hop, T) print(f" mel frames={T} ({T / FPS:.2f}s) turn frames={turn_Ts}") cond = build_cond_smoothed( [t["emotion"] for t in turns], np.asarray([t["vad"] for t in turns], dtype=np.float32), turn_Ts, vad_smooth_sigma=args.vad_smooth_sigma, ) assert cond.shape == (T, 19), cond.shape device = torch.device(args.device) audio_t = torch.from_numpy(mel.astype(np.float32)).unsqueeze(0).to(device) cond_t = torch.from_numpy(cond).unsqueeze(0).to(device) meta_turns = [{ "turn_idx": ti, "emotion": t["emotion"], "vad": list(t["vad"]), "text": t.get("text", ""), "speaker": t.get("speaker", "A"), } for ti, t in enumerate(turns)] for variant in args.variants: ckpt = CKPT_DIR / f"best_expression_{variant}.pt" if not ckpt.exists(): print(f" ✗ {variant}: {ckpt} not found — skipped") continue model, _cfg = load_model(ckpt, device) with torch.no_grad(): pred = model(audio_t, cond_t).squeeze(0).cpu().numpy().astype(np.float32) if not args.no_crisp: pred = crisp_mouth(pred, threshold=args.crisp_threshold, scale=args.crisp_scale, pre_smooth_sigma=args.crisp_sigma, mouth_close_sigma=args.crisp_mouthclose_sigma) if not args.no_brow_smooth: pred = smooth_brows(pred, min_cutoff=args.brow_min_cutoff, beta=args.brow_beta, d_cutoff=args.brow_d_cutoff) if not args.no_blinks: pred = inject_blinks(pred, scenario_id=sid, mean_interval_s=args.blink_interval, expressive_cap=args.blink_expressive_cap) base = f"{sid}_pred_{variant}_dataset" out_json = args.viewer_dir / f"{base}.json" out_json.write_text(json.dumps({ "scenario_id": base, "fps": FPS, "num_frames": int(pred.shape[0]), "names": ARKIT_52_NAMES, "turns": meta_turns, "blendshapes": np.round(pred, 4).tolist(), }, ensure_ascii=False)) print(f" ✓ {variant}: {out_json.name} " f"browInnerUp max {pred[:, 0].max():.3f} jawOpen max {pred[:, 24].max():.3f}") if __name__ == "__main__": main()