| import numpy as np |
| import torch |
|
|
|
|
| class SileroVAD: |
| def __init__(self, path=None): |
| from silero_vad import load_silero_vad |
| self.vad = load_silero_vad(onnx=True) |
|
|
| def reset(self): |
| self.vad.reset_states() |
|
|
| def __call__(self, chunk, sr=16000): |
| if chunk.shape[-1] not in (256, 512): |
| return 0.0 |
| t = torch.from_numpy(chunk.reshape(1, -1).astype(np.float32)) |
| return float(self.vad(t, sr)) |
|
|
|
|
| class RealtimeSession: |
| def __init__(self, vad_path, sr=16000, threshold=0.5, min_speech_ms=128, min_silence_ms=800): |
| self.vad, self.sr, self.threshold = SileroVAD(vad_path), sr, threshold |
| self.min_speech, self.min_silence = int(sr * min_speech_ms / 1000), int(sr * min_silence_ms / 1000) |
| self.reset() |
|
|
| def reset(self): |
| self.vad.reset() |
| self.buffer, self.ring, self.speaking, self.generating, self.interrupt = [], [], False, False, False |
| self.speech_samples = self.silence_samples = self.tail_silence = 0 |
|
|
| def push_chunk(self, chunk, W=512): |
| for i in range(0, max(len(chunk), 1), W): |
| w = chunk[i:i + W] |
| if len(w) < W: |
| w = np.pad(w, (0, W - len(w))) |
| prob = self.vad(w, self.sr) |
| if prob > self.threshold: |
| self.silence_samples = self.tail_silence = 0 |
| self.speech_samples += len(w) |
| self.buffer.append(w) |
| if self.speech_samples >= self.min_speech and not self.speaking: |
| self.speaking = True |
| self.buffer = self.ring + self.buffer |
| self.ring = [] |
| if self.generating and self.speaking: |
| self.interrupt = True |
| return 'interrupt' |
| elif self.speaking: |
| self.silence_samples += len(w) |
| self.tail_silence += 1 |
| self.buffer.append(w) |
| if self.silence_samples >= self.min_silence: |
| if self.tail_silence > 1: |
| del self.buffer[-(self.tail_silence - 1):] |
| self.speaking, self.speech_samples, self.silence_samples, self.tail_silence = False, 0, 0, 0 |
| return 'speech_end' |
| else: |
| if self.speech_samples > 0: |
| self.buffer.clear() |
| self.speech_samples = 0 |
| self.ring = [w] |
| return 'listening' |
|
|
| def get_audio(self): |
| audio = np.concatenate(self.buffer) if self.buffer else np.array([], dtype=np.float32) |
| self.buffer.clear() |
| return audio |
|
|
