vad.py

"""
vad.py — Detector de voz con red neuronal (Silero VAD v5, ONNX, licencia MIT).

A diferencia del detector por energía, distingue la voz de otros sonidos fuertes (un cláxon, una
puerta, un golpe): sirve para encontrar ruidos sueltos en las pausas sin tocar la voz.
Trabaja a 16 kHz en ventanas de 512 muestras (32 ms).
"""

from __future__ import annotations

import os
import threading

import numpy as np

SR = 16000
WIN = 512
CONTEXT = 64
HOP_S = WIN / SR

_lock = threading.Lock()
_session = None


def _load():
    global _session
    with _lock:
        if _session is None:
            import onnxruntime as ort

            from models import silero_vad_path

            opts = ort.SessionOptions()
            opts.log_severity_level = 3
            opts.intra_op_num_threads = 1  # modelo diminuto: un hilo es lo más rápido
            opts.inter_op_num_threads = 1
            _session = ort.InferenceSession(str(silero_vad_path()), opts, providers=["CPUExecutionProvider"])
    return _session


def speech_prob(mono: np.ndarray, sr: int) -> np.ndarray:
    """Probabilidad de voz (0–1) por ventana de 32 ms."""
    import soxr

    sess = _load()
    x = soxr.resample(np.asarray(mono, dtype=np.float32), sr, SR, quality="HQ") if sr != SR else mono
    x = np.asarray(x, dtype=np.float32)
    n = int(np.ceil(len(x) / WIN))
    x = np.pad(x, (0, n * WIN - len(x)))
    state = np.zeros((2, 1, 128), dtype=np.float32)
    ctx = np.zeros(CONTEXT, dtype=np.float32)
    sr_arr = np.array(SR, dtype=np.int64)
    out = np.empty(n, dtype=np.float32)
    for i in range(n):
        chunk = x[i * WIN : (i + 1) * WIN]
        inp = np.concatenate([ctx, chunk])[None, :]
        p, state = sess.run(None, {"input": inp, "state": state, "sr": sr_arr})
        out[i] = p[0, 0]
        ctx = chunk[-CONTEXT:]
    return out


def available() -> bool:
    try:
        _load()
        return True
    except Exception as e:
        print(f"[VAD] No disponible: {e}")
        return False