!pip install pyannote.audio
!pip freeze | grep -i pyannote

pyannote.audio==3.3.2
pyannote.core==5.0.0
pyannote.database==5.1.3
pyannote.metrics==3.2.1
pyannote.pipeline==3.0.1

!wget https://huggingface.co/kotoba-tech/kotoba-whisper-v2.2/resolve/main/sample_audio/sample_diarization_japanese.mp3

!ffmpeg -i sample_diarization_japanese.mp3 sample_diarization_japanese.wav

from IPython.display import Audio, display

display(Audio("sample_diarization_japanese.mp3", autoplay=True))

from pyannote.audio import Pipeline
from google.colab import userdata
import torch

pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token=userdata.get('HF_TOKEN')
)

pipeline.to(torch.device("cuda"))

diarization = pipeline("sample_diarization_japanese.mp3")

for turn, _, speaker in diarization.itertracks(yield_label=True):
    print(f"start={turn.start:.1f}s stop={turn.end:.1f}s speaker_{speaker}")

start=0.0s stop=13.7s speaker_SPEAKER_01
start=13.7s stop=18.3s speaker_SPEAKER_02
start=18.7s stop=21.9s speaker_SPEAKER_02
start=22.2s stop=24.9s speaker_SPEAKER_00

from pyannote.core import notebook, Segment

notebook.crop = Segment(0, 30)  # 可視化の範囲を0〜30秒に限定
diarization

with open("audio.sample_diarization_japanese.rttm", "w") as rttm:
    diarization.write_rttm(rttm)

from pyannote.database.util import load_rttm

past_result = load_rttm("sample_diarization_japanese.rttm")["sample_diarization_japanese"]
for turn, _, speaker in past_result.itertracks(yield_label=True):
    print(f"start={turn.start:.1f}s stop={turn.end:.1f}s speaker_{speaker}")
display(past_result)

from huggingface_hub import HfApi

available_pipelines = [p.modelId for p in HfApi().list_models(filter="pyannote-audio-pipeline")]
list(filter(lambda p: p.startswith("pyannote/"), available_pipelines))

['pyannote/speaker-diarization-3.1',
 'pyannote/speaker-diarization',
 'pyannote/overlapped-speech-detection',
 'pyannote/voice-activity-detection',
 'pyannote/speaker-diarization-3.0',
 'pyannote/speech-separation-ami-1.0',
 'pyannote/speaker-segmentation']

from pyannote.audio import Pipeline

pipeline = Pipeline.from_pretrained(
    "pyannote/voice-activity-detection",
    use_auth_token=userdata.get('HF_TOKEN')
)

output = pipeline("sample_diarization_japanese.wav")

for speech in output.get_timeline().support():
    print(speech)

[ 00:00:00.165 -->  00:00:09.953]
[ 00:00:10.307 -->  00:00:21.884]
[ 00:00:22.356 -->  00:00:25.022]

output

mkdir pyannote-audio-vad-work && cd pyannote-audio-vad-work

mise use python@3.12

cat << 'EOS' >> .mise.toml

[env]
_.python.venv = { path = ".venv", create = true }
EOS

pip install pyannote.audio sounddevice numpy torch

import numpy as np
import sounddevice as sd
from pyannote.audio import Model
from pyannote.core import SlidingWindow, SlidingWindowFeature
import torch
import os

class MicrophoneBuffer:
    def __init__(self, sample_rate=16000, duration=5.0, step=1.0):
        self.sample_rate = sample_rate
        self.duration = duration
        self.step = step
        self.buffer_size = int(sample_rate * duration)
        self.step_size = int(sample_rate * step)
        self.buffer = np.zeros(self.buffer_size, dtype=np.float32)

    def stream_audio(self):
        """Start streaming audio from the microphone."""
        # マイクデバイスの番号は以下で確認できる
        # import sounddevice as sd
        # print(sd.query_devices())
        with sd.InputStream(device=4, channels=1, samplerate=self.sample_rate, callback=self._audio_callback):
            print("Listening to the microphone. Press Ctrl+C to stop.")
            while True:
                yield self._get_current_buffer()

    def _audio_callback(self, indata, frames, time, status):
        """Callback to update the buffer."""
        if status:
            print(f"Audio input error: {status}")
        self.buffer = np.roll(self.buffer, -frames)
        self.buffer[-frames:] = indata[:, 0]

    def _get_current_buffer(self):
        """Return the current buffer as a SlidingWindowFeature."""
        resolution = SlidingWindow(start=0.0, duration=1.0 / self.sample_rate, step=1.0 / self.sample_rate)
        return SlidingWindowFeature(self.buffer.copy().reshape(-1, 1), resolution)

class VoiceActivityDetection:
    def __init__(self):
        print("Loading VAD model...")
        self.model = Model.from_pretrained(
            "pyannote/segmentation",
            use_auth_token=os.environ["HF_TOKEN"]
        )
        self.model.eval()

    def __call__(self, current_buffer: SlidingWindowFeature):
        with torch.no_grad():
            # numpy.ndarray を torch.Tensor に変換
            waveform_tensor = torch.from_numpy(current_buffer.data.T[np.newaxis]).float()

            # モデルに渡してセグメンテーション結果を取得
            segmentation = self.model(waveform_tensor).numpy()[0]

        # モデル出力から直接フレーム情報を取得
        num_frames = segmentation.shape[0]
        frame_duration = current_buffer.sliding_window.duration / num_frames

        resolution = SlidingWindow(
            start=current_buffer.sliding_window.start,
            duration=frame_duration,
            step=frame_duration
        )

        # 音声活動確率を計算
        speech_probability = np.max(segmentation, axis=-1, keepdims=True)
        return SlidingWindowFeature(speech_probability, resolution)


def main():
    sample_rate = 16000
    buffer_duration = 1.0
    buffer_step = 0.5

    # Initialize microphone buffer and VAD
    buffer = MicrophoneBuffer(sample_rate=sample_rate, duration=buffer_duration, step=buffer_step)
    vad = VoiceActivityDetection()

    # 状態を初期化
    previous_state = None

    try:
        for current_buffer in buffer.stream_audio():
            result = vad(current_buffer)

            # スピーチ確率の平均を計算
            speech_prob = result.data.mean()

            # 有音・無音の閾値を設定
            threshold = 0.9
            current_state = "Speech" if speech_prob > threshold else "Silence"

            # 状態が変化したときのみ出力
            if current_state != previous_state:
                print(f"Detected: {current_state}")
                previous_state = current_state
    except KeyboardInterrupt:
        print("\nStopped listening to the microphone.")


if __name__ == "__main__":
    main()

Loading VAD model...
(snip)
Listening to the microphone. Press Ctrl+C to stop.
Detected: Silence
Detected: Speech
Detected: Silence
Detected: Speech
Detected: Silence
Detected: Speech
Detected: Silence
Detected: Speech
Detected: Silence
Detected: Speech
Detected: Silence
Detected: Speech
Detected: Silence
^C
Stopped listening to the microphone.