Add VAD examples using ALSA for recording (#739)
This commit is contained in:
Fangjun Kuang
107
python-api-examples/vad-alsa.py
Executable file
107
python-api-examples/vad-alsa.py
Executable file
@@ -0,0 +1,107 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
This script works only on Linux. It uses ALSA for recording.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
import sherpa_onnx
|
||||
|
||||
|
||||
def get_args():
|
||||
parser = argparse.ArgumentParser(
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--silero-vad-model",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to silero_vad.onnx",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--device-name",
|
||||
type=str,
|
||||
required=True,
|
||||
help="""
|
||||
The device name specifies which microphone to use in case there are several
|
||||
on your system. You can use
|
||||
|
||||
arecord -l
|
||||
|
||||
to find all available microphones on your computer. For instance, if it outputs
|
||||
|
||||
**** List of CAPTURE Hardware Devices ****
|
||||
card 3: UACDemoV10 [UACDemoV1.0], device 0: USB Audio [USB Audio]
|
||||
Subdevices: 1/1
|
||||
Subdevice #0: subdevice #0
|
||||
|
||||
and if you want to select card 3 and the device 0 on that card, please use:
|
||||
|
||||
plughw:3,0
|
||||
|
||||
as the device_name.
|
||||
""",
|
||||
)
|
||||
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = get_args()
|
||||
if not Path(args.silero_vad_model).is_file():
|
||||
raise RuntimeError(
|
||||
f"{args.silero_vad_model} does not exist. Please download it from "
|
||||
"https://github.com/snakers4/silero-vad/blob/master/files/silero_vad.onnx"
|
||||
)
|
||||
|
||||
device_name = args.device_name
|
||||
print(f"device_name: {device_name}")
|
||||
alsa = sherpa_onnx.Alsa(device_name)
|
||||
|
||||
sample_rate = 16000
|
||||
samples_per_read = int(0.1 * sample_rate) # 0.1 second = 100 ms
|
||||
|
||||
config = sherpa_onnx.VadModelConfig()
|
||||
config.silero_vad.model = args.silero_vad_model
|
||||
config.sample_rate = sample_rate
|
||||
|
||||
vad = sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=30)
|
||||
|
||||
print("Started! Please speak. Press Ctrl C to exit")
|
||||
|
||||
printed = False
|
||||
k = 0
|
||||
try:
|
||||
while True:
|
||||
samples = alsa.read(samples_per_read) # a blocking read
|
||||
|
||||
vad.accept_waveform(samples)
|
||||
|
||||
if vad.is_speech_detected() and not printed:
|
||||
print("Detected speech")
|
||||
printed = True
|
||||
|
||||
if not vad.is_speech_detected():
|
||||
printed = False
|
||||
|
||||
while not vad.empty():
|
||||
samples = vad.front.samples
|
||||
duration = len(samples) / sample_rate
|
||||
filename = f"seg-{k}-{duration:.3f}-seconds.wav"
|
||||
k += 1
|
||||
sherpa_onnx.write_wave(filename, samples, sample_rate)
|
||||
print(f"Duration: {duration:.3f} seconds")
|
||||
print(f"Saved to {filename}")
|
||||
print("----------")
|
||||
|
||||
vad.pop()
|
||||
except KeyboardInterrupt:
|
||||
print("\nCaught Ctrl + C. Exit")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
125
python-api-examples/vad-microphone.py
Executable file
125
python-api-examples/vad-microphone.py
Executable file
@@ -0,0 +1,125 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import sounddevice as sd
|
||||
except ImportError:
|
||||
print("Please install sounddevice first. You can use")
|
||||
print()
|
||||
print(" pip install sounddevice")
|
||||
print()
|
||||
print("to install it")
|
||||
sys.exit(-1)
|
||||
|
||||
import sherpa_onnx
|
||||
|
||||
|
||||
def get_args():
|
||||
parser = argparse.ArgumentParser(
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--silero-vad-model",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to silero_vad.onnx",
|
||||
)
|
||||
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = get_args()
|
||||
if not Path(args.silero_vad_model).is_file():
|
||||
raise RuntimeError(
|
||||
f"{args.silero_vad_model} does not exist. Please download it from "
|
||||
"https://github.com/snakers4/silero-vad/blob/master/files/silero_vad.onnx"
|
||||
)
|
||||
|
||||
mic_sample_rate = 16000
|
||||
if "SHERPA_ONNX_MIC_SAMPLE_RATE" in os.environ:
|
||||
mic_sample_rate = int(os.environ.get("SHERPA_ONNX_MIC_SAMPLE_RATE"))
|
||||
print(f"Change microphone sample rate to {mic_sample_rate}")
|
||||
|
||||
sample_rate = 16000
|
||||
samples_per_read = int(0.1 * sample_rate) # 0.1 second = 100 ms
|
||||
|
||||
config = sherpa_onnx.VadModelConfig()
|
||||
config.silero_vad.model = args.silero_vad_model
|
||||
config.sample_rate = sample_rate
|
||||
|
||||
vad = sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=30)
|
||||
|
||||
# python3 -m sounddevice
|
||||
# can also be used to list all devices
|
||||
|
||||
devices = sd.query_devices()
|
||||
if len(devices) == 0:
|
||||
print("No microphone devices found")
|
||||
print(
|
||||
"If you are using Linux and you are sure there is a microphone "
|
||||
"on your system, please use "
|
||||
"./vad-alsa.py"
|
||||
)
|
||||
sys.exit(0)
|
||||
|
||||
print(devices)
|
||||
|
||||
if "SHERPA_ONNX_MIC_DEVICE" in os.environ:
|
||||
input_device_idx = int(os.environ.get("SHERPA_ONNX_MIC_DEVICE"))
|
||||
sd.default.device[0] = input_device_idx
|
||||
print(f'Use selected device: {devices[input_device_idx]["name"]}')
|
||||
else:
|
||||
input_device_idx = sd.default.device[0]
|
||||
print(f'Use default device: {devices[input_device_idx]["name"]}')
|
||||
|
||||
print("Started! Please speak. Press Ctrl C to exit")
|
||||
|
||||
printed = False
|
||||
k = 0
|
||||
try:
|
||||
with sd.InputStream(
|
||||
channels=1, dtype="float32", samplerate=mic_sample_rate
|
||||
) as s:
|
||||
while True:
|
||||
samples, _ = s.read(samples_per_read) # a blocking read
|
||||
samples = samples.reshape(-1)
|
||||
|
||||
if mic_sample_rate != sample_rate:
|
||||
import librosa
|
||||
|
||||
samples = librosa.resample(
|
||||
samples, orig_sr=mic_sample_rate, target_sr=sample_rate
|
||||
)
|
||||
|
||||
vad.accept_waveform(samples)
|
||||
|
||||
if vad.is_speech_detected() and not printed:
|
||||
print("Detected speech")
|
||||
printed = True
|
||||
|
||||
if not vad.is_speech_detected():
|
||||
printed = False
|
||||
|
||||
while not vad.empty():
|
||||
samples = vad.front.samples
|
||||
duration = len(samples) / sample_rate
|
||||
filename = f"seg-{k}-{duration:.3f}-seconds.wav"
|
||||
k += 1
|
||||
sherpa_onnx.write_wave(filename, samples, sample_rate)
|
||||
print(f"Duration: {duration:.3f} seconds")
|
||||
print(f"Saved to {filename}")
|
||||
print("----------")
|
||||
|
||||
vad.pop()
|
||||
except KeyboardInterrupt:
|
||||
print("\nCaught Ctrl + C. Exit")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
138
python-api-examples/vad-remove-non-speech-segments-alsa.py
Executable file
138
python-api-examples/vad-remove-non-speech-segments-alsa.py
Executable file
@@ -0,0 +1,138 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
This file shows how to remove non-speech segments
|
||||
and merge all speech segments into a large segment
|
||||
and save it to a file.
|
||||
|
||||
Different from ./vad-remove-non-speech-segments.py, this file supports only
|
||||
Linux.
|
||||
|
||||
Usage
|
||||
|
||||
python3 ./vad-remove-non-speech-segments-alsa.py \
|
||||
--silero-vad-model silero_vad.onnx
|
||||
|
||||
Please visit
|
||||
https://github.com/snakers4/silero-vad/blob/master/files/silero_vad.onnx
|
||||
to download silero_vad.onnx
|
||||
|
||||
For instance,
|
||||
|
||||
wget https://github.com/snakers4/silero-vad/raw/master/files/silero_vad.onnx
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import sherpa_onnx
|
||||
import soundfile as sf
|
||||
|
||||
|
||||
def assert_file_exists(filename: str):
|
||||
assert Path(filename).is_file(), (
|
||||
f"{filename} does not exist!\n"
|
||||
"Please refer to "
|
||||
"https://k2-fsa.github.io/sherpa/onnx/pretrained_models/index.html to download it"
|
||||
)
|
||||
|
||||
|
||||
def get_args():
|
||||
parser = argparse.ArgumentParser(
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--silero-vad-model",
|
||||
type=str,
|
||||
required=True,
|
||||
help="Path to silero_vad.onnx",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--device-name",
|
||||
type=str,
|
||||
required=True,
|
||||
help="""
|
||||
The device name specifies which microphone to use in case there are several
|
||||
on your system. You can use
|
||||
|
||||
arecord -l
|
||||
|
||||
to find all available microphones on your computer. For instance, if it outputs
|
||||
|
||||
**** List of CAPTURE Hardware Devices ****
|
||||
card 3: UACDemoV10 [UACDemoV1.0], device 0: USB Audio [USB Audio]
|
||||
Subdevices: 1/1
|
||||
Subdevice #0: subdevice #0
|
||||
|
||||
and if you want to select card 3 and the device 0 on that card, please use:
|
||||
|
||||
plughw:3,0
|
||||
|
||||
as the device_name.
|
||||
""",
|
||||
)
|
||||
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = get_args()
|
||||
assert_file_exists(args.silero_vad_model)
|
||||
|
||||
device_name = args.device_name
|
||||
print(f"device_name: {device_name}")
|
||||
alsa = sherpa_onnx.Alsa(device_name)
|
||||
|
||||
sample_rate = 16000
|
||||
samples_per_read = int(0.1 * sample_rate) # 0.1 second = 100 ms
|
||||
|
||||
config = sherpa_onnx.VadModelConfig()
|
||||
config.silero_vad.model = args.silero_vad_model
|
||||
config.sample_rate = sample_rate
|
||||
|
||||
window_size = config.silero_vad.window_size
|
||||
|
||||
buffer = []
|
||||
vad = sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=30)
|
||||
|
||||
all_samples = []
|
||||
|
||||
print("Started! Please speak. Press Ctrl C to exit")
|
||||
|
||||
try:
|
||||
while True:
|
||||
samples = alsa.read(samples_per_read) # a blocking read
|
||||
samples = np.array(samples)
|
||||
|
||||
buffer = np.concatenate([buffer, samples])
|
||||
|
||||
all_samples = np.concatenate([all_samples, samples])
|
||||
|
||||
while len(buffer) > window_size:
|
||||
vad.accept_waveform(buffer[:window_size])
|
||||
buffer = buffer[window_size:]
|
||||
except KeyboardInterrupt:
|
||||
print("\nCaught Ctrl + C. Saving & Exiting")
|
||||
|
||||
speech_samples = []
|
||||
while not vad.empty():
|
||||
speech_samples.extend(vad.front.samples)
|
||||
vad.pop()
|
||||
|
||||
speech_samples = np.array(speech_samples, dtype=np.float32)
|
||||
|
||||
filename_for_speech = time.strftime("%Y%m%d-%H%M%S-speech.wav")
|
||||
sf.write(filename_for_speech, speech_samples, samplerate=sample_rate)
|
||||
|
||||
filename_for_all = time.strftime("%Y%m%d-%H%M%S-all.wav")
|
||||
sf.write(filename_for_all, all_samples, samplerate=sample_rate)
|
||||
|
||||
print(f"Saved to {filename_for_speech} and {filename_for_all}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -66,6 +66,11 @@ def main():
|
||||
devices = sd.query_devices()
|
||||
if len(devices) == 0:
|
||||
print("No microphone devices found")
|
||||
print(
|
||||
"If you are using Linux and you are sure there is a microphone "
|
||||
"on your system, please use "
|
||||
"./vad-remove-non-speech-segments-alsa.py"
|
||||
)
|
||||
sys.exit(0)
|
||||
|
||||
print(devices)
|
||||
@@ -89,7 +94,7 @@ def main():
|
||||
|
||||
all_samples = []
|
||||
|
||||
print("Started! Please speak")
|
||||
print("Started! Please speak. Press Ctrl C to exit")
|
||||
|
||||
try:
|
||||
with sd.InputStream(channels=1, dtype="float32", samplerate=sample_rate) as s:
|
||||
|
||||
Reference in New Issue
Block a user