kabosu-coreを使って日本語の文章でg2p処理をする

初めに

新しいTTSの前処理のライブラリがでていたので、試してみます

github.com

開発環境

環境構築

まずはuvを使って環境構築をします

uv venv --python 3.12
.venv\Scripts\activate

必要なライブラリをインストールします

uv pip install -r requirements.txt

次に処理にyomikataを使っているので、必要なモデルをダウンロードします

uv run python -m yomikata download

日本語文章でg2pを実行

動く環境ができたので、以下のようなサンプルスクリプトを使って実際に日本語の文章で実行してみます

import kabosu_core

def display_phonemes(text: str):
    """音素列を表示"""
    print(f"\n■ 入力テキスト: {text}")
    print("-" * 60)

    # 音素列を取得
    phonemes = kabosu_core.g2p(text, kana=False)
    print(f"【音素列】")
    print(f"  {phonemes}")

    # 音素をリスト形式でも表示
    phoneme_list = phonemes.split()
    print(f"\n【音素リスト】 (合計 {len(phoneme_list)} 音素)")
    print(f"  {phoneme_list}")

    return phonemes


def display_kana(text: str):
    """カナ表記を表示"""
    print(f"\n■ 入力テキスト: {text}")
    print("-" * 60)

    # カナ表記を取得
    kana = kabosu_core.g2p(text, kana=True)
    print(f"【カナ表記】")
    print(f"  {kana}")

    return kana


def main():
    """メイン関数"""
    import sys

    if len(sys.argv) <= 1:
        print("エラー: テキストを指定してください")
        print("\n使用方法: python demo_g2p.py \"変換したいテキスト\"")
        print("\n例:")
        print("  python demo_g2p.py \"こんにちは、今日は良い天気ですね\"")
        sys.exit(1)

    # コマンドライン引数からテキストを取得
    text = " ".join(sys.argv[1:])
    display_kana(text)
    display_phonemes(text)


if __name__ == "__main__":
    main()

実行は以下のようにします

uv run python demo_g2p.py "こんにちは、今日は良い天気ですね"

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

[09/18/25 21:48:16] INFO     Running on cpu                                                                                                                                      dbert.py:41
                    INFO     Loaded model from directory C:\Users\yuta\Desktop\Private\kabosu-core\.venv\Lib\site-packages\yomikata\dbert-artifacts                             dbert.py:124

■ 入力テキスト: こんにちは、今日は良い天気ですね
------------------------------------------------------------
【カナ表記】
  コンニチハ、キョウハヨイテンキデスネ

■ 入力テキスト: こんにちは、今日は良い天気ですね
------------------------------------------------------------
【音素列】
  k o N n i ch i w a pau ky o u w a y o i t e N k i d e s U n e

【音素リスト】 (合計 29 音素)
  ['k', 'o', 'N', 'n', 'i', 'ch', 'i', 'w', 'a', 'pau', 'ky', 'o', 'u', 'w', 'a', 'y', 'o', 'i', 't', 'e', 'N', 'k', 'i', 'd', 'e', 's', 'U', 'n', 'e']

FireRedTTS2をWindowsで動かす

初めに

長時間会話音声生成をするための音声合成ライブラリが出ていたので、試していきます

開発環境

環境構築

リポジトリのReadMeにはcondaで環境構築をしていますが、今回はuvを使っていきます

uv venv -p 3.11 

ReadMeに記載通りライブラリを入れていきます

uv pip install torch==2.7.1 torchvision==0.22.1 torchaudio==2.7.1 --index-url https://download.pytorch.org/whl/cu126

# Step 2. Install Dependencies
uv pip install -e .
uv pip install -r requirements.txt

以下でモデルのダウンロードを行います

git lfs install
git clone https://huggingface.co/FireRedTeam/FireRedTTS2 pretrained_models/FireRedTTS2

デモの起動

以下でデモが起動します

python gradio_demo.py --pretrained-dir "./pretrained_models/FireRedTTS2"

ランダムボイスの実行

ランダムボイスの場合は、リファレンス音声は必要ないため生成するためのテキストをルールに従って記載をしていきます

[S1]こんにちは、今日はいい天気ですね。[S2]そうですね、散歩に行きませんか?[S1]いいアイデアですね!公園に行きましょう。[S2]では、準備してきます。

