wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: tighten language_to_<service>_language return types to plain str

Browse Source 
These provider-specific helpers are all thin wrappers around
`resolve_language(...)`, which itself returns `str` — never `None`.
The `str | None` annotations were misleading and were producing
spurious pyright errors at the call sites that assigned the result
into a `str` field. Update each helper's signature to `str` and
rewrite the `Returns:` docstring to describe the actual fallback
behaviour (resolve to base or full code, with a warning).

Importantly, the per-class `language_to_service_language(...)`
methods on `STTService`/`TTSService` subclasses keep `str | None` as
their return type. That signature is an extension hook for future
and/or third-party subclasses that may genuinely not be able to
produce a code for some languages, even though all in-tree first-
party services currently return a string.

Also includes one small unrelated tightening in azure/stt.py: wrap
`self._settings.language` with `assert_given(...)` so the truthy
fallback to `language_to_azure_language(Language.EN_US)` doesn't
silently swallow a NotGiven sentinel.

Net: -3 pyright errors (full-config run: 785 -> 782).
This commit is contained in:
Paul Kompfner
2026-04-28 16:02:47 -04:00
parent bec407ce3a
commit 2a731336be
28 changed files with 131 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/pipecat/services/asyncai/tts.py
View File

@@ -42,14 +42,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_async_language(language: Language) -> str | None:
def language_to_async_language(language: Language) -> str:
"""Convert a Language enum to Async language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Async language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.EN: "en",

6
src/pipecat/services/aws/tts.py
View File

@@ -37,14 +37,16 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_aws_language(language: Language) -> str | None:
def language_to_aws_language(language: Language) -> str:
"""Convert a Language enum to AWS Polly language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding AWS Polly language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic

6
src/pipecat/services/azure/common.py
View File

