https://github.com/k2-fsa/sherpa-onnx/blob/master/python-api-examples/simulate-streaming-sense-voice-microphone.py 的修改版本

simulate-streaming-sense-voice-microphone.py

#!/usr/bin/env python3
#
# Copyright (c)  2025  Xiaomi Corporation

"""
This file demonstrates how to use sherpa-onnx Python APIs
with VAD and non-streaming SenseVoice for real-time speech recognition
from a microphone.

Requirements:

wget https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2025-09-09.tar.bz2
wget https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/silero_vad.onnx

Usage:

./python-api-examples/simulate-streaming-sense-voice-microphone.py  \
  --silero-vad-model=./silero_vad.onnx \
  --sense-voice=./sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17/model.onnx \
  --tokens=./sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17/tokens.txt
"""
import argparse
import queue
import sys
import threading
import time
from pathlib import Path
import numpy as np
import os
import tkinter as tk

script_path = os.path.realpath(__file__)
script_dir = os.path.dirname(script_path)

try:
    import sounddevice as sd
    import sherpa_onnx
except ImportError:
    print("正在安装需要的python包:\n", file=sys.stderr)
    print("尝试执行：  pip install sounddevice sherpa_onnx==1.12.19\n", file=sys.stderr)
    ret = os.system(f"{sys.executable} -m pip install sounddevice sherpa_onnx==1.12.19")
    if ret == 0:
        import sounddevice as sd
        import sherpa_onnx
    else:
        print("安装python包失败!!", file=sys.stderr)
        sys.exit(-1)

vad_model_url = 'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/silero_vad.onnx'
vad_model_path = os.path.join(script_dir, "silero_vad.onnx")
onnx_model_url = 'https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2025-09-09.tar.bz2'
onnx_model_path = os.path.join(script_dir, "sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2025-09-09", "model.int8.onnx")
tokens_txt_path = os.path.join(script_dir, "sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2025-09-09", "tokens.txt")

killed = False
recording_thread = None
sample_rate = 16000  # Please don't change it

# buffer saves audio samples to be played
samples_queue = queue.Queue()


def get_args():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    
    parser.add_argument(
        "--device",
        type=int,
        default=-1,
        help="输入设备的编号",
    )

    parser.add_argument(
        "--silero-vad-model",
        type=str,
        help="silero_vad.onnx 文件路径",
        default=vad_model_path
    )

    parser.add_argument(
        "--tokens",
        type=str,
        help="模型的tokens.txt文件路径",
        default=tokens_txt_path
    )

    parser.add_argument(
        "--sense-voice",
        default=onnx_model_path,
        type=str,
        help="SenseVoice模型的model.onnx文件",
    )

    parser.add_argument(
        "--num-threads",
        type=int,
        default=2,
        help="用于推理的线程数，默认为2",
    )

    parser.add_argument(
        "--hr-lexicon",
        type=str,
        default="",
        help="If not empty, it is the lexicon.txt for homophone replacer",
    )

    parser.add_argument(
        "--hr-rule-fsts",
        type=str,
        default="",
        help="If not empty, it is the replace.fst for homophone replacer",
    )

    return parser.parse_args()


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 try_download_model(vad_model_path, onnx_model_path):
    if not os.path.exists(vad_model_path):
        print(f"请下载文件到：{vad_model_path}", file=sys.stderr)
        print(f"下载链接：{vad_model_url}", file=sys.stderr)
    if not os.path.exists(onnx_model_path):
        print(f"请下载并解压文件夹到：{onnx_model_path}", file=sys.stderr)
        print(f"下载链接：{onnx_model_url}", file=sys.stderr)

def create_recognizer(args):
    assert_file_exists(args.sense_voice)
    recognizer = sherpa_onnx.OfflineRecognizer.from_sense_voice(
        model=args.sense_voice,
        tokens=args.tokens,
        num_threads=args.num_threads,
        use_itn=False,
        debug=False,
        hr_rule_fsts=args.hr_rule_fsts,
        hr_lexicon=args.hr_lexicon,
    )

    return recognizer


def start_recording():
    # You can use any value you like for samples_per_read
    samples_per_read = int(0.1 * sample_rate)  # 0.1 second = 100 ms

    with sd.InputStream(channels=1, dtype="float32", samplerate=sample_rate) as s:
        while not killed:
            samples, _ = s.read(samples_per_read)  # a blocking read
            samples = samples.reshape(-1)
            samples = np.copy(samples)
            samples_queue.put(samples)

class MyPrinter:
    def __init__(self):
        self.prev_result = ""

    def do_print(self, result):
        if result and self.prev_result != result:
            self.prev_result = result
            print(result, end='\n', flush=True)

    def on_endpoint(self):
        print("\n", end="", flush=True)