ボイスクローン

こちらは特定の話者の音声を渡して、ゼロショットにて対話音声を作成する機能です (言語設定を英語にすると正常に動きませんでした)

windowsだとsoundfileを入れないとエラーになったので以下を実行しておきます

uv pip install soundfile  

精度に関してはゼロショットなので良くはなかったです

Qwen2-Audio-7B-Instructで音声とテキストから感情を判定する

初めに

いままで試してきた感情の判定は、音声データもしくはテキストデータの片方から判定を行っていました。 今回はマルチモーダルを用いて両方のデータから判定を行っていきます

開発環境

環境構築

必要なライブラリをインストールします

uv pip install --extra-index-url https://download.pytorch.org/whl/cu124 --pre torch torchvision torchaudio
uv pip install --upgrade git+https://github.com/huggingface/transformers
uv pip install accelerate sentencepiece soundfile

必要に応じて

uv pip install bitsandbytes

判定の実行

指定したラベルのみを返すように 「番号 → 日本語ラベル」で問題なければ 数値ラベル方式を使って実装をしていきます

import argparse, pathlib, torch, torchaudio
from transformers import (AutoProcessor, Qwen2AudioForConditionalGeneration,
                          LogitsProcessor, LogitsProcessorList)

MODEL_ID = "Qwen/Qwen2-Audio-7B-Instruct"
LABELS   = ["喜び", "怒り", "悲しみ", "恐れ", "驚き", "嫌悪", "中立", "その他"]
NUM_IDS  = list(range(len(LABELS)))          # [0,1,2,3,4,5,6,7]

# ---------- 1) CLI ----------
ap = argparse.ArgumentParser()
ap.add_argument("audio")
ap.add_argument("text")
ap.add_argument("--8bit", action="store_true")
args = ap.parse_args()

# ---------- 2) processor / model ----------
proc = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
tok  = proc.tokenizer
load_kw = dict(device_map="auto", trust_remote_code=True)
load_kw["load_in_8bit" if args.__dict__["8bit"] else "torch_dtype"] = (
    True if args.__dict__["8bit"] else torch.float16
)
model = Qwen2AudioForConditionalGeneration.from_pretrained(MODEL_ID, **load_kw).eval()

# ---------- 3) 音声 ----------
wave, sr = torchaudio.load(args.audio)
target_sr = proc.feature_extractor.sampling_rate
if sr != target_sr:
    wave = torchaudio.functional.resample(wave, sr, target_sr)
wav_np = wave.squeeze().cpu().numpy()

# ---------- 4) プロンプト ----------
system_msg = (
    "日本語の感情を分類します。"
    "次の数値から **1 つだけ**返してください。\n"
    + "\n".join(f"{i}:{lbl}" for i, lbl in enumerate(LABELS))
)
prompt = proc.apply_chat_template(
    [
        {"role": "system", "content": system_msg},
        {"role": "user", "content": [
            {"type": "audio", "audio": wav_np},
            {"type": "text",  "text": args.text},
            {"type": "text",  "text": "感情番号は?"}
        ]}
    ],
    tokenize=False, add_generation_prompt=True,
)

inputs = proc(text=prompt, audio=wav_np, sampling_rate=target_sr,
              return_tensors="pt").to(model.device)

# ---------- 5) ロジット制限 (0〜7 のみ) ----------
class AllowedNums(LogitsProcessor):
    def __init__(self, ids): self.ids = torch.tensor(ids)
    def __call__(self, input_ids, scores):
        mask = torch.full_like(scores, float("-inf"))
        mask[:, self.ids] = 0
        return scores + mask

proc_list = LogitsProcessorList([AllowedNums(NUM_IDS)])

gen = model.generate(**inputs,
                     logits_processor=proc_list,
                     max_new_tokens=1,
                     temperature=0.0, do_sample=False)

num_id = gen[0, -1].item()
print(LABELS[num_id])          # 最終的に日本語ラベルで表示

以下のように実行します

python .\run_audio.py '.\Kanjyou Ikari11.wav' "イタリア旅行で彼は、いくつか景勝の地として有名な都市、例えば、ナポリやフィレンツェを訪れた。"

YoichiTakenaka/deverta-v3-japanese-large-Anticipationでテキストの感情を判定する

開発環境

環境構築

