File size: 2,637 Bytes
45cf443 f268594 45cf443 f268594 45cf443 f268594 45cf443 f268594 45cf443 f268594 45cf443 f268594 45cf443 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | 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
|