【AI Shift Advent Calendar 2024】VAD・VAPを用いた発話終了検知

CEJC/
├── 00readme.html
├── data
├── metaInfo
├── rdb
└── tool

CEJC/data/C001/C001_001
├── C001_001-luu.TextGrid
├── C001_001-luu.csv
├── C001_001-luu.eaf
├── C001_001-morphLUW.csv
├── C001_001-morphSUW.csv
├── C001_001-transUnit.TextGrid
├── C001_001-transUnit.csv
├── C001_001-transUnit.eaf
├── C001_001_IC01.wav
├── C001_001_IC02.wav
├── C001_001_IC03.wav
├── C001_001_IC04.wav
├── C001_001_IC05.wav
└── C001_001_IC0A.wav

python train.py \
   --data_train_path train.csv \
   --data_val_path valid.csv \
   --data_test_path test.csv \
   --vap_encoder_type cpc \
   --vap_cpc_model_pt ../asset/cpc/60k_epoch4-d0f474de.pt \
   --vap_freeze_encoder 1 \
   --vap_channel_layers 1 \
   --vap_cross_layers 3 \
   --vap_context_limit -1 \
   --vap_context_limit_cpc_sec -1 \
   --vap_frame_hz 5 \
   --event_frame_hz 5 \
   --opt_early_stopping 0 \
   --opt_save_top_k 5 \
   --opt_max_epochs 25 \
   --opt_saved_dir ./trained_model/ \
   --data_batch_size 8 \
   --devices "0,1,2,3" \
   --seed 0

from rvap.vap_main.vap_main import VAPRealTime
from pathlib import Path
import matplotlib.pyplot as plt
from matplotlib.gridspec import GridSpec
import soundfile as sf
import numpy as np

def process_audio_file_with_plot(single_channel_audio_path, score_threshold=0.5):
    device = 'cpu'
    vap = VAPRealTime(vap_model, cpc_model, device, frame_rate, context_len_sec)
    
    # 音声データの読み込み
    audio_data, sample_rate = sf.read(single_channel_audio_path)
    print(audio_data.shape, audio_data.ndim)
    if audio_data.ndim == 2 and audio_data.shape[1] != 1:
        import librosa
        audio_data = librosa.to_mono(audio_data.T)
    
    assert sample_rate == 16000, "The sample rate of the input audio file must be 16kHz."

    x1 = audio_data
    x2 = np.zeros_like(x1)
    
    # 結果を格納するリスト
    p_now_list = []
    p_future_list = []
    turn_now_list = []    # p_nowに基づくターン
    turn_future_list = [] # p_futureに基づくターン
    frame_indices = []
    
    vap_processor = VAPRealTime(vap_model, cpc_model, device, frame_rate, context_len_sec)
    
    current_x1 = np.zeros(vap_processor.frame_contxt_padding)
    current_x2 = np.zeros(vap_processor.frame_contxt_padding)
    
    FRAME_SIZE_INPUT = 160

    # プロット用のfigureを作成
    fig = plt.figure(figsize=(15, 10))
    gs = GridSpec(3, 1, height_ratios=[1, 1, 0.5])

    # 波形プロット用のサブプロット
    ax1 = fig.add_subplot(gs[0])
    ax1.grid(True)  # グリッドを表示
    # VAP結果プロット用のサブプロット
    ax2 = fig.add_subplot(gs[1])
    ax2.grid(True)  # グリッドを表示
    # ターン表示用のサブプロット
    ax3 = fig.add_subplot(gs[2])
    ax3.grid(True)  # グリッドを表示

    for i in range(0, len(x1), FRAME_SIZE_INPUT):
        x1_frame = x1[i:i + FRAME_SIZE_INPUT]
        x2_frame = x2[i:i + FRAME_SIZE_INPUT]
        
        if len(x1_frame) < FRAME_SIZE_INPUT:
            break
        
        current_x1 = np.concatenate([current_x1, x1_frame])
        current_x2 = np.concatenate([current_x2, x2_frame])
        
        if len(current_x1) < vap_processor.audio_frame_size:
            continue
        
        vap_processor.process_vap(current_x1, current_x2)
        
        # 結果を保存
        frame_indices.append(i / sample_rate)  # 時間（秒）に変換
        p_now_list.append(vap_processor.result_p_now[0])
        p_future_list.append(vap_processor.result_p_future[0])
        
        # ターン情報を決定 (閾値を0.5とする)
        turn_now = 1 if vap_processor.result_p_now[0] > score_threshold else 0
        turn_future = 1 if vap_processor.result_p_future[0] > score_threshold else 0
        turn_now_list.append(turn_now)
        turn_future_list.append(turn_future)
        
        current_x1 = current_x1[-vap_processor.frame_contxt_padding:]
        current_x2 = current_x2[-vap_processor.frame_contxt_padding:]

    # 波形のプロット
    time = np.arange(len(x1)) / sample_rate
    ax1.plot(time, x1)
    ax1.set_title('Audio Waveform')
    ax1.set_ylabel('Amplitude')
    ax1.set_xticklabels([])  # x軸のラベルを非表示

    # VAP結果のプロット
    ax2.plot(frame_indices, p_now_list, label='p_now', color='blue')
    ax2.plot(frame_indices, p_future_list, label='p_future', color='red')
    ax2.set_title('Voice Activity Projection (VAP) Result')
    ax2.set_ylabel('Probability')
    ax2.legend()
    ax2.grid(True)
    ax2.set_ylim(0, 1)
    ax2.set_xticklabels([])  # x軸のラベルを非表示

    # ターンの表示
    ax3.plot(frame_indices, turn_now_list, label='Turn (p_now)', color='blue')
    ax3.plot(frame_indices, turn_future_list, label='Turn (p_future)', color='red', linestyle='--')
    ax3.set_ylim(-0.5, 1.5)
    ax3.set_yticks([0, 1])
    ax3.set_yticklabels(['Bot', 'User'])
    ax3.set_title('Turn State (Score Threshold: {})'.format(score_threshold))
    
    ax3.set_xlabel('Time (s)')
    ax3.legend()
    ax3.grid(True)

    # x軸の範囲を統一
    xlim = (0, time[-1])
    ax1.set_xlim(xlim)
    ax2.set_xlim(xlim)
    ax3.set_xlim(xlim)

    plt.tight_layout()
    plt.show()

