| | import torchaudio |
| | from VAD.vad_iterator import VADIterator |
| | from baseHandler import BaseHandler |
| | import numpy as np |
| | import torch |
| | from rich.console import Console |
| |
|
| | from utils.utils import int2float |
| | from df.enhance import enhance, init_df |
| | import logging |
| |
|
| | logger = logging.getLogger(__name__) |
| |
|
| | console = Console() |
| |
|
| |
|
| | class VADHandler(BaseHandler): |
| | """ |
| | Handles voice activity detection. When voice activity is detected, audio will be accumulated until the end of speech is detected and then passed |
| | to the following part. |
| | """ |
| |
|
| | def setup( |
| | self, |
| | should_listen, |
| | thresh=0.3, |
| | sample_rate=16000, |
| | min_silence_ms=1000, |
| | min_speech_ms=500, |
| | max_speech_ms=float("inf"), |
| | speech_pad_ms=30, |
| | audio_enhancement=False, |
| | ): |
| | self.should_listen = should_listen |
| | self.sample_rate = sample_rate |
| | self.min_silence_ms = min_silence_ms |
| | self.min_speech_ms = min_speech_ms |
| | self.max_speech_ms = max_speech_ms |
| | self.model, _ = torch.hub.load("snakers4/silero-vad", "silero_vad") |
| | self.iterator = VADIterator( |
| | self.model, |
| | threshold=thresh, |
| | sampling_rate=sample_rate, |
| | min_silence_duration_ms=min_silence_ms, |
| | speech_pad_ms=speech_pad_ms, |
| | ) |
| | self.audio_enhancement = audio_enhancement |
| | if audio_enhancement: |
| | self.enhanced_model, self.df_state, _ = init_df() |
| |
|
| | def process(self, audio_chunk): |
| | audio_int16 = np.frombuffer(audio_chunk, dtype=np.int16) |
| | audio_float32 = int2float(audio_int16) |
| | vad_output = self.iterator(torch.from_numpy(audio_float32)) |
| | if vad_output is not None and len(vad_output) != 0: |
| | console.print("VAD: end of speech detected") |
| | logger.debug("VAD: end of speech detected") |
| | array = torch.cat(vad_output).cpu().numpy() |
| | duration_ms = len(array) / self.sample_rate * 1000 |
| | if duration_ms < self.min_speech_ms or duration_ms > self.max_speech_ms: |
| | console.print( |
| | f"audio input of duration: {len(array) / self.sample_rate}s, skipping" |
| | ) |
| | logger.debug( |
| | f"audio input of duration: {len(array) / self.sample_rate}s, skipping" |
| | ) |
| | else: |
| | self.should_listen.clear() |
| | logger.debug("Stop listening") |
| | if self.audio_enhancement: |
| | if self.sample_rate != self.df_state.sr(): |
| | audio_float32 = torchaudio.functional.resample( |
| | torch.from_numpy(array), |
| | orig_freq=self.sample_rate, |
| | new_freq=self.df_state.sr(), |
| | ) |
| | enhanced = enhance( |
| | self.enhanced_model, |
| | self.df_state, |
| | audio_float32.unsqueeze(0), |
| | ) |
| | enhanced = torchaudio.functional.resample( |
| | enhanced, |
| | orig_freq=self.df_state.sr(), |
| | new_freq=self.sample_rate, |
| | ) |
| | else: |
| | enhanced = enhance( |
| | self.enhanced_model, self.df_state, audio_float32 |
| | ) |
| | array = enhanced.numpy().squeeze() |
| | yield array |
| |
|
| | @property |
| | def min_time_to_debug(self): |
| | return 0.00001 |
| |
|