以下で環境構築をして、必要なライブラリをインストールします

uv venv
.\.venv\Scripts\activate
uv pip install torch --index-url https://download.pytorch.org/whl/cu124
uv pip install transformers pandas deberta-emotion-predictor

ラベル判定

以下のように引数から指定したテキストを判定するコードを作成します

import argparse
import torch
from deberta_emotion_predictor import DeBERTaEmotionPredictor
import warnings
import os
import sys

class EmotionAnalyzer:
    """
    日本語テキストの感情分析を行うための本番用クラス。
    モデルのロードを一度だけ行い、複数のテキストを効率的に処理します。
    """
    def __init__(self, device: str = None, verbose: bool = True):
        """
        アナライザーを初期化し、モデルをメモリにロードします。
        
        Args:
            device (str, optional): 使用するデバイス ('cuda' or 'cpu')。Noneの場合は自動検出。
            verbose (bool, optional): モデルロード時にメッセージを表示するかどうか。
        """
        self.verbose = verbose
        if self.verbose:
            print("--- 感情分析モデルの準備を開始 ---")
            
        try:
            if device is None:
                self.device = "cuda" if torch.cuda.is_available() else "cpu"
            else:
                self.device = device
            
            if self.verbose:
                print(f"使用デバイス: {self.device}")
                print("モデルをロードしています...(初回はダウンロードに時間がかかります)")
                
            self.predictor = DeBERTaEmotionPredictor(device=self.device)
            
            # 感情ラベルと確率カラム名をクラス変数として定義
            self.emotion_labels = ['Joy', 'Sadness', 'Anticipation', 'Surprise', 'Anger', 'Fear', 'Disgust', 'Trust']
            self.prob_columns = [f"{label}_positive_probability" for label in self.emotion_labels]

            if self.verbose:
                print("--- モデル準備完了 ---")

        except Exception as e:
            print(f"モデルの初期化中に致命的なエラーが発生しました: {e}", file=sys.stderr)
            # モデルロードの失敗は致命的なので、プログラムを終了させる
            raise

    def analyze(self, text: str) -> list[tuple[str, float]]:
        """
        単一のテキストから感情を分析し、結果を確率の高い順に返します。

        Args:
            text (str): 分析したい日本語の文章。

        Returns:
            list[tuple[str, float]]: (感情ラベル, 確率) のタプルが格納されたリスト。
                                      例: [('Joy', 0.98), ('Trust', 0.01), ...]
        """
        if not isinstance(text, str) or not text.strip():
            return []
            
        try:
            results_df = self.predictor.predict_emotions(text)
            
            if results_df.empty:
                return []
            
            result_series = results_df.iloc[0]
            
            # 扱いやすいように、簡単な感情名と確率のペアを作成
            emotion_probabilities = {}
            for label, col_name in zip(self.emotion_labels, self.prob_columns):
                emotion_probabilities[label] = result_series[col_name]

            # 確率の高い順にソートしてリストとして返す
            sorted_results = sorted(emotion_probabilities.items(), key=lambda item: item[1], reverse=True)
            return sorted_results

        except Exception as e:
            print(f"テキスト「{text[:30]}...」の推論中にエラーが発生しました: {e}", file=sys.stderr)
            return []

