Spaces:
Running on Zero
Running on Zero
File size: 6,623 Bytes
f1ef7e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """
Transcribes call_1_mixed.wav (stereo: ch0=agent, ch1=customer).
Pipeline
--------
1. Load stereo with soundfile — no ffmpeg, no memory thrashing
2. Mono mix → faster-whisper base/int8 with word_timestamps=True + Silero VAD
3. Per-segment speaker attribution via channel energy (RMS ch0 vs ch1)
4. Write frontend/transcript_data.json in the format the dashboard expects
Why this works
--------------
- Isolated-channel files are 84% silence → Whisper timestamp drift
- Mono mix is only 42.5% silence → stable chunk-level VAD → accurate timestamps
- word_timestamps=True uses DTW in a single pass — no second model, no OOM
- Channel energy attribution is near-perfect because the stereo channels
are exact agent/customer isolations (correlation = 1.000 each)
"""
import os
import json
import time
import numpy as np
import soundfile as sf
from faster_whisper import WhisperModel
# ── Paths ────────────────────────────────────────────────────────────────────
ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
MIXED = os.path.join(ROOT, "data", "apptek", "call_1_mixed.wav")
OUT = os.path.join(ROOT, "frontend", "transcript_data.json")
SR = 16000
# ── Audio helpers ─────────────────────────────────────────────────────────────
def load_stereo(path):
data, sr = sf.read(path, dtype="float32")
assert sr == SR, f"Expected {SR} Hz, got {sr}"
assert data.ndim == 2 and data.shape[1] == 2, f"Expected stereo, got {data.shape}"
return data # (n_samples, 2) — ch0 = agent, ch1 = customer
def speaker_for_segment(stereo, start_s, end_s, pad_s=0.05):
"""
Returns 'agent' or 'customer' by comparing RMS energy of ch0 vs ch1
over [start_s - pad_s, end_s + pad_s]. Minimum window: 100 ms.
"""
s = max(0, int((start_s - pad_s) * SR))
e = min(len(stereo), int((end_s + pad_s) * SR))
if e - s < int(0.1 * SR): # guarantee at least 100 ms
mid = (s + e) // 2
half = int(0.05 * SR)
s, e = max(0, mid - half), min(len(stereo), mid + half)
chunk = stereo[s:e]
rms0 = np.sqrt(np.mean(chunk[:, 0] ** 2))
rms1 = np.sqrt(np.mean(chunk[:, 1] ** 2))
return "agent" if rms0 >= rms1 else "customer"
# ── Main ──────────────────────────────────────────────────────────────────────
def run(mixed_path=MIXED, out_path=OUT, model_size="base", call_id="call_1"):
# 1. Load audio
print("Loading stereo audio (soundfile — no ffmpeg)...")
stereo = load_stereo(mixed_path)
mono = stereo.mean(axis=1) # shape: (n_samples,)
duration = len(mono) / SR
print(f" {duration:.1f}s | ch0 = agent | ch1 = customer")
# 2. Load model
print(f"\nLoading faster-whisper {model_size}/int8...")
model = WhisperModel(model_size, device="cpu", compute_type="int8")
# 3. Transcribe mono mix
# initial_prompt primes Whisper's vocabulary with domain-specific terms,
# improving accuracy on proper nouns, numbers and banking terminology.
initial_prompt = (
"Banking customer service call. "
"Agent Emily at North Star Bank. Customer James Carter. "
"Topics: checking account, savings account, automatic transfer, "
"Canadian dollars, account number, date of birth, mobile app, "
"online banking, scheduled transfer, four hundred dollars."
)
print("Transcribing (word_timestamps=True, vad_filter=True, initial_prompt=True)...")
t0 = time.time()
seg_gen, info = model.transcribe(
mono,
language="en",
beam_size=5,
word_timestamps=True,
vad_filter=True,
vad_parameters={"min_silence_duration_ms": 500},
initial_prompt=initial_prompt,
)
segments = list(seg_gen)
elapsed = time.time() - t0
print(f" {len(segments)} segments | {elapsed:.1f}s "
f"| lang={info.language} p={info.language_probability:.2f}")
# 4. Build per-speaker word lists
agent_words = []
customer_words = []
no_ts = 0
for seg in segments:
if not seg.words:
continue
speaker = speaker_for_segment(stereo, seg.start, seg.end)
bucket = agent_words if speaker == "agent" else customer_words
for w in seg.words:
if w.start is None or w.end is None:
no_ts += 1
continue
bucket.append({
"word": w.word.strip(),
"start": round(w.start, 3),
"end": round(w.end, 3),
})
print(f"\n Agent words : {len(agent_words)}")
print(f" Customer words : {len(customer_words)}")
if no_ts:
print(f" Skipped (no timestamp): {no_ts}")
# 5. Spot-check a few lines
def fmt(words, n=10):
return " " + " ".join(w["word"] for w in words[:n])
print("\n--- Agent start ---")
print(fmt(agent_words))
print("\n--- Customer start ---")
print(fmt(customer_words))
# 6. Sanity check near the 4:48 mark (288 s)
target = 288.0
agent_set = set(id(w) for w in agent_words)
window = [w for w in (agent_words + customer_words)
if abs(w["start"] - target) < 15]
window.sort(key=lambda w: w["start"])
if window:
print("\n--- Words near 4:48 ---")
for w in window:
src = "AGT" if id(w) in agent_set else "CST"
print(f" [{src}] {w['start']:7.3f}s {w['word']}")
# 7. Save
payload = {
"model": f"faster-whisper {model_size}/int8 + stereo channel attribution",
"call": call_id,
"agent": agent_words,
"customer": customer_words,
}
os.makedirs(os.path.dirname(out_path), exist_ok=True)
with open(out_path, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
print(f"\nSaved -> {out_path}")
return payload
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--model", default="base", help="faster-whisper model size (tiny/base/small/medium)")
parser.add_argument("--call", default="call_1")
args = parser.parse_args()
run(model_size=args.model, call_id=args.call)
|