Initial commit

Check in a shit ton of code.

Most of the audio processing logic in `app.py` is lifted/ported from
github.com/yum_food/TaSTT. I made some adjustments to make it work
better (removing normalization, adding volume filters) and also increase
fidelity.

6 files changed, 890 insertions, 0 deletions

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..afc1131
--- /dev/null
+++ b/README.md
@@ -0,0 +1,65 @@
+## yapBox
+
+A black box for your yapping.
+
+This app records your mic. It uses silero-vad to split audio into contiguous
+segments of speech, and saves them to disk as .wav files. Metadata is
+saved to a corresponding .yaml file.
+
+What's a black box? Wikipedia says this:
+```
+A flight recorder is an electronic recording device placed in an aircraft for
+the purpose of facilitating the investigation of aviation accidents and
+incidents. The device may often be referred to colloquially as a "black box",
+an outdated name which has become a misnomer—they are now required to be
+painted bright orange, to aid in their recovery after accidents.
+```
+
+This is a CLI app. It is not polished and requires a little elbow grease to
+use properly. The intent is to assist people who want to gather high-quality
+training data of human voices. Use responsibly.
+
+## Compatibility
+
+This application is designed for Windows 10. Functionality on any other
+platform is purely coincidental.
+
+## Running
+
+Download the latest release and double click `app.bat` in File Explorer.
+
+Read the output and change the mic to whatever you're using. To change mics,
+edit app.py. Any text editor works, including Notepad.
+
+## Building from source
+
+First install python 3.10.9. Make sure that Powershell is using that version by
+typing this (leave out the $, it's used to differentiate between commands and
+output):
+```
+$ python.exe --version
+Python 3.10.9
+```
+
+Then open Powershell and run package.ps1:
+```
+$ cd pkg
+$ ./package.ps1
+```
+
+All dependencies should download themselves. It will use the host Python to
+install dependencies into the app's environment.
+
+## Ethics
+
+We are living in the wild west of AI. You can clone anyone's voice and
+plausibly reproduce it using projects like
+[rvc-beta](https://github.com/RVC-Project/Retrieval-based-Voice-Conversion-WebUI/releases).
+Legislation has not caught
+up to this yet. Cloning someone's voice without their consent is, at best,
+ethically dubious. This tool makes that process easier. In the absence of a
+legal framework, you must make your own choices as to what is right. Take this
+seriously. When in doubt, follow Kant's [universalization
+principle](https://en.wikipedia.org/wiki/Universalizability) and the [golden
+rule](https://en.wikipedia.org/wiki/Golden_Rule).
+
diff --git a/app.bat b/app.bat
new file mode 100644
index 0000000..79f7c1f
--- /dev/null
+++ b/app.bat
@@ -0,0 +1,3 @@
+start "yapBox" .\Python\python.exe app.py
+exit
+
diff --git a/app.py b/app.py
new file mode 100644
index 0000000..3cb3816
--- /dev/null
+++ b/app.py
@@ -0,0 +1,417 @@
+from datetime import datetime
+from pydub import AudioSegment
+
+import math
+import numpy as np
+import os
+import pyaudio
+import sys
+import time
+import typing
+import vad
+import wave
+
+class AudioStream():
+    FORMAT = pyaudio.paInt16
+    # Size of each frame (audio sample), in bytes. If you change FORMAT, make
+    # sure this stays up to date!
+    FRAME_SZ = 2
+    # Frames per second.
+    FPS = 16000
+    CHANNELS = 1
+    def __init__(self):
+        pass
+
+    def getSamples(self) -> bytes:
+        raise NotImplementedError("getSamples is not implemented!")
+
+class MicStream(AudioStream):
+    CHUNK_SZ = 1024
+
+    def __init__(self, which_mic: str, fps: int = AudioStream.FPS):
+        self.p = pyaudio.PyAudio()
+        self.stream = None
+        self.sample_rate = None
+        # Each time pyaudio gives us audio data, it's in the form of a chunk of
+        # samples. We keep these in a list to keep the audio callback as light
+        # as possible. Whenever downstream layers want data, we collapse the
+        # list into a single array of data (a bytes object).
+        self.chunks = []
+        # If set, incoming frames are simply discarded.
+        self.paused = False
+        self.fps = fps
+
+        print(f"Finding mic {which_mic}", file=sys.stderr)
+        self.dumpMicDevices()
+
+        got_match = False
+        device_index = -1
+        focusrite_str = "Focusrite"
+        index_str = "Digital Audio Interface"
+        if which_mic == "index":
+            target_str = index_str
+        elif which_mic == "focusrite":
+            target_str = focusrite_str
+        else:
+            print(f"Mic {which_mic} requested, treating it as a numerical " +
+                    "device ID", file=sys.stderr)
+            device_index = int(which_mic)
+            got_match = True
+        if not got_match:
+            info = self.p.get_host_api_info_by_index(0)
+            numdevices = info.get('deviceCount')
+            for i in range(0, numdevices):
+                if (self.p.get_device_info_by_host_api_device_index(0, i).get('maxInputChannels')) > 0:
+                    device_name = self.p.get_device_info_by_host_api_device_index(0, i).get('name')
+                    if target_str in device_name:
+                        print(f"Got matching mic: {device_name}",
+                                file=sys.stderr)
+                        device_index = i
+                        got_match = True
+                        break
+        if not got_match:
+            raise KeyError(f"Mic {which_mic} not found")
+
+        info = self.p.get_device_info_by_host_api_device_index(0, device_index)
+        print(f"Found mic {which_mic}: {info['name']}", file=sys.stderr)
+        self.sample_rate = int(info['defaultSampleRate'])
+        print(f"Mic sample rate: {self.sample_rate}", file=sys.stderr)
+
+        self.stream = self.p.open(
+                rate=self.sample_rate,
+                channels=self.CHANNELS,
+                format=self.FORMAT,
+                input=True,
+                frames_per_buffer=MicStream.CHUNK_SZ,
+                input_device_index=device_index,
+                stream_callback=self.onAudioFramesAvailable)
+
+        self.stream.start_stream()
+
+        AudioStream.__init__(self)
+
+    def pause(self, state: bool = True):
+        self.paused = state
+
+    def dumpMicDevices(self):
+        info = self.p.get_host_api_info_by_index(0)
+        numdevices = info.get('deviceCount')
+
+        for i in range(0, numdevices):
+            if (self.p.get_device_info_by_host_api_device_index(0, i).get('maxInputChannels')) > 0:
+                device_name = self.p.get_device_info_by_host_api_device_index(0, i).get('name')
+                print("Input Device id ", i, " - ", device_name)
+
+    def onAudioFramesAvailable(self,
+            frames,
+            frame_count,
+            time_info,
+            status_flags):
+        if self.paused:
+            # Don't literally pause, just start returning silence. This allows
+            # the `min_segment_age_s` check to work while paused.
+            n_frames = int(frame_count * self.fps /
+                    float(self.sample_rate))
+            self.chunks.append(np.zeros(n_frames,
+                dtype=np.int16).tobytes())
+            return (frames, pyaudio.paContinue)
+
+        decimated = b''
+        # In pyaudio, a `frame` is a single sample of audio data.
+        frame_len = self.FRAME_SZ
+        next_frame = 0.0
+        # The mic probably has a higher sample rate than Whisper wants, so
+        # decrease the sample rate by dropping samples. Note that this
+        # algorithm only works if the mic's rate is higher than whisper's
+        # expected rate.
+        keep_every = float(self.sample_rate) / self.fps
+        for i in range(frame_count):
+            if i >= next_frame:
+                decimated += frames[i*frame_len:(i+1)*frame_len]
+                next_frame += keep_every
+        self.chunks.append(decimated)
+
+        return (frames, pyaudio.paContinue)
+
+    # Get audio data and the corresponding timestamp.
+    def getSamples(self) -> bytes:
+        chunks = self.chunks
+        self.chunks = []
+        result = b''.join(chunks)
+        return result
+
+class AudioCollector:
+    def __init__(self, stream: AudioStream):
+        self.stream = stream
+        self.frames = b''
+        # Note: by design, this is the only spot where we anchor our timestamps
+        # against the real world. This is done to make it possible to profile
+        # test cases which read from disk (at much faster than real speed) in
+        # the same way that we profile real-time data.
+        self.wall_ts = time.time()
+
+    def getAudio(self) -> bytes:
+        frames = self.stream.getSamples()
+        if frames:
+            self.frames += frames
+        return self.frames
+
+    def dropAudioPrefix(self, dur_s: float) -> bytes:
+        n_bytes = int(dur_s * self.stream.fps) * self.stream.FRAME_SZ
+        n_bytes = min(n_bytes, len(self.frames))
+        cut_portion = self.frames[:n_bytes]
+        self.frames = self.frames[n_bytes:]
+        self.wall_ts += float(n_bytes / self.stream.FRAME_SZ) / self.stream.fps
+        return cut_portion
+
+    def dropAudioPrefixByFrames(self, dur_frames: int) -> bytes:
+        n_bytes = dur_frames * self.stream.FRAME_SZ
+        n_bytes = min(n_bytes, len(self.frames))
+        cut_portion = self.frames[:n_bytes]
+        self.frames = self.frames[n_bytes:]
+        self.wall_ts += float(n_bytes / self.stream.FRAME_SZ) / self.stream.fps
+        return cut_portion
+
+    def keepLast(self, dur_s: float) -> bytes:
+        drop_len = max(0, self.duration() - dur_s)
+        return self.dropAudioPrefix(drop_len)
+
+    def dropAudio(self):
+        self.wall_ts += self.duration()
+        cut_portion = self.frames
+        self.frames = b''
+        return cut_portion
+
+    def duration(self):
+        return len(self.frames) / (self.stream.fps * self.stream.FRAME_SZ)
+
+    def begin(self):
+        return self.wall_ts
+
+    def now(self):
+        return self.begin() + self.duration()
+
+class AudioCollectorFilter:
+    def __init__(self, parent: AudioCollector):
+        self.parent = parent
+        self.stream = self.parent.stream
+
+    def getAudio(self) -> bytes:
+        return self.parent.getAudio()
+    def dropAudioPrefix(self, dur_s: float):
+        return self.parent.dropAudioPrefix(dur_s)
+    def dropAudioPrefixByFrames(self, dur_frames: int):
+        return self.parent.dropAudioPrefixByFrames(dur_frames)
+    def keepLast(self, dur_s):
+        return self.parent.keepLast(dur_s)
+    def dropAudio(self):
+        return self.parent.dropAudio()
+    def duration(self):
+        return self.parent.duration()
+    def begin(self):
+        return self.parent.begin()
+    def now(self):
+        return self.parent.now()
+
+class NormalizingAudioCollector(AudioCollectorFilter):
+    def __init__(self, parent: AudioCollector):
+        AudioCollectorFilter.__init__(self, parent)
+
+    def getAudio(self) -> bytes:
+        audio = self.parent.getAudio()
+
+        audio = AudioSegment(audio, sample_width=self.stream.FRAME_SZ,
+                frame_rate=self.stream.fps, channels=self.stream.CHANNELS)
+        audio = audio.normalize()
+
+        frames = np.array(audio.get_array_of_samples())
+        frames = np.int16(frames).tobytes()
+
+        return frames
+
+class CompressingAudioCollector(AudioCollectorFilter):
+    def __init__(self, parent: AudioCollector):
+        AudioCollectorFilter.__init__(self, parent)
+
+    def getAudio(self) -> bytes:
+        audio = self.parent.getAudio()
+
+        audio = AudioSegment(audio,
+                sample_width=self.stream.FRAME_SZ,
+                frame_rate=self.stream.fps,
+                channels=self.stream.CHANNELS)
+        # subtle compression has a slight positive effect on my benchmark
+        audio = audio.compress_dynamic_range(threshold=-10, ratio=2.0)
+
+        frames = np.array(audio.get_array_of_samples())
+        frames = np.int16(frames).tobytes()
+
+        return frames
+
+class AudioSegmenter:
+    def __init__(self,
+            min_silence_ms=250,
+            max_speech_s=5,
+            stream: AudioStream = None):
+        self.vad_options = vad.VadOptions(
+                min_silence_duration_ms=min_silence_ms,
+                max_speech_duration_s=max_speech_s)
+        self.stream = stream
+        pass
+
+    def segmentAudio(self, audio: bytes):
+        audio = np.frombuffer(audio,
+                dtype=np.int16).flatten().astype(np.float32) / 32768.0
+        return vad.get_speech_timestamps(audio, vad_options=self.vad_options)
+
+    # Returns the stable cutoff (if any) and whether there are any segments.
+    def getStableCutoff(self, audio: bytes) -> typing.Tuple[int, bool]:
+        min_delta_frames = int((self.vad_options.min_silence_duration_ms *
+                self.stream.fps) / 1000)
+        cutoff = None
+
+        last_end = None
+        segments = self.segmentAudio(audio)
+
+        for i in range(len(segments)):
+            s = segments[i]
+            #print(f"s: {s}")
+            #print(f"last_end: {last_end}")
+
+            if last_end:
+                delta_frames = s['start'] - last_end
+                #print(f"delta frames: {delta_frames}")
+                if delta_frames > min_delta_frames:
+                    cutoff = s['start']
+            else:
+                last_end = s['end']
+
+            if i == len(segments) - 1:
+                now = int(len(audio) / self.stream.FRAME_SZ)
+                delta_frames = now - s['end']
+                if delta_frames > min_delta_frames:
+                    cutoff = now - int(min_delta_frames / 2)
+
+        return (cutoff, len(segments) > 0)
+
+def install_in_venv(pkgs: typing.List[str]) -> bool:
+    pkgs_str = " ".join(pkgs)
+    print(f"Installing {pkgs_str}")
+    pip_proc = subprocess.Popen(
+            f"Resources/Python/python.exe -m pip install {pkgs_str} --no-warn-script-location".split(),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE)
+    pip_stdout, pip_stderr = pip_proc.communicate()
+    pip_stdout = pip_stdout.decode("utf-8")
+    pip_stderr = pip_stderr.decode("utf-8")
+    print(pip_stdout, file=sys.stderr)
+    print(pip_stderr, file=sys.stderr)
+    if pip_proc.returncode != 0:
+        print(f"`pip install {pkgs_str}` exited with {pip_proc.returncode}",
+                file=sys.stderr)
+        return False
+    return True
+
+def saveAudio(audio: bytes, path: str, stream: AudioStream):
+    with wave.open(path, 'wb') as wf:
+        print(f"Saving audio to {path}", file=sys.stderr)
+        wf.setnchannels(stream.CHANNELS)
+        wf.setsampwidth(stream.FRAME_SZ)
+        wf.setframerate(stream.fps)
+        wf.writeframes(audio)
+
+def concatenate_wav_files(output_path):
+    # List all .wav files in the CWD
+    wav_files = [f for f in os.listdir('.') if f.endswith('.wav')]
+
+    # Initialize parameters for wave file
+    params = None
+
+    # Open the output file
+    with wave.open(output_path, 'wb') as output_wav:
+        for wav_file in wav_files:
+            print(f"Processing {wav_file}")
+            with wave.open(wav_file, 'rb') as input_wav:
+                # Check if parameters are the same for each file
+                if params is None:
+                    params = input_wav.getparams()
+                    output_wav.setparams(params)
+
+                # Read and write frames
+                frames = input_wav.readframes(input_wav.getnframes())
+                output_wav.writeframes(frames)
+
+if __name__ == "__main__":
+    abspath = os.path.abspath(__file__)
+    dname = os.path.dirname(abspath)
+    os.chdir(dname)
+    print(f"Set cwd to {os.getcwd()}", file=sys.stderr)
+
+    concatenate_wav_files("concatenated.wav")
+    sys.exit(0)
+
+    stream = MicStream("index")
+    stream_hd = MicStream("index", fps=44100)
+
+    collector = AudioCollector(stream)
+    #collector = NormalizingAudioCollector(collector)
+    collector = CompressingAudioCollector(collector)
+
+    collector_hd = AudioCollector(stream_hd)
+    #collector_hd = NormalizingAudioCollector(collector_hd)
+    collector_hd = CompressingAudioCollector(collector_hd)
+
+    min_silence_ms = 1000
+    max_speech_s = 30
+    segmenter = AudioSegmenter(
+            min_silence_ms=min_silence_ms,
+            max_speech_s=max_speech_s,
+            stream=stream)
+
+    while True:
+        audio = collector.getAudio()
+        collector_hd.getAudio()
+        stable_cutoff, has_audio = segmenter.getStableCutoff(audio)
+
+        #print(f"has audio: {has_audio}")
+        #print(f"stable cutoff: {stable_cutoff}")
+
+        if has_audio and stable_cutoff:
+            commit_audio = collector.dropAudioPrefixByFrames(stable_cutoff)
+            print(f"stable cutoff: {stable_cutoff}")
+            hd_cutoff = int(math.floor(stable_cutoff * stream_hd.fps /
+                stream.fps))
+            print(f"hd cutoff: {hd_cutoff}")
+            commit_audio_hd = collector_hd.dropAudioPrefixByFrames(hd_cutoff)
+            print(f"hd audio len: {len(commit_audio_hd)}")
+
+            # Calculate naive measure of volume
+            audio_v = AudioSegment(commit_audio_hd,
+                    sample_width=stream_hd.FRAME_SZ,
+                    frame_rate=stream_hd.fps,
+                    channels=stream_hd.CHANNELS)
+            audio_v = np.array(audio_v.get_array_of_samples())
+            audio_v = np.int16(audio_v)
+            audio_v = np.sqrt(np.mean(np.square(audio_v)))
+            audio_v /= np.sqrt(len(commit_audio_hd) / stream_hd.FRAME_SZ)
+            audio_v = math.log(audio_v, 10)
+            print(f"volume: {audio_v}")
+            # cutoff is a fine-tuned value based on volumes seen while in vr
+            # (index mic)
+            if audio_v < -1.3 or audio_v > -0.8:
+                # Discard sample
+                print("Discarding too-quiet/too-loud segment")
+                collector.keepLast(1.0)
+                collector_hd.keepLast(1.0)
+                continue
+
+
+            ts = datetime.fromtimestamp(time.time())
+            filename = str(ts.strftime('%Y_%m_%d__%H-%M-%S')) + ".wav"
+            saveAudio(commit_audio_hd, filename, stream_hd)
+
+        if not has_audio:
+            #print("VAD detects no audio, skip transcription", file=sys.stderr)
+            collector.keepLast(1.0)
+            collector_hd.keepLast(1.0)
+
diff --git a/pkg/package.ps1 b/pkg/package.ps1
new file mode 100644
index 0000000..694cdea
--- /dev/null
+++ b/pkg/package.ps1
@@ -0,0 +1,85 @@
+param(
+  [switch]$skip_zip = $false,
+  [string]$release = "Release",
+  [string]$install_pip = $true
+)
+
+echo "Skip zip: $skip_zip"
+echo "Release: $release"
+echo "Install pip: $install_pip"
+
+$PSDefaultParameterValues['Out-File:Encoding'] = 'utf8'
+
+$install_dir = "yapBox"
+
+if (Test-Path $install_dir) {
+  rm -Recurse -Force $install_dir
+}
+
+$py_dir = "Python"
+
+if (Test-Path $py_dir) {
+  rm -Recurse $py_dir
+}
+if (-Not (Test-Path $py_dir)) {
+  echo "Fetching python"
+
+  $PYTHON_3_10_9_URL = "https://www.python.org/ftp/python/3.10.9/python-3.10.9-embed-amd64.zip"
+  $PYTHON_URL = $PYTHON_3_10_9_URL
+  $PYTHON_FILE = $(Split-Path -Path $PYTHON_URL -Leaf)
+
+  if (-Not (Test-Path $PYTHON_FILE)) {
+    Invoke-WebRequest $PYTHON_URL -OutFile $PYTHON_FILE
+  }
+
+  mkdir Python
+  Expand-Archive $PYTHON_FILE -DestinationPath Python
+
+  echo ".." >> Python/python310._pth
+  echo "import site" >> Python/python310._pth
+}
+
+$pip_path = "$py_dir/get-pip.py"
+
+if (Test-Path $pip_path) {
+  rm -Force $pip_path
+}
+
+if (-Not (Test-Path $pip_path)) {
+  echo "Fetching pip"
+
+  $PIP_URL = "https://bootstrap.pypa.io/get-pip.py"
+  $PIP_FILE = $(Split-Path -Path $PIP_URL -Leaf)
+
+  if (-Not (Test-Path $PIP_FILE)) {
+    Invoke-WebRequest $PIP_URL -OutFile $PIP_FILE
+  }
+
+  mv $PIP_FILE $pip_path
+}
+
+if ($install_pip) {
+  ./Python/python.exe Python/get-pip.py
+
+  echo "Installing requirements"
+  echo "Assuming host has python 3.10.9 installed" # TODO test for this
+  python -m pip install -r ../requirements.txt --target Python/Lib/site-packages
+}
+
+if (-Not (Test-Path "silero-vad")) {
+  git clone "https://github.com/snakers4/silero-vad"
+}
+
+mkdir $install_dir > $null
+mkdir $install_dir/Models
+cp ../*.py $install_dir/
+cp ../*.bat $install_dir/
+cp ../*.txt $install_dir/
+cp -Recurse Python $install_dir/Python
+cp "silero-vad/files/silero_vad.onnx" $install_dir/Models/
+cp "silero-vad/LICENSE" $install_dir/Models/silero_vad.onnx.LICENSE
+
+if (-Not $skip_zip) {
+  Compress-Archive -Path "$install_dir" -DestinationPath "$install_dir.zip" -Force
+}
+
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..79d8212
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,6 @@
+ctranslate2
+numpy
+pyaudio
+pydub
+onnxruntime
+
diff --git a/vad.py b/vad.py
new file mode 100644
index 0000000..5bc4331
--- /dev/null
+++ b/vad.py
@@ -0,0 +1,314 @@
+# MIT License
+# 
+# Copyright (c) 2023 Guillaume Klein
+# 
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+# 
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import bisect
+import functools
+import os
+import warnings
+
+from typing import List, NamedTuple, Optional
+
+import numpy as np
+
+
+# The code below is adapted from https://github.com/snakers4/silero-vad.
+class VadOptions(NamedTuple):
+    """VAD options.
+
+    Attributes:
+      threshold: 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.
+      min_speech_duration_ms: Final speech chunks shorter min_speech_duration_ms are thrown out.
+      max_speech_duration_s: Maximum duration of speech chunks in seconds. Chunks longer
+        than max_speech_duration_s will be split at the timestamp of the last silence that
+        lasts more than 100ms (if any), to prevent aggressive cutting. Otherwise, they will be
+        split aggressively just before max_speech_duration_s.
+      min_silence_duration_ms: In the end of each speech chunk wait for min_silence_duration_ms
+        before separating it
+      window_size_samples: Audio chunks of window_size_samples size are fed to the silero VAD model.
+        WARNING! Silero VAD models were trained using 512, 1024, 1536 samples for 16000 sample rate.
+        Values other than these may affect model performance!!
+      speech_pad_ms: Final speech chunks are padded by speech_pad_ms each side
+    """
+
+    threshold: float = 0.5
+    min_speech_duration_ms: int = 250
+    max_speech_duration_s: float = float("inf")
+    min_silence_duration_ms: int = 2000
+    window_size_samples: int = 1024
+    speech_pad_ms: int = 400
+
+
+def get_speech_timestamps(
+    audio: np.ndarray,
+    vad_options: Optional[VadOptions] = None,
+    **kwargs,
+) -> List[dict]:
+    """This method is used for splitting long audios into speech chunks using silero VAD.
+
+    Args:
+      audio: One dimensional float array.
+      vad_options: Options for VAD processing.
+      kwargs: VAD options passed as keyword arguments for backward compatibility.
+
+    Returns:
+      List of dicts containing begin and end samples of each speech chunk.
+    """
+    if vad_options is None:
+        vad_options = VadOptions(**kwargs)
+
+    threshold = vad_options.threshold
+    min_speech_duration_ms = vad_options.min_speech_duration_ms
+    max_speech_duration_s = vad_options.max_speech_duration_s
+    min_silence_duration_ms = vad_options.min_silence_duration_ms
+    window_size_samples = vad_options.window_size_samples
+    speech_pad_ms = vad_options.speech_pad_ms
+
+    if window_size_samples not in [512, 1024, 1536]:
+        warnings.warn(
+            "Unusual window_size_samples! Supported window_size_samples:\n"
+            " - [512, 1024, 1536] for 16000 sampling_rate"
+        )
+
+    sampling_rate = 16000
+    min_speech_samples = sampling_rate * min_speech_duration_ms / 1000
+    speech_pad_samples = sampling_rate * speech_pad_ms / 1000
+    max_speech_samples = (
+        sampling_rate * max_speech_duration_s
+        - window_size_samples
+        - 2 * speech_pad_samples
+    )
+    min_silence_samples = sampling_rate * min_silence_duration_ms / 1000
+    min_silence_samples_at_max_speech = sampling_rate * 98 / 1000
+
+    audio_length_samples = len(audio)
+
+    model = get_vad_model()
+    state = model.get_initial_state(batch_size=1)
+
+    speech_probs = []
+    for current_start_sample in range(0, audio_length_samples, window_size_samples):
+        chunk = audio[current_start_sample : current_start_sample + window_size_samples]
+        if len(chunk) < window_size_samples:
+            chunk = np.pad(chunk, (0, int(window_size_samples - len(chunk))))
+        speech_prob, state = model(chunk, state, sampling_rate)
+        speech_probs.append(speech_prob)
+
+    triggered = False
+    speeches = []
+    current_speech = {}
+    neg_threshold = threshold - 0.15
+
+    # to save potential segment end (and tolerate some silence)
+    temp_end = 0
+    # to save potential segment limits in case of maximum segment size reached
+    prev_end = next_start = 0
+
+    for i, speech_prob in enumerate(speech_probs):
+        if (speech_prob >= threshold) and temp_end:
+            temp_end = 0
+            if next_start < prev_end:
+                next_start = window_size_samples * i
+
+        if (speech_prob >= threshold) and not triggered:
+            triggered = True
+            current_speech["start"] = window_size_samples * i
+            continue
+
+        if (
+            triggered
+            and (window_size_samples * i) - current_speech["start"] > max_speech_samples
+        ):
+            if prev_end:
+                current_speech["end"] = prev_end
+                speeches.append(current_speech)
+                current_speech = {}
+                # previously reached silence (< neg_thres) and is still not speech (< thres)
+                if next_start < prev_end:
+                    triggered = False
+                else:
+                    current_speech["start"] = next_start
+                prev_end = next_start = temp_end = 0
+            else:
+                current_speech["end"] = window_size_samples * i
+                speeches.append(current_speech)
+                current_speech = {}
+                prev_end = next_start = temp_end = 0
+                triggered = False
+                continue
+
+        if (speech_prob < neg_threshold) and triggered:
+            if not temp_end:
+                temp_end = window_size_samples * i
+            # condition to avoid cutting in very short silence
+            if (window_size_samples * i) - temp_end > min_silence_samples_at_max_speech:
+                prev_end = temp_end
+            if (window_size_samples * i) - temp_end < min_silence_samples:
+                continue
+            else:
+                current_speech["end"] = temp_end
+                if (
+                    current_speech["end"] - current_speech["start"]
+                ) > min_speech_samples:
+                    speeches.append(current_speech)
+                current_speech = {}
+                prev_end = next_start = temp_end = 0
+                triggered = False
+                continue
+
+    if (
+        current_speech
+        and (audio_length_samples - current_speech["start"]) > min_speech_samples
+    ):
+        current_speech["end"] = audio_length_samples
+        speeches.append(current_speech)
+
+    for i, speech in enumerate(speeches):
+        if i == 0:
+            speech["start"] = int(max(0, speech["start"] - speech_pad_samples))
+        if i != len(speeches) - 1:
+            silence_duration = speeches[i + 1]["start"] - speech["end"]
+            if silence_duration < 2 * speech_pad_samples:
+                speech["end"] += int(silence_duration // 2)
+                speeches[i + 1]["start"] = int(
+                    max(0, speeches[i + 1]["start"] - silence_duration // 2)
+                )
+            else:
+                speech["end"] = int(
+                    min(audio_length_samples, speech["end"] + speech_pad_samples)
+                )
+                speeches[i + 1]["start"] = int(
+                    max(0, speeches[i + 1]["start"] - speech_pad_samples)
+                )
+        else:
+            speech["end"] = int(
+                min(audio_length_samples, speech["end"] + speech_pad_samples)
+            )
+
+    return speeches
+
+
+def collect_chunks(audio: np.ndarray, chunks: List[dict]) -> np.ndarray:
+    """Collects and concatenates audio chunks."""
+    if not chunks:
+        return np.array([], dtype=np.float32)
+
+    return np.concatenate([audio[chunk["start"] : chunk["end"]] for chunk in chunks])
+
+
+class SpeechTimestampsMap:
+    """Helper class to restore original speech timestamps."""
+
+    def __init__(self, chunks: List[dict], sampling_rate: int, time_precision: int = 2):
+        self.sampling_rate = sampling_rate
+        self.time_precision = time_precision
+        self.chunk_end_sample = []
+        self.total_silence_before = []
+
+        previous_end = 0
+        silent_samples = 0
+
+        for chunk in chunks:
+            silent_samples += chunk["start"] - previous_end
+            previous_end = chunk["end"]
+
+            self.chunk_end_sample.append(chunk["end"] - silent_samples)
+            self.total_silence_before.append(silent_samples / sampling_rate)
+
+    def get_original_time(
+        self,
+        time: float,
+        chunk_index: Optional[int] = None,
+    ) -> float:
+        if chunk_index is None:
+            chunk_index = self.get_chunk_index(time)
+
+        total_silence_before = self.total_silence_before[chunk_index]
+        return round(total_silence_before + time, self.time_precision)
+
+    def get_chunk_index(self, time: float) -> int:
+        sample = int(time * self.sampling_rate)
+        return min(
+            bisect.bisect(self.chunk_end_sample, sample),
+            len(self.chunk_end_sample) - 1,
+        )
+
+
+@functools.lru_cache
+def get_vad_model():
+    """Returns the VAD model instance."""
+    abspath = os.path.abspath(__file__)
+    my_dir = os.path.dirname(abspath)
+
+    path = os.path.join(my_dir, "Models/silero_vad.onnx")
+    return SileroVADModel(path)
+
+
+class SileroVADModel:
+    def __init__(self, path):
+        try:
+            import onnxruntime
+        except ImportError as e:
+            raise RuntimeError(
+                "Applying the VAD filter requires the onnxruntime package"
+            ) from e
+
+        opts = onnxruntime.SessionOptions()
+        opts.inter_op_num_threads = 1
+        opts.intra_op_num_threads = 1
+        opts.log_severity_level = 4
+
+        self.session = onnxruntime.InferenceSession(
+            path,
+            providers=["CPUExecutionProvider"],
+            sess_options=opts,
+        )
+
+    def get_initial_state(self, batch_size: int):
+        h = np.zeros((2, batch_size, 64), dtype=np.float32)
+        c = np.zeros((2, batch_size, 64), dtype=np.float32)
+        return h, c
+
+    def __call__(self, x, state, sr: int):
+        if len(x.shape) == 1:
+            x = np.expand_dims(x, 0)
+        if len(x.shape) > 2:
+            raise ValueError(
+                f"Too many dimensions for input audio chunk {len(x.shape)}"
+            )
+        if sr / x.shape[1] > 31.25:
+            raise ValueError("Input audio chunk is too short")
+
+        h, c = state
+
+        ort_inputs = {
+            "input": x,
+            "h": h,
+            "c": c,
+            "sr": np.array(sr, dtype="int64"),
+        }
+
+        out, h, c = self.session.run(None, ort_inputs)
+        state = (h, c)
+
+        return out, state