# %pip install faster-whisper librosa soundfile pandas numpy matplotlibNotebook 1 — From audio to role-labeled transcripts
MDIG2026 Summer School · NLP track (1 / 5)
The Balance Corpus has rich audio but no speech transcripts — so before we can do any language analysis we have to make them. That is this notebook’s whole job, and it is a genuinely reusable skill: speech → text is the front door to almost all spoken-language NLP.
What we exploit
Each trial’s .wav is stereo, and the two channels are separate lavalier microphones — one per speaker (one lavalier per person, so the channels are near-independent). So we get speaker separation almost for free — no diarization, no GPU. The mics do bleed into each other a little, so each channel faintly picks up the other speaker too; we clean that up in §4 by keeping each utterance only on the channel where it is loudest.
The metadata never records which channel is which speaker (and mic wiring can differ between recording sessions), so in §2 we recover that mapping from the data. Combined with metadata.csv — which tells us who the clue-giver was on each trial — every transcript becomes role-labeled (clue_giver / guesser), which Notebooks 2–4 depend on.
Output
Per-trial CSVs in transcripts/, plus a combined transcripts_all.csv (one row per speech segment: who, which role, start, end, text).
Runtime. Everything here is CPU-friendly. To keep the first run fast we transcribe a small
MAX_TRIALSsubset; flip it toNoneto do all 80 (~25 min of audio).
0 · Install (run once)
Skip if these are already in your environment.
1 · Configuration & corpus location
# --- Make sure model downloads can verify TLS ---
# (Some Anaconda/Windows installs set SSL_CERT_FILE to a file that doesn't exist,
# which makes Hugging Face model downloads fail with a confusing error. Fix it.)
import os, certifi
if not os.path.exists(os.environ.get("SSL_CERT_FILE", "")):
os.environ["SSL_CERT_FILE"] = certifi.where()
os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
# --- Locate the corpus (works no matter where Jupyter is launched from) ---
def find_corpus(start=None):
"""Walk upward until we find the folder holding metadata.csv + audios/."""
d = os.path.abspath(start or os.getcwd())
for _ in range(8):
if os.path.isfile(os.path.join(d, "metadata.csv")) and os.path.isdir(os.path.join(d, "audios")):
return d
parent = os.path.dirname(d)
if parent == d:
break
d = parent
raise FileNotFoundError(
"Could not find the BalanceCorpus root (a folder with metadata.csv and audios/).\n"
"Open this notebook from inside the cloned repository."
)
CORPUS = find_corpus()
TRANSCRIPTS = os.path.join(CORPUS, "scripts", "nlp_taboo", "transcripts")
os.makedirs(TRANSCRIPTS, exist_ok=True)
print("Corpus root :", CORPUS)
print("Transcripts :", TRANSCRIPTS)import glob
import pandas as pd
# metadata.csv drives everything: one row per trial, with the target word,
# its five taboo words, who the clue-giver was, and the board/ground condition.
meta = pd.read_csv(os.path.join(CORPUS, "metadata.csv"), encoding="utf-8-sig")
meta["pair_id"] = meta["pair_id"].astype(str)
def parse_stem(path):
"""103_203_12_1_20250113_152455_doughnut_board.wav -> fields."""
p = os.path.basename(path).rsplit(".", 1)[0].split("_")
return dict(pair_id=f"{p[0]}_{p[1]}", trial_number=int(p[2]),
target_word=p[6], condition=p[7],
stem=os.path.basename(path).rsplit(".", 1)[0])
# Only two dyads currently have media on disk (80 trials); the join is exact.
wavs = sorted(glob.glob(os.path.join(CORPUS, "audios", "*", "*.wav")))
rows = []
for w in wavs:
r = parse_stem(w); r["audio_path"] = w
m = meta[(meta.pair_id == r["pair_id"]) & (meta.trial_number == r["trial_number"])]
if len(m) != 1:
continue
m = m.iloc[0]
r.update(clue_giver_id=int(m.clue_giver_id),
participant_1_id=int(m.participant_1_id),
participant_2_id=int(m.participant_2_id),
taboos=[str(m[f"taboo_{i}"]) for i in range(1, 6)])
rows.append(r)
trials = pd.DataFrame(rows)
print(f"{len(trials)} trials with audio | "
f"{trials.condition.value_counts().to_dict()} | "
f"pairs: {sorted(trials.pair_id.unique())}")
trials.head(3)2 · Recover the microphone → speaker mapping from the data
Which stereo channel is which speaker is not in the metadata, so we recover it. The idea: on any trial the clue-giver does most of the talking, and metadata tells us who the clue-giver was. For each pair we therefore pick the channel↔︎participant assignment that best makes the clue-giver the louder channel — aggregated over all 40 of the pair’s trials, so a single noisy trial can’t fool us. (Doing this per pair means it stays correct even if a future session was wired the other way round.)
import wave
import numpy as np
def channels(path):
"""Split a stereo wav into (left, right, samplerate)."""
w = wave.open(path, "rb"); sr = w.getframerate()
raw = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16).astype(np.float64)
return raw[0::2], raw[1::2], sr
def voiced_energy(x, sr, frame=0.05, pct=60):
"""Mean energy of the louder (voiced) frames — ignores silence between words."""
n = max(int(sr * frame), 1)
fr = x[: len(x) // n * n].reshape(-1, n)
e = np.sqrt((fr ** 2).mean(axis=1))
keep = e >= np.percentile(e, pct)
return float(e[keep].mean()) if keep.any() else float(e.mean())
def calibrate(pair_df):
"""Return {'L': 'participant_1'|'participant_2', 'R': ...} for one pair, + confidence."""
louder_is_left, clue_giver_is_p1 = [], []
for _, r in pair_df.iterrows():
L, R, sr = channels(r.audio_path)
louder_is_left.append(voiced_energy(L, sr) > voiced_energy(R, sr))
clue_giver_is_p1.append(r.clue_giver_id == r.participant_1_id)
louder_is_left = np.array(louder_is_left); clue_giver_is_p1 = np.array(clue_giver_is_p1)
# If Left==participant_1, the clue-giver (louder) is on the left exactly when cg is p1:
agree_L_is_p1 = np.mean(louder_is_left == clue_giver_is_p1)
if agree_L_is_p1 >= 0.5:
return {"L": "participant_1", "R": "participant_2"}, agree_L_is_p1
return {"L": "participant_2", "R": "participant_1"}, 1 - agree_L_is_p1
# Calibrate every pair and remember the mapping
pair_map = {}
for pair, g in trials.groupby("pair_id"):
mapping, conf = calibrate(g)
pair_map[pair] = mapping
print(f"{pair}: left = {mapping['L']}, right = {mapping['R']} "
f"(clue-giver was the louder channel on {conf:.0%} of trials)")
print("\nWe use this recovered mapping to attach a speaker id to each channel; the "
"clue-giver/guesser *role* then comes straight from metadata.")3 · Load the ASR model
We use faster-whisper (a fast CPU re-implementation of OpenAI Whisper). small + int8 is a good accuracy/speed trade-off on a laptop. The game is played in English, but with many non-native speakers — Whisper is multilingual and robust to accents, so we set language="en" (change to None to auto-detect per channel).
try:
from faster_whisper import WhisperModel
except ModuleNotFoundError:
raise ModuleNotFoundError("faster-whisper is not installed — run the install cell in "
"Section 0, then re-run this cell.")
MODEL_SIZE = "small" # "tiny"/"base" = faster & rougher, "small" = recommended
LANGUAGE = "en" # None -> auto-detect each channel
asr = WhisperModel(MODEL_SIZE, device="cpu", compute_type="int8")
print(f"Loaded faster-whisper '{MODEL_SIZE}' on CPU.")4 · Transcribe a trial — and remove microphone crosstalk
We transcribe both channels, then for every utterance compare its loudness on the two channels and keep it only on the louder one. That drops the faint bleed of the other speaker (which Whisper otherwise happily transcribes on both channels), so each utterance is attributed to exactly one speaker. This matters a lot for Notebook 3, where we compare the two speakers — without it, the same words would appear on both sides and fake “convergence”.
import librosa
def load_stereo_16k(path):
"""Both channels as float32 arrays at 16 kHz (what Whisper wants)."""
y, _ = librosa.load(path, sr=16000, mono=False) # (2, n) stereo, or (n,) if mono
if y.ndim == 1: # a mono file -> one speaker on both
y = np.vstack([y, y])
return [np.ascontiguousarray(y[c], dtype=np.float32) for c in (0, 1)]
def _rms(a, s, e, sr=16000):
seg = a[int(s * sr):int(e * sr)]
return float(np.sqrt(np.mean(seg ** 2))) if len(seg) else 0.0
def transcribe_trial(path):
"""Return {0: [segs], 1: [segs]}, language — with cross-channel bleed removed."""
chans = load_stereo_16k(path)
kept, lang = {0: [], 1: []}, None
for ch in (0, 1):
segs, info = asr.transcribe(chans[ch], language=LANGUAGE, vad_filter=True, beam_size=1)
lang = info.language
for s in segs:
if not s.text.strip():
continue
# keep the utterance only where it is loudest (this speaker's own mic)
if _rms(chans[ch], s.start, s.end) >= _rms(chans[1 - ch], s.start, s.end):
kept[ch].append(dict(start=round(s.start, 2), end=round(s.end, 2),
text=s.text.strip()))
return kept, lang
def channel_of(trial_row, want_role):
"""Which channel (0=L, 1=R) carries the clue_giver / guesser on this trial?"""
m = pair_map[trial_row.pair_id]
for ch, side in [(0, "L"), (1, "R")]:
who = m[side]
pid = trial_row.participant_1_id if who == "participant_1" else trial_row.participant_2_id
if ("clue_giver" if pid == trial_row.clue_giver_id else "guesser") == want_role:
return ch
return 0
# Demo on the first trial: the clue-giver channel after crosstalk removal
r0 = trials.iloc[0]
segs0, lang = transcribe_trial(r0.audio_path)
print(f"{r0.stem} (target = '{r0.target_word}', detected lang = {lang})\n"
f"clue-giver channel, bleed removed:\n")
for s in segs0[channel_of(r0, "clue_giver")]:
print(f"[{s['start']:5.1f}-{s['end']:5.1f}] {s['text']}")5 · Transcribe the whole corpus → transcripts/
For every trial we transcribe both channels, attach the speaker id and their role (clue_giver / guesser) from metadata, and save one CSV per trial plus a combined table. Already-transcribed trials are skipped, so you can stop and resume.
MAX_TRIALS = 8 # <-- set to None to transcribe all 80 trials
subset = trials if MAX_TRIALS is None else trials.head(MAX_TRIALS)
all_rows = []
COLS = ["trial_id", "pair_id", "trial_number", "condition", "target_word",
"participant_id", "role", "channel", "seg", "start", "end", "text", "language"]
for i, r in subset.reset_index(drop=True).iterrows():
out_csv = os.path.join(TRANSCRIPTS, r.stem + ".csv")
if os.path.exists(out_csv):
all_rows.append(pd.read_csv(out_csv) if os.path.getsize(out_csv) else pd.DataFrame(columns=COLS))
print(f"[{i+1}/{len(subset)}] cached {r.stem}");
continue
print(f"[{i+1}/{len(subset)}] {r.stem} ...", end=" ", flush=True)
rows = []
m = pair_map[r.pair_id]
segs, lang = transcribe_trial(r.audio_path) # both channels, bleed removed
for ch, side in [(0, "L"), (1, "R")]:
who = m[side] # recovered in section 2
pid = r.participant_1_id if who == "participant_1" else r.participant_2_id
role = "clue_giver" if pid == r.clue_giver_id else "guesser"
for k, s in enumerate(segs[ch]):
rows.append(dict(trial_id=r.stem, pair_id=r.pair_id, trial_number=r.trial_number,
condition=r.condition, target_word=r.target_word,
participant_id=pid, role=role, channel="LR"[ch],
seg=k, start=s["start"], end=s["end"], text=s["text"],
language=lang))
df = pd.DataFrame(rows, columns=COLS) # explicit columns -> header even if empty
df.to_csv(out_csv, index=False, encoding="utf-8")
all_rows.append(df)
print(f"{len(df)} segments")
nonempty = [d for d in all_rows if len(d)] # skip trials with no speech
combined = pd.concat(nonempty, ignore_index=True) if nonempty else pd.DataFrame(columns=COLS)
combined.to_csv(os.path.join(TRANSCRIPTS, "transcripts_all.csv"), index=False, encoding="utf-8")
print(f"\nSaved {len(combined)} segments across {combined.trial_id.nunique()} trials "
f"-> transcripts/transcripts_all.csv")6 · Sanity check
import matplotlib.pyplot as plt
words = (combined.assign(nwords=combined.text.str.split().str.len())
.groupby("role").nwords.sum())
print("Words spoken by role:", words.to_dict())
fig, ax = plt.subplots(figsize=(6, 3))
(combined.assign(nwords=combined.text.str.split().str.len())
.groupby(["role"]).nwords.sum().plot.bar(ax=ax, color=["#c44", "#48a"]))
ax.set_ylabel("total words"); ax.set_title("Clue-givers talk more than guessers")
plt.tight_layout(); plt.show()Recap & handoff
- We turned raw stereo audio into role-labeled, timestamped transcripts — no GPU, no diarization.
- The trick was recognizing (and verifying) that each stereo channel is one speaker.
transcripts/transcripts_all.csvis now the shared input for:- Notebook 2 — Semantic homing: how the clue-giver’s words close in on the hidden target.
- Notebook 3 — Lexical alignment: how the two partners’ language converges.
Run this on all 80 trials (
MAX_TRIALS = None) before Notebooks 2–3 for the full picture.