def select_input_device(devices):
    """弹出tkinter窗口让用户选择输入设备"""
    # 过滤有输入通道的设备
    input_devices = [(i, d) for i, d in enumerate(devices) if d['max_input_channels'] > 0 and d['hostapi'] == 0]

    if not input_devices:
        return None

    selected_device = [None]

    def on_ok():
        selected_device[0] = device_var.get()
        root.destroy()

    root = tk.Tk()
    root.title("选择输入设备")

    device_var = tk.IntVar(value=input_devices[0][0])

    for idx, device in input_devices:
        rb = tk.Radiobutton(
            root,
            text=f"{idx}: {device['name']}",
            variable=device_var,
            value=idx
        )
        rb.pack(anchor='w', padx=20, pady=5)

    tk.Button(root, text="确定", command=on_ok).pack(pady=10)

    root.mainloop()

    return selected_device[0]

def main():
    devices = sd.query_devices()
    if len(devices) == 0:
        print("没有任何输入设备！", file=sys.stderr)
        sys.exit(0)

    args = get_args()

    print(devices, file=sys.stderr)

    # 如果命令行没有指定设备，弹出选择框
    if args.device < 0:
        selected = select_input_device(devices)
        if selected is not None:
            sd.default.device[0] = selected
    else:
        sd.default.device[0] = args.device

    # 如果你想要选择其他的输入设备，请解除下面这行的注释，并将 xxx 改为设备的序号
    # sd.default.device[0] = xxx
    # 注意要选设备结尾的in大于零的，比如 (2 in, 0 out) 表示两声道输入，没有输出声道，说明是录音设备。
    # 如果想要识别系统声音，尝试启用“立体声混音”，并设置识别设备为它。

    default_input_device_idx = sd.default.device[0]
    print(f'使用输入设备: {devices[default_input_device_idx]["name"]}', file=sys.stderr)

    try_download_model(args.silero_vad_model, args.silero_vad_model)
    assert_file_exists(args.tokens)
    assert_file_exists(args.silero_vad_model)

    assert args.num_threads > 0, args.num_threads

    print("正在启动识别器，请稍后", file=sys.stderr)
    recognizer = create_recognizer(args)

    config = sherpa_onnx.VadModelConfig()
    config.silero_vad.model = args.silero_vad_model
    config.silero_vad.threshold = 0.5
    config.silero_vad.min_silence_duration = 0.1  # seconds
    config.silero_vad.min_speech_duration = 0.25  # seconds
    # If the current segment is larger than this value, then it increases
    # the threshold to 0.9 internally. After detecting this segment,
    # it resets the threshold to its original value.
    config.silero_vad.max_speech_duration = 8  # seconds
    config.sample_rate = sample_rate

    window_size = config.silero_vad.window_size

    vad = sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=100)

    print("识别已启动，请说话", file=sys.stderr)

    buffer = []

    global recording_thread
    recording_thread = threading.Thread(target=start_recording)
    recording_thread.start()

    # display = sherpa_onnx.Display()
    printer = MyPrinter()

    started = False
    started_time = None

    offset = 0
    while not killed:
        samples = samples_queue.get()  # a blocking read

        buffer = np.concatenate([buffer, samples])
        while offset + window_size < len(buffer):
            vad.accept_waveform(buffer[offset : offset + window_size])
            if not started and vad.is_speech_detected():
                started = True
                started_time = time.time()
            offset += window_size

        if not started:
            if len(buffer) > 10 * window_size:
                offset -= len(buffer) - 10 * window_size
                buffer = buffer[-10 * window_size :]

        if started and time.time() - started_time > 0.2:
            stream = recognizer.create_stream()
            stream.accept_waveform(sample_rate, buffer)
            recognizer.decode_stream(stream)
            text = stream.result.text.strip()
            if text:
                printer.do_print(text)
                # display.update_text(text)
                # display.display()

            started_time = time.time()

        while not vad.empty():
            # In general, this while loop is executed only once
            stream = recognizer.create_stream()
            stream.accept_waveform(sample_rate, vad.front.samples)

            vad.pop()
            recognizer.decode_stream(stream)

            text = stream.result.text.strip()

            # display.update_text(text)
            printer.do_print(text)

            buffer = []
            offset = 0
            started = False
            started_time = None

            # display.finalize_current_sentence()
            # display.display()
            printer.on_endpoint()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        killed = True
        if recording_thread:
            recording_thread.join()
        print("\n检测到 Ctrl + C. 正在退出", file=sys.stderr)