@@ -9,14 +9,16 @@
from pipecat.transcriptions.language import Language, resolve_language
def language_to_azure_language(language: Language) -> str | None:
def language_to_azure_language(language: Language) -> str:
"""Convert a Language enum to Azure language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Azure language code, or None if not supported.
The corresponding Azure language code. If ``language`` is not in the
verified mapping, falls back to the full language code string and logs
a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans

6
src/pipecat/services/azure/stt.py
View File

@@ -182,9 +182,9 @@ class AzureSTTService(STTService):
changed = await super()._update_settings(delta)
if "language" in changed:
self._speech_config.speech_recognition_language = (
self._settings.language or language_to_azure_language(Language.EN_US)
)
self._speech_config.speech_recognition_language = assert_given(
self._settings.language
) or language_to_azure_language(Language.EN_US)
if self._audio_stream:
await self._disconnect()
await self._connect()

7
src/pipecat/services/camb/tts.py
View File

@@ -44,14 +44,17 @@ MODEL_SAMPLE_RATES: dict[str, int] = {
}
def language_to_camb_language(language: Language) -> str | None:
def language_to_camb_language(language: Language) -> str:
"""Convert a Pipecat Language enum to Camb.ai language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Camb.ai language code (BCP-47 format), or None if not supported.
The corresponding Camb.ai language code (BCP-47 format). If
``language`` is not in the verified mapping, falls back to the base
language code (e.g., ``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.EN: "en-us",

7
src/pipecat/services/cartesia/tts.py
View File

@@ -62,14 +62,17 @@ class GenerationConfig(BaseModel):
emotion: str | None = None
def language_to_cartesia_language(language: Language) -> str | None:
def language_to_cartesia_language(language: Language) -> str:
"""Convert a Language enum to Cartesia language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Cartesia language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

2
src/pipecat/services/deepgram/flux/base.py
View File

@@ -32,7 +32,7 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_deepgram_flux_language(language: Language) -> str | None:
def language_to_deepgram_flux_language(language: Language) -> str:
"""Convert a Pipecat Language to a Deepgram Flux language code.
Only honored by the ``flux-general-multi`` model. Locale variants

6
src/pipecat/services/elevenlabs/stt.py
View File

@@ -54,7 +54,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_elevenlabs_language(language: Language) -> str | None:
def language_to_elevenlabs_language(language: Language) -> str:
"""Convert a Language enum to ElevenLabs language code.
Source:
@@ -64,7 +64,9 @@ def language_to_elevenlabs_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding ElevenLabs language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.AF: "afr", # Afrikaans

7
src/pipecat/services/elevenlabs/tts.py
View File

@@ -69,14 +69,17 @@ ELEVENLABS_MULTILINGUAL_MODELS = {
}
def language_to_elevenlabs_language(language: Language) -> str | None:
def language_to_elevenlabs_language(language: Language) -> str:
"""Convert a Language enum to ElevenLabs language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding ElevenLabs language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

7
src/pipecat/services/fal/stt.py
View File

@@ -28,14 +28,17 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
def language_to_fal_language(language: Language) -> str | None:
def language_to_fal_language(language: Language) -> str:
"""Convert a Language enum to Fal's Wizper language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Fal Wizper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",

9
src/pipecat/services/gladia/stt.py
View File

@@ -56,14 +56,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_gladia_language(language: Language) -> str | None:
def language_to_gladia_language(language: Language) -> str:
"""Convert a Language enum to Gladia's language code format.
Args:
language: The Language enum value to convert.
Returns:
The Gladia language code string or None if not supported.
The corresponding Gladia language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",
@@ -361,7 +364,7 @@ class GladiaSTTService(WebsocketSTTService):
language: The Language enum value to convert.
Returns:
The Gladia language code string or None if not supported.
The Gladia language code string, or None if not supported.
"""
return language_to_gladia_language(language)

6
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -111,7 +111,7 @@ MAX_CONSECUTIVE_FAILURES = 3
CONNECTION_ESTABLISHED_THRESHOLD = 10.0 # seconds
def language_to_gemini_language(language: Language) -> str | None:
def language_to_gemini_language(language: Language) -> str:
"""Maps a Language enum value to a Gemini Live supported language code.
Source:
@@ -121,7 +121,9 @@ def language_to_gemini_language(language: Language) -> str | None:
language: The language enum value to convert.
Returns:
The Gemini language code string, or None if the language is not supported.
The Gemini language code string. If ``language`` is not in the
verified mapping, falls back to the full language code string and logs
a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic

7
src/pipecat/services/google/stt.py
View File

@@ -60,14 +60,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_google_stt_language(language: Language) -> str | None:
def language_to_google_stt_language(language: Language) -> str:
"""Maps Language enum to Google Speech-to-Text V2 language codes.
Args:
language: Language enum value.
Returns:
Optional[str]: Google STT language code or None if not supported.
The corresponding Google STT language code. If ``language`` is not
in the verified mapping, falls back to the full language code string
and logs a warning (via
``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans

12
src/pipecat/services/google/tts.py
View File

@@ -60,7 +60,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_google_tts_language(language: Language) -> str | None:
def language_to_google_tts_language(language: Language) -> str:
"""Convert a Language enum to Google TTS language code.
Source:
@@ -70,7 +70,9 @@ def language_to_google_tts_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding Google TTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic
@@ -219,7 +221,7 @@ def language_to_google_tts_language(language: Language) -> str | None:
return resolve_language(language, LANGUAGE_MAP, use_base_code=False)
def language_to_gemini_tts_language(language: Language) -> str | None:
def language_to_gemini_tts_language(language: Language) -> str:
"""Convert a Language enum to Gemini TTS language code.
Source:
@@ -229,7 +231,9 @@ def language_to_gemini_tts_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding Gemini TTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Afrikaans (Preview)

7
src/pipecat/services/gradium/stt.py
View File

@@ -79,14 +79,17 @@ def _input_format_from_encoding(encoding: str, sample_rate: int) -> str:
return encoding
def language_to_gradium_language(language: Language) -> str | None:
def language_to_gradium_language(language: Language) -> str:
"""Convert a Language enum to Gradium's language code format.
Args:
language: The Language enum value to convert.
Returns:
The Gradium language code string or None if not supported.
The corresponding Gradium language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.DE: "de",

7
src/pipecat/services/lmnt/tts.py
View File

@@ -37,14 +37,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_lmnt_language(language: Language) -> str | None:
def language_to_lmnt_language(language: Language) -> str:
"""Convert a Language enum to LMNT language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding LMNT language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

6
src/pipecat/services/minimax/tts.py
View File

@@ -31,14 +31,16 @@ from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.tracing.service_decorators import traced_tts
def language_to_minimax_language(language: Language) -> str | None:
def language_to_minimax_language(language: Language) -> str:
"""Convert a Language enum to MiniMax language format.
Args:
language: The Language enum value to convert.
Returns:
The corresponding MiniMax language name, or None if not supported.
The corresponding MiniMax language name. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.AF: "Afrikaans",

7
src/pipecat/services/neuphonic/tts.py
View File

@@ -44,14 +44,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_neuphonic_lang_code(language: Language) -> str | None:
def language_to_neuphonic_lang_code(language: Language) -> str:
"""Convert a Language enum to Neuphonic language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Neuphonic language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.DE: "de",

6
src/pipecat/services/nvidia/stt.py
View File

@@ -49,7 +49,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_nvidia_nemotron_speech_language(language: Language) -> str | None:
def language_to_nvidia_nemotron_speech_language(language: Language) -> str:
"""Maps Language enum to NVIDIA Nemotron Speech ASR language codes.
Source:
@@ -59,7 +59,9 @@ def language_to_nvidia_nemotron_speech_language(language: Language) -> str | Non
language: Language enum value.
Returns:
str | None: NVIDIA Nemotron Speech language code or None if not supported.
The NVIDIA Nemotron Speech language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
# Arabic

6
src/pipecat/services/sarvam/tts.py
View File

@@ -216,14 +216,16 @@ def get_speakers_for_model(model: str) -> list[str]:
return list(TTS_MODEL_CONFIGS["bulbul:v2"].speakers)
def language_to_sarvam_language(language: Language) -> str | None:
def language_to_sarvam_language(language: Language) -> str:
"""Convert Pipecat Language enum to Sarvam AI language codes.
Args:
language: The Language enum value to convert.
Returns:
The corresponding Sarvam AI language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the full language code string and
logs a warning (via ``resolve_language(..., use_base_code=False)``).
"""
LANGUAGE_MAP = {
Language.BN: "bn-IN", # Bengali

7
src/pipecat/services/smallest/tts.py
View File

@@ -51,14 +51,17 @@ class SmallestTTSModel(StrEnum):
LIGHTNING_V3_1 = "lightning-v3.1"
def language_to_smallest_tts_language(language: Language) -> str | None:
def language_to_smallest_tts_language(language: Language) -> str:
"""Convert a Language enum to a Smallest TTS language string.
Args:
language: The Language enum value to convert.
Returns:
The Smallest language code string, or None if unsupported.
The corresponding Smallest language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

5
src/pipecat/services/stt_service.py
View File

@@ -269,7 +269,10 @@ class STTService(AIService):
language: The language to convert.
Returns:
The service-specific language identifier, or None if not supported.
The service-specific language identifier. Return ``None`` to
indicate an unsupported language. This optional return is an
extension hook for future or third-party subclasses; as of
2026-04-28, first-party services return a string.
"""
return Language(language)

5
src/pipecat/services/tts_service.py
View File

@@ -467,7 +467,10 @@ class TTSService(AIService):
language: The language to convert.
Returns:
The service-specific language identifier, or None if not supported.
The service-specific language identifier. Return ``None`` to
indicate an unsupported language. This optional return is an
extension hook for future or third-party subclasses; as of
2026-04-28, first-party services return a string.
"""
return Language(language)

9
src/pipecat/services/whisper/base_stt.py
View File

@@ -40,7 +40,7 @@ class BaseWhisperSTTSettings(STTSettings):
temperature: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
def language_to_whisper_language(language: Language) -> str | None:
def language_to_whisper_language(language: Language) -> str:
"""Maps pipecat Language enum to Whisper API language codes.
Language support for Whisper API.
@@ -50,7 +50,10 @@ def language_to_whisper_language(language: Language) -> str | None:
language: A Language enum value representing the input language.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AF: "af",
@@ -235,7 +238,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
language: The Language enum value to convert.
Returns:
str or None: The corresponding service language code, or None if not supported.
The corresponding service language code, or None if not supported.
"""
return language_to_whisper_language(language)

9
src/pipecat/services/whisper/stt.py
View File

@@ -97,14 +97,17 @@ class MLXModel(Enum):
LARGE_V3_TURBO_Q4 = "mlx-community/whisper-large-v3-turbo-q4"
def language_to_whisper_language(language: Language) -> str | None:
def language_to_whisper_language(language: Language) -> str:
"""Maps pipecat Language enum to Whisper language codes.
Args:
language: A Language enum value representing the input language.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
Note:
Only includes languages officially supported by Whisper.
@@ -300,7 +303,7 @@ class WhisperSTTService(SegmentedSTTService):
language: The Language enum value to convert.
Returns:
str or None: The corresponding Whisper language code, or None if not supported.
The corresponding Whisper language code, or None if not supported.
"""
return language_to_whisper_language(language)

7
src/pipecat/services/xai/stt.py
View File

@@ -44,7 +44,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_xai_stt_language(language: Language) -> str | None:
def language_to_xai_stt_language(language: Language) -> str:
"""Convert a Language enum to the xAI STT language code.
xAI STT accepts two-letter language codes (e.g. ``en``, ``fr``, ``de``,
@@ -54,7 +54,10 @@ def language_to_xai_stt_language(language: Language) -> str | None:
language: The Language enum value to convert.
Returns:
The corresponding xAI STT language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar",

7
src/pipecat/services/xai/tts.py
View File

@@ -49,14 +49,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_xai_language(language: Language) -> str | None:
def language_to_xai_language(language: Language) -> str:
"""Convert a Language enum to xAI language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding xAI language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.AR: "ar-EG",

7
src/pipecat/services/xtts/tts.py
View File

@@ -37,14 +37,17 @@ from pipecat.utils.tracing.service_decorators import traced_tts
# https://github.com/coqui-ai/xtts-streaming-server
def language_to_xtts_language(language: Language) -> str | None:
def language_to_xtts_language(language: Language) -> str:
"""Convert a Language enum to XTTS language code.
Args:
language: The Language enum value to convert.
Returns:
The corresponding XTTS language code, or None if not supported.
The corresponding service language code. If ``language`` is not in
the verified mapping, falls back to the base language code (e.g.,
``en`` from ``en-US``) and logs a warning (via
``resolve_language(..., use_base_code=True)``).
"""
LANGUAGE_MAP = {
Language.CS: "cs",