Add chunk-based Gemini transcription with VAD timestamp anchoring

+8

observe/transcribe/__init__.py

··· 135 135 audio: "np.ndarray", 136 136 sample_rate: int, 137 137 config: dict, 138 + speech_segments: list[tuple[float, float]] | None = None, 138 139 ) -> list[dict]: 139 140 """Dispatch transcription to the specified backend. 140 141 ··· 143 144 audio: Audio buffer (float32, mono) 144 145 sample_rate: Sample rate in Hz (typically 16000) 145 146 config: Backend-specific configuration dict 147 + speech_segments: Optional VAD speech segments for chunk-based transcription. 148 + Currently only used by the Gemini backend for timestamp anchoring. 146 149 147 150 Returns: 148 151 List of statement dicts with id, start, end, text, and optionally words 149 152 """ 150 153 backend_mod = get_backend(backend) 154 + 155 + # Pass speech_segments to backends that support it (currently only Gemini) 156 + if backend == "gemini" and speech_segments is not None: 157 + return backend_mod.transcribe(audio, sample_rate, config, speech_segments) 158 + 151 159 return backend_mod.transcribe(audio, sample_rate, config) 152 160 153 161

+25 -7

observe/transcribe/gemini.md

··· 4 4 label: Audio Transcription (Gemini) 5 5 group: Observe 6 6 --- 7 - You are accurately transcribing audio and indentifying distinct. Return a JSON object with individual speech segments that represent separate statements or sentences. 7 + You are transcribing audio clips from a continuous recording. Each clip is labeled with its start time and duration. Your task is to extract all statements from each clip, identify speakers, and produce a sequential transcript. 8 + 9 + ## Input Format 10 + 11 + You will receive multiple audio clips, each preceded by a label like: 12 + `Clip starting at 01:23 (15s):` 13 + 14 + This means the clip begins at 1 minute 23 seconds into the original recording and is 15 seconds long. 8 15 9 16 ## Output Format 10 17 ··· 13 20 ```json 14 21 { 15 22 "segments": [ 16 - {"start": "MM:SS", "end": "MM:SS", "speaker": "Speaker 1", "text": "<text>"}, 23 + {"start": "01:23", "speaker": "Speaker 1", "text": "First statement in clip"}, 24 + {"start": "01:28", "speaker": "Speaker 2", "text": "Response from another person"}, 25 + {"start": "01:35", "speaker": "Speaker 1", "text": "Next statement"}, 17 26 ... 18 27 ] 19 28 } ··· 21 30 22 31 ## Guidelines 23 32 24 - ### Segments 25 - - Create new segment when speaker changes or there's a natural pause or new sentence. 26 - - Timestamps as MM:SS (e.g., "01:23", "00:05"). 27 - - Label speakers consistently: "Speaker 1", "Speaker 2", etc. 28 - - Transcribe exactly what you hear with professional accuracy, this is an important task. 33 + ### Timestamps 34 + - Output absolute timestamps as MM:SS (time in the original recording) 35 + - Calculate by adding the offset within the clip to the clip's start time 36 + - Example: If clip starts at 01:23 and someone speaks 5 seconds in, output "01:28" 37 + 38 + ### Statements 39 + - Extract EVERY statement from each clip - clips may contain multiple speakers and sentences 40 + - Create a new segment when the speaker changes or at natural sentence boundaries 41 + - Transcribe exactly what you hear with professional accuracy 42 + 43 + ### Speaker Identification 44 + - Label speakers consistently across ALL clips: "Speaker 1", "Speaker 2", etc. 45 + - Use voice characteristics to track the same speaker across different clips 46 + - Assign speaker numbers in order of first appearance

+172 -69

observe/transcribe/gemini.py

