Enhance speaker assignment in transcription: Introduced interval overlap calculations and smoothing techniques for improved accuracy in speaker labeling. Added methods for determining dominant speakers and stabilizing segment boundaries.

app.py

@@ -568,41 +568,114 @@ class WhisperTranscriber:
         """Assign speakers to words and segments based on overlap with diarization segments."""
         if not diarization_segments:
             return transcription_results
-        # simple helper to find speaker at given time
+        # Helper: find the diarization speaker active at time t, or closest
         def speaker_at(t: float):
-            for seg in diarization_segments:
-                if seg["start"] <= t < seg["end"]:
-                    return seg["speaker"]
+            for dseg in diarization_segments:
+                if float(dseg["start"]) <= t < float(dseg["end"]):
+                    return dseg["speaker"]
             # if not inside, return closest segment's speaker
             closest = None
-            best = float("inf")
-            for seg in diarization_segments:
-                if t < seg["start"]:
-                    d = seg["start"] - t
-                elif t > seg["end"]:
-                    d = t - seg["end"]
+            best_dist = float("inf")
+            for dseg in diarization_segments:
+                if t < float(dseg["start"]):
+                    d = float(dseg["start"]) - t
+                elif t > float(dseg["end"]):
+                    d = t - float(dseg["end"])
                 else:
                     d = 0.0
-                if d < best:
-                    best = d
-                    closest = seg
+                if d < best_dist:
+                    best_dist = d
+                    closest = dseg
             return closest["speaker"] if closest else "SPEAKER_00"
 
+        # Helper: overlap length between two intervals
+        def interval_overlap(a_start: float, a_end: float, b_start: float, b_end: float) -> float:
+            return max(0.0, min(a_end, b_end) - max(a_start, b_start))
+
+        # Helper: choose speaker for an interval by maximum overlap with diarization
+        def best_speaker_for_interval(start_t: float, end_t: float) -> str:
+            best_spk = None
+            best_ov = -1.0
+            for dseg in diarization_segments:
+                ov = interval_overlap(float(start_t), float(end_t), float(dseg["start"]), float(dseg["end"]))
+                if ov > best_ov:
+                    best_ov = ov
+                    best_spk = dseg["speaker"]
+            if best_ov > 0.0 and best_spk is not None:
+                return best_spk
+            # fallback to nearest by midpoint
+            mid = (float(start_t) + float(end_t)) / 2.0
+            return speaker_at(mid)
+
         for seg in transcription_results:
-            # Assign per-word speakers
+            # Assign per-word speakers using overlap, then smooth and stabilize boundaries
             if seg.get("words"):
-                speaker_counts = {}
-                for w in seg["words"]:
-                    mid = (float(w["start"]) + float(w["end"])) / 2.0
-                    spk = speaker_at(mid)
-                    w["speaker"] = spk
-                    speaker_counts[spk] = speaker_counts.get(spk, 0) + (float(w["end"]) - float(w["start"]))
-                # Segment speaker = speaker with max accumulated word duration
-                if speaker_counts:
-                    seg["speaker"] = max(speaker_counts.items(), key=lambda kv: kv[1])[0]
+                words = seg["words"]
+                # 1) Initial assignment by overlap
+                for w in words:
+                    w_start = float(w["start"])
+                    w_end = float(w["end"])
+                    w["speaker"] = best_speaker_for_interval(w_start, w_end)
+
+                # 2) Small median filter (window=3) to fix isolated outliers
+                if len(words) >= 3:
+                    smoothed = [words[i]["speaker"] for i in range(len(words))]
+                    for i in range(1, len(words) - 1):
+                        prev_spk = words[i - 1]["speaker"]
+                        curr_spk = words[i]["speaker"]
+                        next_spk = words[i + 1]["speaker"]
+                        if prev_spk == next_spk and curr_spk != prev_spk:
+                            smoothed[i] = prev_spk
+                    for i in range(len(words)):
+                        words[i]["speaker"] = smoothed[i]
+
+                # 3) Determine dominant speaker by summed word durations
+                speaker_dur = {}
+                total_word_dur = 0.0
+                for w in words:
+                    dur = max(0.0, float(w["end"]) - float(w["start"]))
+                    total_word_dur += dur
+                    spk = w.get("speaker", "SPEAKER_00")
+                    speaker_dur[spk] = speaker_dur.get(spk, 0.0) + dur
+                if speaker_dur:
+                    dominant_speaker = max(speaker_dur.items(), key=lambda kv: kv[1])[0]
+                else:
+                    dominant_speaker = speaker_at((float(seg["start"]) + float(seg["end"])) / 2.0)
+
+                # 4) Boundary stabilization: relabel tiny prefix/suffix runs to dominant
+                seg_duration = max(1e-6, float(seg["end"]) - float(seg["start"]))
+                max_boundary_sec = 0.5  # hard cap for how much to relabel at edges
+                max_boundary_frac = 0.2  # or up to 20% of the segment duration
+
+                # prefix
+                prefix_dur = 0.0
+                prefix_count = 0
+                for w in words:
+                    if w.get("speaker") == dominant_speaker:
+                        break
+                    prefix_dur += max(0.0, float(w["end"]) - float(w["start"]))
+                    prefix_count += 1
+                if prefix_count > 0 and prefix_dur <= min(max_boundary_sec, max_boundary_frac * seg_duration):
+                    for i in range(prefix_count):
+                        words[i]["speaker"] = dominant_speaker
+
+                # suffix
+                suffix_dur = 0.0
+                suffix_count = 0
+                for w in reversed(words):
+                    if w.get("speaker") == dominant_speaker:
+                        break
+                    suffix_dur += max(0.0, float(w["end"]) - float(w["start"]))
+                    suffix_count += 1
+                if suffix_count > 0 and suffix_dur <= min(max_boundary_sec, max_boundary_frac * seg_duration):
+                    for i in range(len(words) - suffix_count, len(words)):
+                        words[i]["speaker"] = dominant_speaker
+
+                # 5) Final segment speaker
+                seg["speaker"] = dominant_speaker
             else:
-                mid = (float(seg["start"]) + float(seg["end"])) / 2.0
-                seg["speaker"] = speaker_at(mid)
+                # No word timings: choose by overlap with diarization over the whole segment
+                seg["speaker"] = best_speaker_for_interval(float(seg["start"]), float(seg["end"]))
         return transcription_results
     
     def group_segments_by_speaker(self, segments, max_gap=1.0, max_duration=30.0):