Whisper_mic に faster_whisper を組込む 【Python】

スポンサーリンク
スポンサーリンク

はじめに

前回は、Whisper_mic の使用方法について説明しました。

今回は、Whisper の高速版である faster_wisper を、Whisper_mic に組込む方法について説明していきます。

前提条件

前提条件は以下の通りです。

  • Python == 3.10.6
  • Windows11

faster_whisper のインストール

faster_whisper は以下のコマンドでインストールできます。

python -m pip install faster-whisper
python -m pip install SpeechRecognition

今回は faster_whisper == 0.8.0, SpeechRecognition == 3.10.0 を使用します。

Whisper_mic の準備

前回と同じように準備します。

git clone https://github.com/mallorbc/whisper_mic.git
cd whisper_mic/whisper_mic

プログラムの変更

faster_whisper に対応できるようにプログラムを変更していきます。

cli.py main関数

def main(model: str, english: bool, verbose: bool, energy:  int, pause: float, dynamic_energy: bool, save_file: bool, device: str, loop: bool, dictate: bool,mic_index:Optional[int],list_devices:bool) -> None:
    if list_devices:
        print("Possible devices: ",sr.Microphone.list_microphone_names())
        return
    mic = WhisperMic(model=model, english=english, verbose=verbose, energy=energy, pause=pause, dynamic_energy=dynamic_energy, save_file=save_file, device=device,mic_index=mic_index)
    if not loop:
        result = mic.listen()
        for r in result:
            print(r.text)
        # print("You said: " + result)
    else:
        mic.listen_loop(dictate=dictate)

if __name__ == "__main__":
    main()

faster_whisper は結果をジェネレータで返すので、 for 文で取得しています。

whisper_mic.py

import torch
from faster_whisper import WhisperModel
# import whisper
import queue
import speech_recognition as sr
import threading
# import io
import numpy as np
# from pydub import AudioSegment
# import os
import tempfile
import time
import platform
import pynput.keyboard

# from whisper_mic.utils import get_logger
from utils import get_logger


class WhisperMic:
    def __init__(self,model="small",device=("cuda" if torch.cuda.is_available() else "cpu"),english=False,verbose=False,energy=300,pause=0.8,dynamic_energy=False,save_file=False, model_root="~/.cache/whisper",mic_index=None):
        self.logger = get_logger("whisper_mic", "info")
        self.energy = energy
        self.pause = pause
        self.dynamic_energy = dynamic_energy
        self.save_file = save_file
        self.verbose = verbose
        self.english = english
        self.keyboard = pynput.keyboard.Controller()

        self.platform = platform.system()

        if self.platform == "darwin":
            if device == "mps":
                self.logger.warning("Using MPS for Mac, this does not work but may in the future")
                device = "mps"
                device = torch.device(device)

        if (model != "large" and model != "large-v2") and self.english:
            model = model + ".en"
        
        # self.audio_model = whisper.load_model(model).to(device)
        model_size = "medium"
        self.audio_model = WhisperModel(model_size, device="cuda", compute_type="int8_float16")
        self.temp_dir = tempfile.mkdtemp() if save_file else None

        self.audio_queue = queue.Queue()
        self.result_queue: "queue.Queue[str]" = queue.Queue()

        self.break_threads = False
        self.mic_active = False

        self.banned_results = [""," ","\n",None]

        self.setup_mic(mic_index)


    def setup_mic(self, mic_index):
        if mic_index is None:
            self.logger.info("No mic index provided, using default")
        self.source = sr.Microphone(sample_rate=16000, device_index=mic_index)

        self.recorder = sr.Recognizer()
        self.recorder.energy_threshold = self.energy
        self.recorder.pause_threshold = self.pause
        self.recorder.dynamic_energy_threshold = self.dynamic_energy

        with self.source:
            self.recorder.adjust_for_ambient_noise(self.source)

        self.recorder.listen_in_background(self.source, self.record_callback, phrase_time_limit=2)
        self.logger.info("Mic setup complete, you can now talk")


    def preprocess(self, data):
        return torch.from_numpy(np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0)

    def get_all_audio(self, min_time: float = -1.):
        audio = bytes()
        got_audio = False
        time_start = time.time()
        while not got_audio or time.time() - time_start < min_time:
            while not self.audio_queue.empty():
                audio += self.audio_queue.get()
                got_audio = True

        data = sr.AudioData(audio,16000,2)
        data = data.get_raw_data()
        return data


    def record_callback(self,_, audio: sr.AudioData) -> None:
        data = audio.get_raw_data()
        self.audio_queue.put_nowait(data)


    def transcribe_forever(self) -> None:
        while True:
            # if self.break_threads:
            #     break
            self.transcribe()


    def transcribe(self,data=None, realtime: bool = False) -> None:
        if data is None:
            audio_data = self.get_all_audio()
        else:
            audio_data = data
        audio_data = self.preprocess(audio_data)
        if self.english:
            result = self.audio_model.transcribe(audio_data,language='english')
        else:
            if isinstance(audio_data, torch.Tensor):
                audio_data = audio_data.detach().numpy()
            result = self.audio_model.transcribe(audio=audio_data, language="ja", without_timestamps=True)

        predicted_text = result[0]
        if predicted_text not in self.banned_results:
            self.result_queue.put_nowait(predicted_text)

        # predicted_text = result["text"]
        # if not self.verbose:
        #     if predicted_text not in self.banned_results:
        #         self.result_queue.put_nowait(predicted_text)
        # else:
        #     if predicted_text not in self.banned_results:
        #         self.result_queue.put_nowait(result)

        # if self.save_file:
        #     os.remove(audio_data)


    def listen_loop(self, dictate: bool = False) -> None:
        # threading.Thread(target=self.transcribe_forever).start()
        threading.Thread(target=self.transcribe_forever, daemon=True).start()
        while True:
            result = self.result_queue.get()
            if dictate:
                self.keyboard.type(result)
            else:
                for r in result:
                    print(r.text)
                    # print(type(r.text))
                    if r.text == "終わり" or r.text == "おわり":
                        print("プログラムを終了します")
                        import sys
                        sys.exit()
                        
                # print(result)

    def listen(self, timeout: int = 3):
        audio_data = self.get_all_audio(timeout)
        self.transcribe(data=audio_data)
        while True:
            if not self.result_queue.empty():
                return self.result_queue.get()

    def toggle_microphone(self) -> None:
        #TO DO: make this work
        self.mic_active = not self.mic_active
        if self.mic_active:
            print("Mic on")
        else:
            print("turning off mic")
            self.mic_thread.join()
            print("Mic off")

