From 1612d8b2d48d4b28ab6eda02e3da2f8d08e7eb8b Mon Sep 17 00:00:00 2001 From: leo <260626284+cameledev@users.noreply.github.com> Date: Thu, 7 May 2026 11:26:59 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(audio)=20assign=20users=20to=20diariz?= =?UTF-8?q?ation=20speaker=20results=20using=20VAD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a new user assignment mechanism to for more friendly output than the current (SPEAKER_0, SPEAKER_1, ...). Use the VAD metadata to compare speech intervals with those returned by WhisperX. User with the highest overlap score above a defined threshold is assigned to each segment. This method allows for multi-speaker scenarios for a single account. --- CHANGELOG.md | 1 + src/agents/metadata_collector.py | 4 +- src/backend/core/api/viewsets.py | 1 + .../core/recording/event/notification.py | 6 +- src/summary/summary/core/celery_worker.py | 73 +++ src/summary/summary/core/config.py | 3 + src/summary/summary/core/file_service.py | 22 + src/summary/summary/core/user_assign.py | 297 ++++++++++ src/summary/tests/unit/test_user_assign.py | 528 ++++++++++++++++++ 9 files changed, 929 insertions(+), 6 deletions(-) create mode 100644 src/summary/summary/core/user_assign.py create mode 100644 src/summary/tests/unit/test_user_assign.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 18dcb244..c3967239 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to - 🔒️(backend) add validation of Room.configuration - ✨(helm) add support multiple transcribe worker / endpoint #1247 - ✨(backend) make LiveKit Egress recording encoding configurable #1288 +- ✨(summary) add speaker-to-participant assignment ### Changed diff --git a/src/agents/metadata_collector.py b/src/agents/metadata_collector.py index 5d4acd08..d03f31de 100644 --- a/src/agents/metadata_collector.py +++ b/src/agents/metadata_collector.py @@ -177,7 +177,7 @@ class MetadataCollector: def save(self): """Serialize collected events and upload as JSON to S3.""" - logger.info("Persisting metadata…") + logger.info("Persisting metadata...") participants = [] for k, v in self.participants.items(): @@ -372,7 +372,7 @@ async def entrypoint(ctx: JobContext): await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) async def cleanup(): - logger.info("Shutting down metadata collector…") + logger.info("Shutting down metadata collector...") await metadata_collector.aclose() ctx.add_shutdown_callback(cleanup) diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index fc1308ec..e883e1ea 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -363,6 +363,7 @@ class RoomViewSet( ): try: MetadataCollectorService().start(recording) + logger.debug("Started MetadataCollectorService") except MetadataCollectorException: logger.warning("Failed to start MetadataCollectorService") diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index fee4922a..f147fb86 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -213,7 +213,7 @@ class NotificationService: payload = { "owner_id": str(owner_access.user.id), "recording_filename": recording.key, - "metadata_filename": metadata_filename, # For future use + "metadata_filename": metadata_filename, "email": owner_access.user.email, "sub": owner_access.user.sub, "room": recording.room.name, @@ -222,9 +222,7 @@ class NotificationService: "download_link": f"{get_recording_download_base_url()}/{recording.id}", "context_language": owner_access.user.language, "recording_start_at": (started_at.isoformat() if started_at else None), - "recording_end_at": ( - ended_at.isoformat() if ended_at else None - ), # For future use + "recording_end_at": (ended_at.isoformat() if ended_at else None), } headers = { diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index 8b23fd1a..19c56f35 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -4,6 +4,7 @@ import json import time +from datetime import datetime import openai import sentry_sdk @@ -39,6 +40,7 @@ from summary.core.shared_models import ( webhook_payload_adapter, ) from summary.core.transcript_formatter import TranscriptFormatter +from summary.core.user_assign import resolve_speaker_identities from summary.core.webhook_service import ( call_webhook_v2, submit_content, @@ -160,6 +162,65 @@ def transcribe_audio( return transcription +def resolve_speaker_identities_and_apply_to( + transcription, recording_start_at, recording_end_at, metadata_filename, task_id +): + """Assign users to detected speakers and rewrite the transcriptions. + + Args: + transcription: output of meet-whisperx after transcription and diarization + recording_start_at: sourced from LiveKit FileInfo via the egress_ended webhook + recording_end_at: sourced from LiveKit FileInfo via the egress_ended webhook + metadata_filename: name of metadata file containing VAD information in S3 + task_id: current task id, for logging purposes + """ + recording_start_dt = ( + datetime.fromisoformat(recording_start_at) if recording_start_at else None + ) + recording_end_dt = ( + datetime.fromisoformat(recording_end_at) if recording_end_at else None + ) + + logger.debug( + "recording_start_dt: %s ; recording_end_dt: %s", + recording_start_dt, + recording_end_dt, + ) + if (recording_start_dt is None) or (recording_end_dt is None): + logger.debug("Skipping resolve_speaker_identities") + return transcription + + logger.debug("Running resolve_speaker_identities") + try: + metadata = file_service.read_json(metadata_filename) + speaker_mapping = resolve_speaker_identities( + metadata, + transcription, + recording_start_dt, + recording_end_dt, + ) + new_transcription = speaker_mapping.apply_to(transcription.model_dump()) + return new_transcription + + except FileServiceException as exc: + logger.error( + "Error reading metadata for task %s; skipping speaker assignment." + " Error: %s", + task_id, + exc, + ) + return transcription + + except Exception as exc: + logger.exception( + "resolve_speaker_identities failed for task %s; skipping" + " speaker assignment. Error: %s", + task_id, + exc, + ) + return transcription + + def format_transcript( transcription, context_language: str | None, @@ -269,6 +330,18 @@ def process_audio_transcribe_summarize_v2( if transcription is None: return + # Assign speakers and rewrite transcription/diarization output + if settings.is_resolve_speaker_identities_enabled and ( + metadata_filename is not None + ): + transcription = resolve_speaker_identities_and_apply_to( + transcription, + recording_start_at, + recording_end_at, + metadata_filename, + task_id, + ) + # Format output content, title = format_transcript( transcription, diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index dfd83fde..d5099f1f 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -103,6 +103,9 @@ class Settings(BaseSettings): # Transcription processing hallucination_patterns: List[str] = ["Vap'n'Roll Thierry"] + # Speaker to user assignment + is_resolve_speaker_identities_enabled: bool = True + # Webhook-related settings webhook_max_retries: int = 2 webhook_status_forcelist: List[int] = [502, 503, 504] diff --git a/src/summary/summary/core/file_service.py b/src/summary/summary/core/file_service.py index 8fd356d5..8df30f34 100644 --- a/src/summary/summary/core/file_service.py +++ b/src/summary/summary/core/file_service.py @@ -229,6 +229,28 @@ class FileService: os.remove(output_path) raise RuntimeError("Failed to extract audio.") from e + def read_json(self, object_name: str) -> dict: + """Read and parse a JSON file from MinIO storage.""" + logger.info("Reading JSON: %s", object_name) + + if not object_name: + raise ValueError("Invalid object_name") + + response = None + try: + response = self._minio_client.get_object(self._bucket_name, object_name) + return json.loads(response.read()) + except (MinioException, S3Error) as e: + raise FileServiceException( + "Unexpected error while reading JSON object." + ) from e + except (json.JSONDecodeError, UnicodeDecodeError) as e: + raise FileServiceException("Invalid JSON content.") from e + finally: + if response: + response.close() + response.release_conn() + @contextmanager def prepare_audio_file( self, diff --git a/src/summary/summary/core/user_assign.py b/src/summary/summary/core/user_assign.py new file mode 100644 index 00000000..6aab94e8 --- /dev/null +++ b/src/summary/summary/core/user_assign.py @@ -0,0 +1,297 @@ +"""Assign WhisperX diarization speakers to participant identities. + +Uses per-stream VAD events to match generic SPEAKER_XX labels provided +by diarization to real user id's by computing time overlap between +diarization segments and VAD intervals. + +Multiple speakers can map to the same participant (e.g. two people sharing +one microphone). A participant with no matching speaker gets no assignment. +""" + +import logging +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +logger = logging.getLogger(__name__) + +# Minimum fraction of a speaker's total duration that must overlap with a +# participant's VAD to accept the assignment. +DEFAULT_OVERLAP_THRESHOLD = 0.5 + + +@dataclass +class Interval: + """A time interval in seconds relative to recording start.""" + + start: float + end: float + + +@dataclass +class SpeakerAssignment: + """Maps a diarization speaker label to a participant.""" + + speaker_label: str + participant_id: str + participant_name: str + score: float + + +@dataclass +class AssignmentResult: + """Result of speaker-to-participant assignment.""" + + assignments: list[SpeakerAssignment] = field(default_factory=list) + unassigned_speakers: list[str] = field(default_factory=list) + + def apply_to(self, diarization: dict[str, Any]) -> dict[str, Any]: + """Return a copy of diarization with speaker labels replaced by names. + + Replaces `"speaker"` fields in segments and word_segments with the + assigned participant name. Unassigned speakers are left as-is. + + Args: + diarization: WhisperX dict with `segments` and optionally + `word_segments`. + + Returns: + New dict with speaker labels replaced. + """ + speaker_to_name = { + a.speaker_label: a.participant_name for a in self.assignments + } + + name_to_speaker_count = defaultdict(int) + for name in speaker_to_name.values(): + name_to_speaker_count[name] += 1 + + def _replace_speaker(item: dict[str, Any]) -> dict[str, Any]: + if "speaker" in item and item["speaker"] in speaker_to_name: + name = speaker_to_name[item["speaker"]] + suffix = ( + f" ({item['speaker']})" if name_to_speaker_count[name] > 1 else "" + ) # Add suffix only if there are multiple detected speakers per user + return {**item, "speaker": f"{name}{suffix}"} + return item + + def _process_segment( + item: dict[str, Any], include_words: bool = False + ) -> dict[str, Any]: + new_item = _replace_speaker(item) + if include_words and "words" in item: + new_item["words"] = [_replace_speaker(w) for w in item["words"]] + return new_item + + result: dict[str, Any] = {} + for key, value in diarization.items(): + if key not in ("segments", "word_segments"): + result[key] = value + continue + result[key] = [ + _process_segment(item, include_words=(key == "segments")) + for item in value + ] + return result + + +def _merge_intervals(intervals: list[Interval]) -> list[Interval]: + """Return a list of non-overlapping intervals sorted by start time.""" + if not intervals: + return [] + sorted_intervals = sorted(intervals, key=lambda interval: interval.start) + merged: list[Interval] = [ + Interval(sorted_intervals[0].start, sorted_intervals[0].end) + ] + for interval in sorted_intervals[1:]: + if interval.start <= merged[-1].end: + merged[-1].end = max(merged[-1].end, interval.end) + else: + merged.append(Interval(interval.start, interval.end)) + return merged + + +def _total_duration(intervals: list[Interval]) -> float: + """Return the sum of all interval durations.""" + return sum(interval.end - interval.start for interval in intervals) + + +def _overlap_duration( + a_intervals: list[Interval], + b_intervals: list[Interval], +) -> float: + """Compute total overlap between two merged interval lists, sorted by start time.""" + overlap = 0.0 + i = j = 0 + while i < len(a_intervals) and j < len(b_intervals): + a = a_intervals[i] + b = b_intervals[j] + lo = max(a.start, b.start) + hi = min(a.end, b.end) + if lo < hi: + overlap += hi - lo + if a.end <= b.end: + i += 1 + else: + j += 1 + return overlap + + +def _build_participant_timelines( + metadata: dict[str, Any], + recording_start_datetime: datetime, + recording_end_datetime: datetime | None = None, +) -> tuple[dict[str, list[Interval]], dict[str, str]]: + """Build VAD interval timelines for each participant. + + Args: + metadata: Dict with `events` and `participants` keys. + recording_start_datetime: UTC datetime used as t=0 reference. + recording_end_datetime: UTC datetime of recording end. When provided, + any open speech_start without a matching speech_end is closed at + this time (the participant is assumed to be speaking until the end). + + Returns: + participant_id → merged VAD intervals + (seconds relative to recording_start_datetime). + participant_id → display name. + Intervals are in seconds relative to recording_start_datetime. + Events before recording start are clamped to 0. + """ + events = metadata.get("events", []) + participants_info = { + p["participantId"]: p.get("name", p["participantId"]) + for p in metadata.get("participants", []) + } + + ref_epoch = recording_start_datetime.timestamp() + + open_starts: dict[str, float] = {} + intervals: dict[str, list[Interval]] = {} + + for event in events: + pid = event["participant_id"] + ts = datetime.fromisoformat(event["timestamp"]).timestamp() - ref_epoch + etype = event["type"] + + if etype == "speech_start": + open_starts[pid] = max(ts, 0.0) + elif etype == "speech_end": + start = open_starts.pop(pid, None) + if start is not None: + end = max(ts, 0.0) + if end > start: + intervals.setdefault(pid, []).append(Interval(start, end)) + + # Close any speech_start that was never matched by a speech_end. + # Assume the participant kept speaking until the recording ended. + if recording_end_datetime is not None and open_starts: + recording_end = recording_end_datetime.timestamp() - ref_epoch + for pid, start in open_starts.items(): + end = max(recording_end, 0.0) + if end > start: + intervals.setdefault(pid, []).append(Interval(start, end)) + + for pid, pid_intervals in intervals.items(): + intervals[pid] = _merge_intervals(pid_intervals) + + return intervals, participants_info + + +def _build_speaker_timelines( + transcription: Any, +) -> dict[str, list[Interval]]: + """Build interval timelines from WhisperX transcription segments.""" + intervals: dict[str, list[Interval]] = {} + + segments = transcription.segments if hasattr(transcription, "segments") else [] + for segment in segments: + speaker = segment.get("speaker") + if speaker is None: + continue + intervals.setdefault(speaker, []).append( + Interval(segment["start"], segment["end"]) + ) + + for speaker, speaker_intervals in intervals.items(): + intervals[speaker] = _merge_intervals(speaker_intervals) + + return intervals + + +def resolve_speaker_identities( + metadata: dict[str, Any], + transcription: Any, + recording_start_datetime: datetime, + recording_end_datetime: datetime, + overlap_threshold: float = DEFAULT_OVERLAP_THRESHOLD, +) -> AssignmentResult: + """Match WhisperX speaker labels to participants. + + Args: + metadata: User metadata with `events` and `participants`. + transcription: WhisperX Transcription object with a `segments` attribute. + recording_start_datetime: UTC datetime for t=0 reference. + recording_end_datetime: UTC datetime of recording end. Open speech + intervals are closed at this time. + overlap_threshold: Minimum overlap/speaker_duration to accept. + + Returns: + AssignmentResult with per-speaker assignments and unassigned + speakers. + """ + participant_timelines, participant_names = _build_participant_timelines( + metadata, recording_start_datetime, recording_end_datetime + ) + speaker_timelines = _build_speaker_timelines(transcription) + + logger.debug( + "Assignment inputs: %d participants, %d speakers", + len(participant_timelines), + len(speaker_timelines), + ) + + result = AssignmentResult() + + for speaker, speaker_intervals in speaker_timelines.items(): + speaker_duration = _total_duration(speaker_intervals) + if speaker_duration == 0: + result.unassigned_speakers.append(speaker) + continue + + best_pid: str | None = None + best_score: float = 0.0 + + for pid, part_intervals in participant_timelines.items(): + overlap = _overlap_duration(speaker_intervals, part_intervals) + score = overlap / speaker_duration + if score > best_score: + best_score = score + best_pid = pid + + if best_pid is not None and best_score >= overlap_threshold: + result.assignments.append( + SpeakerAssignment( + speaker_label=speaker, + participant_id=best_pid, + participant_name=participant_names.get(best_pid, best_pid), + score=best_score, + ) + ) + logger.info( + "Assigned %s -> %s (score=%.3f)", + speaker, + participant_names.get(best_pid, best_pid), + best_score, + ) + else: + result.unassigned_speakers.append(speaker) + logger.info( + "Speaker %s unassigned (best=%.3f, threshold=%.3f)", + speaker, + best_score, + overlap_threshold, + ) + + return result diff --git a/src/summary/tests/unit/test_user_assign.py b/src/summary/tests/unit/test_user_assign.py new file mode 100644 index 00000000..9fd36c84 --- /dev/null +++ b/src/summary/tests/unit/test_user_assign.py @@ -0,0 +1,528 @@ +"""Tests for the speaker-to-user assignment service.""" + +import math +from dataclasses import dataclass, field +from datetime import datetime + +from summary.core.user_assign import ( + AssignmentResult, + Interval, + SpeakerAssignment, + _merge_intervals, + _overlap_duration, + _total_duration, + resolve_speaker_identities, +) + + +@dataclass +class FakeTranscription: + """Mimics the OpenAI Transcription pydantic model for testing.""" + + segments: list = field(default_factory=list) + + +RECORDING_START = datetime.fromisoformat("2026-03-17T15:30:33.000001") +RECORDING_END = datetime.fromisoformat("2026-03-17T15:31:33.000001") + +METADATA_SINGLE_USER = { + "events": [ + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "participant_connected", + "timestamp": "2026-03-17T15:30:33.000001", + }, + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:36.039456", + }, + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:36.589114", + }, + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:38.887518", + }, + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:39.438141", + }, + { + "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "type": "participant_disconnected", + "timestamp": "2026-03-17T15:30:43.223255", + }, + ], + "participants": [ + { + "participantId": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb", + "name": "cameledev", + } + ], +} + +DIARIZATION_SINGLE_SPEAKER = FakeTranscription( + segments=[ + { + "start": 1.363, + "end": 3.545, + "text": " The stale smell.", + "speaker": "SPEAKER_00", + }, + { + "start": 4.466, + "end": 6.247, + "text": "It takes heat.", + "speaker": "SPEAKER_00", + }, + ], +) + +USER_ID = "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb" + + +class TestMergeIntervals: + """Tests for _merge_intervals.""" + + def test_empty(self): + """Empty input returns empty list.""" + assert _merge_intervals([]) == [] + + def test_no_overlap(self): + """Non-overlapping intervals stay separate.""" + result = _merge_intervals([Interval(1, 2), Interval(3, 4)]) + assert len(result) == 2 + + def test_overlap(self): + """Overlapping intervals are merged.""" + result = _merge_intervals([Interval(1, 3), Interval(2, 4)]) + assert len(result) == 1 + assert result[0].start == 1 + assert result[0].end == 4 + + def test_adjacent(self): + """Adjacent intervals are merged.""" + result = _merge_intervals([Interval(1, 2), Interval(2, 3)]) + assert len(result) == 1 + assert result[0].start == 1 + assert result[0].end == 3 + + def test_unsorted(self): + """Unsorted input is sorted before merging.""" + result = _merge_intervals([Interval(5, 6), Interval(1, 2)]) + assert len(result) == 2 + assert result[0].start == 1 + + +class TestOverlapDuration: + """Tests for _overlap_duration.""" + + def test_no_overlap(self): + """Disjoint intervals have zero overlap.""" + a = [Interval(1, 2)] + b = [Interval(3, 4)] + assert math.isclose(_overlap_duration(a, b), 0.0) + + def test_full_overlap(self): + """Identical intervals have full overlap.""" + a = [Interval(1, 3)] + b = [Interval(1, 3)] + assert math.isclose(_overlap_duration(a, b), 2.0) + + def test_partial_overlap(self): + """Partially overlapping intervals.""" + a = [Interval(1, 3)] + b = [Interval(2, 4)] + assert math.isclose(_overlap_duration(a, b), 1.0) + + def test_multiple_intervals(self): + """Multiple intervals with a spanning interval.""" + a = [Interval(1, 3), Interval(5, 7)] + b = [Interval(2, 6)] + assert math.isclose(_overlap_duration(a, b), 2.0) + + def test_empty(self): + """Empty input yields zero overlap.""" + assert math.isclose(_overlap_duration([], [Interval(1, 2)]), 0.0) + assert math.isclose(_overlap_duration([Interval(1, 2)], []), 0.0) + + +class TestTotalDuration: + """Tests for _total_duration.""" + + def test_basic(self): + """Sum of durations for multiple intervals.""" + ivs = [Interval(0, 1), Interval(2, 5)] + assert math.isclose(_total_duration(ivs), 4.0) + + def test_empty(self): + """Empty input returns zero.""" + assert math.isclose(_total_duration([]), 0.0) + + +class TestResolveSpeakerIdentities: + """Tests for resolve_speaker_identities.""" + + def test_single_speaker_single_user(self): + """Single speaker assigned to single user with low threshold.""" + result = resolve_speaker_identities( + METADATA_SINGLE_USER, + DIARIZATION_SINGLE_SPEAKER, + RECORDING_START, + RECORDING_END, + overlap_threshold=0.2, + ) + assert len(result.assignments) == 1 + assert result.assignments[0].speaker_label == "SPEAKER_00" + assert result.assignments[0].participant_id == USER_ID + assert result.assignments[0].participant_name == "cameledev" + assert result.assignments[0].score > 0 + assert result.unassigned_speakers == [] + + def test_no_vad_events(self): + """Participant with no speech leaves speakers unassigned.""" + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "participant_connected", + "timestamp": "2026-03-17T15:30:33.000000", + }, + ], + "participants": [{"participantId": "user-a", "name": "Silent User"}], + } + result = resolve_speaker_identities( + metadata, + DIARIZATION_SINGLE_SPEAKER, + RECORDING_START, + RECORDING_END, + ) + assert len(result.assignments) == 0 + assert "SPEAKER_00" in result.unassigned_speakers + + def test_multiple_speakers_same_user(self): + """Two speakers from same mic both assigned to same user.""" + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:35.000000", + }, + { + "participant_id": "user-a", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:50.000000", + }, + ], + "participants": [{"participantId": "user-a", "name": "Shared Mic"}], + } + transcription = FakeTranscription( + segments=[ + {"start": 1.0, "end": 3.0, "speaker": "SPEAKER_00"}, + {"start": 5.0, "end": 7.0, "speaker": "SPEAKER_01"}, + ], + ) + result = resolve_speaker_identities( + metadata, transcription, RECORDING_START, RECORDING_END + ) + assert len(result.assignments) == 2 + pids = {a.participant_id for a in result.assignments} + assert pids == {"user-a"} + + def test_two_users_two_speakers(self): + """Each speaker maps to correct user by VAD overlap.""" + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:34.000001", + }, + { + "participant_id": "user-a", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:37.000001", + }, + { + "participant_id": "user-b", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:38.000001", + }, + { + "participant_id": "user-b", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:41.000001", + }, + ], + "participants": [ + {"participantId": "user-a", "name": "Alice"}, + {"participantId": "user-b", "name": "Bob"}, + ], + } + transcription = FakeTranscription( + segments=[ + {"start": 1.5, "end": 3.5, "speaker": "SPEAKER_00"}, + {"start": 5.5, "end": 7.5, "speaker": "SPEAKER_01"}, + ], + ) + result = resolve_speaker_identities( + metadata, transcription, RECORDING_START, RECORDING_END + ) + assert len(result.assignments) == 2 + by_speaker = {a.speaker_label: a for a in result.assignments} + assert by_speaker["SPEAKER_00"].participant_name == "Alice" + assert by_speaker["SPEAKER_01"].participant_name == "Bob" + + def test_overlapping_speech_two_users(self): + """Simultaneous speech from two users still assigns each speaker correctly.""" + # user-a speaks from t=1s to t=6s, user-b speaks from t=3s to t=8s + # (3s overlap where both are speaking) + # SPEAKER_00 diarization covers t=1.5–5.5 (mostly user-a) + # SPEAKER_01 diarization covers t=4.0–7.5 (mostly user-b) + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:34.000001", + }, + { + "participant_id": "user-b", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:36.000001", + }, + { + "participant_id": "user-a", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:39.000001", + }, + { + "participant_id": "user-b", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:41.000001", + }, + ], + "participants": [ + {"participantId": "user-a", "name": "Alice"}, + {"participantId": "user-b", "name": "Bob"}, + ], + } + transcription = FakeTranscription( + segments=[ + {"start": 1.5, "end": 5.5, "speaker": "SPEAKER_00"}, + {"start": 4.0, "end": 7.5, "speaker": "SPEAKER_01"}, + ], + ) + result = resolve_speaker_identities( + metadata, + transcription, + RECORDING_START, + RECORDING_END, + overlap_threshold=0.3, + ) + assert len(result.assignments) == 2 + by_speaker = {a.speaker_label: a for a in result.assignments} + assert by_speaker["SPEAKER_00"].participant_name == "Alice" + assert by_speaker["SPEAKER_01"].participant_name == "Bob" + assert result.unassigned_speakers == [] + + def test_below_threshold(self): + """Speaker with minimal overlap stays unassigned.""" + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:34.000001", + }, + { + "participant_id": "user-a", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:35.169950", + }, + ], + "participants": [{"participantId": "user-a", "name": "Brief User"}], + } + transcription = FakeTranscription( + segments=[ + {"start": 1.0, "end": 10.0, "speaker": "SPEAKER_00"}, + ], + ) + result = resolve_speaker_identities( + metadata, + transcription, + RECORDING_START, + RECORDING_END, + overlap_threshold=0.5, + ) + assert len(result.assignments) == 0 + assert "SPEAKER_00" in result.unassigned_speakers + + def test_events_before_recording_start_clamped(self): + """Speech events before recording start are clamped to t=0.""" + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:31.000001", # before RECORDING_START + }, + { + "participant_id": "user-a", + "type": "speech_end", + "timestamp": "2026-03-17T15:30:36.000001", # after RECORDING_START + }, + ], + "participants": [{"participantId": "user-a", "name": "Early User"}], + } + transcription = FakeTranscription( + segments=[ + {"start": 0.0, "end": 3.0, "speaker": "SPEAKER_00"}, + ], + ) + result = resolve_speaker_identities( + metadata, transcription, RECORDING_START, RECORDING_END + ) + assert len(result.assignments) == 1 + assert result.assignments[0].participant_name == "Early User" + + def test_empty_diarization(self): + """No segments produces empty result.""" + result = resolve_speaker_identities( + METADATA_SINGLE_USER, + FakeTranscription(segments=[]), + RECORDING_START, + RECORDING_END, + ) + assert result == AssignmentResult() + + def test_segment_without_speaker_ignored(self): + """Segments missing speaker key are skipped.""" + transcription = FakeTranscription( + segments=[ + {"start": 1.0, "end": 3.0, "text": "no speaker"}, + ], + ) + result = resolve_speaker_identities( + METADATA_SINGLE_USER, transcription, RECORDING_START, RECORDING_END + ) + assert result == AssignmentResult() + + def test_unclosed_speech_closed_at_recording_end(self): + """Open speech_start without speech_end is closed at recording end.""" + recording_end = datetime.fromisoformat("2026-03-17T15:30:43.000001") + metadata = { + "events": [ + { + "participant_id": "user-a", + "type": "speech_start", + "timestamp": "2026-03-17T15:30:35.000001", + }, + # No speech_end — participant kept speaking until recording stopped + ], + "participants": [{"participantId": "user-a", "name": "Still Talking"}], + } + transcription = FakeTranscription( + segments=[ + {"start": 2.0, "end": 9.0, "speaker": "SPEAKER_00"}, + ], + ) + result = resolve_speaker_identities( + metadata, + transcription, + RECORDING_START, + recording_end, + overlap_threshold=0.5, + ) + assert len(result.assignments) == 1 + assert result.assignments[0].participant_name == "Still Talking" + assert result.unassigned_speakers == [] + + +class TestApply: + """Tests for AssignmentResult.apply_to.""" + + def test_replaces_speakers_in_segments_and_words(self): + """Speaker labels replaced in segments, words, and word_segments.""" + diarization = { + "segments": [ + { + "start": 1.0, + "end": 3.0, + "text": "Hello", + "speaker": "SPEAKER_00", + "words": [ + { + "word": "Hello", + "start": 1.0, + "end": 1.5, + "speaker": "SPEAKER_00", + }, + ], + }, + { + "start": 4.0, + "end": 6.0, + "text": "World", + "speaker": "SPEAKER_01", + "words": [ + { + "word": "World", + "start": 4.0, + "end": 4.5, + "speaker": "SPEAKER_01", + }, + ], + }, + ], + "word_segments": [ + {"word": "Hello", "start": 1.0, "end": 1.5, "speaker": "SPEAKER_00"}, + {"word": "World", "start": 4.0, "end": 4.5, "speaker": "SPEAKER_01"}, + ], + } + assignment = AssignmentResult( + assignments=[ + SpeakerAssignment("SPEAKER_00", "id-a", "Alice", 0.9), + SpeakerAssignment("SPEAKER_01", "id-b", "Bob", 0.8), + ], + ) + result = assignment.apply_to(diarization) + + assert result["segments"][0]["speaker"] == "Alice" + assert result["segments"][0]["words"][0]["speaker"] == "Alice" + assert result["segments"][1]["speaker"] == "Bob" + assert result["word_segments"][0]["speaker"] == "Alice" + assert result["word_segments"][1]["speaker"] == "Bob" + + def test_unassigned_speakers_unchanged(self): + """Unassigned speaker labels are left as-is.""" + diarization = { + "segments": [ + {"start": 1.0, "end": 3.0, "speaker": "SPEAKER_02"}, + ], + } + assignment = AssignmentResult( + assignments=[ + SpeakerAssignment("SPEAKER_00", "id-a", "Alice", 0.9), + ], + unassigned_speakers=["SPEAKER_02"], + ) + result = assignment.apply_to(diarization) + assert result["segments"][0]["speaker"] == "SPEAKER_02" + + def test_preserves_extra_keys(self): + """Non-segment keys in diarization are preserved.""" + diarization = { + "segments": [], + "language": "en", + "custom_field": 42, + } + result = AssignmentResult().apply_to(diarization) + assert result["language"] == "en" + assert result["custom_field"] == 42