Add a simple "VAD" alternative - periodic VAD

This is a very simple alternative to using Silero Vad, where every
5 minutes (0:00 - 5:00, 5:00 - 10:00, etc.) is marked as a
"speech" segment, and whisper is then run on each
segment individually.

The upside is that all potential dialogue will be sent to
Whisper for transcribing, but it is also more likely to
get into an infinite sentence loop (although limited to each
5 minute segment).

Creating an artifical break every 5 minutes may also break
up a sentence accidentally, causing the sentence (or a word) to be
incorrectly transcribed.

app.py

@@ -15,7 +15,7 @@ import gradio as gr
 from download import ExceededMaximumDuration, downloadUrl
 
 from utils import slugify, write_srt, write_vtt
-from vad import VadTranscription
+from vad import VadPeriodicTranscription, VadSileroTranscription
 
 # Limitations (set to -1 to disable)
 DEFAULT_INPUT_AUDIO_MAX_DURATION = 600 # seconds
@@ -67,15 +67,23 @@ class UI:
                     model = whisper.load_model(selectedModel)
                     model_cache[selectedModel] = model
 
+                # Callable for processing an audio file
+                whisperCallable = lambda audio : model.transcribe(audio, language=selectedLanguage, task=task)
+
                 # The results
                 if (vad == 'silero-vad'):
                     # Use Silero VAD
                     if (self.vad_model is None):
-                        self.vad_model = VadTranscription()
-                    result = self.vad_model.transcribe(source, lambda audio : model.transcribe(audio, language=selectedLanguage, task=task))
+                        self.vad_model = VadSileroTranscription()
+                    result = self.vad_model.transcribe(source, whisperCallable)
+                elif (vad == 'periodic-vad'):
+                    # Very simple VAD - mark every 5 minutes as speech. This makes it less likely that Whisper enters an infinite loop, but
+                    # it may create a break in the middle of a sentence, causing some artifacts.
+                    periodic_vad = VadPeriodicTranscription(periodic_duration=60 * 5)
+                    result = periodic_vad.transcribe(source, whisperCallable)
                 else:
                     # Default VAD
-                    result = model.transcribe(source, language=selectedLanguage, task=task)
+                    result = whisperCallable(source)
 
                 text = result["text"]
 
@@ -176,7 +184,7 @@ def createUi(inputAudioMaxDuration, share=False, server_name: str = None):
         gr.Audio(source="upload", type="filepath", label="Upload Audio"), 
         gr.Audio(source="microphone", type="filepath", label="Microphone Input"),
         gr.Dropdown(choices=["transcribe", "translate"], label="Task"),
