Faster-whisperマイクのコード解説

Faster-whisperマイクのコード解説

コードを読むのにChatGPTを使用
これによりコードの解説が得られることで理解しやすくなる

git clone https://github.com/personabb/colab_AI_sample.git

でリポジトリクローン

cd colab_AI_sample 
cd colab_fasterwhisper_sample

そして
main.pyの中の

    while True:
        audio_data = recorder.speech2audio()
        text = fasterWhispermodel.audio2text(audio_data)
        print(text)

これで
recorder.speech2audio()メソッドで音声を録音して、録音データをaudio_dataに格納し、fasterWhispermodel.audio2text()メソッドにて、音声ファイルを文字起こしして、textとして表示

つまり
printではなく
文字を送信するメソッドを作成すれば処理は完成するはず

module_recoder.py
発話の録音用モジュール

import numpy as np
import sounddevice as sd

import os
import configparser
# ファイルの存在チェック用モジュール
import errno

class Recorderconfig:
    def __init__(self, config_ini_path = './configs/config.ini'):
        # iniファイルの読み込み
        self.config_ini = configparser.ConfigParser()
        
        # 指定したiniファイルが存在しない場合、エラー発生
        if not os.path.exists(config_ini_path):
            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), config_ini_path)
        
        self.config_ini.read(config_ini_path, encoding='utf-8')
        Recorder_items = self.config_ini.items('Recorder')
        self.Recorder_config_dict = dict(Recorder_items)

class Recorder:
    def __init__(self, config_ini_path = './configs/config.ini'):
            
            Recorder_config = Recorderconfig(config_ini_path = config_ini_path)
            config_dict = Recorder_config.Recorder_config_dict
            
            self.fs = int(config_dict["fs"])
            self.silence_threshold = float(config_dict["silence_threshold"])
            self.min_duration = float(config_dict["min_duration"])
            self.amplitude_threshold = float(config_dict["amplitude_threshold"])
            self.start_threshold = float(config_dict["start_threshold"])

    def speech2audio(self):
        record_Flag = False

        non_recorded_data = []
        recorded_audio = []
        silent_time = 0
        input_time = 0
        start_threshold = 0.3
        all_time = 0
        
        with sd.InputStream(samplerate=self.fs, channels=1) as stream:
            while True:
                data, overflowed = stream.read(int(self.fs * self.min_duration))
                all_time += 1
                if all_time == 10:
                    print("stand by ready OK")
                elif all_time >=10:
                    if np.max(np.abs(data) > self.amplitude_threshold) and not record_Flag:
                        input_time += self.min_duration
                        if input_time >= start_threshold:
                            record_Flag = True
                            print("recording...")
                            recorded_audio=non_recorded_data[int(-1*start_threshold*10)-2:]  

                    else:
                        input_time = 0

                    if overflowed:
                        print("Overflow occurred. Some samples might have been lost.")
                    if record_Flag:
                        recorded_audio.append(data)

                    else:
                        non_recorded_data.append(data)

                    if np.all(np.abs(data) < self.amplitude_threshold):
                        silent_time += self.min_duration
                        if (silent_time >= self.silence_threshold) and record_Flag:
                            print("finished")
                            record_Flag = False
                            break
                    else:
                        silent_time = 0

        audio_data = np.concatenate(recorded_audio, axis=0)

        return audio_data

が全文

class Recorderconfig:
    def __init__(self, config_ini_path = './configs/config.ini'):
        # iniファイルの読み込み
        self.config_ini = configparser.ConfigParser()
        
        # 指定したiniファイルが存在しない場合、エラー発生
        if not os.path.exists(config_ini_path):
            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), config_ini_path)
        
        self.config_ini.read(config_ini_path, encoding='utf-8')
        Recorder_items = self.config_ini.items('Recorder')
        self.Recorder_config_dict = dict(Recorder_items)

この中で

config_ini_path = './configs/config.ini'

で指定されている設定ファイルをRecorder_config_dictとして読み込む

辞書型で読み込んでいるため、設定ファイルの中身をpythonの辞書として読み込むことが可能

self.Recorder_config_dict = dict(Recorder_items)

読み込んでいるのは

[Recorder]

fs=16000
silence_threshold=0.5
min_duration=0.1
amplitude_threshold=0.05
start_threshold = 0.3

の部分
それが

Recorder_items = self.config_ini.items('Recorder')

次に
Recorderクラスのinitメソッド

class Recorder:
    def __init__(self, config_ini_path = './configs/config.ini'):
            
            Recorder_config = Recorderconfig(config_ini_path = config_ini_path)
            config_dict = Recorder_config.Recorder_config_dict
            
            self.fs = int(config_dict["fs"])
            self.silence_threshold = float(config_dict["silence_threshold"])
            self.min_duration = float(config_dict["min_duration"])
            self.amplitude_threshold = float(config_dict["amplitude_threshold"])
            self.start_threshold = float(config_dict["start_threshold"])

