はじめに
この記事はWhisperを使ったリアルタイム音声認識を試みた(Macbook)｜Pythonの続き
前回はWhisperの導入をメインに書いた
ソースコードはClaudeに書かせて、少し手直ししたものだった
今回はその中身について解読していく
main()関数
def main():
transcriber = WhisperTranscriber(
model_size='base',
language='japanese'
)
transcriber.run()WhisperTranscriberクラスのインスタンスtranscriberを生成し、run()メソッドを実行している
インスタンスを生成する際に、引数にmodel_size、languageを指定する
WhisperTranscriberクラス
コンストラクタ(__init__)メソッド
def __init__(self, model_size='base', language='japanese'):
"""
Whisper音声認識クラス
"""
self.model = whisper.load_model(model_size, device="mps")
for param in self.model.parameters():
print(f"モデルのデバイス: {param.device}")
break # 最初のパラメータだけ確認すれば十分
# 音声入力設定
self.pyaudio = pyaudio.PyAudio()
self.sample_rate = 16000
self.chunk_duration = 3
self.chunks_per_inference = int(self.sample_rate * self.chunk_duration)
# スレッディング用キュー
self.audio_queue = queue.Queue()
self.stop_event = threading.Event()
self.language = languageインスタンス変数の定義が主な内容
whisperのモデルの読み込みもここで行われる
さらにforループを１周だけ実行し、読み込んだモデルが使用するデバイスを表示する
表示されるデバイスとは”cpu”のこと（環境次第では”cuda”にも）
run()メソッド
def run(self):
"""
音声認識プロセス実行
"""
self.start_recording()
transcribe_thread = threading.Thread(target=self.transcribe_audio)
transcribe_thread.start()
try:
transcribe_thread.join()
except KeyboardInterrupt:
print("\n音声認識を終了")
self.stop_event.set()
self.stream.stop_stream()
self.stream.close()
self.pyaudio.terminate()録音プロセスと認識プロセスをスレッドを用いて並列に実行している
ctrl + Cで終了するエラー処理も実装されている
start_recording()メソッド
def start_recording(self):
"""
音声録音開始
"""
self.stream = self.pyaudio.open(
format=pyaudio.paFloat32,
channels=1,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunks_per_inference,
stream_callback=self.audio_callback
)
print("音声録音を開始")音声録音を行う部分
録音はpyaudioを用いている
音声入力が検出されるたびに、audio_callback()メソッドが呼ばれる
今回は実装されていないが、input_device_indexの指定で入力デバイスの指定も可能そう
audio_callback()メソッド
def audio_callback(self, in_data, frame_count, time_info, status):
"""
音声データ処理用コールバック
"""
audio_chunk = np.frombuffer(in_data, dtype=np.float32)
self.audio_queue.put(audio_chunk)
return (None, pyaudio.paContinue)音声データをキューに格納する
pyaudioから受け取るin_dataはバイナリ形式で、それをfloat32のnumpy形式に変換している
return部分では、録音を続ける指示を出している
transcribe_audio()メソッド
def transcribe_audio(self):
"""
リアルタイム音声文字起こし
"""
if self.model is None:
print("モデルの初期化に失敗しました")
return
audio_data = []
while not self.stop_event.is_set():
try:
chunk = self.audio_queue.get(timeout=1)
audio_data.extend(chunk)
if len(audio_data) >= self.chunks_per_inference:
# テンソル変換
audio_np = np.array(audio_data[:self.chunks_per_inference])
# 推論を実行
result = self.model.transcribe(
audio_np,
language=self.language,
fp16=False
)
print(f"文字起こし結果: {result['text']}")
audio_data = audio_data[self.chunks_per_inference:]
except queue.Empty:
continue
except Exception as e:
print(f"推論中にエラー: {e}")音声認識を行い、結果を標準出力する部分
1秒ごとにキューから音声データを取り出し、audio_dataに格納する
audio_dataが指定されたフレーム数以上になったら推論を開始する
推論するためにリストからnumpy配列に変換し（テンソル変換）、model.transcribe()メソッドでテキスト化する
テキストを出力後、キューが空ならばループを続行する
おわりに
これだけのものを1 分で書き上げるClaudeすごい
全体像
import whisper
import pyaudio
import numpy as np
import threading
import queue
import time
import torch
class WhisperTranscriber:
def __init__(self, model_size='base', language='japanese'):
"""
Whisper音声認識クラス
"""
self.model = whisper.load_model(model_size, device="mps")
for param in self.model.parameters():
print(f"モデルのデバイス: {param.device}")
break # 最初のパラメータだけ確認すれば十分
# 音声入力設定
self.pyaudio = pyaudio.PyAudio()
self.sample_rate = 16000
self.chunk_duration = 3
self.chunks_per_inference = int(self.sample_rate * self.chunk_duration)
# スレッディング用キュー
self.audio_queue = queue.Queue()
self.stop_event = threading.Event()
self.language = language
def transcribe_audio(self):
"""
リアルタイム音声文字起こし
"""
if self.model is None:
print("モデルの初期化に失敗しました")
return
audio_data = []
while not self.stop_event.is_set():
try:
chunk = self.audio_queue.get(timeout=1)
audio_data.extend(chunk)
if len(audio_data) >= self.chunks_per_inference:
# テンソル変換
audio_np = np.array(audio_data[:self.chunks_per_inference])
# 推論を実行
result = self.model.transcribe(
audio_np,
language=self.language,
fp16=False
)
print(f"文字起こし結果: {result['text']}")
audio_data = audio_data[self.chunks_per_inference:]
except queue.Empty:
continue
except Exception as e:
print(f"推論中にエラー: {e}")
def audio_callback(self, in_data, frame_count, time_info, status):
"""
音声データ処理用コールバック
"""
audio_chunk = np.frombuffer(in_data, dtype=np.float32)
self.audio_queue.put(audio_chunk)
return (None, pyaudio.paContinue)
def start_recording(self):
"""
音声録音開始
"""
self.stream = self.pyaudio.open(
format=pyaudio.paFloat32,
channels=1,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunks_per_inference,
stream_callback=self.audio_callback
)
print("音声録音を開始")
def run(self):
"""
音声認識プロセス実行
"""
self.start_recording()
transcribe_thread = threading.Thread(target=self.transcribe_audio)
transcribe_thread.start()
try:
transcribe_thread.join()
except KeyboardInterrupt:
print("\n音声認識を終了")
self.stop_event.set()
self.stream.stop_stream()
self.stream.close()
self.pyaudio.terminate()
def main():
transcriber = WhisperTranscriber(
model_size='base',
language='japanese'
)
transcriber.run()
if __name__ == "__main__":
main()
コメントを残す