Processing: Speaker diarization

The balance data consist of interaction between two people: a cluegiver and a guesser. This means that audio records both of them. For some analysis, we might want/need to separate them. For instance:

• in text transcript, we will need to know what words can be attributed to whom;
• for acoustic envelope, we need to separate the two voices.

In this script, we perform so-called diarization, which is the process of separating the audio into segments and attributing each segment to a speaker. As an output, we have a csv file for each wav file that tells us timestamps and the speaker.

The code is adapted from Marianne de Heer Kloots’ tutorial: https://www.envisionbox.org/embedded_Speaker_diarization_pyannote.html

# packages
import os
import glob
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy.signal import butter, filtfilt, sosfilt
import librosa
import IPython.display as ipd
import seaborn as sns
from pyannote.audio import Pipeline
from pyannote.core.notebook import Notebook
import torch
from IPython.display import Audio
import soundfile as sf
notebook = Notebook()

curfolder = os.getcwd()
datasetfolder = os.path.join(curfolder, '..', '..')

# BalanceCorpus
audiofolder = os.path.join(datasetfolder, 'audios', '*') # last level for participant folder
wavtotrack = glob.glob(os.path.join(audiofolder, "*.wav"))
balancemeta = pd.read_csv(os.path.join(datasetfolder, 'metadata.csv'))
balancedemo = pd.read_csv(os.path.join(datasetfolder, 'demographics.csv'))
print('Number of audio files found: ', len(wavtotrack))


# Create folder to save processed acoustic time series
ACfolder_processed = os.path.join(curfolder, '..', '..', 'TS_acoustics')
if not os.path.exists(ACfolder_processed):
    os.makedirs(ACfolder_processed)

Number of audio files found:  120

def audio_player(audio_file):
  return ipd.display(Audio(audio_file))

from huggingface_hub import login
login()

Example on one file

for wav in wavtotrack:
    
    basename = os.path.basename(wav)
    print("working on ", basename)
    
    audio_path = wav

    audio, sr = sf.read(audio_path, dtype="float32", always_2d=True)
    waveform = torch.from_numpy(audio.T)  # (channel, time)

    pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-community-1" # You need to get your own token at HuggingFaces 
    )
    # add: ,token="XXXXXXX"

    # send pipeline to GPU if available
    device = torch.device("cuda:0") if torch.cuda.is_available() else torch.device("cpu")
    pipeline.to(device)

    output = pipeline(
        {"waveform": waveform, "sample_rate": sr},
        min_speakers=1, max_speakers=2
    )

    diarization = output.speaker_diarization  # <- this is the Annotation

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

    # plot timeline
    fig, ax = plt.subplots(figsize=(20,2))
    notebook.plot_annotation(diarization, ax=ax)
    ax.set_xlim(0, turn.end)
    plt.show()

    # play the recording
    audio_player(audio_path)

    rows = []

    # Create csv documentation
    for turn, _, speaker in diarization.itertracks(yield_label=True):
        rows.append({
            "speaker": speaker,
            "start": round(turn.start, 3),
            "end": round(turn.end, 3),
            "duration": round(turn.end - turn.start, 3)
        })

    diar_df = pd.DataFrame(rows)
    diarname = f"{basename}_diarization.csv"
    diar_df.to_csv(os.path.join("diarization", diarname), index=False)
    diar_df.head(10)

    break

working on  103_203_12_1_20250113_152455_doughnut_board.wav

start=1.3s stop=4.6s speaker=SPEAKER_00
start=5.7s stop=10.0s speaker=SPEAKER_00
start=7.7s stop=7.8s speaker=SPEAKER_01
start=11.2s stop=13.9s speaker=SPEAKER_00
start=14.7s stop=16.0s speaker=SPEAKER_00
start=15.6s stop=15.7s speaker=SPEAKER_01
start=16.2s stop=16.6s speaker=SPEAKER_00

Full loop

for wav in wavtotrack:
    
    if '116_216' not in wav:
        continue
    
    basename = os.path.basename(wav)
    print("working on ", basename)
    
    audio_path = wav

    audio, sr = sf.read(audio_path, dtype="float32", always_2d=True)
    waveform = torch.from_numpy(audio.T)  # (channel, time)

    pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-community-1" # You need to get your own token at HuggingFaces 
    )
    # add: ,token="XXXXXXX"

    # send pipeline to GPU if available
    device = torch.device("cuda:0") if torch.cuda.is_available() else torch.device("cpu")
    pipeline.to(device)

    output = pipeline(
        {"waveform": waveform, "sample_rate": sr},
        min_speakers=1, max_speakers=2
    )

    diarization = output.speaker_diarization  # <- this is the Annotation

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

    rows = []

    # Create csv documentation
    for turn, _, speaker in diarization.itertracks(yield_label=True):
        rows.append({
            "speaker": speaker,
            "start": round(turn.start, 3),
            "end": round(turn.end, 3),
            "duration": round(turn.end - turn.start, 3)
        })

    diar_df = pd.DataFrame(rows)
    diarname = f"{basename}_diarization.csv"
    diar_df.to_csv(os.path.join("diarization", diarname), index=False)