まず、設定ファイルの内容をconfig_dictに格納
これは辞書型のため、config_dict[“device”]のような形で設定ファイルの内容を文字列として取得
すべての文字を文字列として取得するため、int型やbool型にしたい場合は、適宜型変更をする必要があることに注意

self.fs = int(config_dict["fs"]):

ではサンプリングレート fsの設定

fs=16000

がその対象
この時文字列として読み込んでいるので
Int でキャストしている

つまり

            self.silence_threshold = float(config_dict["silence_threshold"])
            self.min_duration = float(config_dict["min_duration"])
            self.amplitude_threshold = float(config_dict["amplitude_threshold"])
            self.start_threshold = float(config_dict["start_threshold"])

の部分で

config_dict[]

で設定値を読み込んで
それぞれfloat int などにキャストして値を設定している

続いて、設定ファイルから各種値をインスタンス変数に格納する

Recorderクラスのspeech2audioメソッド

class Recorder:
　　　・・・
    def speech2audio(self):
        record_Flag = False

        non_recorded_data = []
        recorded_audio = []
        silent_time = 0
        input_time = 0
        start_threshold = 0.3
        all_time = 0
        
        with sd.InputStream(samplerate=self.fs, channels=1) as stream:
            while True:
                data, overflowed = stream.read(int(self.fs * self.min_duration))
                all_time += 1
                if all_time == 10:
                    print("stand by ready OK")
                elif all_time >=10:
                    if np.max(np.abs(data) > self.amplitude_threshold) and not record_Flag:
                        input_time += self.min_duration
                        if input_time >= start_threshold:
                            record_Flag = True
                            print("recording...")
                            recorded_audio=non_recorded_data[int(-1*start_threshold*10)-2:]  

                    else:
                        input_time = 0

                    if overflowed:
                        print("Overflow occurred. Some samples might have been lost.")
                    if record_Flag:
                        recorded_audio.append(data)

                    else:
                        non_recorded_data.append(data)

                    if np.all(np.abs(data) < self.amplitude_threshold):
                        silent_time += self.min_duration
                        if (silent_time >= self.silence_threshold) and record_Flag:
                            print("finished")
                            record_Flag = False
                            break
                    else:
                        silent_time = 0

        audio_data = np.concatenate(recorded_audio, axis=0)

        return audio_data

上記の部分では，ユーザの発話をマイクを通して取得

「stand by ready OK」と表示されてから，マイクに入力された音声の大きさが閾値（self.amplitude_threshold）以上だった場合に，録音が開始
録音が開始されたら、「recording…」と表示

その後、マイクに入力された音声の大きさが閾値以下になって0.5秒（self.silence_threshold）経過したら録音が停止する

録音が停止したら、「finished」と表示

初期設定:
* record_Flag : 録音中かどうかを示すフラグ。初期値は False（録音していない）。
* non_recorded_data : 録音を開始する前に収集したデータを保持するリスト。
* recorded_audio : 録音中の音声データを保持するリスト。
* silent_time : 無音が続いた時間をカウントする変数。
* input_time : 入力が続いた時間をカウントする変数。
* start_threshold : 録音を開始するためのしきい値。
* all_time : 全体の時間をカウントする変数

with sd.InputStream(samplerate=self.fs, channels=1) as stream:: 

サンプリングレート self.fs とチャンネル数 1（モノラル）で音声入力ストリームを開始

            while True:

でループで音声を取得する

データの読み取り: data, overflowed = stream.read(int(self.fs * self.min_duration))
* self.min_duration に基づいて、一定時間分の音声データを読み取ります。
* overflowed はバッファがオーバーフローしたかどうかを示します。
スタンバイメッセージの表示:
* 最初の10回のループで “stand by ready OK” を表示します（初期化期間を意味するようです）。
が

                data, overflowed = stream.read(int(self.fs * self.min_duration))
                all_time += 1
                if all_time == 10:
                    print("stand by ready OK")

の部分

if np.max(np.abs(data) > self.amplitude_threshold) and not record_Flag:
    input_time += self.min_duration
    if input_time >= start_threshold:
        record_Flag = True
        print("recording...")
        recorded_audio = non_recorded_data[int(-1 * start_threshold * 10) - 2:]
else:
    input_time = 0

では

np.max(np.abs(data) > self.amplitude_threshold):

* これは、現在の音声データの振幅が self.amplitude_threshold（振幅のしきい値）を超えているかどうかをチェックします。np.abs(data) は音声データの絶対値（振幅）を計算し、np.max() はその最大値を取得します。
not record_Flag:
* record_Flag が False の場合、つまりまだ録音が開始されていない場合にこの条件が真になります。

次に入力時間の追加

