"""dualturn_endpointing — Speech endpoint detector for two-channel audio. Combines the dualturn transformer model with a trained logistic regression classifier to answer one question in real-time: "Has the user finished speaking? → ST (yes) or CL (no)" ═══════════════════════════════════════════════════════════════════════════════ MODELS ═══════════════════════════════════════════════════════════════════════════════ 1. dualturn backbone (best.pt, ~534 MB) - Mimi encoder (24 kHz stereo → discrete tokens @ 12.5 Hz) - Transformer: predicts VAD, EOT, BOT per channel every 80 ms 2. endpoint_clf (endpoint_clf.pkl, ~370 KB) - 10-feature Logistic Regression (GBT in bundle) - Input: last value of each of the 10 dualturn signals at the VAD offset - Output: P(ST) — probability that the user has finished their turn - Threshold: 0.30 (tuned to maximise ST recall on held-out test set) ═══════════════════════════════════════════════════════════════════════════════ DECISION LOGIC (every 80 ms) ═══════════════════════════════════════════════════════════════════════════════ audio (dual channel, 24 kHz) │ ▼ Mimi + dualturn backbone signals every 80 ms: vad_user, vad_agent, eot_user, eot_agent, bot_user, bot_agent, fvad_user_short, fvad_user_long, fvad_agent_short, fvad_agent_long │ ▼ watch vad_user crossing 0.5 (VAD offset = user stopped speaking) anchor detected? │ ├─ NO → emit signals, no action │ └─ YES (agent silent) → endpoint_clf.predict_proba([10 signal values]) P(ST) >= 0.30 → ST ✓ agent should respond now P(ST) < 0.30 → CL user paused mid-sentence, wait ═══════════════════════════════════════════════════════════════════════════════ USAGE — offline (WAV file) ═══════════════════════════════════════════════════════════════════════════════ from endpointing import DualTurnEndpointing model = DualTurnEndpointing.from_pretrained("anyreach/dualturn-endpointing") # Process a stereo WAV (user=ch0, agent=ch1) frames, endpoints = model.process_file("call.wav", user_channel=0, agent_channel=1) for ep in endpoints: print(f"t={ep['t_s']:.2f}s action={ep['action']} P(ST)={ep['p_st']:.3f}") ═══════════════════════════════════════════════════════════════════════════════ USAGE — streaming (real-time, 80 ms chunks) ═══════════════════════════════════════════════════════════════════════════════ model = DualTurnEndpointing.from_pretrained("anyreach/dualturn-endpointing") stream = model.stream(user_channel=0, agent_channel=1) # Feed 80 ms of 24 kHz stereo PCM float32 every tick for chunk_stereo in audio_source(): # shape: (2, 1920) result = stream.push(chunk_stereo) if result: print(f" {result['action']} P(ST)={result['p_st']:.3f}") """ from __future__ import annotations import sys from collections import deque from pathlib import Path from typing import Optional import joblib import numpy as np import torch import torchaudio # ───────────────────────────────────────────────────────────────────────────── # Constants # ───────────────────────────────────────────────────────────────────────────── MIMI_SR = 24_000 # Mimi encoder sample rate MIMI_RATE_HZ = 12.5 # frame rate after encoding (one frame = 80 ms) MIMI_MS = 80.0 SIGNAL_KEYS = [ "vad_user", "vad_agent", "eot_user", "eot_agent", "bot_user", "bot_agent", "fvad_user_short", "fvad_user_long", "fvad_agent_short", "fvad_agent_long", ] VAD_THRESHOLD = 0.50 # vad_user edge detection threshold AGENT_VOICE_MIN = 0.15 # min agent VAD to consider agent voicing HISTORY_SECONDS = 2.0 # rolling signal window for feature extraction DEFAULT_ST_THRESHOLD = 0.30 # ───────────────────────────────────────────────────────────────────────────── # Main class # ───────────────────────────────────────────────────────────────────────────── class DualTurnEndpointing: """ Speech endpoint detector combining dualturn backbone + logistic regression. Detects when the user has finished speaking (ST) vs paused mid-sentence (CL). Only fires at VAD offset anchors while agent is silent. Parameters ---------- backbone_path : str | Path Path to dualturn checkpoint (best.pt). classifier_path : str | Path Path to endpoint_clf.pkl (sklearn bundle). device : str "cuda" or "cpu". st_threshold : float P(ST) threshold. Loaded from classifier bundle if not overridden. """ def __init__( self, backbone_path: str | Path, classifier_path: str | Path, device: str = "cuda", st_threshold: Optional[float] = None, ): self.device = device # ── Load dualturn backbone ──────────────────────────────────────── repo_root = Path(__file__).resolve().parent / "src" if str(repo_root) not in sys.path: sys.path.insert(0, str(repo_root)) print(f"[endpointing] Loading backbone: {backbone_path}") from evaluation.verify_v2_model import load_model from evaluation.run_model_on_audio import encode_with_mimi, run_inference from dualturn.data.labeling import silero_probs_native, clean_anti_flicker, SILERO_RATE_HZ self._encode_with_mimi = encode_with_mimi self._run_inference = run_inference self._silero_probs = silero_probs_native self._anti_flicker = clean_anti_flicker self._silero_sr = int(SILERO_RATE_HZ) self.model, self.input_mode = load_model(str(backbone_path), device=device) self.model.eval() # ── Load endpoint classifier ────────────────────────────────────── print(f"[endpointing] Loading classifier: {classifier_path}") bundle = joblib.load(str(classifier_path)) self._clf = bundle["model"] self._feat_names = bundle["feature_names"] # 10 *_last features # Threshold: prefer caller arg, then bundle recommendation, then default if st_threshold is not None: self.st_threshold = st_threshold elif "recommended_threshold" in bundle: self.st_threshold = bundle["recommended_threshold"] else: self.st_threshold = DEFAULT_ST_THRESHOLD print(f"[endpointing] Ready — clf={bundle.get('clf_type','?')} " f"features={len(self._feat_names)} st_threshold={self.st_threshold}") # ───────────────────────────────────────────────────────────────────────── # Factory: load from HuggingFace Hub # ───────────────────────────────────────────────────────────────────────── @classmethod def from_pretrained( cls, repo_id: str = "anyreach/dualturn-endpointing", device: str = "cuda", st_threshold: Optional[float] = None, cache_dir: Optional[str] = None, ) -> "DualTurnEndpointing": """ Download model files from HuggingFace Hub and initialise. Usage: model = DualTurnEndpointing.from_pretrained("anyreach/dualturn-endpointing") """ from huggingface_hub import hf_hub_download print(f"[endpointing] Downloading from {repo_id} …") backbone_path = hf_hub_download(repo_id, "best.pt", cache_dir=cache_dir) classifier_path = hf_hub_download(repo_id, "endpoint_clf.pkl", cache_dir=cache_dir) # Download src package files needed by the backbone src_files = [ "src/evaluation/verify_v2_model.py", "src/evaluation/run_model_on_audio.py", "src/dualturn/data/labeling.py", ] for f in src_files: try: hf_hub_download(repo_id, f, cache_dir=cache_dir) except Exception: pass # some may not exist; backbone path is enough return cls( backbone_path = backbone_path, classifier_path = classifier_path, device = device, st_threshold = st_threshold, ) # ───────────────────────────────────────────────────────────────────────── # Offline: process a stereo WAV file end-to-end # ───────────────────────────────────────────────────────────────────────── def process_file( self, audio_path: str | Path, user_channel: int = 0, agent_channel: int = 1, ) -> tuple[list[dict], list[dict]]: """ Process a dual-channel WAV file offline. Returns ------- frames : list[dict] One dict per 80 ms frame with all 10 signals + action_at_frame. endpoints : list[dict] Sparse — one dict per ST/CL decision (at VAD offset anchors). """ wav, sr = torchaudio.load(str(audio_path)) if wav.shape[0] == 1: raise ValueError("Audio must be stereo (2 channels).") if sr != MIMI_SR: wav = torchaudio.functional.resample(wav, sr, MIMI_SR) user_wav = wav[user_channel].unsqueeze(0) agent_wav = wav[agent_channel].unsqueeze(0) stereo = torch.stack([wav[user_channel], wav[agent_channel]], dim=0) print(f"[endpointing] Processing {wav.shape[1]/MIMI_SR:.1f}s audio …") return self._run(stereo, user_wav, agent_wav) # ───────────────────────────────────────────────────────────────────────── # Streaming: push 80 ms chunks in real time # ───────────────────────────────────────────────────────────────────────── def stream(self, user_channel: int = 0, agent_channel: int = 1) -> "_StreamHandle": """ Return a streaming handle. Call `.push(chunk)` every 80 ms. chunk : np.ndarray shape (2, 1920) — stereo 24 kHz float32 PCM rows: [user_channel, agent_channel] Returns dict or None: { "t_s": float, # time of anchor "action": "ST"|"CL", "p_st": float, # P(ST) from classifier "signals": dict, # all 10 signal values at anchor } """ return _StreamHandle(self, user_channel, agent_channel) # ───────────────────────────────────────────────────────────────────────── # Internal: full offline inference pass # ───────────────────────────────────────────────────────────────────────── def _run(self, stereo, user_wav, agent_wav): # Step 1: Mimi encode + backbone with torch.no_grad(): tokens = self._encode_with_mimi( stereo.unsqueeze(0).to(self.device), self.model, self.input_mode ) preds = self._run_inference(tokens, self.model) n_frames = preds["vad_user"].shape[0] # Step 2: FVAD (Silero) per channel u16 = torchaudio.functional.resample(user_wav, MIMI_SR, self._silero_sr)[0].numpy() a16 = torchaudio.functional.resample(agent_wav, MIMI_SR, self._silero_sr)[0].numpy() fvad_us = _resample_to(self._anti_flicker(self._silero_probs(u16, self._silero_sr), alpha=0.3), n_frames) fvad_ul = _resample_to(self._anti_flicker(self._silero_probs(u16, self._silero_sr), alpha=0.7), n_frames) fvad_as = _resample_to(self._anti_flicker(self._silero_probs(a16, self._silero_sr), alpha=0.3), n_frames) fvad_al = _resample_to(self._anti_flicker(self._silero_probs(a16, self._silero_sr), alpha=0.7), n_frames) # Step 3: Frame loop history: deque[dict] = deque(maxlen=int(HISTORY_SECONDS * MIMI_RATE_HZ) + 5) frames, endpoints = [], [] prev_above = False for i in range(n_frames): t_s = i * MIMI_MS / 1000.0 sig = { "vad_user": float(preds["vad_user"][i]), "vad_agent": float(preds["vad_agent"][i]), "eot_user": float(preds["eot_user"][i]), "eot_agent": float(preds["eot_agent"][i]), "bot_user": float(preds["bot_user"][i]), "bot_agent": float(preds["bot_agent"][i]), "fvad_user_short": float(fvad_us[i]), "fvad_user_long": float(fvad_ul[i]), "fvad_agent_short": float(fvad_as[i]), "fvad_agent_long": float(fvad_al[i]), } history.append(sig) cur_above = sig["vad_user"] > VAD_THRESHOLD recent_agent = float(np.mean([h["vad_agent"] for h in list(history)[-10:]])) agent_voicing = recent_agent > AGENT_VOICE_MIN action_at_frame = None # VAD offset + agent silent → ST/CL endpoint decision if prev_above and not cur_above and not agent_voicing: ep = self._classify(t_s, sig) endpoints.append(ep) action_at_frame = ep["action"] prev_above = cur_above frames.append({"t_s": t_s, "action_at_frame": action_at_frame, **sig}) return frames, endpoints def _classify(self, t_s: float, sig: dict) -> dict: """Run endpoint_clf on the 10 signal values at the VAD offset.""" feats = np.array([sig[k] for k in SIGNAL_KEYS], dtype=np.float32).reshape(1, -1) # feature names are *_last — same order as SIGNAL_KEYS proba = self._clf.predict_proba(feats)[0] classes = list(self._clf.classes_) p_st = float(proba[classes.index(1)] if 1 in classes else proba[-1]) endpoint = p_st >= self.st_threshold print(f"[endpointing] t={t_s:.2f}s VAD-offset → " f"endpoint={endpoint} P={p_st:.3f} thr={self.st_threshold}") return { "t_s": t_s, "endpoint": endpoint, # True → user done, agent responds "p_endpoint": p_st, # P(user is done) ∈ [0, 1] "signals": sig, } # ───────────────────────────────────────────────────────────────────────────── # Streaming handle # ───────────────────────────────────────────────────────────────────────────── class _StreamHandle: """ Stateful streaming inference handle. Push 80 ms stereo chunks; get ST/CL decisions at VAD offsets. """ def __init__(self, model: DualTurnEndpointing, user_ch: int, agent_ch: int): self._model = model self._user_ch = user_ch self._agent_ch = agent_ch self._history: deque[dict] = deque(maxlen=int(HISTORY_SECONDS * MIMI_RATE_HZ) + 5) self._prev_above = False self._t_s = 0.0 self._last_decision_t = -999.0 self._refractory_s = 0.4 # minimum seconds between decisions # Audio buffers — accumulate 3 frames (240 ms) before running model # for stable context; you can lower to 1 for ultra-low latency self._buf_user: list[np.ndarray] = [] self._buf_agent: list[np.ndarray] = [] self._BATCH = 3 # run inference every 3 × 80 ms = 240 ms def push(self, chunk_stereo: np.ndarray) -> Optional[dict]: """ Push one 80 ms stereo chunk (shape: (2, 1920) float32 at 24 kHz). Returns a decision dict if a VAD offset was detected, else None. """ self._buf_user.append(chunk_stereo[self._user_ch]) self._buf_agent.append(chunk_stereo[self._agent_ch]) if len(self._buf_user) < self._BATCH: return None # Run backbone on buffered chunk user_wav = torch.from_numpy(np.concatenate(self._buf_user)).unsqueeze(0) agent_wav = torch.from_numpy(np.concatenate(self._buf_agent)).unsqueeze(0) stereo = torch.stack([user_wav[0], agent_wav[0]], dim=0) self._buf_user.clear(); self._buf_agent.clear() model = self._model with torch.no_grad(): tokens = model._encode_with_mimi( stereo.unsqueeze(0).to(model.device), model.model, model.input_mode ) preds = model._run_inference(tokens, model.model) n_frames = preds["vad_user"].shape[0] # FVAD u16 = torchaudio.functional.resample(user_wav, MIMI_SR, model._silero_sr)[0].numpy() a16 = torchaudio.functional.resample(agent_wav, MIMI_SR, model._silero_sr)[0].numpy() fvad_us = _resample_to(model._anti_flicker(model._silero_probs(u16, model._silero_sr), 0.3), n_frames) fvad_ul = _resample_to(model._anti_flicker(model._silero_probs(u16, model._silero_sr), 0.7), n_frames) fvad_as = _resample_to(model._anti_flicker(model._silero_probs(a16, model._silero_sr), 0.3), n_frames) fvad_al = _resample_to(model._anti_flicker(model._silero_probs(a16, model._silero_sr), 0.7), n_frames) decision = None for i in range(n_frames): sig = { "vad_user": float(preds["vad_user"][i]), "vad_agent": float(preds["vad_agent"][i]), "eot_user": float(preds["eot_user"][i]), "eot_agent": float(preds["eot_agent"][i]), "bot_user": float(preds["bot_user"][i]), "bot_agent": float(preds["bot_agent"][i]), "fvad_user_short": float(fvad_us[i]), "fvad_user_long": float(fvad_ul[i]), "fvad_agent_short": float(fvad_as[i]), "fvad_agent_long": float(fvad_al[i]), } self._history.append(sig) cur_above = sig["vad_user"] > VAD_THRESHOLD recent_agent = float(np.mean([h["vad_agent"] for h in list(self._history)[-10:]])) agent_voicing = recent_agent > AGENT_VOICE_MIN if (self._prev_above and not cur_above and not agent_voicing and self._t_s - self._last_decision_t >= self._refractory_s): decision = model._classify(self._t_s, sig) self._last_decision_t = self._t_s self._prev_above = cur_above self._t_s += MIMI_MS / 1000.0 return decision # ───────────────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────────────── def _resample_to(arr: np.ndarray, n: int) -> np.ndarray: idx = np.linspace(0, len(arr) - 1, n) return np.interp(idx, np.arange(len(arr)), arr) # ───────────────────────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": import argparse, json p = argparse.ArgumentParser(description="DualTurn endpoint detection") p.add_argument("--audio", required=True) p.add_argument("--backbone", default=None, help="Path to best.pt (or use --from-hf)") p.add_argument("--classifier", default=None, help="Path to endpoint_clf.pkl") p.add_argument("--from-hf", default="anyreach/dualturn-endpointing", help="HuggingFace repo id (used if --backbone not set)") p.add_argument("--user-channel", type=int, default=0) p.add_argument("--agent-channel", type=int, default=1) p.add_argument("--device", default="cuda") p.add_argument("--st-threshold", type=float, default=None) p.add_argument("--out-json", default=None) args = p.parse_args() if args.backbone: model = DualTurnEndpointing( backbone_path = args.backbone, classifier_path = args.classifier, device = args.device, st_threshold = args.st_threshold, ) else: model = DualTurnEndpointing.from_pretrained( repo_id = args.from_hf, device = args.device, st_threshold = args.st_threshold, ) frames, endpoints = model.process_file( args.audio, user_channel=args.user_channel, agent_channel=args.agent_channel ) print(f"\n{'='*60}") print(f" {len(endpoints)} endpoint decisions in {frames[-1]['t_s']:.1f}s") print(f"{'='*60}") counts = {"ST": 0, "CL": 0} n_ep = sum(1 for ep in endpoints if ep["endpoint"]) for ep in endpoints: flag = "✓" if ep["endpoint"] else "·" print(f" {flag} t={ep['t_s']:6.2f}s endpoint={ep['endpoint']} P={ep['p_endpoint']:.3f}") print(f"\n endpoints={n_ep}/{len(endpoints)}") if args.out_json: Path(args.out_json).write_text(json.dumps({"frames": frames, "endpoints": endpoints}, indent=2)) print(f" Saved → {args.out_json}")