投稿者: admin_wp

import vosk
import pyaudio
import json
import numpy as np
import sounddevice as sd
import queue
import threading
import time

class VoskSpeechRecognizer:
    def __init__(self, model_path='./vosk-model-ja-0.22'):
        # モデルの初期化
        vosk.SetLogLevel(-1)
        self.model = vosk.Model(model_path)
        self.recognizer = vosk.KaldiRecognizer(self.model, 16000)
        
        # キュー設定
        self.audio_queue = queue.Queue()
        self.stop_event = threading.Event()
        
        # マイク設定
        self.sample_rate = 16000
        self.channels = 1
        
        # スレッド準備
        self.recording_thread = threading.Thread(target=self._record_audio)
        self.recognition_thread = threading.Thread(target=self._recognize_audio)
        
    def _record_audio(self):
        """
        連続的な音声録音スレッド
        """
        with sd.InputStream(
            samplerate=self.sample_rate, 
            channels=self.channels,
            dtype='int16',
            # device=2,
            callback=self._audio_callback
        ):
            while not self.stop_event.is_set():
                sd.sleep(100)
    
    def _audio_callback(self, indata, frames, time, status):
        """
        音声入力のコールバック関数
        """
        if status:
            print(status)
        self.audio_queue.put(indata.copy())
    
    def _recognize_audio(self):
        """
        連続的な音声認識スレッド
        """
        while not self.stop_event.is_set():
            try:
                audio_chunk = self.audio_queue.get(timeout=0.5)
                if self.recognizer.AcceptWaveform(audio_chunk.tobytes()):
                    result = json.loads(self.recognizer.Result())
                    text = result.get('text', '').strip()
                    if text:
                        print(f"{text}")
            except queue.Empty:
                continue
    
    def start_recognition(self):
        """
        音声認識の開始
        """
        self.stop_event.clear()
        self.recording_thread.start()
        self.recognition_thread.start()
    
    def stop_recognition(self):
        """
        音声認識の停止
        """
        self.stop_event.set()
        self.recording_thread.join()
        self.recognition_thread.join()

def main():
    recognizer = VoskSpeechRecognizer()
    
    try:
        print("音声認識を開始します。Ctrl+Cで終了できます。")
        recognizer.start_recognition()
        
        # 無限ループを防ぐ
        while True:
            time.sleep(1)
    
    except KeyboardInterrupt:
        print("\n音声認識を終了します...")
    finally:
        recognizer.stop_recognition()

if __name__ == "__main__":
    main()

$ python
Python 3.11.2 (main, Nov 30 2024, 21:22:50) [GCC 12.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import sounddevice as sd
>>> print(sd.query_devices())  # すべてのデバイス一覧を表示
   0 SF-558: USB Audio (hw:0,0), ALSA (1 in, 0 out)
   1 UACDemoV1.0: USB Audio (hw:2,0), ALSA (0 in, 2 out)
   2 sysdefault, ALSA (128 in, 0 out)
   3 spdif, ALSA (1 in, 0 out)
   4 lavrate, ALSA (128 in, 0 out)
   5 samplerate, ALSA (128 in, 0 out)
   6 speexrate, ALSA (128 in, 0 out)
   7 pulse, ALSA (32 in, 32 out)
   8 speex, ALSA (1 in, 0 out)
   9 upmix, ALSA (8 in, 0 out)
  10 vdownmix, ALSA (6 in, 0 out)
* 11 default, ALSA (32 in, 32 out)