input_time += self.min_duration

上記の条件が真の場合、input_time に self.min_duration を加算します。これにより、音声がしきい値を超えた時間がカウントされます。

if input_time >= start_threshold:

input_time が start_threshold を超えた場合、録音を開始する条件が満たされたとみなされます。start_threshold は録音開始のためのしきい値です。

録音の開始

record_Flag = True
print("recording...")
recorded_audio = non_recorded_data[int(-1 * start_threshold * 10) - 2:]

record_Flag = True に設定することで、録音が開始されます。

recorded_audio = non_recorded_data[int(-1 * start_threshold * 10) - 2:]:

* 録音の開始時に、直前の音声データを recorded_audio に追加します。これは、しきい値を超える前の短い期間の音声も記録するためです。int(-1 * start_threshold * 10) – 2 は、start_threshold 秒間のデータを遡って取得するインデックスを計算しています。-2 の部分は余裕を持たせるために追加されている可能性があります

module_whisper.py
音声の文字起こし用のモジュール

from faster_whisper import WhisperModel
import numpy as np
import torch

import os
import configparser
# ファイルの存在チェック用モジュール
import errno

class FasterWhisperconfig:
    def __init__(self, config_ini_path = './configs/config.ini'):
        # iniファイルの読み込み
        self.config_ini = configparser.ConfigParser()
        
        # 指定したiniファイルが存在しない場合、エラー発生
        if not os.path.exists(config_ini_path):
            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), config_ini_path)
        
        self.config_ini.read(config_ini_path, encoding='utf-8')
        FasterWhisper_items = self.config_ini.items('FasterWhisper')
        self.FasterWhisper_config_dict = dict(FasterWhisper_items)

class FasterWhisperModel:
    def __init__(self,device = None, config_ini_path = './configs/config.ini'):
        FasterWhisper_config = FasterWhisperconfig(config_ini_path = config_ini_path)
        config_dict = FasterWhisper_config.FasterWhisper_config_dict

        if device is not None:
            self.DEVICE = device
        else:
            device = config_dict["device"]

            self.DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
            if device != "auto":
                self.DEVICE = device
            
        self.BEAM_SIZE = int(config_dict["gpu_beam_size"]) if self.DEVICE == "cuda" else int(config_dict["cpu_beam_size"])
        self.language = config_dict["language"]
        self.COMPUTE_TYPE = config_dict["gpu_compute_type"] if self.DEVICE == "cuda" else config_dict["cpu_compute_type"]
        self.MODEL_TYPE = config_dict["gpu_model_type"] if self.DEVICE == "cuda" else config_dict["cpu_model_type"]
        self.kotoba_chunk_length = int(config_dict["chunk_length"])
        self.kotoba_condition_on_previous_text = config_dict["condition_on_previous_text"]
        if self.kotoba_condition_on_previous_text == "True":
            self.kotoba_condition_on_previous_text = True
        else:
            self.kotoba_condition_on_previous_text = False

        if config_dict["use_kotoba"] == "True":
            self.use_kotoba = True
        else:
            self.use_kotoba = False

        if not self.use_kotoba:
            self.model = WhisperModel(self.MODEL_TYPE, device=self.DEVICE, compute_type=self.COMPUTE_TYPE)
        else:
            self.MODEL_TYPE = config_dict["kotoba_model_type"]
            #self.model = WhisperModel(self.MODEL_TYPE, device=self.DEVICE, compute_type=self.cotoba_compute_type)
            self.model = WhisperModel(self.MODEL_TYPE)


    def audio2text(self, data):
        result = ""
        data = data.flatten().astype(np.float32)
        if not self.use_kotoba:
            segments, _ = self.model.transcribe(data, beam_size=self.BEAM_SIZE,language=self.language)
        else:
            segments, _ = self.model.transcribe(data, beam_size=self.BEAM_SIZE,language=self.language, chunk_length=self.kotoba_chunk_length, condition_on_previous_text=self.kotoba_condition_on_previous_text)
        
        for segment in segments:
            result += segment.text
        
        return result
            
    def audioFile2text(self, file_path):
        result = ""
        if not self.use_kotoba:
            segments, _ = self.model.transcribe(file_path, beam_size=self.BEAM_SIZE,language=self.language)
        else:
            segments, _ = self.model.transcribe(file_path, beam_size=self.BEAM_SIZE,language=self.language, chunk_length=self.kotoba_chunk_length, condition_on_previous_text=self.kotoba_condition_on_previous_text)
        
        for segment in segments:
            result += segment.text

        return result
FasterWhisperconfigクラス