··· 6 6 This module provides cloud-based speech-to-text transcription using Google's 7 7 Gemini API with speaker diarization (identifies who said what). 8 8 9 + When VAD speech segments are provided, audio is sent as labeled clips with 10 + explicit timestamps. Gemini returns absolute MM:SS timestamps which are then 11 + mapped back to the audio timeline. This anchors output to known clip boundaries 12 + rather than relying solely on Gemini's internal clock. 13 + 9 14 Enrichment (topics, setting, emotion, corrections) is handled separately by 10 15 the enrich step, same as other backends. This keeps the transcription focused 11 16 and avoids hallucinations from entity name hints in the prompt. ··· 35 40 SPEAKER_PATTERN = re.compile(r"(?:speaker\s*)?(\d+)", re.IGNORECASE) 36 41 37 42 43 + def _format_timestamp(seconds: float) -> str: 44 + """Format seconds as MM:SS timestamp string. 45 + 46 + Args: 47 + seconds: Time in seconds 48 + 49 + Returns: 50 + Formatted string like "01:23" or "12:05" 51 + """ 52 + minutes = int(seconds // 60) 53 + secs = int(seconds % 60) 54 + return f"{minutes:02d}:{secs:02d}" 55 + 56 + 38 57 def _parse_timestamp(ts: str) -> float | None: 39 - """Parse MM:SS or HH:MM:SS timestamp to seconds. 40 - 41 - Handles various formats: 42 - - "1:23" -> 83.0 43 - - "0:05" -> 5.0 44 - - "1:05:30" -> 3930.0 45 - - "5" -> 5.0 (just seconds) 58 + """Parse MM:SS timestamp to seconds. 46 59 47 60 Args: 48 - ts: Timestamp string 61 + ts: Timestamp string like "01:23" or "1:23" 49 62 50 63 Returns: 51 64 Seconds as float, or None if unparseable ··· 59 72 60 73 try: 61 74 parts = ts.split(":") 62 - if len(parts) == 1: 63 - # Just seconds 64 - seconds = float(parts[0]) 65 - elif len(parts) == 2: 66 - # MM:SS 75 + if len(parts) == 2: 67 76 minutes = int(parts[0]) 68 77 seconds = float(parts[1]) 69 - seconds += minutes * 60 70 - elif len(parts) == 3: 71 - # HH:MM:SS 72 - hours = int(parts[0]) 73 - minutes = int(parts[1]) 74 - seconds = float(parts[2]) 75 - seconds += hours * 3600 + minutes * 60 76 - else: 77 - return None 78 + return max(0.0, minutes * 60 + seconds) 79 + elif len(parts) == 1: 80 + # Just seconds 81 + return max(0.0, float(parts[0])) 82 + except (ValueError, TypeError): 83 + pass 78 84 79 - # Clamp negative to 0 80 - return max(0.0, seconds) 81 - 82 - except (ValueError, TypeError): 83 - return None 85 + return None 84 86 85 87 86 88 def _parse_speaker(speaker: str | int | None) -> int | None: ··· 121 123 return None 122 124 123 125 124 - def _normalize_segments( 126 + def _build_chunk_contents( 127 + audio: np.ndarray, 128 + sample_rate: int, 129 + speech_segments: list[tuple[float, float]], 130 + prompt_text: str, 131 + ) -> list: 132 + """Build interleaved content list with labeled audio clips. 133 + 134 + Creates a list of [prompt, label1, audio1, label2, audio2, ...] where 135 + each label tells Gemini the clip's start time and duration. 136 + 137 + Args: 138 + audio: Full audio buffer (float32, mono) 139 + sample_rate: Sample rate in Hz 140 + speech_segments: List of (start, end) tuples from VAD 141 + prompt_text: The transcription prompt 142 + 143 + Returns: 144 + List of content parts for Gemini API 145 + """ 146 + contents: list = [prompt_text] 147 + 148 + for start, end in speech_segments: 149 + # Extract audio chunk 150 + start_sample = int(start * sample_rate) 151 + end_sample = int(end * sample_rate) 152 + chunk_audio = audio[start_sample:end_sample] 153 + 154 + # Skip empty chunks 155 + if len(chunk_audio) == 0: 156 + continue 157 + 158 + # Add label with start time and duration 159 + timestamp = _format_timestamp(start) 160 + duration = int(end - start) 161 + contents.append(f"Clip starting at {timestamp} ({duration}s):") 162 + 163 + # Add audio bytes 164 + audio_bytes = audio_to_flac_bytes(chunk_audio, sample_rate) 165 + contents.append(types.Part.from_bytes(data=audio_bytes, mime_type="audio/flac")) 166 + 167 + return contents 168 + 169 + 170 + def _find_segment_for_timestamp( 171 + timestamp_seconds: float, 172 + speech_segments: list[tuple[float, float]], 173 + ) -> tuple[float, float]: 174 + """Find the VAD segment that contains or is nearest to a timestamp. 175 + 176 + Args: 177 + timestamp_seconds: Absolute timestamp in seconds 178 + speech_segments: List of (start, end) tuples from VAD 179 + 180 + Returns: 181 + The (start, end) tuple of the matching or nearest segment 182 + """ 183 + # Check if timestamp falls within any segment 184 + for start, end in speech_segments: 185 + if start <= timestamp_seconds <= end: 186 + return (start, end) 187 + 188 + # Find nearest segment 189 + min_distance = float("inf") 190 + nearest = speech_segments[0] 191 + 192 + for start, end in speech_segments: 193 + # Distance to segment (0 if inside, otherwise distance to nearest edge) 194 + if timestamp_seconds < start: 195 + distance = start - timestamp_seconds 196 + else: 197 + distance = timestamp_seconds - end 198 + 199 + if distance < min_distance: 200 + min_distance = distance 201 + nearest = (start, end) 202 + 203 + return nearest 204 + 205 + 206 + def _normalize_chunked_segments( 125 207 segments: list[dict], 126 - audio_duration: float, 127 - ) -> tuple[list[dict], int]: 128 - """Convert Gemini segments to standard statement format. 208 + speech_segments: list[tuple[float, float]], 209 + ) -> list[dict]: 210 + """Convert Gemini segments with MM:SS timestamps to standard statement format. 129 211 130 - Segments with empty text are dropped. IDs are assigned sequentially 131 - to the remaining segments. 212 + Parses absolute timestamps from Gemini output and maps them to VAD segments. 213 + Falls back to segment boundaries if timestamp parsing fails. 132 214 133 215 Args: 134 - segments: Raw segments from Gemini response 135 - audio_duration: Total audio duration in seconds 216 + segments: Raw segments from Gemini response with "start" timestamps 217 + speech_segments: Original VAD segments with (start, end) times 136 218 137 219 Returns: 138 - Tuple of (statements list, count of segments with invalid timestamps) 220 + List of statements with proper timestamps 139 221 """ 140 222 statements = [] 141 - invalid_count = 0 142 223 statement_id = 1 224 + 225 + # Calculate overall time range for clamping 226 + min_time = speech_segments[0][0] if speech_segments else 0.0 227 + max_time = speech_segments[-1][1] if speech_segments else 0.0 143 228 144 229 for seg in segments: 145 230 # Get and strip text - skip if empty ··· 147 232 if not text: 148 233 continue 149 234 150 - # Parse timestamps 151 - start = _parse_timestamp(seg.get("start", "")) 152 - end = _parse_timestamp(seg.get("end", "")) 153 - 154 - # Track invalid timestamps but still include segment 155 - has_valid_timestamps = start is not None and end is not None 156 - if not has_valid_timestamps: 157 - invalid_count += 1 235 + # Parse timestamp from Gemini output 236 + raw_timestamp = seg.get("start", "") 237 + parsed_time = _parse_timestamp(raw_timestamp) 158 238 159 - # Clamp to audio duration if timestamps are valid 160 - if start is not None and start > audio_duration: 161 - start = audio_duration 162 - if end is not None and end > audio_duration: 163 - end = audio_duration 239 + if parsed_time is not None: 240 + # Clamp to valid range 241 + start = max(min_time, min(parsed_time, max_time)) 242 + # Find the segment this timestamp belongs to for the end time 243 + seg_start, seg_end = _find_segment_for_timestamp(start, speech_segments) 244 + end = seg_end 245 + else: 246 + # Fallback: use first segment boundaries 247 + seg_start, seg_end = speech_segments[0] if speech_segments else (0.0, 0.0) 248 + start = seg_start 249 + end = seg_end 164 250 165 251 # Build statement 166 252 statement = { ··· 179 265 180 266 statements.append(statement) 181 267 182 - return statements, invalid_count 268 + return statements 183 269 184 270 185 271 def transcribe( 186 272 audio: np.ndarray, 187 273 sample_rate: int, 188 274 config: dict, 275 + speech_segments: list[tuple[float, float]] | None = None, 189 276 ) -> list[dict]: 190 277 """Transcribe audio using Gemini API. 191 278 192 - This is the standard backend interface. It converts audio to FLAC, 193 - sends to Gemini with a transcription prompt, and returns normalized 194 - statements with speaker diarization. 279 + When speech_segments is provided (from VAD), sends audio as labeled clips 280 + with explicit timestamps. Gemini returns absolute MM:SS timestamps which 281 + are mapped back to the audio timeline. 195 282 196 283 Args: 197 284 audio: Audio buffer (float32, mono) 198 285 sample_rate: Sample rate in Hz (typically 16000) 199 286 config: Backend configuration dict (currently unused) 287 + speech_segments: Optional list of (start, end) tuples from VAD. 288 + When provided, enables clip-based transcription for better 289 + timestamp accuracy. 200 290 201 291 Returns: 202 292 List of statements with id, start, end, text, speaker. 203 293 """ 204 294 audio_duration = len(audio) / sample_rate 295 + use_chunks = speech_segments is not None and len(speech_segments) > 0 205 296 206 - logger.info(f"Transcribing audio with Gemini ({audio_duration:.1f}s)...") 297 + if use_chunks: 298 + logger.info( 299 + f"Transcribing audio with Gemini ({audio_duration:.1f}s, " 300 + f"{len(speech_segments)} clips)..." 301 + ) 302 + else: 303 + logger.info(f"Transcribing audio with Gemini ({audio_duration:.1f}s)...") 304 + 207 305 t0 = time.perf_counter() 208 - 209 - # Convert audio to FLAC bytes 210 - audio_bytes = audio_to_flac_bytes(audio, sample_rate) 211 306 212 307 # Load prompt from gemini.md 213 308 prompt_text = load_prompt("gemini", base_dir=Path(__file__).parent).text 214 309 215 - # Build contents: prompt text + audio 216 - contents: list = [ 217 - prompt_text, 218 - types.Part.from_bytes(data=audio_bytes, mime_type="audio/flac"), 219 - ] 310 + # Build contents based on mode 311 + if use_chunks: 312 + contents = _build_chunk_contents( 313 + audio, sample_rate, speech_segments, prompt_text 314 + ) 315 + else: 316 + # Legacy single-audio mode (for backwards compatibility) 317 + audio_bytes = audio_to_flac_bytes(audio, sample_rate) 318 + contents = [ 319 + prompt_text, 320 + types.Part.from_bytes(data=audio_bytes, mime_type="audio/flac"), 321 + ] 220 322 221 323 # Call Gemini via think.models.generate() 222 324 response_text = generate( ··· 244 346 segments = [] 245 347 246 348 # Normalize to standard statement format 247 - statements, invalid_count = _normalize_segments(segments, audio_duration) 248 - 249 - if invalid_count > 0: 250 - logger.warning( 251 - f" {invalid_count} segment(s) had invalid timestamps " 252 - "(will be saved but not embedded)" 349 + if use_chunks: 350 + statements = _normalize_chunked_segments(segments, speech_segments) 351 + else: 352 + # Legacy mode 353 + statements = _normalize_chunked_segments( 354 + segments, 355 + [(0.0, audio_duration)], # Single chunk covering entire audio 253 356 ) 254 357 255 358 logger.info(

+25 -5

observe/transcribe/main.py

··· 461 461 except json.JSONDecodeError: 462 462 logging.warning(f"Invalid SEGMENT_META JSON: {segment_meta_str[:100]}") 463 463 464 - # Use reduced audio for STT if available, otherwise full buffer 465 - if reduced_audio is not None: 464 + # Gemini uses chunk-based transcription with VAD segments for timestamp accuracy 465 + # Other backends use reduced audio with post-hoc timestamp restoration 466 + use_gemini_chunks = backend == "gemini" and vad_result.speech_segments 467 + 468 + if use_gemini_chunks: 469 + # Gemini: use full audio buffer with VAD segments for chunking 470 + stt_buffer = audio_buffer 471 + elif reduced_audio is not None: 472 + # Other backends: use reduced audio 466 473 stt_buffer = reduced_audio 467 474 else: 468 475 stt_buffer = audio_buffer 469 476 470 477 try: 471 478 # Dispatch to STT backend 472 - statements = stt_transcribe(backend, stt_buffer, SAMPLE_RATE, backend_config) 479 + if use_gemini_chunks: 480 + # Pass VAD segments to Gemini for chunk-based transcription 481 + statements = stt_transcribe( 482 + backend, 483 + stt_buffer, 484 + SAMPLE_RATE, 485 + backend_config, 486 + speech_segments=vad_result.speech_segments, 487 + ) 488 + else: 489 + statements = stt_transcribe( 490 + backend, stt_buffer, SAMPLE_RATE, backend_config 491 + ) 473 492 474 493 # Get model info for metadata (dynamic import based on backend) 475 494 backend_module = get_backend(backend) ··· 535 554 stt_buffer, statements, SAMPLE_RATE, model_info.get("device", "cpu") 536 555 ) 537 556 538 - # Restore original timestamps if audio was reduced 539 - if reduction: 557 + # Restore original timestamps if audio was reduced (non-Gemini backends only) 558 + # Gemini with chunks already has timestamps in original audio time 559 + if reduction and not use_gemini_chunks: 540 560 statements = restore_statement_timestamps(statements, reduction) 541 561 logging.info( 542 562 f" Restored timestamps from reduced audio "

+157 -110

tests/test_transcribe_gemini.py

··· 3 3 4 4 """Tests for the Gemini STT backend.""" 5 5 6 + import numpy as np 7 + 6 8 from observe.transcribe.gemini import ( 7 - _normalize_segments, 9 + _build_chunk_contents, 10 + _find_segment_for_timestamp, 11 + _format_timestamp, 12 + _normalize_chunked_segments, 8 13 _parse_speaker, 9 14 _parse_timestamp, 10 15 get_model_info, 11 16 ) 12 17 13 18 19 + class TestFormatTimestamp: 20 + """Tests for _format_timestamp function.""" 21 + 22 + def test_basic_formatting(self): 23 + """Basic timestamp formatting.""" 24 + assert _format_timestamp(0) == "00:00" 25 + assert _format_timestamp(5) == "00:05" 26 + assert _format_timestamp(65) == "01:05" 27 + assert _format_timestamp(3600) == "60:00" 28 + 29 + def test_minutes_and_seconds(self): 30 + """Minutes and seconds formatting.""" 31 + assert _format_timestamp(90) == "01:30" 32 + assert _format_timestamp(125) == "02:05" 33 + assert _format_timestamp(599) == "09:59" 34 + 35 + 14 36 class TestParseTimestamp: 15 37 """Tests for _parse_timestamp function.""" 16 38 17 39 def test_mm_ss_format(self): 18 - """Standard MM:SS format.""" 19 - assert _parse_timestamp("1:23") == 83.0 40 + """MM:SS format.""" 41 + assert _parse_timestamp("01:23") == 83.0 20 42 assert _parse_timestamp("0:05") == 5.0 21 43 assert _parse_timestamp("10:30") == 630.0 22 - 23 - def test_hh_mm_ss_format(self): 24 - """HH:MM:SS format.""" 25 - assert _parse_timestamp("1:05:30") == 3930.0 26 - assert _parse_timestamp("0:10:00") == 600.0 27 - assert _parse_timestamp("2:00:00") == 7200.0 28 44 29 45 def test_just_seconds(self): 30 - """Just seconds (no colons).""" 46 + """Just seconds.""" 31 47 assert _parse_timestamp("5") == 5.0 32 - assert _parse_timestamp("0") == 0.0 33 48 assert _parse_timestamp("123") == 123.0 34 49 35 - def test_zero_timestamp(self): 36 - """Zero timestamps.""" 37 - assert _parse_timestamp("0:00") == 0.0 38 - assert _parse_timestamp("0:00:00") == 0.0 39 - assert _parse_timestamp("0") == 0.0 40 - 41 - def test_whitespace_handling(self): 42 - """Whitespace should be stripped.""" 43 - assert _parse_timestamp(" 1:23 ") == 83.0 44 - assert _parse_timestamp("\t0:05\n") == 5.0 45 - 46 - def test_fractional_seconds(self): 47 - """Fractional seconds.""" 48 - assert _parse_timestamp("1:23.5") == 83.5 49 - assert _parse_timestamp("0:05.25") == 5.25 50 - 51 50 def test_invalid_returns_none(self): 52 51 """Invalid timestamps return None.""" 53 52 assert _parse_timestamp("") is None 54 53 assert _parse_timestamp(None) is None 55 54 assert _parse_timestamp("invalid") is None 56 - assert _parse_timestamp("abc:def") is None 57 - assert _parse_timestamp("1:2:3:4") is None # Too many parts 58 55 59 - def test_negative_clamped_to_zero(self): 60 - """Negative values are clamped to 0.""" 61 - # This is an edge case - negative minutes/seconds shouldn't happen 62 - # but if somehow parsed, we clamp to 0 63 - assert _parse_timestamp("-5") == 0.0 56 + def test_whitespace_stripped(self): 57 + """Whitespace is stripped.""" 58 + assert _parse_timestamp(" 01:23 ") == 83.0 64 59 65 60 66 61 class TestParseSpeaker: ··· 99 94 assert _parse_speaker("") is None 100 95 101 96 102 - class TestNormalizeSegments: 103 - """Tests for _normalize_segments function.""" 97 + class TestFindSegmentForTimestamp: 98 + """Tests for _find_segment_for_timestamp function.""" 104 99 105 - def test_basic_normalization(self): 106 - """Basic segment normalization.""" 100 + def test_timestamp_inside_segment(self): 101 + """Timestamp inside a segment returns that segment.""" 102 + segments = [(0.0, 10.0), (15.0, 25.0), (30.0, 40.0)] 103 + assert _find_segment_for_timestamp(5.0, segments) == (0.0, 10.0) 104 + assert _find_segment_for_timestamp(20.0, segments) == (15.0, 25.0) 105 + assert _find_segment_for_timestamp(35.0, segments) == (30.0, 40.0) 106 + 107 + def test_timestamp_at_boundary(self): 108 + """Timestamp at segment boundary returns that segment.""" 109 + segments = [(0.0, 10.0), (15.0, 25.0)] 110 + assert _find_segment_for_timestamp(0.0, segments) == (0.0, 10.0) 111 + assert _find_segment_for_timestamp(10.0, segments) == (0.0, 10.0) 112 + assert _find_segment_for_timestamp(15.0, segments) == (15.0, 25.0) 113 + 114 + def test_timestamp_in_gap_returns_nearest(self): 115 + """Timestamp in gap returns nearest segment.""" 116 + segments = [(0.0, 10.0), (20.0, 30.0)] 117 + # 12 is closer to segment ending at 10 than starting at 20 118 + assert _find_segment_for_timestamp(12.0, segments) == (0.0, 10.0) 119 + # 18 is closer to segment starting at 20 120 + assert _find_segment_for_timestamp(18.0, segments) == (20.0, 30.0) 121 + 122 + def test_timestamp_before_first(self): 123 + """Timestamp before first segment returns first.""" 124 + segments = [(10.0, 20.0), (30.0, 40.0)] 125 + assert _find_segment_for_timestamp(5.0, segments) == (10.0, 20.0) 126 + 127 + def test_timestamp_after_last(self): 128 + """Timestamp after last segment returns last.""" 129 + segments = [(0.0, 10.0), (15.0, 25.0)] 130 + assert _find_segment_for_timestamp(50.0, segments) == (15.0, 25.0) 131 + 132 + 133 + class TestNormalizeChunkedSegments: 134 + """Tests for _normalize_chunked_segments function.""" 135 + 136 + def test_parses_mm_ss_timestamps(self): 137 + """Parses MM:SS timestamps from Gemini output.""" 138 + segments = [ 139 + {"start": "00:05", "speaker": "Speaker 1", "text": "Hello"}, 140 + {"start": "00:12", "speaker": "Speaker 2", "text": "Hi there"}, 141 + ] 142 + speech_segments = [(0.0, 10.0), (10.0, 20.0)] 143 + 144 + statements = _normalize_chunked_segments(segments, speech_segments) 145 + 146 + assert len(statements) == 2 147 + assert statements[0]["start"] == 5.0 148 + assert statements[0]["text"] == "Hello" 149 + assert statements[0]["speaker"] == 1 150 + assert statements[1]["start"] == 12.0 151 + 152 + def test_clamps_timestamp_to_valid_range(self): 153 + """Clamps timestamps to valid range.""" 107 154 segments = [ 108 - { 109 - "start": "0:05", 110 - "end": "0:12", 111 - "speaker": "Speaker 1", 112 - "text": "Hello there", 113 - } 155 + {"start": "10:00", "speaker": "Speaker 1", "text": "Way too late"}, 114 156 ] 157 + speech_segments = [(0.0, 10.0), (15.0, 25.0)] 115 158 116 - statements, invalid_count = _normalize_segments(segments, 60.0) 159 + statements = _normalize_chunked_segments(segments, speech_segments) 117 160 118 - assert len(statements) == 1 119 - assert invalid_count == 0 120 - stmt = statements[0] 121 - assert stmt["id"] == 1 122 - assert stmt["start"] == 5.0 123 - assert stmt["end"] == 12.0 124 - assert stmt["text"] == "Hello there" 125 - assert stmt["speaker"] == 1 126 - assert stmt["words"] is None 161 + # Should clamp to max_time (25.0) 162 + assert statements[0]["start"] == 25.0 127 163 128 - def test_multiple_segments(self): 129 - """Multiple segments get sequential IDs.""" 164 + def test_fallback_on_invalid_timestamp(self): 165 + """Falls back to first segment on invalid timestamp.""" 130 166 segments = [ 131 - {"start": "0:00", "end": "0:10", "text": "First"}, 132 - {"start": "0:15", "end": "0:25", "text": "Second"}, 133 - {"start": "0:30", "end": "0:40", "text": "Third"}, 167 + {"start": "invalid", "speaker": "Speaker 1", "text": "Test"}, 134 168 ] 169 + speech_segments = [(5.0, 15.0), (20.0, 30.0)] 135 170 136 - statements, invalid_count = _normalize_segments(segments, 60.0) 171 + statements = _normalize_chunked_segments(segments, speech_segments) 137 172 138 - assert len(statements) == 3 139 - assert invalid_count == 0 140 - assert [s["id"] for s in statements] == [1, 2, 3] 173 + # Falls back to first segment start 174 + assert statements[0]["start"] == 5.0 141 175 142 - def test_clamps_to_audio_duration(self): 143 - """Timestamps beyond audio duration are clamped.""" 176 + def test_assigns_end_from_containing_segment(self): 177 + """End time comes from the segment containing the start.""" 144 178 segments = [ 145 - {"start": "0:00", "end": "2:00", "text": "Goes past end"} # 120s > 60s 179 + {"start": "00:22", "speaker": "Speaker 1", "text": "In second segment"}, 146 180 ] 181 + speech_segments = [(0.0, 10.0), (20.0, 30.0)] 147 182 148 - statements, invalid_count = _normalize_segments(segments, 60.0) 183 + statements = _normalize_chunked_segments(segments, speech_segments) 149 184 150 - assert len(statements) == 1 151 - assert statements[0]["end"] == 60.0 # Clamped to duration 185 + assert statements[0]["start"] == 22.0 186 + assert statements[0]["end"] == 30.0 # End of containing segment 152 187 153 - def test_invalid_timestamps_counted(self): 154 - """Invalid timestamps are counted but segment is still included.""" 188 + def test_empty_text_dropped(self): 189 + """Segments with empty text are dropped.""" 155 190 segments = [ 156 - {"start": "invalid", "end": "0:10", "text": "Bad start"}, 157 - {"start": "0:00", "end": "bad", "text": "Bad end"}, 158 - {"start": "0:00", "end": "0:10", "text": "Good one"}, 191 + {"start": "00:05", "text": "First"}, 192 + {"start": "00:10", "text": ""}, 193 + {"start": "00:15", "text": " "}, 194 + {"start": "00:20", "text": "Last"}, 159 195 ] 196 + speech_segments = [(0.0, 30.0)] 160 197 161 - statements, invalid_count = _normalize_segments(segments, 60.0) 198 + statements = _normalize_chunked_segments(segments, speech_segments) 162 199 163 - # All segments included, but 2 have invalid timestamps 164 - assert len(statements) == 3 165 - assert invalid_count == 2 166 - assert statements[0]["start"] is None 167 - assert statements[1]["end"] is None 168 - assert statements[2]["start"] == 0.0 169 - assert statements[2]["end"] == 10.0 200 + assert len(statements) == 2 201 + assert statements[0]["text"] == "First" 202 + assert statements[1]["text"] == "Last" 170 203 171 - def test_missing_optional_fields(self): 172 - """Missing optional fields are handled gracefully.""" 173 - segments = [{"start": "0:00", "end": "0:10", "text": "Minimal"}] 204 + def test_sequential_ids(self): 205 + """Statements get sequential IDs.""" 206 + segments = [ 207 + {"start": "00:05", "text": "First"}, 208 + {"start": "00:10", "text": "Second"}, 209 + {"start": "00:15", "text": "Third"}, 210 + ] 211 + speech_segments = [(0.0, 30.0)] 174 212 175 - statements, invalid_count = _normalize_segments(segments, 60.0) 213 + statements = _normalize_chunked_segments(segments, speech_segments) 176 214 177 - assert len(statements) == 1 178 - assert "speaker" not in statements[0] 215 + assert [s["id"] for s in statements] == [1, 2, 3] 179 216 180 217 def test_empty_segments(self): 181 218 """Empty segments list.""" 182 - statements, invalid_count = _normalize_segments([], 60.0) 219 + statements = _normalize_chunked_segments([], [(0.0, 10.0)]) 183 220 assert statements == [] 184 - assert invalid_count == 0 185 221 186 - def test_whitespace_text_stripped(self): 187 - """Text is stripped of whitespace.""" 188 - segments = [{"start": "0:00", "end": "0:10", "text": " Hello "}] 189 222 190 - statements, invalid_count = _normalize_segments(segments, 60.0) 223 + class TestBuildChunkContents: 224 + """Tests for _build_chunk_contents function.""" 191 225 192 - assert statements[0]["text"] == "Hello" 226 + def test_basic_chunking(self): 227 + """Basic chunk content building.""" 228 + audio = np.zeros(16000 * 30, dtype=np.float32) # 30s of audio 229 + speech_segments = [(0.0, 10.0), (15.0, 25.0)] 193 230 194 - def test_empty_text_dropped(self): 195 - """Segments with empty text are dropped.""" 196 - segments = [ 197 - {"start": "0:00", "end": "0:10", "text": "First"}, 198 - {"start": "0:15", "end": "0:20", "text": ""}, 199 - {"start": "0:25", "end": "0:30", "text": " "}, # whitespace only 200 - {"start": "0:35", "end": "0:40", "text": "Last"}, 201 - ] 231 + contents = _build_chunk_contents(audio, 16000, speech_segments, "Test prompt") 232 + 233 + # Should have: prompt + (label + audio) * 2 = 5 items 234 + assert len(contents) == 5 235 + assert contents[0] == "Test prompt" 236 + assert contents[1] == "Clip starting at 00:00 (10s):" 237 + assert contents[3] == "Clip starting at 00:15 (10s):" 238 + 239 + def test_duration_in_label(self): 240 + """Label includes duration.""" 241 + audio = np.zeros(16000 * 30, dtype=np.float32) 242 + speech_segments = [(5.0, 12.0)] # 7 second clip 243 + 244 + contents = _build_chunk_contents(audio, 16000, speech_segments, "Prompt") 245 + 246 + assert contents[1] == "Clip starting at 00:05 (7s):" 247 + 248 + def test_skips_empty_chunks(self): 249 + """Empty audio chunks are skipped.""" 250 + audio = np.zeros(16000 * 10, dtype=np.float32) 251 + # Second segment has start == end (empty) 252 + speech_segments = [(0.0, 5.0), (5.0, 5.0), (7.0, 10.0)] 202 253 203 - statements, invalid_count = _normalize_segments(segments, 60.0) 254 + contents = _build_chunk_contents(audio, 16000, speech_segments, "Prompt") 204 255 205 - # Only non-empty segments kept, IDs renumbered 206 - assert len(statements) == 2 207 - assert statements[0]["text"] == "First" 208 - assert statements[0]["id"] == 1 209 - assert statements[1]["text"] == "Last" 210 - assert statements[1]["id"] == 2 256 + # Should have: prompt + 2 valid chunks * 2 = 5 items 257 + assert len(contents) == 5 211 258 212 259 213 260 class TestGetModelInfo:

Configure Feed

Configure Feed