whisper_mic.py は全文を記載しました。

変更点のみ説明していきます。

2 – 3 行目

from faster_whisper import WhisperModel
# import whisper

whisper ではなく faster_whisper を import します。

42 – 44 行目

# self.audio_model = whisper.load_model(model).to(device)
model_size = "medium"
self.audio_model = WhisperModel(model_size, device="cuda", compute_type="int8_float16")

モデルは medium を使用します。
compute_type は最も高速な int8 を指定します。

97 – 101 行目

def transcribe_forever(self) -> None:
    while True:
        # if self.break_threads:
        #     break
        self.transcribe()

self.break_threads は、スレッドスタート時に daemon = True としたので、不要です。

104 – 130 行目

def transcribe(self,data=None, realtime: bool = False) -> None:
    if data is None:
        audio_data = self.get_all_audio()
    else:
        audio_data = data
    audio_data = self.preprocess(audio_data)
    if self.english:
        result = self.audio_model.transcribe(audio_data,language='english')
    else:
        if isinstance(audio_data, torch.Tensor):
            audio_data = audio_data.detach().numpy()
        result = self.audio_model.transcribe(audio=audio_data, language="ja", without_timestamps=True)

    predicted_text = result[0]
    if predicted_text not in self.banned_results:
        self.result_queue.put_nowait(predicted_text)

    # predicted_text = result["text"]
    # if not self.verbose:
    #     if predicted_text not in self.banned_results:
    #         self.result_queue.put_nowait(predicted_text)
    # else:
    #     if predicted_text not in self.banned_results:
    #         self.result_queue.put_nowait(result)

    # if self.save_file:
    #     os.remove(audio_data)

whisper の入力は torch.Tensor で、faster_whisper の入力は numpy.ndarray なので、.detach().numpy() 形式に入力を変換します。

また、faster_whisper は日本語を指定し、より高速になる without_timestamps を有効にします。

133 – 149 行目

def listen_loop(self, dictate: bool = False) -> None:
    # threading.Thread(target=self.transcribe_forever).start()
    threading.Thread(target=self.transcribe_forever, daemon=True).start()
    while True:
        result = self.result_queue.get()
        if dictate:
            self.keyboard.type(result)
        else:
            for r in result:
                print(r.text)
                # print(type(r.text))
                if r.text == "終わり" or r.text == "おわり":
                    print("プログラムを終了します")
                    import sys
                    sys.exit()
                        
            # print(result)

threading.Thread() に daemon=True 追加しました。
sys.exit() 時にプログラム全体が終了するようになります。

また、result はジェネレータなので、for 文で一つずつ取得します。
話者が “おわり” と発音したらプログラムを終了するようにしました。

faster_whisper の Whisper_mic を動かしてみる

まずは loop 指定せず動かします。

python cli.py

結果は以下の通りです。

[09/17/23 08:51:31] INFO     No mic index provided, using default                                      whisper_mic.py:60
[09/17/23 08:51:33] INFO     Mic setup complete, you can now talk                                      whisper_mic.py:72
おはようございます。

次に loop 指定して動かします。

python cli.py --loop

結果は以下のようになります。

[09/17/23 10:34:38] INFO     No mic index provided, using default                                      whisper_mic.py:60
[09/17/23 10:34:40] INFO     Mic setup complete, you can now talk                                      whisper_mic.py:72
おはようございます。
寝ています
終わり
プログラムを終了します

しっかり検出できていますね！

おわりに

今回は faster_whisper を Whisper_mic に組込む方法について説明しました。

高速で正確に検出できています。
Whisper_mic の音声処理のおかげでノイズ等も除去され、誤検出が減っています。

スポンサーリンク

コメント

タイトルとURLをコピーしました