class FasterWhisperconfig:
    def __init__(self, config_ini_path = './configs/config.ini'):
        # iniファイルの読み込み
        self.config_ini = configparser.ConfigParser()
        
        # 指定したiniファイルが存在しない場合、エラー発生
        if not os.path.exists(config_ini_path):
            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), config_ini_path)
        
        self.config_ini.read(config_ini_path, encoding='utf-8')
        FasterWhisper_items = self.config_ini.items('FasterWhisper')
        self.FasterWhisper_config_dict = dict(FasterWhisper_items)

ここではconfig_ini_path = ‘./configs/config.ini’で指定されている設定ファイルをFasterWhisper_config_dictとして読み込んでいます。
辞書型で読み込んでいるため、設定ファイルの中身をpythonの辞書として読み込むことが可能

これにより

[FasterWhisper]
device = auto
language = ja

gpu_model_type = large-v3
gpu_beam_size = 1
gpu_compute_type = float16

cpu_model_type = small
cpu_beam_size = 1
cpu_compute_type = int8

use_kotoba = True
kotoba_model_type = kotoba-tech/kotoba-whisper-v1.0-faster
chunk_length = 15
condition_on_previous_text = False

の部分を読み込んで辞書にしている

FasterWhisperModelクラスのinitメソッド

class FasterWhisperModel:
    def __init__(self,device = None, config_ini_path = './configs/config.ini'):
        FasterWhisper_config = FasterWhisperconfig(config_ini_path = config_ini_path)
        config_dict = FasterWhisper_config.FasterWhisper_config_dict

        if device is not None:
            self.DEVICE = device
        else:
            device = config_dict["device"]

            self.DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
            if device != "auto":
                self.DEVICE = device
            
        self.BEAM_SIZE = int(config_dict["gpu_beam_size"]) if self.DEVICE == "cuda" else int(config_dict["cpu_beam_size"])
        self.language = config_dict["language"]
        self.COMPUTE_TYPE = config_dict["gpu_compute_type"] if self.DEVICE == "cuda" else config_dict["cpu_compute_type"]
        self.MODEL_TYPE = config_dict["gpu_model_type"] if self.DEVICE == "cuda" else config_dict["cpu_model_type"]
        self.kotoba_chunk_length = int(config_dict["chunk_length"])
        self.kotoba_condition_on_previous_text = config_dict["condition_on_previous_text"]
        if self.kotoba_condition_on_previous_text == "True":
            self.kotoba_condition_on_previous_text = True
        else:
            self.kotoba_condition_on_previous_text = False

        if config_dict["use_kotoba"] == "True":
            self.use_kotoba = True
        else:
            self.use_kotoba = False

        if not self.use_kotoba:
            self.model = WhisperModel(self.MODEL_TYPE, device=self.DEVICE, compute_type=self.COMPUTE_TYPE)
        else:
            self.MODEL_TYPE = config_dict["kotoba_model_type"]
            #self.model = WhisperModel(self.MODEL_TYPE, device=self.DEVICE, compute_type=self.cotoba_compute_type)
            self.model = WhisperModel(self.MODEL_TYPE)

これは

        FasterWhisper_config = FasterWhisperconfig(config_ini_path = config_ini_path)
        config_dict = FasterWhisper_config.FasterWhisper_config_dict

で
まず、設定ファイルの内容をconfig_dictに格納しています。これは辞書型のため、config_dict[“device”]のような形で設定ファイルの内容を文字列として取得することができます。
あくまで、すべての文字を文字列として取得するため、int型やbool型にしたい場合は、適宜型変更をする必要があることに注意

続いて下記の順番で処理を行います。
* モデルを動作させるdeviceを指定する
* 設定ファイルの各種設定を取得する
* モデルを定義する。
* 設定ファイルに合わせて、適切なモデルを定義する

FasterWhisperModelクラスのaudioFile2textメソッド

class FasterWhisperModel:
    ・・・
    def audio2text(self, data):
        result = ""
        data = data.flatten().astype(np.float32)
        if not self.use_kotoba:
            segments, _ = self.model.transcribe(data, beam_size=self.BEAM_SIZE,language=self.language)
        else:
            segments, _ = self.model.transcribe(data, beam_size=self.BEAM_SIZE,language=self.language, chunk_length=self.kotoba_chunk_length, condition_on_previous_text=self.kotoba_condition_on_previous_text)
        
        for segment in segments:
            result += segment.text
        
        return result

faster-whisperモデルのtranscribeメソッドを呼び出して、音声認識をしています。
faster-whisperモデルのtranscribeメソッドは、ファイル名を引数にした場合は、そのファイルを読み込んで処理を行い。numpyデータやbinaryデータを引数にした場合は、そのデータをそのまま利用してくれます

設定ファイルで指定したモデルに合わせて、適切な引数を使っています。
faster-whisperは30秒以上の音声に関しては、音声を分割して処理をするため、分割されて生成されたテキストをresult変数に格納して、return

data = data.flatten().astype(np.float32)

音声データをfaster-whisperモデルに入力するために、データを整形