1#!/usr/bin/env python3
2"""Transcribe a session's mic audio into a Markdown transcript with speakers.
3
4 python session_transcript.py <audio> <markers.json|none> <out.md> <title>
5
6Each segment is matched against the enrolled voice library (You + named guests);
7unmatched voices are clustered into distinct Speaker 2/3/… Writes:
8 - <out.md> meeting-minutes transcript, paragraphs labelled by speaker
9 - transcript.json segments + labels (for fast re-labelling after naming)
10 - speakers.json detected speakers + a sample clip + centroid (for the UI)
11"""
12import json
13import os
14import subprocess
15import sys
16
17import numpy as np
18import mlx_whisper
19
20sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
21import speaker_id as sid
22from transcript_render import PAUSE_SPLIT, render
23
24DIARIZE_PY = os.path.expanduser("~/.clover-diarize/.venv/bin/python")
25
26
27def assign_speakers(segments, data, sr):
28 """Cluster segments into voices, then match each *cluster* to the library —
29 far more robust than per-segment matching, so one person stays one speaker.
30
31 Returns (labels, speakers): labels[i] is the speaker for segment i; speakers
32 is per-cluster metadata for the review UI."""
33 from scipy.cluster.hierarchy import fcluster, linkage
34
35 library = [(v["name"], np.asarray(v["centroid"], dtype=np.float32)) for v in sid.load_library()]
36 embs = [sid.embed_segment(data, sr, float(s["start"]), float(s["end"])) for s in segments]
37 valid = [i for i, e in enumerate(embs) if e is not None]
38 if not valid:
39 return [None] * len(segments), []
40
41 # Agglomerative clustering on cosine distance (average linkage).
42 if len(valid) == 1:
43 cids = [1]
44 else:
45 Z = linkage(np.stack([embs[i] for i in valid]), method="average", metric="cosine")
46 cids = fcluster(Z, t=sid.CLUSTER_DIST, criterion="distance")
47 # Cap runaway fragmentation: if acoustic distance alone invents too many
48 # voices, collapse to at most MAX_SPEAKERS by cutting the tree higher.
49 if len(set(cids)) > sid.MAX_SPEAKERS:
50 cids = fcluster(Z, t=sid.MAX_SPEAKERS, criterion="maxclust")
51
52 members = {}
53 for pos, cid in enumerate(cids):
54 members.setdefault(int(cid), []).append(valid[pos])
55
56 # Label each cluster: known name if its centroid matches the library, else
57 # Speaker N (numbered by first appearance).
58 cluster_label, cluster_centroid, unknown_n = {}, {}, 2
59 for cid, idxs in sorted(members.items(), key=lambda kv: min(kv[1])):
60 cen = np.mean([embs[i] for i in idxs], axis=0)
61 cen /= np.linalg.norm(cen)
62 cluster_centroid[cid] = cen
63 name, sim = None, -1.0
64 for nm, c in library:
65 d = float(cen @ c)
66 if d > sim:
67 sim, name = d, nm
68 if sim >= sid.MATCH_THRESHOLD:
69 cluster_label[cid] = name
70 else:
71 cluster_label[cid] = f"Speaker {unknown_n}"
72 unknown_n += 1
73
74 labels = [None] * len(segments)
75 for cid, idxs in members.items():
76 for i in idxs:
77 labels[i] = cluster_label[cid]
78 prev = None # short (un-embedded) segments inherit the previous speaker
79 for i in range(len(labels)):
80 if labels[i] is None:
81 labels[i] = prev
82 else:
83 prev = labels[i]
84
85 speakers = []
86 for cid, idxs in sorted(members.items(), key=lambda kv: min(kv[1])):
87 lab = cluster_label[cid]
88 unknown = lab.startswith("Speaker ")
89 s = segments[idxs[0]]
90 speakers.append({
91 "label": lab, "unknown": unknown,
92 "sample": {"start": s["start"], "end": s["end"]},
93 "centroid": cluster_centroid[cid].tolist() if unknown else None,
94 })
95 return labels, speakers
96
97
98def run_diarization(audio):
99 """Precise speaker turns via the pyannote venv, or None if unavailable."""
100 if not os.path.exists(DIARIZE_PY):
101 return None
102 script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "diarize.py")
103 try:
104 out = subprocess.run([DIARIZE_PY, script, audio], capture_output=True, text=True, timeout=3600)
105 turns = json.loads(out.stdout.strip().splitlines()[-1])
106 return turns or None
107 except Exception:
108 return None
109
110
111def name_speakers(turns, data, sr):
112 """Name each diarized speaker against the voice library (ECAPA)."""
113 library = [(v["name"], np.asarray(v["centroid"], dtype=np.float32)) for v in sid.load_library()]
114 by_spk, sample = {}, {}
115 for tr in turns:
116 e = sid.embed_segment(data, sr, tr["start"], tr["end"])
117 if e is not None:
118 by_spk.setdefault(tr["speaker"], []).append(e)
119 sample.setdefault(tr["speaker"], tr)
120
121 names, meta, unknown_n = {}, [], 2
122 for spk in sorted(sample, key=lambda s: sample[s]["start"]):
123 cen = None
124 if by_spk.get(spk):
125 cen = np.mean(by_spk[spk], axis=0)
126 cen /= np.linalg.norm(cen)
127 name, sim = None, -1.0
128 if cen is not None:
129 for nm, c in library:
130 d = float(cen @ c)
131 if d > sim:
132 sim, name = d, nm
133 if cen is not None and sim >= sid.MATCH_THRESHOLD:
134 names[spk], unknown = name, False
135 else:
136 names[spk], unknown = f"Speaker {unknown_n}", True
137 unknown_n += 1
138 meta.append({
139 "label": names[spk], "unknown": unknown,
140 "sample": {"start": sample[spk]["start"], "end": sample[spk]["end"]},
141 "centroid": cen.tolist() if (unknown and cen is not None) else None,
142 })
143 return names, meta
144
145
146def _speaker_at(turns, t):
147 for tr in turns:
148 if tr["start"] <= t < tr["end"]:
149 return tr["speaker"]
150 return min(turns, key=lambda tr: min(abs(tr["start"] - t), abs(tr["end"] - t)))["speaker"]
151
152
153def build_from_diarization(segments, turns, data, sr):
154 """Split forced-aligned words at diarization boundaries (precise) and label
155 each by the named speaker. Returns (final_segments, labels, speakers)."""
156 names, meta = name_speakers(turns, data, sr)
157 words = [dict(w) for s in segments for w in s.get("words", [])]
158 if not words:
159 return None
160
161 final = []
162 for w in words:
163 name = names.get(_speaker_at(turns, (float(w["start"]) + float(w["end"])) / 2))
164 if (final and final[-1]["label"] == name
165 and float(w["start"]) - final[-1]["end"] <= PAUSE_SPLIT):
166 final[-1]["end"] = float(w["end"])
167 final[-1]["text"] += " " + w["word"]
168 final[-1]["words"].append(w)
169 else:
170 final.append({
171 "start": float(w["start"]), "end": float(w["end"]),
172 "text": w["word"], "label": name, "words": [w],
173 })
174 return final, [s["label"] for s in final], meta
175
176
177def main():
178 audio, markers_path, out_path, title = sys.argv[1:5]
179 # "solo" (default): one speaker, no diarization — right for journaling and
180 # improv, where clustering just shatters your voice into fake speakers.
181 # "multi": diarize + name, for the occasional session with other people.
182 mode = sys.argv[5] if len(sys.argv) > 5 else "solo"
183 folder = os.path.dirname(out_path)
184
185 segments = mlx_whisper.transcribe(
186 audio, path_or_hf_repo="mlx-community/whisper-large-v3-turbo"
187 ).get("segments", [])
188 segments = [{"start": s["start"], "end": s["end"], "text": s["text"]} for s in segments]
189
190 data, sr = sid.load_audio(audio)
191 try:
192 import forced_align as fa
193
194 for s in segments:
195 s["words"] = fa.align_segment(data, sr, float(s["start"]), float(s["end"]), s["text"])
196 except Exception:
197 for s in segments:
198 s["words"] = []
199
200 if mode != "multi":
201 # Solo: everyone is "You". render() hides the single label, and the
202 # non-unknown speaker keeps the Speakers-review popup from firing.
203 labels = ["You"] * len(segments)
204 final_segments = [dict(s, label="You") for s in segments]
205 speakers = [{"label": "You", "unknown": False,
206 "sample": {"start": 0.0, "end": 0.0}, "centroid": None}]
207 else:
208 # Precise mode (pyannote) when available, else cluster-then-match fallback.
209 diarized = None
210 turns = run_diarization(audio)
211 if turns:
212 diarized = build_from_diarization(segments, turns, data, sr)
213 if diarized:
214 final_segments, labels, speakers = diarized
215 else:
216 try:
217 labels, speakers = assign_speakers(segments, data, sr)
218 except Exception:
219 labels, speakers = [None] * len(segments), []
220 final_segments = [dict(s, label=l) for s, l in zip(segments, labels)]
221
222 markers = []
223 if markers_path and markers_path != "none" and os.path.exists(markers_path):
224 for mk in json.load(open(markers_path)).get("markers", []):
225 markers.append((float(mk.get("offsetSeconds", 0)), mk.get("text")))
226 markers.sort(key=lambda x: x[0])
227
228 render(title, final_segments, labels, markers, out_path)
229 json.dump({"title": title, "segments": final_segments},
230 open(os.path.join(folder, "transcript.json"), "w"))
231 json.dump({"speakers": speakers}, open(os.path.join(folder, "speakers.json"), "w"))
232
233
234if __name__ == "__main__":
235 main()