Repository: livekit/agents
File: livekit-plugins/livekit-plugins-sarvam/livekit/plugins/sarvam/stt.py (Python)

Code before:
    """

    language: str  # BCP-47 language code, e.g., "hi-IN", "en-IN"
    api_key: str
    model: SarvamSTTModels | str = "saarika:v2.5"
    base_url: str | None = None
    streaming_url: str | None = None
    prompt: str | None = None  # Optional prompt for STT translate (saaras models only)
    high_vad_sensitivity: bool | None = None
    sample_rate: int = 16000
    flush_signal: bool | None = None
    input_audio_codec: str | None = None

    def __post_init__(self) -> None:
        """Set URLs based on model if not explicitly provided."""
        if self.base_url is None or self.streaming_url is None:
            base_url, streaming_url = _get_urls_for_model(self.model)
            if self.base_url is None:
                self.base_url = base_url
            if self.streaming_url is None:
                self.streaming_url = streaming_url
        if self.sample_rate <= 0:
            raise ValueError("sample_rate must be greater than zero")


def _get_option_value(default_value: T, override_value: NotGivenOr[T]) -> T:
    """Helper to get option value with NOT_GIVEN handling."""
    if isinstance(override_value, type(NOT_GIVEN)):
        return default_value
    return cast(T, override_value)


def _get_urls_for_model(model: str) -> tuple[str, str]:
    """Get base URL and streaming URL based on model type.

    Args:
        model: The Sarvam model name

    Returns:
        Tuple of (base_url, streaming_url)
    """
    if model.startswith("saaras:"):
        return SARVAM_STT_TRANSLATE_BASE_URL, SARVAM_STT_TRANSLATE_STREAMING_URL
    else:  # saarika models
        return SARVAM_STT_BASE_URL, SARVAM_STT_STREAMING_URL


def _calculate_audio_duration(
    buffer: AudioBuffer,
) -> float:  # TODO: Copied from livekit/agents/utils/audio.py, check if it can be reused
    """Calculate audio duration from buffer."""

Review comment (suggestion):
`_get_option_value` is unnecessary, you can use

https://github.com/livekit/agents/blob/694720d18d753faca95c3a241f53ed0041088a0b/livekit-agents/livekit/agents/utils/misc.py#L25

You can also use this pattern:
`val = val or default`

Code after:
    language: str  # BCP-47 language code, e.g., "hi-IN", "en-IN"
    api_key: str
    model: SarvamSTTModels | str = "saarika:v2.5"
    base_url: str | None = None
    streaming_url: str | None = None
    prompt: str | None = None  # Optional prompt for STT translate (saaras models only)
    high_vad_sensitivity: bool | None = None
    sample_rate: int = 16000
    flush_signal: bool | None = None
    input_audio_codec: str | None = None

    def __post_init__(self) -> None:
        """Set URLs based on model if not explicitly provided."""
        if self.base_url is None or self.streaming_url is None:
            base_url, streaming_url = _get_urls_for_model(self.model)
            if self.base_url is None:
                self.base_url = base_url
            if self.streaming_url is None:
                self.streaming_url = streaming_url
        if self.sample_rate <= 0:
            raise ValueError("sample_rate must be greater than zero")


def _get_urls_for_model(model: str) -> tuple[str, str]:
    """Get base URL and streaming URL based on model type.

    Args:
        model: The Sarvam model name

    Returns:
        Tuple of (base_url, streaming_url)
    """
    if model.startswith("saaras:"):
        return SARVAM_STT_TRANSLATE_BASE_URL, SARVAM_STT_TRANSLATE_STREAMING_URL
    else:  # saarika models
        return SARVAM_STT_BASE_URL, SARVAM_STT_STREAMING_URL


def _calculate_audio_duration(
    buffer: AudioBuffer,
) -> float:  # TODO: Copied from livekit/agents/utils/audio.py, check if it can be reused
    """Calculate audio duration from buffer."""
    try:
        if isinstance(buffer, list):
            # Calculate total duration from all frames
            total_samples = sum(frame.samples_per_channel for frame in buffer)
            if buffer and total_samples > 0:
                sample_rate = buffer[0].sample_rate
                return total_samples / sample_rate
        elif hasattr(buffer, "duration"):
            return buffer.duration / 1000.0  # buffer.duration is in ms