diff --git a/litellm/llms/elevenlabs/audio_transcription/transformation.py b/litellm/llms/elevenlabs/audio_transcription/transformation.py index 04d2717d64f..2a548e13dbb 100644 --- a/litellm/llms/elevenlabs/audio_transcription/transformation.py +++ b/litellm/llms/elevenlabs/audio_transcription/transformation.py @@ -2,6 +2,7 @@ Translates from OpenAI's `/v1/audio/transcriptions` to ElevenLabs's `/v1/speech-to-text` """ +import math from collections.abc import Iterable, Mapping from typing import Final @@ -26,6 +27,52 @@ from ..common_utils import ElevenLabsException _JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True)) _JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True)) +_FORM_OPENAI_PARAMS: Final = frozenset({"language", "temperature"}) + + +def _finite_float(value: object) -> float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + try: + number: Final = float(value) + except OverflowError: + return None + return number if math.isfinite(number) else None + + +def _timestamp(value: object) -> float: + seconds: Final = _finite_float(value) + if seconds is None or seconds < 0 or _finite_float(seconds * 1000) is None: + return 0.0 + return seconds + + +def _time_span(item: Mapping[str, object]) -> Mapping[str, float]: + start: Final = _timestamp(item.get("start")) + return {"start": start, "end": max(start, _timestamp(item.get("end")))} + + +def _speaker_fields(item: Mapping[str, object]) -> Mapping[str, str]: + speaker: Final = item.get("speaker_id") + return {"speaker": speaker} if isinstance(speaker, str) else {} + + +def _word_fields(item: Mapping[str, object]) -> Mapping[str, object]: + logprob: Final = _finite_float(item.get("logprob")) + return { + "word": item.get("text", ""), + **_time_span(item), + **({} if logprob is None else {"logprob": logprob}), + **_speaker_fields(item), + } + + +def _audio_event_fields(item: Mapping[str, object]) -> Mapping[str, object]: + return { + "text": item.get("text", ""), + **_time_span(item), + **_speaker_fields(item), + } class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig): @@ -34,7 +81,11 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig): return litellm.LlmProviders.ELEVENLABS.value def get_supported_openai_params(self, model: str) -> list[OpenAIAudioTranscriptionOptionalParams]: - return ["language", "temperature"] + return ["language", "temperature", "response_format"] + + @property + def supports_subtitle_synthesis(self) -> bool: + return True def map_openai_params( self, @@ -82,7 +133,7 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig): # Add OpenAI Compatible Parameters ######################################################### for key, value in optional_params.items(): - if key in self.get_supported_openai_params(model) and value is not None: + if key in _FORM_OPENAI_PARAMS and value is not None: # Convert values to strings for form data, but skip None values form_data[key] = str(value) @@ -132,19 +183,20 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig): response["task"] = "transcribe" response["language"] = response_object.get("language_code", "unknown") - # Map ElevenLabs words to OpenAI format if "words" in response_object: - response["words"] = [] - for word_data in _JSON_OBJECTS.validate_python(response_object["words"]): - # Only include actual words, skip spacing and audio events - if word_data.get("type") == "word": - response["words"].append( - { - "word": word_data.get("text", ""), - "start": word_data.get("start", 0), - "end": word_data.get("end", 0), - } - ) + items: Final = tuple(_JSON_OBJECTS.validate_python(response_object["words"])) + response["words"] = [_word_fields(item) for item in items if item.get("type") == "word"] + response["audio_events"] = [ + _audio_event_fields(item) for item in items if item.get("type") == "audio_event" + ] + + language_probability: Final = _finite_float(response_object.get("language_probability")) + if language_probability is not None: + response["language_probability"] = language_probability + + duration: Final = _finite_float(response_object.get("audio_duration_secs")) + if duration is not None and duration >= 0: + response["duration"] = duration # Store full response in hidden params response.hidden_params = response_json diff --git a/tests/unit/llms/elevenlabs/audio_transcription/test_transformation.py b/tests/unit/llms/elevenlabs/audio_transcription/test_transformation.py index 86252a1a95d..3a6baa276ef 100644 --- a/tests/unit/llms/elevenlabs/audio_transcription/test_transformation.py +++ b/tests/unit/llms/elevenlabs/audio_transcription/test_transformation.py @@ -1,3 +1,7 @@ +import io +import json +import re +import wave from collections.abc import Iterator from typing import Final @@ -40,36 +44,269 @@ def _transform(payload: object) -> TranscriptionResponse: ) -def test_transform_audio_transcription_response_keeps_only_spoken_words(): +def _transform_raw(content: bytes) -> TranscriptionResponse: + return ElevenLabsAudioTranscriptionConfig().transform_audio_transcription_response( + raw_response=httpx.Response(200, content=content) + ) + + +def _word_with(**fields: object) -> dict[str, object]: + return {"type": "word", "text": "hi", "start": 0.0, "end": 0.5, **fields} + + +NUMBER_CASES: Final = ( + ("-0.35", -0.35), + ("0", 0.0), + ("0.0", 0.0), + ("-2", -2.0), + ("-1e-300", -1e-300), + ("0.3", 0.3), + ("1", 1.0), +) +NOT_FINITE_NUMBER_LITERALS: Final = ( + "NaN", + "Infinity", + "-Infinity", + "true", + "false", + '"-0.5"', + "null", + "1" + "0" * 400, + "-1" + "0" * 400, + "[1]", +) + + +def test_transform_audio_transcription_response_keeps_spoken_words_with_native_metadata(): payload = { "language_code": "en", - "text": "Hello world", + "text": "Hello (laughter) world", "words": [ - {"type": "word", "text": "Hello", "start": 0.0, "end": 0.4, "speaker_id": "speaker_0"}, - {"type": "spacing", "text": " ", "start": 0.4, "end": 0.5}, - {"type": "audio_event", "text": "(laughter)", "start": 0.5, "end": 0.9}, + {"type": "word", "text": "Hello", "start": 0.0, "end": 0.4, "speaker_id": "speaker_0", "logprob": -0.35}, + {"type": "spacing", "text": " ", "start": 0.4, "end": 0.5, "logprob": -0.1}, + {"type": "audio_event", "text": "(laughter)", "start": 0.5, "end": 0.9, "speaker_id": "speaker_1"}, {"type": "word", "text": "world", "start": 0.9, "end": 1.3}, ], } response = _transform(payload) - assert response.text == "Hello world" + assert response.text == "Hello (laughter) world" assert response["task"] == "transcribe" assert response["language"] == "en" assert response["words"] == [ - {"word": "Hello", "start": 0.0, "end": 0.4}, + {"word": "Hello", "start": 0.0, "end": 0.4, "logprob": -0.35, "speaker": "speaker_0"}, {"word": "world", "start": 0.9, "end": 1.3}, ] + assert response["audio_events"] == [{"text": "(laughter)", "start": 0.5, "end": 0.9, "speaker": "speaker_1"}] + assert response._hidden_params == payload + + +@pytest.mark.parametrize(("literal", "expected"), NUMBER_CASES) +def test_transform_audio_transcription_response_passes_finite_logprob_through(literal: str, expected: float): + body: Final = b'{"words": [{"type": "word", "text": "a", "start": 0, "end": 1, "logprob": %s}]}' % literal.encode() + + logprob: Final = _transform_raw(body)["words"][0]["logprob"] + + assert logprob == expected + assert isinstance(logprob, float) + + +@pytest.mark.parametrize("literal", NOT_FINITE_NUMBER_LITERALS) +def test_transform_audio_transcription_response_omits_unusable_logprob(literal: str): + body: Final = b'{"words": [{"type": "word", "text": "a", "start": 0, "end": 1, "logprob": %s}]}' % literal.encode() + + response: Final = _transform_raw(body) + + assert response["words"] == [{"word": "a", "start": 0, "end": 1}] + assert "NaN" not in json.dumps(response["words"]) and "Infinity" not in json.dumps(response["words"]) + + +def test_transform_audio_transcription_response_omits_absent_logprob_and_keeps_siblings(): + body: Final = ( + b'{"words": [{"type": "word", "text": "a", "logprob": -1.5}, {"type": "word", "text": "b"},' + b' {"type": "word", "text": "c", "logprob": NaN}]}' + ) + + response: Final = _transform_raw(body) + + assert response["words"][0]["logprob"] == -1.5 + assert all("logprob" not in w for w in response["words"][1:]) + + +@pytest.mark.parametrize(("literal", "expected"), NUMBER_CASES) +def test_transform_audio_transcription_response_passes_finite_language_probability_through( + literal: str, expected: float +): + response: Final = _transform_raw(b'{"language_probability": %s}' % literal.encode()) + + assert response["language_probability"] == expected + assert isinstance(response["language_probability"], float) + + +@pytest.mark.parametrize("literal", NOT_FINITE_NUMBER_LITERALS) +def test_transform_audio_transcription_response_omits_unusable_language_probability(literal: str): + response: Final = _transform_raw(b'{"text": "t", "language_probability": %s}' % literal.encode()) + + assert "language_probability" not in response + + +def test_transform_audio_transcription_response_omits_absent_language_probability(): + assert "language_probability" not in _transform({"text": "t"}) + + +@pytest.mark.parametrize( + ("literal", "expected"), + [("0", 0.0), ("0.0", 0.0), ("11.0", 11.0), ("17", 17.0), ("1e-300", 1e-300), ("3600.5", 3600.5)], +) +def test_transform_audio_transcription_response_exposes_valid_duration(literal: str, expected: float): + response: Final = _transform_raw(b'{"text": "t", "audio_duration_secs": %s}' % literal.encode()) + + assert response["duration"] == expected + assert isinstance(response["duration"], float) + + +@pytest.mark.parametrize( + "literal", + ["-0.01", "-3", "-1e-300", "-Infinity", "NaN", "Infinity", "true", "false", '"12"', "null", "1" + "0" * 400, "[1]"], +) +def test_transform_audio_transcription_response_omits_invalid_duration(literal: str): + body: Final = ( + b'{"text": "t", "audio_duration_secs": %s, "words": [{"type": "word", "end": 2.5}]}' % literal.encode() + ) + + assert "duration" not in _transform_raw(body) + + +def test_transform_audio_transcription_response_omits_absent_duration(): + assert "duration" not in _transform({"text": "t", "words": [_word_with(end=2.5)]}) + + +@pytest.mark.parametrize("speaker", ["speaker_0", "speaker_12", ""]) +def test_transform_audio_transcription_response_exposes_string_speaker_on_words_and_events(speaker: str): + response: Final = _transform( + {"words": [_word_with(speaker_id=speaker), {"type": "audio_event", "text": "(x)", "speaker_id": speaker}]} + ) + + assert response["words"][0]["speaker"] == speaker + assert response["audio_events"][0]["speaker"] == speaker + + +@pytest.mark.parametrize("speaker", [None, 3, 0, True, ["a"], {"id": "a"}, 1.5]) +def test_transform_audio_transcription_response_omits_non_string_speaker(speaker: object): + response: Final = _transform( + {"words": [_word_with(speaker_id=speaker), {"type": "audio_event", "text": "(x)", "speaker_id": speaker}]} + ) + + assert "speaker" not in response["words"][0] + assert "speaker" not in response["audio_events"][0] + + +def test_transform_audio_transcription_response_omits_speaker_when_missing(): + response: Final = _transform({"words": [_word_with(), {"type": "audio_event", "text": "(x)"}]}) + + assert "speaker" not in response["words"][0] + assert "speaker" not in response["audio_events"][0] + + +def test_transform_audio_transcription_response_keeps_audio_events_in_order_apart_from_words_and_spacing(): + payload = { + "text": "a [music] b [applause]", + "words": [ + {"type": "audio_event", "text": "[music]", "start": 0.0, "end": 1.0, "logprob": -0.2}, + {"type": "word", "text": "a", "start": 1.0, "end": 1.2}, + {"type": "spacing", "text": " ", "start": 1.2, "end": 1.3}, + {"type": "word", "text": "b", "start": 1.3, "end": 1.5}, + {"type": "audio_event", "text": "[applause]", "start": 1.5, "end": 2.5, "speaker_id": "speaker_1"}, + {"type": "audio_event", "text": "[cough]", "start": 2.5, "end": 2.6}, + ], + } + + response: Final = _transform(payload) + + assert response["audio_events"] == [ + {"text": "[music]", "start": 0.0, "end": 1.0}, + {"text": "[applause]", "start": 1.5, "end": 2.5, "speaker": "speaker_1"}, + {"text": "[cough]", "start": 2.5, "end": 2.6}, + ] + assert [w["word"] for w in response["words"]] == ["a", "b"] + assert response.text == "a [music] b [applause]" + + +def test_transform_audio_transcription_response_defaults_missing_audio_event_fields_like_words(): + response: Final = _transform({"words": [{"type": "audio_event"}, {"type": "word"}]}) + + assert response["audio_events"] == [{"text": "", "start": 0, "end": 0}] + assert response["words"] == [{"word": "", "start": 0, "end": 0}] + + +@pytest.mark.parametrize("words", [[], [{"type": "spacing", "text": " "}], [{"type": "word", "text": "a"}]]) +def test_transform_audio_transcription_response_has_empty_audio_events_when_words_present_without_events( + words: list[dict[str, object]], +): + assert _transform({"text": "t", "words": words})["audio_events"] == [] + + +@pytest.mark.parametrize("payload", [{}, {"text": "t"}, {"text": "t", "audio_duration_secs": 3.0}]) +def test_transform_audio_transcription_response_has_no_audio_events_without_words(payload: dict[str, object]): + assert "audio_events" not in _transform(payload) + + +def test_transform_audio_transcription_response_output_is_strict_json_for_hostile_numbers(): + body: Final = ( + b'{"language_probability": NaN, "audio_duration_secs": Infinity, "words": [' + b'{"type": "word", "text": "a", "logprob": -Infinity, "speaker_id": "s"},' + b'{"type": "audio_event", "text": "(x)", "start": 0, "end": 1, "logprob": NaN}]}' + ) + + response: Final = _transform_raw(body) + + strict: Final = json.dumps({k: response[k] for k in ("words", "audio_events", "language", "task")}, allow_nan=False) + assert "language_probability" not in response and "duration" not in response + assert "null" not in strict + + +def test_transform_audio_transcription_response_matches_real_response_shape(): + payload = { + "language_code": "eng", + "language_probability": 0.9746739864349365, + "text": "Four score and seven years ago (applause)", + "words": [ + {"text": "Four", "start": 0.44, "end": 0.7, "type": "word", "speaker_id": "speaker_0", "logprob": -0.00017}, + {"text": " ", "start": 0.7, "end": 0.82, "type": "spacing", "speaker_id": "speaker_0", "logprob": -0.35}, + {"text": "score", "start": 0.82, "end": 1.12, "type": "word", "speaker_id": "speaker_0", "logprob": -0.62}, + { + "text": "(applause)", + "start": 1.2, + "end": 2.0, + "type": "audio_event", + "speaker_id": "speaker_0", + "logprob": -0.9, + }, + ], + "transcription_id": "abc", + "audio_duration_secs": 11.0, + } + + response: Final = _transform(payload) + + assert response["language"] == "eng" + assert response["language_probability"] == 0.9746739864349365 + assert response["duration"] == 11.0 + assert response["words"] == [ + {"word": "Four", "start": 0.44, "end": 0.7, "logprob": -0.00017, "speaker": "speaker_0"}, + {"word": "score", "start": 0.82, "end": 1.12, "logprob": -0.62, "speaker": "speaker_0"}, + ] + assert response["audio_events"] == [{"text": "(applause)", "start": 1.2, "end": 2.0, "speaker": "speaker_0"}] assert response._hidden_params == payload @pytest.mark.parametrize( ("word", "expected"), [ - ({"type": "word"}, [{"word": "", "start": 0, "end": 0}]), - ({"type": "word", "text": None, "start": None, "end": None}, [{"word": None, "start": None, "end": None}]), - ({"type": "word", "text": 7, "start": "0.1", "end": [2]}, [{"word": 7, "start": "0.1", "end": [2]}]), + ({"type": "word"}, [{"word": "", "start": 0.0, "end": 0.0}]), + ({"type": "word", "text": None, "start": None, "end": None}, [{"word": None, "start": 0.0, "end": 0.0}]), + ({"type": "word", "text": 7, "start": "0.1", "end": [2]}, [{"word": 7, "start": 0.0, "end": 0.0}]), ({"text": "untyped"}, []), ({"type": None, "text": "untyped"}, []), ({}, []), @@ -78,7 +315,10 @@ def test_transform_audio_transcription_response_keeps_only_spoken_words(): def test_transform_audio_transcription_response_maps_one_word( word: dict[str, object], expected: list[dict[str, object]] ): - assert _transform({"text": "t", "words": [word]})["words"] == expected + words: Final = _transform({"text": "t", "words": [word]})["words"] + + assert words == expected + assert all(isinstance(w["start"], float) and isinstance(w["end"], float) for w in words) @pytest.mark.parametrize("words", [[], "", {}]) @@ -129,6 +369,293 @@ def test_transform_audio_transcription_response_wraps_malformed_payloads_with_th assert str(exc_info.value).endswith(f"\nResponse: {raw_response.text}") +def _silent_wav(seconds: int) -> bytes: + buffer: Final = io.BytesIO() + with wave.open(buffer, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(8000) + wav.writeframes(b"\x00\x00" * 8000 * seconds) + return buffer.getvalue() + + +@pytest.mark.parametrize("audio", [b"not a decodable audio file", _silent_wav(1)]) +def test_transcription_cost_uses_the_duration_in_the_elevenlabs_response( + audio: bytes, respx_mock: respx.MockRouter +) -> None: + api_base: Final = "http://localhost:12346" + billed_seconds: Final = 7.5 + respx_mock.post(f"{api_base}/v1/speech-to-text").mock( + return_value=httpx.Response( + 200, + json={"language_code": "eng", "text": "hi", "words": [], "audio_duration_secs": billed_seconds}, + ) + ) + rate: Final = litellm.get_model_info("elevenlabs/scribe_v2")["input_cost_per_second"] + assert rate + + response: Final = litellm.transcription( + model="elevenlabs/scribe_v2", file=("a.wav", audio), api_base=api_base, api_key=ELEVENLABS_API_KEY + ) + + assert response["duration"] == billed_seconds + assert response._hidden_params["response_cost"] == pytest.approx(billed_seconds * rate) + + +RESPONSE_FORMATS: Final = ("json", "text", "verbose_json", "srt", "vtt") +SUBTITLE_FORMATS: Final = ("srt", "vtt") +NATIVE_FORMATS: Final = ("json", "text", "verbose_json") +SPOKEN_BODY: Final = { + "language_code": "eng", + "language_probability": 0.97, + "text": "Four score (applause) again", + "audio_duration_secs": 11.0, + "words": [ + {"text": "Four", "start": 0.44, "end": 0.7, "type": "word", "speaker_id": "speaker_0", "logprob": -0.1}, + {"text": " ", "start": 0.7, "end": 0.82, "type": "spacing"}, + {"text": "score", "start": 0.82, "end": 1.12, "type": "word", "speaker_id": "speaker_0", "logprob": -0.2}, + {"text": "(applause)", "start": 1.2, "end": 2.0, "type": "audio_event"}, + {"text": "again", "start": 6.0, "end": 6.5, "type": "word", "speaker_id": "speaker_0", "logprob": -0.3}, + ], +} + + +def _form_values(request: httpx.Request) -> dict[str, bytes]: + boundary: Final = request.headers["content-type"].split("boundary=")[1].encode() + parts: Final = request.content.split(b"--" + boundary) + return { + match.group(1).decode(): match.group(2) + for part in parts + if (match := re.search(rb'name="([^"]+)"\r\n\r\n(.*)\r\n$', part, re.S)) and b"filename=" not in part + } + + +def _transcribe( + respx_mock: respx.MockRouter, body: bytes, **params: object +) -> tuple[TranscriptionResponse, dict[str, bytes]]: + api_base: Final = "http://localhost:12346" + route: Final = respx_mock.post(f"{api_base}/v1/speech-to-text").mock( + return_value=httpx.Response(200, content=body, headers={"content-type": "application/json"}) + ) + response: Final = litellm.transcription( + model="elevenlabs/scribe_v2", + file=("a.wav", _silent_wav(1)), + api_base=api_base, + api_key=ELEVENLABS_API_KEY, + **params, + ) + return response, _form_values(route.calls.last.request) + + +def _transcribe_spoken_body( + respx_mock: respx.MockRouter, **params: object +) -> tuple[TranscriptionResponse, dict[str, bytes]]: + return _transcribe(respx_mock, json.dumps(SPOKEN_BODY).encode(), **params) + + +@pytest.mark.parametrize("response_format", RESPONSE_FORMATS) +def test_elevenlabs_transcription_accepts_response_format_without_drop_params( + response_format: str, monkeypatch: pytest.MonkeyPatch +): + monkeypatch.setattr(litellm, "drop_params", False) + + optional_params: Final = litellm.utils.get_optional_params_transcription( + model="scribe_v2", custom_llm_provider="elevenlabs", response_format=response_format + ) + + assert optional_params["response_format"] == response_format + + +@pytest.mark.parametrize("response_format", RESPONSE_FORMATS) +def test_transform_audio_transcription_request_form_is_unchanged_by_response_format(response_format: str): + config: Final = ElevenLabsAudioTranscriptionConfig() + audio: Final = ("a.wav", b"fake audio data") + base_params: Final = {"language_code": "es", "temperature": 0.5, "diarize": True, "tag_audio_events": False} + + without: Final = config.transform_audio_transcription_request( + model="scribe_v2", audio_file=audio, optional_params=dict(base_params), litellm_params={} + ) + with_format: Final = config.transform_audio_transcription_request( + model="scribe_v2", + audio_file=audio, + optional_params={**base_params, "response_format": response_format}, + litellm_params={}, + ) + + assert with_format.data == without.data + assert without.data["diarize"] == "True" and without.data["temperature"] == "0.5" + assert response_format not in with_format.data.values() + assert "response_format" not in with_format.data + + +@pytest.mark.parametrize("response_format", SUBTITLE_FORMATS) +def test_elevenlabs_transcription_returns_subtitle_document_built_from_the_spoken_words( + response_format: str, respx_mock: respx.MockRouter +): + response, form = _transcribe_spoken_body(respx_mock, response_format=response_format, language="es") + + _, form_without_format = _transcribe_spoken_body(respx_mock, language="es") + assert form == form_without_format + assert form["language_code"] == b"es" + separator: Final = "," if response_format == "srt" else "." + cues: Final = re.findall(r"^(\d{2}:\d{2}:\d{2}[,.]\d{3}) --> (\d{2}:\d{2}:\d{2}[,.]\d{3})$", response.text, re.M) + assert [start for start, _ in cues] == [f"00:00:00{separator}440", f"00:00:06{separator}000"] + assert [end for _, end in cues] == [f"00:00:01{separator}120", f"00:00:06{separator}500"] + assert "Four score" in response.text and "again" in response.text + assert (response_format == "vtt") == response.text.startswith("WEBVTT") + assert response.text != SPOKEN_BODY["text"] + + +@pytest.mark.parametrize("response_format", NATIVE_FORMATS) +def test_elevenlabs_transcription_keeps_native_text_and_metadata_for_non_subtitle_formats( + response_format: str, respx_mock: respx.MockRouter +): + response, form = _transcribe_spoken_body(respx_mock, response_format=response_format, temperature=0.5) + + _, form_without_format = _transcribe_spoken_body(respx_mock, temperature=0.5) + assert form == form_without_format + assert form["temperature"] == b"0.5" + assert response.text == SPOKEN_BODY["text"] + assert [w["word"] for w in response["words"]] == ["Four", "score", "again"] + assert response["words"][0]["speaker"] == "speaker_0" + assert response["audio_events"] == [{"text": "(applause)", "start": 1.2, "end": 2.0}] + assert response["duration"] == SPOKEN_BODY["audio_duration_secs"] + + +PASSING_TIMESTAMPS: Final = (("0", 0.0), ("2", 2.0), ("0.44", 0.44), ("6.5", 6.5), ("1e300", 1e300)) +UNUSABLE_TIMESTAMPS: Final = ("NaN", "Infinity", "-Infinity", "1e308", "-0.5", "true", "false", '"0.1"', "[2]", "null") + + +def _item_body(item_type: str, start: str, end: str) -> bytes: + return b'{"words": [{"type": "%s", "text": "x", "start": %s, "end": %s}]}' % ( + item_type.encode(), + start.encode(), + end.encode(), + ) + + +def _timed_item(response: TranscriptionResponse, item_type: str) -> dict[str, object]: + return response["words" if item_type == "word" else "audio_events"][0] + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +@pytest.mark.parametrize(("literal", "expected"), PASSING_TIMESTAMPS) +def test_transform_audio_transcription_response_passes_usable_start_and_end_through_as_float( + item_type: str, literal: str, expected: float +): + item: Final = _timed_item(_transform_raw(_item_body(item_type, literal, literal)), item_type) + + assert item["start"] == expected and item["end"] == expected + assert isinstance(item["start"], float) and isinstance(item["end"], float) + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +@pytest.mark.parametrize("literal", UNUSABLE_TIMESTAMPS) +def test_transform_audio_transcription_response_zeroes_unusable_start_and_end(item_type: str, literal: str): + item: Final = _timed_item(_transform_raw(_item_body(item_type, literal, literal)), item_type) + + assert item["start"] == 0.0 and item["end"] == 0.0 + assert isinstance(item["start"], float) and isinstance(item["end"], float) + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +def test_transform_audio_transcription_response_zeroes_absent_start_and_end_and_keeps_the_other(item_type: str): + body: Final = b'{"words": [{"type": "%s", "text": "x", "end": 3}]}' % item_type.encode() + + item: Final = _timed_item(_transform_raw(body), item_type) + + assert item["start"] == 0.0 and isinstance(item["start"], float) + assert item["end"] == 3.0 and isinstance(item["end"], float) + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +@pytest.mark.parametrize("literal", UNUSABLE_TIMESTAMPS) +def test_transform_audio_transcription_response_ends_at_start_when_end_is_unusable(item_type: str, literal: str): + item: Final = _timed_item(_transform_raw(_item_body(item_type, "1.0", literal)), item_type) + + assert item["start"] == 1.0 and item["end"] == 1.0 + assert isinstance(item["start"], float) and isinstance(item["end"], float) + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +@pytest.mark.parametrize( + ("start", "end", "expected_end"), + [("1.0", "0.5", 1.0), ("2", "0", 2.0), ("1.0", "1.0", 1.0), ("0.44", "0.68", 0.68), ("1.0", "6.5", 6.5)], +) +def test_transform_audio_transcription_response_never_ends_before_it_starts( + item_type: str, start: str, end: str, expected_end: float +): + item: Final = _timed_item(_transform_raw(_item_body(item_type, start, end)), item_type) + + assert item["start"] == float(start) + assert item["end"] == expected_end + assert isinstance(item["end"], float) + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +def test_transform_audio_transcription_response_ends_at_start_when_end_is_absent(item_type: str): + body: Final = b'{"words": [{"type": "%s", "text": "x", "start": 1.5}]}' % item_type.encode() + + item: Final = _timed_item(_transform_raw(body), item_type) + + assert item["start"] == 1.5 and item["end"] == 1.5 + + +@pytest.mark.parametrize("item_type", ["word", "audio_event"]) +def test_transform_audio_transcription_response_keeps_a_usable_end_when_start_is_unusable(item_type: str): + item: Final = _timed_item(_transform_raw(_item_body(item_type, "Infinity", "2.0")), item_type) + + assert item["start"] == 0.0 and item["end"] == 2.0 + + +def _cue_seconds(clock: str) -> float: + hours, minutes, seconds, millis = (int(part) for part in re.split(r"[:,.]", clock)) + return hours * 3600 + minutes * 60 + seconds + millis / 1000 + + +def _cue_spans(document: str) -> tuple[tuple[float, float], ...]: + cues: Final = re.findall(r"(\d{2}:\d{2}:\d{2}[,.]\d{3}) --> (\d{2}:\d{2}:\d{2}[,.]\d{3})", document) + return tuple((_cue_seconds(start), _cue_seconds(end)) for start, end in cues) + + +@pytest.mark.parametrize("response_format", SUBTITLE_FORMATS) +@pytest.mark.parametrize("literal", ["Infinity", "NaN", "1e308", "-Infinity"]) +def test_elevenlabs_transcription_subtitles_survive_unusable_word_timestamps( + response_format: str, literal: str, respx_mock: respx.MockRouter +): + word: Final = '{"type": "word", "text": "%s", "start": %s, "end": %s}' + body: Final = ( + '{"text": "Four score", "words": [%s, %s]}' + % (word % ("Four", literal, literal), word % ("score", "1.0", literal)) + ).encode() + + response, _ = _transcribe(respx_mock, body, response_format=response_format) + + separator: Final = "," if response_format == "srt" else "." + assert response.text != "Four score" + assert f"00:00:00{separator}000 --> " in response.text + assert "Four" in response.text and "score" in response.text + assert (response_format == "vtt") == response.text.startswith("WEBVTT") + spans: Final = _cue_spans(response.text) + assert spans + assert all(start <= end for start, end in spans) + + +@pytest.mark.parametrize("response_format", SUBTITLE_FORMATS) +def test_elevenlabs_transcription_subtitles_never_end_before_they_start_when_the_end_is_earlier( + response_format: str, respx_mock: respx.MockRouter +): + body: Final = b'{"text": "score", "words": [{"type": "word", "text": "score", "start": 1.0, "end": 0.5}]}' + + response, _ = _transcribe(respx_mock, body, response_format=response_format) + + separator: Final = "," if response_format == "srt" else "." + spans: Final = _cue_spans(response.text) + assert f"00:00:01{separator}000 --> 00:00:01{separator}000" in response.text + assert spans + assert all(start <= end for start, end in spans) + + class TestElevenLabsAudioTranscription: @pytest.mark.usefixtures("elevenlabs_api_key_env") def test_elevenlabs_diarize_parameter_passthrough(self):