Max_speech_duration_s for streaming scenario

[lionsheep24]

Hi all,
I'm planning to use silero-vad for my STT service, but I need to limit the model's audio input length to manage latency and throughput. From what I understand, I should use VADIterator for streaming input, but I noticed that there isn't a max_speech_duration_s parameter for it. I've only seen this parameter in the get_speech_timestamps function.

Is there a way to limit the maximum speech length in a streaming scenario?

Let me share my implementation (V4)

MINIMUM_SAMPLES_FOR_VAD: int = 256 
MIN_SILENCE_SAMPLES: int = 80000  


class VADProcessor:

    def __init__(self, vad_model_path: str):
        self.vad_iterator = self.get_vad_iterator(vad_model_path)

    def get_vad_iterator(self, vad_model_path: str):
        self.vad_model, VADIterator = torch.hub.load(
            source="local", repo_or_dir=vad_model_path, model="silero_vad", onnx=True
        )
        vad_iterator: VADIterator = VADIterator(self.vad_model, sampling_rate=8000)
        return vad_iterator

    def detect_speech(self, audio_chunk: bytes) -> Dict[str, float]:
        audio_tensor: torch.tensor = torch.frombuffer(audio_chunk, dtype=torch.int16).float() / 32768.0
        audio_tensor: torch.tensor = audio_tensor.unsqueeze(0)
        audio_samples: int = audio_tensor.shape[1]
        if audio_samples < MINIMUM_SAMPLES_FOR_VAD:
            padding_length = MIN_SILENCE_SAMPLES - audio_samples
            padded_audio_tensor = F.pad(audio_tensor, (0, padding_length), "constant", 0)
            vad_event: Dict[str, float] = self.vad_iterator(padded_audio_tensor, return_seconds=True)
            speech_prob: float = self.vad_model(padded_audio_tensor, 8000).item()
        else:
            vad_event: Dict[str, float] = self.vad_iterator(audio_tensor, return_seconds=True)
            speech_prob = self.vad_model(audio_tensor, 8000).item()
        return speech_prob, vad_event

    def reset_states(self):
        self.vad_iterator.reset_states()


class VADIterator:
    def __init__(
        self,
        model,
        threshold: float = 0.5,
        sampling_rate: int = 8000,
        min_silence_duration_ms: int = 1000,
        speech_pad_ms: int = 30,
    ):
        """
        Class for stream imitation

        Parameters
        ----------
        model: preloaded .jit silero VAD model

        threshold: float (default - 0.5)
            Speech threshold. Silero VAD outputs speech probabilities for each audio chunk, probabilities ABOVE this value are considered as SPEECH.
            It is better to tune this parameter for each dataset separately, but "lazy" 0.5 is pretty good for most datasets.

        sampling_rate: int (default - 8000)
            Currently silero VAD models support 8000 and 16000 sample rates

        min_silence_duration_ms: int (default - 100 milliseconds)
            In the end of each speech chunk wait for min_silence_duration_ms before separating it

        speech_pad_ms: int (default - 30 milliseconds)
            Final speech chunks are padded by speech_pad_ms each side
        """

        self.model = model
        self.threshold = threshold
        self.sampling_rate = sampling_rate

        if sampling_rate not in [8000, 16000]:
            raise ValueError("VADIterator does not support sampling rates other than [8000, 16000]")

        self.min_silence_samples = sampling_rate * min_silence_duration_ms / 1000
        self.speech_pad_samples = sampling_rate * speech_pad_ms / 1000
        self.reset_states()

    def reset_states(self):
        self.model.reset_states()
        self.triggered = False
        self.temp_end = 0
        self.current_sample = 0

    def __call__(self, x, return_seconds=False):
        """
        x: torch.Tensor
            audio chunk (see examples in repo)

        return_seconds: bool (default - False)
            whether return timestamps in seconds (default - samples)
        """

        if not torch.is_tensor(x):
            try:
                x = torch.Tensor(x)
            except:
                raise TypeError("Audio cannot be casted to tensor. Cast it manually")

        window_size_samples = len(x[0]) if x.dim() == 2 else len(x)
        self.current_sample += window_size_samples

        speech_prob = self.model(x, self.sampling_rate).item()

        if (speech_prob >= self.threshold) and self.temp_end:
            self.temp_end = 0

        if (speech_prob >= self.threshold) and not self.triggered:
            self.triggered = True
            speech_start = self.current_sample - self.speech_pad_samples - window_size_samples
            return {"start": int(speech_start) if not return_seconds else round(speech_start / self.sampling_rate, 1)}

        if (speech_prob < self.threshold - 0.15) and self.triggered:
            if not self.temp_end:
                self.temp_end = self.current_sample
            if self.current_sample - self.temp_end < self.min_silence_samples:
                return None
            else:
                speech_end = self.temp_end + self.speech_pad_samples - window_size_samples
                self.temp_end = 0
                self.triggered = False
                return {"end": int(speech_end) if not return_seconds else round(speech_end / self.sampling_rate, 1)}

        return None


# Example implementation
for audio_chink in stream.read():
     speech_prob, vad_event = vad_processor.detect_speech(audio_chunk)

[snakers4]

Hi,

Is there a way to limit the maximum speech length in a streaming scenario?

In a streaming scenario we really do not know in advance how long the person would speak.
Also different domains may differ very much in the heuristics that are to be applied on the fly.

Hence we just provide the minimal example, scaffolding for a streaming scenario.
It is assumed that for each domain and application an end user would modify / write the streaming post-processing for himself.
The provided iterator is just a basic demonstration.

[Simon-chai]

I have post a similar issue last year ,see #405. And I think it's a worth solving problem. I will have a try.