def main():
    """
    コマンドラインからテキストを受け取り、感情分析を実行して結果を表示するメイン関数。
    """
    parser = argparse.ArgumentParser(
        description="日本語テキストから8つの感情を分析します(本番用コード)。",
        formatter_class=argparse.RawTextHelpFormatter # ヘルプの改行を保持
    )
    parser.add_argument(
        "text",
        nargs='*', # 複数のテキスト引数を受け取れるようにする
        default=[],
        help="分析したい日本語の文章。複数指定可能。\n例: \"これは嬉しい\" \"なんてことだ\""
    )
    parser.add_argument(
        "-f", "--file",
        type=str,
        help="テキストが1行ずつ書かれたファイルへのパス。\nファイル内の各行を個別に分析します。"
    )

    args = parser.parse_args()
    
    # テキストが一つも指定されなかった場合は使い方を表示して終了
    if not args.text and not args.file:
        parser.print_help()
        print("\nエラー: 分析対象のテキストを引数で指定するか、--fileオプションでファイルを指定してください。")
        return

    try:
        analyzer = EmotionAnalyzer()
    except Exception:
        print("アナライザーの起動に失敗したため、処理を終了します。", file=sys.stderr)
        return

    texts_to_process = args.text
    if args.file:
        try:
            with open(args.file, 'r', encoding='utf-8') as f:
                # ファイルから読み込んだ各行の改行文字などを除去し、空行は無視する
                texts_to_process.extend([line.strip() for line in f if line.strip()])
        except FileNotFoundError:
            print(f"エラー: 指定されたファイルが見つかりません: {args.file}", file=sys.stderr)
            return
    
    # 2. テキストごとに感情を分析し、結果を表示
    for i, text in enumerate(texts_to_process):
        print("\n" + "#"*50)
        print(f"分析対象 {i+1}: 「{text}」")
        print("#"*50)
        
        results = analyzer.analyze(text)
        
        if results:
            top_label, top_score = results[0]
            print(f"  最も可能性の高い感情: {top_label} ({top_score:.2%})")
            
            print("\n  --- 各感情の確率 ---")
            for label, score in results:
                print(f"  - {label:<15}: {score:.2%}")
        else:
            print("  感情を判定できませんでした。")
        print("#"*50)

if __name__ == '__main__':
    main()

以下のように実行します

python .\text_emotion_recognition_koala.py "ストラッ
トフォード・オン・エイヴォンは、シェイクスピアの生まれたところですが、毎年多くの観光客が訪れます。"

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

##################################################
分析対象 1: 「ストラットフォード・オン・エイヴォンは、シェイクスピアの生まれたところですが、毎年多くの観光客が訪れます。」
##################################################
感情:Joy の推論を開始 (1/1)
感情:Sadness の推論を開始 (1/1)
感情:Anticipation の推論を開始 (1/1)
感情:Surprise の推論を開始 (1/1)
感情:Anger の推論を開始 (1/1)
感情:Fear の推論を開始 (1/1)
感情:Disgust の推論を開始 (1/1)
感情:Trust の推論を開始 (1/1)
  最も可能性の高い感情: Surprise (59.25%)

  --- 各感情の確率 ---
  - Surprise       : 59.25%
  - Anticipation   : 2.84%
  - Joy            : 0.57%
  - Fear           : 0.43%
  - Sadness        : 0.30%
  - Disgust        : 0.23%
  - Trust          : 0.12%
  - Anger          : 0.01%
##################################################

litagin/anime_speech_emotion_classificationを使って音声の感情を判定する

初めに

以下でも音声ファイルの感情判定を行っていますが、こちらとは違うモデルを使って判定を行っていきます

ayousanz.hatenadiary.jp

開発環境

環境構築

以下で必要なライブラリをインストールします

uv pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124
uv pip install transformers huggingface_hub librosa soundfile

感情の判定

以下のコードをファイルを指定して実行します

import argparse
import os
import torch
import torchaudio
import librosa
import numpy as
import torch.nn.functional as F
from transformers import AutoModelForAudioClassification, AutoFeatureExtractor
from huggingface_hub import HfFolder
import warnings

warnings.filterwarnings("ignore")

