音声認識バッチ処理システムの作成_ローカル実装
前書き
試行錯誤のための仕組み作りの重要度が高めなAIエンジニアです。
個人的に変更可能な値が複数ある場合、設定ファイルを用意する派です。
argsも便利ですが、実行コマンドはなるべく短くしたいと考えています。
前回、Google Colaboratory上でReazonSpeechとWhisperを利用して音声認識を実行するまでの流れをまとめました。
今回は音声認識を実施したい音ファイルがある場合に設定ファイルの内容に応じて音声認識を実行するバッチ処理システムを作成していきます。
オンライン処理のシステムではない点にご注意ください。
音声認識バッチ処理システムの作成
この章では音声認識バッチ処理システムの環境構築、ディレクトリ構成、全体の流れを説明していきます。音声認識のモデルとしてReazonSpeechとWhisperを利用できる仕組みにしています。
環境構築
まずは音声認識用の環境を構築します。
今回はMacでconda環境を使い、環境構築していきます。
参考までに私が実装したPCの性能をまとめておきます。
項目 | 詳細 |
---|---|
マシン名 | MacBook Pro |
チップ | Apple M1 Pro |
メモリ | 16GB |
OS | mac OS Sonomaバージョン14.6.1 |
HomebrewとMiniforgeをインストールしてbrewコマンドとcondaコマンドが使用可能な前提で進めていきます。
参考までに
HomebrewとMiniforgeのインストールのやり方は以下の記事に記載しております。
conda環境を作成して起動します。
# 音声認識用のconda環境をspeech_recognitionという名前で作成
% conda create -n speech_recognition python=3.10
# speech_recognition環境を起動
% conda activate speech_recognition
必要なライブラリをインストールしていきます。
(speech_recognition)% brew install ffmpeg
# 自分が作業するworkspaceに移動
(speech_recognition)% cd workspace
# reazonspeechのコードをクローン
(speech_recognition)% git clone https://github.com/reazon-research/reazonspeech
# huggingface-hubのインストール
# (何も指定しないとバージョンの関係で動かなかったため、バージョンを指定)
(speech_recognition)% pip install huggingface-hub==0.23.2
# ReazonSpeechに必要なライブラリのインストール
(speech_recognition)% pip install --no-warn-conflicts reazonspeech/pkg/nemo-asr
# Whisperに必要なライブラリのインストール
(speech_recognition)% pip install git+https://github.com/openai/whisper.git
ディレクトリ構成
以下のようなディレクトリ構成にしました。
workspace/
└── speech_recognition/
├── 00_data/
│ └── 01_input_data/
│ ├── 01_test.wav
│ ├── 02_test.wav
│ ├── ...
│ └── xx_test.wav
├── 01_code/
│ ├── main.py
│ ├── speech_recognition_setting.yaml
│ └── using_main/
│ └── speech_recognition
└── reazonspeech
全体の流れ
全体の流れは以下のような流れにしました。
実装コード
ここからは各ファイルの説明と実装コードを記載していきます。
mainファイル
main.pyにはSpeechRecognitionをインスタンス化して実行する処理を記載します。
import using_main.speech_recognition as speech_recognition
def main():
speech_recog = speech_recognition.SpeechRecognition()
result = speech_recog.speech_recognition()
return
if __name__ == "__main__":
main()
設定ファイル
speech_recognition_setting.yamlには音声認識に関する設定を記載します。
dir内にディレクトリの設定をまとめます。
input_dir内のファイルすべてに音声認識を実行するか、指定ファイルだけ選ぶようにするのかも選択できるようにします。
音声認識を実行するモデルも複数選択できるようにします。
whisperのモデルは複数あるので"whisper_base", "whisper_large"のように先頭に"whisper_"という文字列をつけて指定することにします。
CPUで実行するか、GPUで実行するかという点も選択できるようにします。
dir:
base_dir: "/workspace/"
data_dir: "00_data"
input_dir: "01_input_data"
code_dir: "01_code"
# input_dir内のファイル全てに実行する場合
target_type: "all"
# 指定したファイルのみ実行する場合
# target_type: "designation"
# file_name_list: ["xx_test.wav"]
# 音声認識を実行するモデルのリスト
model_list: ["reazon", "whisper_base", "whisper_large"]
# デバイスの設定
reazon_device: "cpu" # "cuda:0" or "mps" or "cpu"
whisper_device: "cpu" # "cuda:0" or "mps" or "cpu"
SpeechRecognitionファイル
speech_recognition.pyには設定ファイルの内容に基づき、音声認識を実施する処理を記載します。
import os
import glob
import yaml
import whisper
from reazonspeech.nemo.asr import transcribe, audio_from_path, load_model
class SpeechRecognition:
"""
音声認識を実施するクラス
Attributes
----------
setting_dict : dict
設定情報を格納した辞書
base_dir_path : str
ベースディレクトリのパス
data_dir_path : str
データディレクトリのパス
input_dir_path : str
入力ディレクトリのパス
"""
def __init__(self):
"""初期化処理"""
self.setting_dict = self.__read_setting_yaml()
self.base_dir_path = self.setting_dict["dir"]["base_dir"]
self.data_dir_path = self.__generate_data_dir_path()
self.input_dir_path = self.__generate_input_dir_path()
return
def __read_setting_yaml(self):
"""設定情報をyamlファイルから読み込む"""
with open("/workspace/01_code/speech_recognition_setting.yaml", "r") as yml:
setting_dict = yaml.safe_load(yml)
return setting_dict
def __generate_data_dir_path(self):
"""データディレクトリのパスを生成する"""
data_dir = self.setting_dict["dir"]["data_dir"]
data_dir_path = os.path.join(self.base_dir_path, data_dir)
return data_dir_path
def __generate_input_dir_path(self):
"""入力ディレクトリのパスを生成する"""
input_dir = self.setting_dict["dir"]["input_dir"]
input_dir_path = os.path.join(self.data_dir_path, input_dir)
return input_dir_path
def __get_all_file_path_list(self, dir_path):
"""input_dir内のすべてのファイルパスをリストで取得する"""
key = dir_path + "/*.wav"
file_path_list = glob.glob(key)
file_path_list.sort()
return file_path_list
def __generate_designation_file_path_list(self, dir_path, file_name_list):
"""指定のファイル名リストに対応するファイルパスリストを取得する"""
file_path_list = []
for file_name in file_name_list:
file_path = os.path.join(dir_path, file_name)
file_path_list.append(file_path)
return file_path_list
def __generate_target_file_path_list(self, setting_dict):
"""対象ファイルパスリストを作成する"""
target_type = setting_dict["target_type"]
if target_type == "all":
target_file_path_list = self.__get_all_file_path_list(self.input_dir_path)
elif target_type == "designation":
file_name_list = setting_dict["file_name_list"]
target_file_path_list = self.__generate_designation_file_path_list(self.input_dir_path, file_name_list)
else:
raise ValueError("target_type is not supported.")
return target_file_path_list
def __exec_reazon(self, file_path, setting_dict):
"""Reazonで指定の音声ファイルを音声認識する"""
audio = audio_from_path(file_path)
# ReazonSpeechモデルをHugging Faceから取得
device = setting_dict["reazon_device"]
model = load_model(device=device)
result = transcribe(model, audio)
return result
def __print_speech_segment_reazon(self, result):
"""Reazonで認識した結果をセグメントごとに表示する"""
for seg in result.segments:
print("%5.2f %5.2f %s" % (seg.start_seconds, seg.end_seconds, seg.text))
return
def __exec_whisper(self, file_path, setting_dict, load_model=None):
"""whisperで音声認識を実施"""
device = setting_dict["whisper_device"]
if load_model is None:
model = whisper.load_model(name="base", device=device)
else:
model = whisper.load_model(name=load_model, device=device)
result = model.transcribe(file_path)
return result
def __print_speech_segment_whisper(self, result):
"""whisperで認識した結果をセグメントごとに表示する"""
for seg in result["segments"]:
id, start, end, text = [seg[key] for key in ["id", "start", "end", "text"]]
print(f"{id:03}: {start:5.1f} - {end:5.1f} | {text}")
return
def __exec_speech_recognition(self, file_path, model, setting_dict):
"""指定の音声ファイルを指定の指定のモデルで音声認識する"""
if model == "reazon":
result = self.__exec_reazon(file_path, setting_dict)
self.__print_speech_segment_reazon(result)
elif "whisper" in model:
model_name = model.replace("whisper_", "")
result = self.__exec_whisper(file_path, load_model=model_name, setting_dict=setting_dict)
self.__print_speech_segment_whisper(result)
else:
raise ValueError("model is not supported.")
return result
def speech_recognition(self):
"""設定に従い、音声認識を実施する"""
file_path_list = self.__generate_target_file_path_list(self.setting_dict)
for file_path in file_path_list:
print("file_path: ", file_path)
model_list = self.setting_dict["model_list"]
for model in model_list:
print("model: ", model)
result = self.__exec_speech_recognition(file_path, model, self.setting_dict)
return result
後書き
簡単ですが設定ファイルの内容に応じて音声認識を実行するバッチ処理システムを作成しました。
使用したいモデルが増えた場合も内部を拡張することで活用の幅を広げることができるのではないかと考えております。
参考
・バッチ処理とオンライン処理の違い
・ディレクトリ構成をマークダウン化
・シーケンス図作成のサイト
・シーケンス図の書き方
・Whisperのmspでの実行
Discussion