from silero_vad import load_silero_vad
import matplotlib.pyplot as plt
from matplotlib.gridspec import GridSpec
import numpy as np
import soundfile as sf
import torch

class AudioBuffer:
    def __init__(self, window_size_samples, sample_rate):
        self.window_size_samples = window_size_samples
        self.buffer = torch.zeros(window_size_samples)
        self.sample_rate = sample_rate

    def update(self, new_chunk):
        # バッファを左にシフトして新しいチャンクを追加
        shift_size = len(new_chunk)
        self.buffer = torch.cat([self.buffer[shift_size:], new_chunk])
        return self.buffer

def plot_realtime_vad_analysis(wav_path, 
                              threshold=0.5,
                              chunk_size_ms=20,
                              silence_threshold_ms=500):
    
    # モデルのロード
    model = load_silero_vad()
    
    # 音声データの読み込み
    audio_data, sample_rate = sf.read(wav_path)
    wav = torch.from_numpy(audio_data).float()
    
    # パラメータ設定
    window_size_samples = 512 if sample_rate == 16000 else 256
    chunk_size_samples = int(chunk_size_ms * sample_rate / 1000)
    silence_threshold_samples = int(silence_threshold_ms * sample_rate / 1000)
    
    # バッファの初期化
    buffer = AudioBuffer(window_size_samples, sample_rate)
    
    # 結果格納用の配列
    vad_results = np.zeros(len(audio_data))
    turn_states = np.zeros(len(audio_data))
    
    # 状態の追跡用変数
    is_speaking = False
    silence_counter = 0
    current_turn = 0
    
    # chunk_size_samplesごとの処理
    for i in range(0, len(audio_data), chunk_size_samples):
        # 現在のチャンクを取得
        chunk_end = min(i + chunk_size_samples, len(audio_data))
        current_chunk = wav[i:chunk_end]
        
        # チャンクが小さい場合はパディング
        if len(current_chunk) < chunk_size_samples:
            padding = torch.zeros(chunk_size_samples - len(current_chunk))
            current_chunk = torch.cat([current_chunk, padding])
        
        # バッファを更新
        current_window = buffer.update(current_chunk)
        
        # VADの判定
        speech_prob = model(current_window, sample_rate).item()
        is_speech = speech_prob >= threshold
        
        # VAD結果の保存
        vad_results[i:chunk_end] = 1 if is_speech else 0
        
        # ターン状態の更新
        if is_speech:
            is_speaking = True
            silence_counter = 0
            current_turn = 1
        else:
            if is_speaking:
                silence_counter += chunk_size_samples
                if silence_counter >= silence_threshold_samples:
                    is_speaking = False
                    current_turn = 0
        
        # ターン状態の保存
        turn_states[i:chunk_end] = current_turn
    
    # プロット
    fig = plt.figure(figsize=(15, 10))
    gs = GridSpec(3, 1, height_ratios=[1, 1, 0.5])

    # 波形プロット
    ax1 = fig.add_subplot(gs[0])
    time = np.arange(len(audio_data)) / sample_rate
    ax1.plot(time, audio_data)
    ax1.set_title('Audio Waveform')
    ax1.set_ylabel('Amplitude')
    ax1.set_xticklabels([])
    ax1.grid(True)

    # VAD結果プロット
    ax2 = fig.add_subplot(gs[1])
    ax2.plot(time, vad_results)
    ax2.set_ylim(-0.5, 1.5)
    ax2.set_yticks([0, 1])
    ax2.set_title(f'Voice Activity Detection (VAD) Result')
    ax2.set_yticklabels(['Silent', 'Speech'])
    ax2.grid(True)

    # ターン状態プロット
    ax3 = fig.add_subplot(gs[2])
    ax3.plot(time, turn_states, 'r-', label='Turn')
    ax3.set_ylim(-0.5, 1.5)
    ax3.set_yticks([0, 1])
    ax3.set_title(f'Turn State (Silence Threshold: {silence_threshold_ms}ms)')
    ax3.set_yticklabels(['Bot', 'User'])
    ax3.set_xlabel('Time (s)')
    ax3.grid(True)
    ax3.legend()

    # x軸の範囲を統一
    xlim = (0, time[-1])
    ax1.set_xlim(xlim)
    ax2.set_xlim(xlim)
    ax3.set_xlim(xlim)

    plt.tight_layout()
    plt.show()