def analyze_anime_emotion(audio_file_path: str):
    """
    litagin/anime_speech_emotion_classificationモデルを使用して音声ファイルの感情を分析する。
    (librosaによる音声読込強化版)
    """
    audio_file_path = os.path.abspath(audio_file_path)

    if not os.path.exists(audio_file_path):
        print(f"エラー: 指定されたファイルが見つかりません。")
        print(f"パスを確認してください: {audio_file_path}")
        return

    print("--- 環境・モデル情報 ---")
    print(f"ライブラリ: transformers (手動制御)")
    
    model_id = "litagin/anime_speech_emotion_classification"
    print(f"モデル: {model_id}")

    try:
        device = "cuda:0" if torch.cuda.is_available() else "cpu"
        print(f"使用デバイス: {device}")
        
        hf_token = HfFolder.get_token()
        if hf_token is None:
            print("\nエラー: Hugging Faceの認証トークンが見つかりません。`huggingface-cli login` を実行してください。")
            return

        print("モデルをロードしています...")
        feature_extractor = AutoFeatureExtractor.from_pretrained(model_id, token=hf_token, trust_remote_code=True)
        model = AutoModelForAudioClassification.from_pretrained(model_id, token=hf_token, trust_remote_code=True).to(device)
        model.eval() 
        
        print("モデルのロードが完了しました。")
        
    except Exception as e:
        print(f"モデルのロード中にエラーが発生しました: {e}")
        return


    print(f"\n音声ファイル '{audio_file_path}' を読み込んでいます...")
    try:
        waveform_np, original_sample_rate = librosa.load(audio_file_path, sr=None, mono=False)
        
        # NumPy配列をPyTorchテンソルに変換
        waveform = torch.from_numpy(waveform_np)

        # librosaがモノラルで1次元配列を返した場合、2次元に変換
        if waveform.ndim == 1:
            waveform = waveform.unsqueeze(0)
        
        # チャンネルが複数ある場合、モノラルに変換
        if waveform.shape[0] > 1:
            waveform = torch.mean(waveform, dim=0, keepdim=True)
        
        # モデルが要求するサンプリングレート(16kHz)に変換
        target_sample_rate = 16000 
        if original_sample_rate != target_sample_rate:
            # librosaはNumPy配列を扱うため、一度テンソルに変換してからリサンプル
            resampler = torchaudio.transforms.Resample(orig_freq=original_sample_rate, new_freq=target_sample_rate)
            waveform = resampler(waveform)
            
    except Exception as e:
        print(f"音声ファイルの読み込みまたは処理中にエラーが発生しました: {e}")
        return

    # 推論の実行
    print("音声の感情を判定しています...")
    try:
        inputs = feature_extractor(
            waveform.squeeze(0).numpy(), 
            sampling_rate=target_sample_rate,
            return_tensors="pt"
        ).to(device)

        with torch.no_grad():
            outputs = model(**inputs)
        
        logits = outputs.logits
        probabilities = F.softmax(logits, dim=1)[0]
        
        print("\n" + "="*40)
        print("  感情判定結果")
        print("="*40)
        
        sorted_probs, sorted_ids = torch.sort(probabilities, descending=True)
        
        top_prediction_id = sorted_ids[0].item()
        top_label = model.config.id2label[top_prediction_id]
        top_score = sorted_probs[0].item()

        print(f"  ファイル: {os.path.basename(audio_file_path)}")
        print(f"  最も可能性の高い感情: {top_label} ({top_score:.2%})")
        
        print("\n  --- 各感情の確率 ---")
        for i in range(len(sorted_probs)):
            label_id = sorted_ids[i].item()
            label_name = model.config.id2label[label_id]
            score = sorted_probs[i].item()
            print(f"  - {label_name:<12}: {score:.2%}")
        
        print("="*40)

    except Exception as e:
        print(f"\n推論中に予期せぬエラーが発生しました: {e}")

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="アニメ風音声の感情認識モデルを使用して、感情を分析します。"
    )
    parser.add_argument(
        "audio_file",
        type=str,
        help="分析したい音声ファイルのパス。"
    )
    
    args = parser.parse_args()
    
    analyze_anime_emotion(args.audio_file)

以下のように実行します

python .\anime_emotion_recognition.py '.\Kanjyou Ikari96.wav'

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

========================================
  感情判定結果
========================================
  ファイル: Kanjyou Ikari96.wav
  最も可能性の高い感情: Angry (38.79%)

  --- 各感情の確率 ---
  - Angry       : 38.79%
  - Happy       : 28.14%
  - Sexual1     : 8.57%
  - Surprised   : 8.50%
  - Embarrassed : 7.67%
  - Sad         : 3.16%
  - Neutral     : 3.14%
  - Disgusted   : 1.43%
  - Fearful     : 0.57%
  - Sexual2     : 0.02%
========================================

FunAudioLLM/SenseVoiceSmallを使って音声の感情を判定する

開発環境

環境構築

以下のライブラリをインストールします

uv pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124
uv pip install funasr

実行

以下のコードを実行します

import argparse
import os
from funasr import AutoModel

