diff --git a/README.md b/README.md index 31f6224..677c816 100644 --- a/README.md +++ b/README.md @@ -123,6 +123,28 @@ python scripts/asr/transcribe_file_offline.py \ --input-file data/examples/en-US_AntiBERTa_for_word_boosting_testing.wav ``` +Streaming, offline, and Realtime file transcription can write finalized results as +JSON, SubRip, or WebVTT. The format is inferred from the output extension unless +`--output-format` is provided. SRT and VTT outputs automatically request word time +offsets from the ASR model. +```bash +python scripts/asr/transcribe_file.py \ + --input-file data/examples/en-US_AntiBERTa_for_word_boosting_testing.wav \ + --output-file transcript.srt + +python scripts/asr/transcribe_file_offline.py \ + --input-file data/examples/en-US_AntiBERTa_for_word_boosting_testing.wav \ + --output-file transcript.json + +python scripts/asr/realtime_asr_client.py \ + --input-file data/examples/en-US_AntiBERTa_for_word_boosting_testing.wav \ + --output-file transcript.vtt +``` + +JSON export remains available when a model does not return word timestamps. SRT and +VTT export require timestamp support and fail with an explicit error if a finalized +result does not contain word offsets. + You can improve transcription of this audio by word boosting. ```bash python scripts/asr/transcribe_file_offline.py \ diff --git a/riva/client/__init__.py b/riva/client/__init__.py index 7656bd6..ad18348 100644 --- a/riva/client/__init__.py +++ b/riva/client/__init__.py @@ -15,6 +15,17 @@ add_custom_configuration_to_config, ) from riva.client.auth import Auth +from riva.client.transcript import ( + TRANSCRIPT_OUTPUT_FORMATS, + Transcript, + TranscriptAlternative, + TranscriptSegment, + TranscriptWord, + collect_streaming_transcript, + resolve_output_format, + transcript_from_offline_response, + write_transcript, +) from riva.client.nlp import ( NLPService, extract_all_text_classes_and_confidences, diff --git a/riva/client/realtime.py b/riva/client/realtime.py index 58f29af..0d2c50f 100644 --- a/riva/client/realtime.py +++ b/riva/client/realtime.py @@ -14,6 +14,8 @@ import ssl from websockets.exceptions import WebSocketException +from riva.client.transcript import Transcript, write_transcript + logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" @@ -151,6 +153,7 @@ def __init__(self, args: argparse.Namespace): self.is_input_playing = False self.input_buffer_size = 1024 # Buffer size for input audio playback self.final_transcript: str = "" + self.transcript = Transcript() self.is_config_updated = False self._force_eou_pending = False @@ -547,7 +550,8 @@ async def receive_responses(self): elif event_type == "conversation.item.input_audio_transcription.completed": is_last_result = event.get("is_last_result", False) interim_final_transcript = event.get("transcript", "") - self.final_transcript = interim_final_transcript + self.transcript.add_realtime_event(event) + self.final_transcript = self.transcript.text if is_last_result: logger.info("Final Transcript: %s", self.final_transcript) @@ -608,6 +612,10 @@ def save_responses(self, output_text_file: str): except Exception as e: logger.error("Error saving text: %s", e) + def save_transcript(self, output_file: str, output_format: Optional[str] = None): + """Save collected finalized responses as JSON, SRT, or WebVTT.""" + write_transcript(self.transcript, output_file, output_format) + async def disconnect(self): """Close the WebSocket connection.""" if self.websocket: diff --git a/riva/client/transcript.py b/riva/client/transcript.py new file mode 100644 index 0000000..ab2e682 --- /dev/null +++ b/riva/client/transcript.py @@ -0,0 +1,271 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: MIT + +"""Normalized ASR transcripts and file export helpers.""" + +import json +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Union + +import riva.client.proto.riva_asr_pb2 as rasr + + +TRANSCRIPT_OUTPUT_FORMATS = ("json", "srt", "vtt") + + +@dataclass +class TranscriptWord: + word: str + start_time_ms: int + end_time_ms: int + confidence: float = 0.0 + speaker_tag: int = 0 + language_code: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "word": self.word, + "start_time_ms": self.start_time_ms, + "end_time_ms": self.end_time_ms, + "confidence": self.confidence, + "speaker_tag": self.speaker_tag, + "language_code": self.language_code, + } + + +@dataclass +class TranscriptAlternative: + transcript: str + confidence: float = 0.0 + words: List[TranscriptWord] = field(default_factory=list) + language_codes: List[str] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "transcript": self.transcript, + "confidence": self.confidence, + "words": [word.to_dict() for word in self.words], + "language_codes": self.language_codes, + } + + +@dataclass +class TranscriptSegment: + transcript: str + start_time_ms: Optional[int] = None + end_time_ms: Optional[int] = None + confidence: float = 0.0 + speaker_tag: Optional[int] = None + language_code: str = "" + words: List[TranscriptWord] = field(default_factory=list) + alternatives: List[TranscriptAlternative] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "transcript": self.transcript, + "start_time_ms": self.start_time_ms, + "end_time_ms": self.end_time_ms, + "confidence": self.confidence, + "speaker_tag": self.speaker_tag, + "language_code": self.language_code, + "words": [word.to_dict() for word in self.words], + "alternatives": [ + alternative.to_dict() for alternative in self.alternatives + ], + } + + +@dataclass +class Transcript: + segments: List[TranscriptSegment] = field(default_factory=list) + + @property + def text(self) -> str: + return " ".join( + segment.transcript.strip() + for segment in self.segments + if segment.transcript.strip() + ) + + def add_grpc_result( + self, + result: Union[rasr.SpeechRecognitionResult, rasr.StreamingRecognitionResult], + ) -> None: + if not result.alternatives: + return + alternatives = [ + _grpc_alternative(alternative) for alternative in result.alternatives + ] + top_alternative = alternatives[0] + self.segments.append(_segment_from_alternative(top_alternative, alternatives)) + + def add_realtime_event(self, event: Dict[str, Any]) -> None: + transcript = event.get("transcript", "") + words = [ + _realtime_word(word) + for word in event.get("words_info", {}).get("words", []) + ] + alternative = TranscriptAlternative(transcript=transcript, words=words) + segment = _segment_from_alternative(alternative, [alternative]) + + # Some servers repeat the final completed event with is_last_result=true. + # Avoid emitting an identical subtitle cue twice. + if self.segments and self.segments[-1].to_dict() == segment.to_dict(): + return + self.segments.append(segment) + + def to_dict(self) -> Dict[str, Any]: + return { + "schema_version": "1.0", + "transcript": self.text, + "segments": [segment.to_dict() for segment in self.segments], + } + + +def collect_streaming_transcript( + responses: Iterable[rasr.StreamingRecognizeResponse], + transcript: Transcript, +) -> Iterable[rasr.StreamingRecognizeResponse]: + """Collect final results while yielding every response unchanged.""" + for response in responses: + for result in response.results: + if result.is_final: + transcript.add_grpc_result(result) + yield response + + +def transcript_from_offline_response(response: rasr.RecognizeResponse) -> Transcript: + transcript = Transcript() + for result in response.results: + transcript.add_grpc_result(result) + return transcript + + +def resolve_output_format( + output_file: Union[str, os.PathLike], + output_format: Optional[str] = None, +) -> str: + if output_format: + normalized_format = output_format.lower() + else: + normalized_format = Path(output_file).suffix.lower().lstrip(".") + if normalized_format not in TRANSCRIPT_OUTPUT_FORMATS: + raise ValueError( + "Unable to determine transcript output format. Use a .json, .srt, or .vtt " + "file extension, or pass --output-format." + ) + return normalized_format + + +def write_transcript( + transcript: Transcript, + output_file: Union[str, os.PathLike], + output_format: Optional[str] = None, +) -> None: + output_format = resolve_output_format(output_file, output_format) + output_path = Path(output_file).expanduser() + if output_format == "json": + rendered = json.dumps(transcript.to_dict(), ensure_ascii=False, indent=2) + "\n" + elif output_format == "srt": + rendered = _render_subtitles(transcript, webvtt=False) + else: + rendered = _render_subtitles(transcript, webvtt=True) + output_path.write_text(rendered, encoding="utf-8") + + +def _grpc_word(word: rasr.WordInfo) -> TranscriptWord: + return TranscriptWord( + word=word.word, + start_time_ms=word.start_time, + end_time_ms=word.end_time, + confidence=word.confidence, + speaker_tag=word.speaker_tag, + language_code=word.language_code, + ) + + +def _grpc_alternative( + alternative: rasr.SpeechRecognitionAlternative, +) -> TranscriptAlternative: + return TranscriptAlternative( + transcript=alternative.transcript, + confidence=alternative.confidence, + words=[_grpc_word(word) for word in alternative.words], + language_codes=list(alternative.language_code), + ) + + +def _realtime_word(word: Dict[str, Any]) -> TranscriptWord: + # Realtime ASR word offsets are seconds; gRPC word offsets are milliseconds. + return TranscriptWord( + word=word.get("word", ""), + start_time_ms=round(float(word.get("start_time", 0.0)) * 1000), + end_time_ms=round(float(word.get("end_time", 0.0)) * 1000), + confidence=float(word.get("confidence", 0.0)), + speaker_tag=int(word.get("speaker_tag", 0)), + language_code=word.get("language_code", ""), + ) + + +def _segment_from_alternative( + alternative: TranscriptAlternative, + alternatives: List[TranscriptAlternative], +) -> TranscriptSegment: + words = alternative.words + speaker_tags = {word.speaker_tag for word in words} + language_codes = {word.language_code for word in words if word.language_code} + return TranscriptSegment( + transcript=alternative.transcript, + start_time_ms=words[0].start_time_ms if words else None, + end_time_ms=words[-1].end_time_ms if words else None, + confidence=alternative.confidence, + speaker_tag=next(iter(speaker_tags)) if len(speaker_tags) == 1 else None, + language_code=next(iter(language_codes)) if len(language_codes) == 1 else "", + words=words, + alternatives=alternatives, + ) + + +def _render_subtitles(transcript: Transcript, webvtt: bool) -> str: + lines = ["WEBVTT", ""] if webvtt else [] + cue_index = 1 + for segment in transcript.segments: + text = " ".join(segment.transcript.split()) + if not text: + continue + if segment.start_time_ms is None or segment.end_time_ms is None: + raise ValueError( + "SRT and VTT export require word time offsets, but a finalized transcript " + "segment did not contain them." + ) + if segment.start_time_ms < 0 or segment.end_time_ms < segment.start_time_ms: + raise ValueError( + "SRT and VTT export require non-negative, ordered timestamps." + ) + + # Transducer models can return equal word timestamps. Keep such cues valid + # without inventing a perceptible duration. + end_time_ms = max(segment.end_time_ms, segment.start_time_ms + 1) + lines.append(str(cue_index)) + lines.append( + "{} --> {}".format( + _format_timestamp(segment.start_time_ms, webvtt), + _format_timestamp(end_time_ms, webvtt), + ) + ) + lines.append(text) + lines.append("") + cue_index += 1 + return "\n".join(lines) + + +def _format_timestamp(timestamp_ms: int, webvtt: bool) -> str: + hours, remainder = divmod(timestamp_ms, 3_600_000) + minutes, remainder = divmod(remainder, 60_000) + seconds, milliseconds = divmod(remainder, 1_000) + separator = "." if webvtt else "," + return "{:02d}:{:02d}:{:02d}{}{:03d}".format( + hours, minutes, seconds, separator, milliseconds + ) diff --git a/scripts/asr/realtime_asr_client.py b/scripts/asr/realtime_asr_client.py index 9c6a5bd..7210b69 100644 --- a/scripts/asr/realtime_asr_client.py +++ b/scripts/asr/realtime_asr_client.py @@ -8,6 +8,7 @@ from riva.client.asr import get_wav_file_parameters, AudioChunkFileIterator from riva.client.realtime import RealtimeClientASR +from riva.client.transcript import TRANSCRIPT_OUTPUT_FORMATS, resolve_output_format from riva.client.argparse_utils import ( add_asr_config_argparse_parameters, add_realtime_config_argparse_parameters, @@ -91,6 +92,15 @@ def parse_args() -> argparse.Namespace: type=str, help="Output text file" ) + parser.add_argument( + "--output-file", + help="Write the finalized transcript to a .json, .srt, or .vtt file. File input only.", + ) + parser.add_argument( + "--output-format", + choices=TRANSCRIPT_OUTPUT_FORMATS, + help="Transcript output format. By default, infer it from --output-file.", + ) parser.add_argument( "--prompt", default="", @@ -114,6 +124,20 @@ def parse_args() -> argparse.Namespace: args = parser.parse_args() + if args.output_format and not args.output_file: + parser.error("--output-format requires --output-file") + if args.output_text and args.output_file: + parser.error("--output-text and --output-file cannot be used together") + if args.output_file: + if args.mic: + parser.error("--output-file currently supports --input-file only") + try: + args.output_format = resolve_output_format(args.output_file, args.output_format) + except ValueError as error: + parser.error(str(error)) + if args.output_format in ("srt", "vtt"): + args.word_time_offsets = True + return args @@ -246,6 +270,8 @@ async def run_transcription(args): # Save results if output file specified if args.output_text: client.save_responses(args.output_text) + elif args.output_file: + client.save_transcript(args.output_file, args.output_format) except KeyboardInterrupt: if hasattr(args, '_interruptible_iterator'): diff --git a/scripts/asr/transcribe_file.py b/scripts/asr/transcribe_file.py index 160f9a5..8ca556e 100644 --- a/scripts/asr/transcribe_file.py +++ b/scripts/asr/transcribe_file.py @@ -2,10 +2,12 @@ # SPDX-License-Identifier: MIT import argparse +import importlib import os import sys import riva.client +from riva.client.transcript import TRANSCRIPT_OUTPUT_FORMATS, resolve_output_format from riva.client.argparse_utils import ( add_asr_config_argparse_parameters, add_connection_argparse_parameters, @@ -31,6 +33,15 @@ def parse_args() -> argparse.Namespace: group.add_argument("--list-models", action="store_true", help="List available models.") group.add_argument("--list-devices", action="store_true", help="List output devices indices") parser.add_argument("--output-seglst", action="store_true", help="Output seglst file for speaker diarization.") + parser.add_argument( + "--output-file", + help="Write the finalized transcript to a .json, .srt, or .vtt file.", + ) + parser.add_argument( + "--output-format", + choices=TRANSCRIPT_OUTPUT_FORMATS, + help="Transcript output format. By default, infer it from --output-file.", + ) parser.add_argument( "--show-intermediate", action="store_true", help="Show intermediate transcripts as they are available." @@ -71,8 +82,15 @@ def parse_args() -> argparse.Namespace: parser = add_connection_argparse_parameters(parser) parser = add_asr_config_argparse_parameters(parser, max_alternatives=True, profanity_filter=True, word_time_offsets=True) args = parser.parse_args() + if args.output_format and not args.output_file: + parser.error("--output-format requires --output-file") + if args.output_file: + try: + args.output_format = resolve_output_format(args.output_file, args.output_format) + except ValueError as error: + parser.error(str(error)) if args.play_audio or args.output_device is not None or args.list_devices: - import riva.client.audio_io + importlib.import_module("riva.client.audio_io") return args @@ -122,7 +140,11 @@ def main() -> int: profanity_filter=args.profanity_filter, enable_automatic_punctuation=args.automatic_punctuation, verbatim_transcripts=not args.no_verbatim_transcripts, - enable_word_time_offsets=args.word_time_offsets or args.speaker_diarization, + enable_word_time_offsets=( + args.word_time_offsets + or args.speaker_diarization + or args.output_format in ("srt", "vtt") + ), ), interim_results=True, ) @@ -158,17 +180,24 @@ def main() -> int: with riva.client.AudioChunkFileIterator( args.input_file, args.file_streaming_chunk, delay_callback, ) as audio_chunk_iterator: + responses = asr_service.streaming_response_generator( + audio_chunks=audio_chunk_iterator, + streaming_config=config, + ) + transcript = None + if args.output_file: + transcript = riva.client.Transcript() + responses = riva.client.collect_streaming_transcript(responses, transcript) riva.client.print_streaming( - responses=asr_service.streaming_response_generator( - audio_chunks=audio_chunk_iterator, - streaming_config=config, - ), + responses=responses, show_intermediate=args.show_intermediate, additional_info="time" if (args.word_time_offsets or args.speaker_diarization) else ("confidence" if args.print_confidence else "no"), word_time_offsets=args.word_time_offsets or args.speaker_diarization, speaker_diarization=args.speaker_diarization, seglst_output_file=seglst_output_file, ) + if transcript is not None: + riva.client.write_transcript(transcript, args.output_file, args.output_format) finally: if sound_callback is not None and sound_callback.opened: sound_callback.close() diff --git a/scripts/asr/transcribe_file_offline.py b/scripts/asr/transcribe_file_offline.py index f637b0b..2866350 100644 --- a/scripts/asr/transcribe_file_offline.py +++ b/scripts/asr/transcribe_file_offline.py @@ -7,6 +7,7 @@ from pathlib import Path import riva.client +from riva.client.transcript import TRANSCRIPT_OUTPUT_FORMATS, resolve_output_format from riva.client.argparse_utils import ( add_asr_config_argparse_parameters, add_connection_argparse_parameters, @@ -30,10 +31,26 @@ def parse_args() -> argparse.Namespace: group.add_argument("--input-file", type=Path, help="A path to a local file to transcribe.") group.add_argument("--list-models", action="store_true", help="List available models.") parser.add_argument("--output-seglst", action="store_true", help="Output seglst file for speaker diarization.") + parser.add_argument( + "--output-file", + help="Write the finalized transcript to a .json, .srt, or .vtt file.", + ) + parser.add_argument( + "--output-format", + choices=TRANSCRIPT_OUTPUT_FORMATS, + help="Transcript output format. By default, infer it from --output-file.", + ) parser = add_connection_argparse_parameters(parser) parser = add_asr_config_argparse_parameters(parser, max_alternatives=True, profanity_filter=True, word_time_offsets=True) args = parser.parse_args() + if args.output_format and not args.output_file: + parser.error("--output-format requires --output-file") + if args.output_file: + try: + args.output_format = resolve_output_format(args.output_file, args.output_format) + except ValueError as error: + parser.error(str(error)) if args.input_file: args.input_file = args.input_file.expanduser() return args @@ -82,7 +99,11 @@ def main() -> int: profanity_filter=args.profanity_filter, enable_automatic_punctuation=args.automatic_punctuation, verbatim_transcripts=not args.no_verbatim_transcripts, - enable_word_time_offsets=args.word_time_offsets or args.speaker_diarization, + enable_word_time_offsets=( + args.word_time_offsets + or args.speaker_diarization + or args.output_format in ("srt", "vtt") + ), ) riva.client.add_word_boosting_to_config(config, args.boosted_lm_words, args.boosted_lm_score) riva.client.add_speaker_diarization_to_config(config, args.speaker_diarization, args.diarization_max_speakers) @@ -104,11 +125,15 @@ def main() -> int: seglst_output_file = None if args.output_seglst: seglst_output_file = os.path.basename(args.input_file).split(".")[0] + response = asr_service.offline_recognize(data, config) riva.client.print_offline( - response=asr_service.offline_recognize(data, config), + response=response, speaker_diarization=args.speaker_diarization, seglst_output_file=seglst_output_file, ) + if args.output_file: + transcript = riva.client.transcript_from_offline_response(response) + riva.client.write_transcript(transcript, args.output_file, args.output_format) if __name__ == "__main__": diff --git a/tests/unit/test_realtime.py b/tests/unit/test_realtime.py index ae773bd..60a6261 100644 --- a/tests/unit/test_realtime.py +++ b/tests/unit/test_realtime.py @@ -120,3 +120,36 @@ def test_update_session_maps_no_verbatim_transcripts_to_itn(): request = client._send_message.await_args.args[0] assert request["session"]["recognition_config"]["enable_verbatim_transcripts"] is False + + +def test_receive_responses_accumulates_completed_transcripts(): + client = RealtimeClientASR(argparse.Namespace(word_time_offsets=False)) + client.websocket = AsyncMock() + client.websocket.recv.side_effect = [ + json.dumps( + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "hello", + "words_info": { + "words": [{"word": "hello", "start_time": 0.0, "end_time": 0.5}] + }, + "is_last_result": False, + } + ), + json.dumps( + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "world", + "words_info": { + "words": [{"word": "world", "start_time": 0.5, "end_time": 1.0}] + }, + "is_last_result": True, + } + ), + ] + + asyncio.run(client.receive_responses()) + + assert client.final_transcript == "hello world" + assert len(client.transcript.segments) == 2 + assert client.transcript.segments[1].end_time_ms == 1000 diff --git a/tests/unit/test_transcript.py b/tests/unit/test_transcript.py new file mode 100644 index 0000000..22e6d7a --- /dev/null +++ b/tests/unit/test_transcript.py @@ -0,0 +1,164 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: MIT + +import json + +import pytest + +import riva.client.proto.riva_asr_pb2 as rasr +from riva.client.transcript import ( + Transcript, + collect_streaming_transcript, + resolve_output_format, + transcript_from_offline_response, + write_transcript, +) + + +def make_alternative(text="hello", start_time=1000, end_time=1500): + return rasr.SpeechRecognitionAlternative( + transcript=text, + confidence=0.9, + words=[ + rasr.WordInfo( + word=text, + start_time=start_time, + end_time=end_time, + confidence=0.8, + speaker_tag=1, + language_code="en-US", + ) + ], + language_code=["en-US"], + ) + + +def test_collect_streaming_transcript_ignores_interim_results(): + transcript = Transcript() + responses = [ + rasr.StreamingRecognizeResponse( + results=[ + rasr.StreamingRecognitionResult( + alternatives=[make_alternative("partial")] + ) + ] + ), + rasr.StreamingRecognizeResponse( + results=[ + rasr.StreamingRecognitionResult( + alternatives=[make_alternative("final")], + is_final=True, + ) + ] + ), + ] + + assert list(collect_streaming_transcript(responses, transcript)) == responses + assert transcript.text == "final" + assert len(transcript.segments) == 1 + + +def test_offline_response_preserves_alternatives(): + response = rasr.RecognizeResponse( + results=[ + rasr.SpeechRecognitionResult( + alternatives=[make_alternative(), make_alternative("yellow")] + ) + ] + ) + + transcript = transcript_from_offline_response(response) + + assert transcript.text == "hello" + assert [ + alternative.transcript for alternative in transcript.segments[0].alternatives + ] == ["hello", "yellow"] + + +def test_realtime_event_converts_seconds_to_milliseconds_and_accumulates(): + transcript = Transcript() + transcript.add_realtime_event( + { + "transcript": "hello", + "words_info": { + "words": [ + { + "word": "hello", + "start_time": 1.25, + "end_time": 1.75, + "confidence": 0.9, + "speaker_tag": 2, + } + ] + }, + } + ) + transcript.add_realtime_event({"transcript": "world", "words_info": {"words": []}}) + + assert transcript.text == "hello world" + assert transcript.segments[0].start_time_ms == 1250 + assert transcript.segments[0].end_time_ms == 1750 + + +def test_realtime_event_deduplicates_repeated_final_event(): + transcript = Transcript() + event = {"transcript": "hello", "words_info": {"words": []}} + + transcript.add_realtime_event(event) + transcript.add_realtime_event(event) + + assert len(transcript.segments) == 1 + + +def test_json_export_is_one_valid_document(tmp_path): + transcript = transcript_from_offline_response( + rasr.RecognizeResponse( + results=[rasr.SpeechRecognitionResult(alternatives=[make_alternative()])] + ) + ) + output_file = tmp_path / "transcript.json" + + write_transcript(transcript, output_file) + + exported = json.loads(output_file.read_text(encoding="utf-8")) + assert exported["schema_version"] == "1.0" + assert exported["transcript"] == "hello" + assert exported["segments"][0]["words"][0]["start_time_ms"] == 1000 + + +@pytest.mark.parametrize( + "extension, expected_timestamp, expected_prefix", + [ + ("srt", "00:00:01,000 --> 00:00:01,500", "1\n"), + ("vtt", "00:00:01.000 --> 00:00:01.500", "WEBVTT\n\n1\n"), + ], +) +def test_subtitle_export(extension, expected_timestamp, expected_prefix, tmp_path): + transcript = transcript_from_offline_response( + rasr.RecognizeResponse( + results=[rasr.SpeechRecognitionResult(alternatives=[make_alternative()])] + ) + ) + output_file = tmp_path / ("transcript." + extension) + + write_transcript(transcript, output_file) + + rendered = output_file.read_text(encoding="utf-8") + assert rendered.startswith(expected_prefix) + assert expected_timestamp in rendered + assert rendered.endswith("hello\n") + + +def test_subtitle_export_rejects_missing_word_timestamps(tmp_path): + transcript = Transcript() + transcript.add_realtime_event({"transcript": "hello"}) + + with pytest.raises(ValueError, match="require word time offsets"): + write_transcript(transcript, tmp_path / "transcript.srt") + + +def test_resolve_output_format_requires_supported_extension(): + assert resolve_output_format("transcript.SRT") == "srt" + assert resolve_output_format("transcript.data", "vtt") == "vtt" + with pytest.raises(ValueError, match="Unable to determine"): + resolve_output_format("transcript.txt")