-        gr.Dropdown(choices=["none", "silero-vad"], label="VAD"),
+        gr.Dropdown(choices=["none", "silero-vad", "periodic-vad"], label="VAD"),
     ], outputs=[
         gr.File(label="Download"),
         gr.Text(label="Transcription"), 

tests/vad_test.py

@@ -0,0 +1,66 @@
+import pprint
+import unittest
+import numpy as np
+import sys
+
+sys.path.append('../whisper-webui')
+
+from vad import AbstractTranscription, VadSileroTranscription
+
+class TestVad(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        super(TestVad, self).__init__(*args, **kwargs)
+        self.transcribe_calls = []
+
+    def test_transcript(self):
+        mock = MockVadTranscription()
+
+        self.transcribe_calls.clear()
+        result = mock.transcribe("mock", lambda segment : self.transcribe_segments(segment))
+
+        self.assertListEqual(self.transcribe_calls, [ 
+            [30, 30],
+            [100, 100]
+        ])
+
+        self.assertListEqual(result['segments'],
+            [{'end': 50.0, 'start': 40.0, 'text': 'Hello world'},
+            {'end': 120.0, 'start': 110.0, 'text': 'Hello world'}]
+        )
+
+    def transcribe_segments(self, segment):
+        self.transcribe_calls.append(segment.tolist())
+
+        # Dummy text
+        return {
+            'text': "Hello world ",
+            'segments': [
+                {
+                    "start": 10.0,
+                    "end": 20.0,
+                    "text": "Hello world "
+                }   
+            ],
+            'language': ""
+        }
+
+class MockVadTranscription(AbstractTranscription):
+    def __init__(self):
+        super().__init__()
+
+    def get_audio_segment(self, str, start_time: str = None, duration: str = None):
+        start_time_seconds = float(start_time.removesuffix("s"))
+        duration_seconds = float(duration.removesuffix("s"))
+
+        # For mocking, this just returns a simple numppy array
+        return np.array([start_time_seconds, duration_seconds], dtype=np.float64)
+
+    def get_transcribe_timestamps(self, audio: str):
+        result = []
+
+        result.append( {  'start': 30, 'end': 60 } )
+        result.append( {  'start': 100, 'end': 200 } )
+        return result
+
+if __name__ == '__main__':
+    unittest.main()

vad.py

@@ -1,6 +1,7 @@
+from abc import ABC, abstractmethod
 from collections import Counter
 from dis import dis
-from typing import Any, Iterator, List, Dict
+from typing import Any, Callable, Iterator, List, Dict, Union
 
 from pprint import pprint
 import torch
@@ -8,71 +9,45 @@ import torch
 import ffmpeg
 import numpy as np
 
+# Defaults
 SPEECH_TRESHOLD = 0.3
 MAX_SILENT_PERIOD = 10 # seconds
-
 SEGMENT_PADDING_LEFT = 1 # Start detected text segment early
 SEGMENT_PADDING_RIGHT = 4 # End detected segments late
 
-def load_audio(file: str, sample_rate: int = 16000, 
-               start_time: str = None, duration: str = None):
-    """
-    Open an audio file and read as mono waveform, resampling as necessary
-
-    Parameters
-    ----------
-    file: str
-        The audio file to open
-
-    sr: int
-        The sample rate to resample the audio if necessary
-
-    start_time: str
-        The start time, using the standard FFMPEG time duration syntax, or None to disable.
-    
-    duration: str
-        The duration, using the standard FFMPEG time duration syntax, or None to disable.
-
-    Returns
-    -------
-    A NumPy array containing the audio waveform, in float32 dtype.
-    """
-    try:
-        inputArgs = {'threads': 0}
-
-        if (start_time is not None):
-            inputArgs['ss'] = start_time
-        if (duration is not None):
-            inputArgs['t'] = duration
 
-        # This launches a subprocess to decode audio while down-mixing and resampling as necessary.
-        # Requires the ffmpeg CLI and `ffmpeg-python` package to be installed.
-        out, _ = (
-            ffmpeg.input(file, **inputArgs)
-            .output("-", format="s16le", acodec="pcm_s16le", ac=1, ar=sample_rate)
-            .run(cmd="ffmpeg", capture_stdout=True, capture_stderr=True)
-        )
-    except ffmpeg.Error as e:
-        raise RuntimeError(f"Failed to load audio: {e.stderr.decode()}")
+class AbstractTranscription(ABC):
+    def __init__(self, segment_padding_left: int = None, segment_padding_right = None, max_silent_period: int = None):
+        self.sampling_rate = 16000
+        self.segment_padding_left = segment_padding_left
+        self.segment_padding_right = segment_padding_right
+        self.max_silent_period = max_silent_period
 
-    return np.frombuffer(out, np.int16).flatten().astype(np.float32) / 32768.0
+    def get_audio_segment(self, str, start_time: str = None, duration: str = None):
+        return load_audio(str, self.sampling_rate, start_time, duration)
 
-class VadTranscription:
-    def __init__(self):
-        self.model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', model='silero_vad')
+    @abstractmethod
+    def get_transcribe_timestamps(self, audio: str):
+        """
+        Get the start and end timestamps of the sections that should be transcribed by this VAD method.
 
-        (self.get_speech_timestamps, _, _, _, _) = utils
+        Parameters
+        ----------
+        audio: str
+            The audio file.
 
-    def transcribe(self, audio: str, whisperCallable):
-        SAMPLING_RATE = 16000
-        wav = load_audio(audio, sample_rate=SAMPLING_RATE)
+        Returns
+        -------
+        A list of start and end timestamps, in fractional seconds.
+        """
+        return 
 
+    def transcribe(self, audio: str, whisperCallable: Callable[[Union[str, np.ndarray, torch.Tensor]], dict[str, Union[dict, Any]]]):
         # get speech timestamps from full audio file
-        sample_timestamps = self.get_speech_timestamps(wav, self.model, sampling_rate=SAMPLING_RATE, threshold=SPEECH_TRESHOLD)
-        seconds_timestamps = self.convert_seconds(sample_timestamps, sampling_rate=SAMPLING_RATE)
+        seconds_timestamps = self.get_transcribe_timestamps(audio)
 
-        padded = self.pad_timestamps(seconds_timestamps, SEGMENT_PADDING_LEFT, SEGMENT_PADDING_RIGHT)
-        merged = self.merge_timestamps(padded, MAX_SILENT_PERIOD)
+        padded = self.pad_timestamps(seconds_timestamps, self.segment_padding_left, self.segment_padding_right)
+        merged = self.merge_timestamps(padded, self.max_silent_period)
 
         print("Timestamps:")
         pprint(merged)
@@ -89,7 +64,7 @@ class VadTranscription:
             segment_start = segment['start']
             segment_duration = segment['end'] - segment_start
 
-            segment_audio = load_audio(audio, sample_rate=SAMPLING_RATE, start_time = str(segment_start) + "s", duration = str(segment_duration) + "s")
+            segment_audio = self.get_audio_segment(audio, start_time = str(segment_start) + "s", duration = str(segment_duration) + "s")
 
             print("Running whisper on " + str(segment_start) + ", duration: " + str(segment_duration))
             segment_result = whisperCallable(segment_audio)
@@ -145,6 +120,9 @@ class VadTranscription:
         return result
 
     def merge_timestamps(self, timestamps: List[Dict[str, Any]], max_distance: float):
+        if max_distance is None:
+            return timestamps
+
         result = []
         current_entry = None
 
@@ -170,7 +148,7 @@ class VadTranscription:
 
         return result
 
-    def convert_seconds(self, timestamps: List[Dict[str, Any]], sampling_rate: int):
+    def multiply_timestamps(self, timestamps: List[Dict[str, Any]], factor: float):
         result = []
 
         for entry in timestamps:
@@ -178,8 +156,91 @@ class VadTranscription:
             end = entry['end']
 
             result.append({
-                'start': start / sampling_rate,
-                'end': end / sampling_rate
+                'start': start * factor,
+                'end': end * factor
             })
         return result
-        
+
+class VadSileroTranscription(AbstractTranscription):
+    def __init__(self):
+        super().__init__(SEGMENT_PADDING_LEFT, SEGMENT_PADDING_RIGHT, MAX_SILENT_PERIOD)
+        self.model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad', model='silero_vad')
+
+        (self.get_speech_timestamps, _, _, _, _) = utils
+
+    def get_transcribe_timestamps(self, audio: str):
+        wav = self.get_audio_segment(audio)
+
+        sample_timestamps = self.get_speech_timestamps(wav, self.model, sampling_rate=self.sampling_rate, threshold=SPEECH_TRESHOLD)
+        seconds_timestamps = self.multiply_timestamps(sample_timestamps, factor=1 / self.sampling_rate) 
+
+        return seconds_timestamps
+
+# A very simple VAD that just marks every N seconds as speech
+class VadPeriodicTranscription(AbstractTranscription):
+    def __init__(self, periodic_duration: int):
+        super().__init__()
+        self.periodic_duration = periodic_duration
+
+    def get_transcribe_timestamps(self, audio: str):
+        # Get duration in seconds
+        audio_duration = float(ffmpeg.probe(audio)["format"]["duration"])
+        result = []
+
+        # Generate a timestamp every N seconds
+        start_timestamp = 0
+
+        while (start_timestamp < audio_duration):
+            end_timestamp = min(start_timestamp + self.periodic_duration, audio_duration)
+            segment_duration = end_timestamp - start_timestamp
+
+            # Minimum duration is 1 second
+            if (segment_duration >= 1):
+                result.append( {  'start': start_timestamp, 'end': end_timestamp } )
+
+            start_timestamp = end_timestamp
+
+        return result
+
+def load_audio(file: str, sample_rate: int = 16000, 
+               start_time: str = None, duration: str = None):
+    """
+    Open an audio file and read as mono waveform, resampling as necessary
+
+    Parameters
+    ----------
+    file: str
+        The audio file to open
+
+    sr: int
+        The sample rate to resample the audio if necessary
+
+    start_time: str
+        The start time, using the standard FFMPEG time duration syntax, or None to disable.
+    
+    duration: str
+        The duration, using the standard FFMPEG time duration syntax, or None to disable.
+
+    Returns
+    -------
+    A NumPy array containing the audio waveform, in float32 dtype.
+    """
+    try:
+        inputArgs = {'threads': 0}
+
+        if (start_time is not None):
+            inputArgs['ss'] = start_time
+        if (duration is not None):
+            inputArgs['t'] = duration
+
+        # This launches a subprocess to decode audio while down-mixing and resampling as necessary.
+        # Requires the ffmpeg CLI and `ffmpeg-python` package to be installed.
+        out, _ = (
+            ffmpeg.input(file, **inputArgs)
+            .output("-", format="s16le", acodec="pcm_s16le", ac=1, ar=sample_rate)
+            .run(cmd="ffmpeg", capture_stdout=True, capture_stderr=True)
+        )
+    except ffmpeg.Error as e:
+        raise RuntimeError(f"Failed to load audio: {e.stderr.decode()}")
+
+    return np.frombuffer(out, np.int16).flatten().astype(np.float32) / 32768.0