def analyze_emotion_with_funasr(audio_file_path: str):
    """
    funasrライブラリとSenseVoiceモデルを使用して音声ファイルの感情を分析する。

    Args:
        audio_file_path (str): 分析したい音声ファイルのパス
    """
    if not os.path.exists(audio_file_path):
        print(f"エラー: 指定されたファイルが見つかりません。")
        print(f"パスを確認してください: {audio_file_path}")
        return

    print("--- 環境・モデル情報 ---")
    print(f"ライブラリ: funasr")
    
    # --- 1. モデルのロード ---
    # funasr の公式ドキュメントに沿ったモデルのロード方法
    model_dir = "iic/SenseVoiceSmall"
    print(f"モデル: {model_dir}")
    print("モデルをロードしています... (初回は時間がかかります)")
    
    try:
        model = AutoModel(
            model=model_dir,
            vad_model="fsmn-vad",
            vad_kwargs={"max_single_segment_time": 30000},
            device="cuda:0",
            # trust_remote_code=True は funasr のバージョンによっては不要な場合があります
        )
    except Exception as e:
        print(f"モデルのロード中にエラーが発生しました: {e}")
        return

    print("モデルのロードが完了しました。")
    print("\n音声の感情を判定しています...")
    
    # --- 2. 感情認識の実行 ---
    try:
        # ★★★ ここが重要 ★★★
        # task="ser" を指定して、感情認識タスクを実行する
        res = model.generate(
            input=audio_file_path,
            task="ser",  # Speech Emotion Recognition
        )
        
        # --- 3. 結果の表示 ---
        print("\n" + "="*30)
        print("  感情判定結果")
        print("="*30)
        
        # funasrのSERタスクの出力形式はリスト形式で返ってくることが多い
        if res and "emotion" in res[0]:
            emotion_label = res[0]["emotion"]
            print(f"  ファイル: {os.path.basename(audio_file_path)}")
            print(f"  判定された感情: {emotion_label}")
        else:
            print("  感情を判定できませんでした。")
            print("  モデルからの生データ:", res) # デバッグ用に生データを表示
        
        print("="*30)

    except Exception as e:
        print(f"\n推論中に予期せぬエラーが発生しました: {e}")


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="funasr と SenseVoiceモデルを使用して、音声ファイルの感情を分析します。"
    )
    parser.add_argument(
        "audio_file",
        type=str,
        help="分析したい音声ファイルのパス。"
    )
    
    args = parser.parse_args()
    
    analyze_emotion_with_funasr(args.audio_file)

これを以下で実行します

python .\emotion_recognition.py .\VOICEACTRESS100_090.wav

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

音声の感情を判定しています...
rtf_avg: 0.038: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  4.29it/s] 
rtf_avg: 0.046: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  3.54it/s] 
rtf_avg: 0.047, time_speech:  6.062, time_escape: 0.286: 100%|████████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00,  3.38it/s] 

==============================
  感情判定結果
==============================
  感情を判定できませんでした。
  モデルからの生データ: [{'key': 'VOICEACTRESS100_090', 'text': '<|ja|><|EMO_UNKNOWN|><|Speech|><|woitn|>戦闘服は両腕を露出し両足がアンダースーツで覆われている'}]
==============================

ニュートラルの場合は、上記のように判定をします。ただしほぼ?テキストから判定をしているみたいなのでボイス側のニュアンスはあまり反映されていない感じがしています

UTMOS-v2 による音質推定値(MOS)をGPUの同時並列で高速に処理を行う

開発環境

区分 バージョン / 詳細 備考
OS / イメージ Ubuntu 22.04 (JupyterLab コンテナ) uname -a で確認
Python 3.12.11 python --version
CUDA Driver / Runtime 12.x / 12.x nvidia-smi で確認
GPU 4 × NVIDIA T4 ID = 0 1 2 3
PyTorch 2.3.1 + cu12 pip show torch
transformers 4.42.1 pip show transformers
utmosv2 0.5.1 MOS 推論モデル
librosa 0.10.2 wav 読み込み
numpy 1.26.x 数値演算
tqdm 4.66.x 進捗バー

処理の実装

import argparse, json, os, time, multiprocessing as mp
from pathlib import Path

import soundfile as sf          # librosa より高速に長さ取得
import torch, utmosv2
from tqdm import tqdm

# ---------- helpers -------------------------------------------------
def get_duration(path: Path) -> float:
    """wav の長さ[sec] を高速に取得 (デコードしない)"""
    info = sf.info(str(path))
    return info.frames / info.samplerate


def update_json(json_path: Path, duration: float, mos: float):
    data = {}
    if json_path.exists():
        try:
            data = json.loads(json_path.read_text(encoding="utf-8"))
        except Exception:
            pass

    data.setdefault("parakeet_jp_transcription", "")
    data.setdefault("anime_whisper_transcription", "")
    data["duration"]  = duration
    data["speechMOS"] = round(float(mos), 6)

    json_path.write_text(json.dumps(data, ensure_ascii=False, indent=2),
                         encoding="utf-8")


# ---------- 1 GPU worker --------------------------------------------
def gpu_worker(rank: int, gpu_id: int, fp16: bool,
               wav_list: list[Path], skip_existing: bool):
    torch.cuda.set_device(gpu_id)
    model = utmosv2.create_model(pretrained=True, device=f"cuda:{gpu_id}")
    if fp16:
        model.half()
    model.eval()

    bar = tqdm(total=len(wav_list), position=rank, ascii=True,
               desc=f"[GPU{gpu_id}]", ncols=80, leave=False)

    for wav in wav_list:
        try:
            jpath = wav.with_suffix(".json")
            if skip_existing and jpath.exists():
                try:
                    if "speechMOS" in json.loads(jpath.read_text()):
                        bar.update()
                        continue
                except Exception:
                    pass

            dur  = get_duration(wav)
            mos  = model.predict(input_path=str(wav))
            update_json(jpath, dur, mos)
        except Exception as e:
            # 例外は表示だけして処理継続
            print(f"[GPU{gpu_id}] {wav} … {e}", flush=True)
        finally:
            bar.update()

    bar.close()


# ---------- main ----------------------------------------------------
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--root", required=True,
                    help="wav root ( .../speaker_id/wav/*.wav )")
    ap.add_argument("--gpus", nargs="+", type=int, default=[0],
                    help="GPU id list (e.g. 0 1 2 3)")
    ap.add_argument("--fp16", action="store_true",
                    help="load model in FP16 to save VRAM")
    ap.add_argument("--skip-existing", action="store_true",
                    help="skip wavs whose JSON already contains speechMOS")
    cfg = ap.parse_args()

    root = Path(cfg.root).expanduser()
    wav_files = list(root.glob("*/wav/*.wav"))
    if not wav_files:
        print("💡 wav が見つかりません。ROOT パスを確認してください。")
        return

    ngpu = len(cfg.gpus)
    print(f"Processing {len(wav_files):,} wav files on {ngpu} GPU(s) …")

    # round-robin でファイルを各 GPU に振り分け
    buckets = [[] for _ in range(ngpu)]
    for idx, w in enumerate(wav_files):
        buckets[idx % ngpu].append(w)

    procs = []
    for rank, (gpu_id, lst) in enumerate(zip(cfg.gpus, buckets)):
        p = mp.Process(target=gpu_worker,
                       args=(rank, gpu_id, cfg.fp16, lst, cfg.skip_existing),
                       daemon=False)
        p.start()
        procs.append(p)

    for p in procs:
        p.join()


if __name__ == "__main__":
    # safetensors 強制 ―― torch.load 脆弱性 CVE-2025-32434 対策
    os.environ["TRANSFORMERS_USE_SAFETENSORS"] = "1"
    from transformers import AutoModel
    _orig = AutoModel.from_pretrained
    AutoModel.from_pretrained = \
        lambda name,*a,**k: _orig(name, *a, use_safetensors=True, **k)

    t0 = time.time()
    main()
    print(f"Done! Elapsed {time.time()-t0:.1f} s")

実行方法

python make_mos_jsons.py \
  --root /data/moe-speech-plus/data \
  --gpus 0 1 2 3 \
  --fp16 

実行時間は 30万ファイルで1-2分程度で終わりました

実装メモ

safetensors 強制ロード

from transformers import AutoModel
_orig = AutoModel.from_pretrained
AutoModel.from_pretrained = lambda n,*a,**k: _orig(n, *a, use_safetensors=True, **k)
os.environ["TRANSFORMERS_USE_SAFETENSORS"] = "1"

1 GPU = 1 Processで処理をする

p = mp.Process(target=gpu_worker,
               args=(rank, gpu_id, cfg.fp16, bucket),
               daemon=False)

子プロセス内で DataLoader を使わない ため “daemon から子生成禁止” エラーを完全排除。 GPU 固定・FP16 で推論 ➜ 約 15 s / 512 kB WAV。