mistral_common.tokens.tokenizers.audio

@staticmethod
def from_audio_chunk(chunk: "AudioChunk") -> "Audio":
    r"""Create an Audio instance from an AudioChunk.

    Args:
        chunk: An AudioChunk with `input_audio` as str (base64) or bytes.

    Returns:
        An instance of the Audio class.
    """
    input_audio = chunk.input_audio
    if isinstance(input_audio, bytes):
        return Audio.from_bytes(input_audio)
    elif isinstance(input_audio, str):
        return Audio.from_base64(input_audio)
    else:
        raise ValueError(f"Unsupported input_audio type: {type(input_audio)}")

@staticmethod
def from_base64(audio_base64: str, strict: bool = True) -> "Audio":
    r"""Create an Audio instance from a base64 encoded string.

    Args:
        audio_base64: The base64 encoded audio data, optionally prefixed
            with a data:audio/<format>;base64, URL prefix (stripped
            automatically).
        strict: If `True`, strictly enforce mono audio; multi-channel audio
            raises an error. If `False`, extra channels are dropped.

    Returns:
        An instance of the Audio class.

    Raises:
        ValueError: If the string is not valid base64 or decoding fails.
    """
    assert_soundfile_installed()

    if re.match(r"^data:audio/\w+;base64,", audio_base64):
        audio_base64 = audio_base64.split(",")[1]

    try:
        audio_bytes = base64.b64decode(audio_base64)
    except Exception as e:
        raise ValueError("base64 decoding failed. Please check the input string is a valid base64.") from e

    return Audio.from_bytes(audio_bytes, strict=strict)

@staticmethod
def from_bytes(audio_bytes: bytes, strict: bool = True) -> "Audio":
    r"""Create an Audio instance from bytes.

    Args:
        audio_bytes: The audio data as bytes in a soundfile-readable
            format (e.g., wav, mp3).
        strict: If `True`, strictly enforce mono audio; multi-channel audio
            raises an error. If `False`, extra channels are averaged down
            to mono.

    Returns:
        An instance of the Audio class.

    Raises:
        ValueError: If the audio is multi-channel and strict is `True`, or
            the bytes cannot be decoded.
    """
    assert_soundfile_installed()

    with io.BytesIO(audio_bytes) as audio_file:
        with sf.SoundFile(audio_file) as f:
            audio_array = f.read(dtype="float32")
            sampling_rate = f.samplerate
            audio_format = f.format

    format_enum = AudioFormat(audio_format)
    format = format_enum.value.lower()

    if audio_array.ndim != 1:
        if strict:
            raise ValueError(f"{audio_array.ndim=}")
        else:
            audio_array = audio_array.mean(axis=1)

    return Audio(audio_array=audio_array, sampling_rate=sampling_rate, format=format)

@staticmethod
def from_file(file: str, strict: bool = True) -> "Audio":
    r"""Create an Audio instance from an audio file.

    Args:
        file: Path to the audio file. A file:// URI prefix is accepted
            and stripped.
        strict: If `True`, strictly enforce mono audio; multi-channel audio
            raises an error. If `False`, extra channels are dropped.

    Returns:
        An instance of the Audio class.

    Raises:
        FileNotFoundError: If the file does not exist.
    """
    assert_soundfile_installed()

    if isinstance(file, str) and file.startswith("file://"):
        file = file[7:]

    if not Path(file).exists():
        raise FileNotFoundError(f"{file=} does not exist")

    with open(file, "rb") as f:
        audio_bytes = f.read()

    return Audio.from_bytes(audio_bytes, strict=strict)

@staticmethod
def from_raw_audio(audio: "RawAudio") -> "Audio":
    r"""Create an Audio instance from a RawAudio object.

    Deprecated: Use `from_base64()` or `from_bytes()` instead. Will be removed in 1.13.0.

    Args:
        audio: The RawAudio object containing audio data.

    Returns:
        An instance of the Audio class.
    """

    warn_once(
        "Audio.from_raw_audio",
        "Audio.from_raw_audio() is deprecated. "
        "Use Audio.from_base64() or Audio.from_bytes() instead. "
        "Will be removed in 1.13.0.",
        DeprecationWarning,
        stacklevel=2,
    )
    if isinstance(audio.data, bytes):
        return Audio.from_bytes(audio.data)
    elif isinstance(audio.data, str):
        return Audio.from_base64(audio.data)
    else:
        raise ValueError(f"Unsupported audio data type: {type(audio.data)}")

@staticmethod
def from_url(url: str, strict: bool = True) -> "Audio":
    r"""Create an Audio instance from a URL.

    Downloads the audio file and decodes it into samples.

    Args:
        url: The URL of the audio file (http or https).
        strict: If `True`, strictly enforce mono audio; multi-channel audio
            raises an error. If `False`, extra channels are dropped.

    Returns:
        An instance of the Audio class.

    Raises:
        ValueError: If the download or decoding fails.
    """
    try:
        response = _requests_lib.get(url)
        response.raise_for_status()
        return Audio.from_bytes(response.content, strict=strict)
    except _requests_lib.RequestException as e:  # Something went wrong with the request.
        raise ValueError(f"Failed to download audio from URL: {url}") from e
    except Exception as e:  # Something went wrong with the audio file.
        raise ValueError(f"Failed to create Audio instance from URL: {url} .") from e

def resample(self, new_sampling_rate: int) -> None:
    r"""Resample audio data to a new sampling rate.

    Mutates this instance's `audio_array` and `sampling_rate` in place. No-op
    if the sampling rate is already the target.

    Args:
        new_sampling_rate: The new sampling rate in Hz.
    """
    if self.sampling_rate == new_sampling_rate:
        return

    assert_soxr_installed()

    self.audio_array = soxr.resample(self.audio_array, self.sampling_rate, new_sampling_rate, quality="HQ")
    self.sampling_rate = new_sampling_rate

def to_base64(self, format: str, prefix: bool = False) -> str:
    r"""Convert the audio data to a base64 encoded string.

    Args:
        format: The format to encode the audio in (e.g., "wav"). Must be
            supported by soundfile.
        prefix: If `True`, prepend a data:audio/<format>;base64, prefix to
            the output string.

    Returns:
        The base64 encoded audio data.

    Raises:
        AssertionError: If format is not supported by soundfile.
    """
    assert_soundfile_installed()

    assert format in EXPECTED_FORMAT_VALUES, f"{format=} not in {EXPECTED_FORMAT_VALUES=}"

    with io.BytesIO() as audio_file:
        sf.write(audio_file, self.audio_array, self.sampling_rate, format=format.upper())
        audio_file.seek(0)
        base64_str = base64.b64encode(audio_file.read()).decode("utf-8")
    if prefix:
        base64_str = f"data:audio/{format.lower()};base64,{base64_str}"
    return base64_str

def delay_len(self, transcription_delay_ms: float) -> int:
    r"""Convert a transcription delay into an audio length in samples.

    Args:
        transcription_delay_ms: The delay in milliseconds.

    Returns:
        The corresponding number of audio samples at this config's sampling rate.
    """
    return int(transcription_delay_ms / 1000.0 * self.sampling_rate)

def get_num_delay_tokens(self, transcription_delay_ms: float | None = None) -> int:
    r"""Compute the number of tokens covering the streaming transcription delay.

    Args:
        transcription_delay_ms: The delay in milliseconds. If `None`, falls back
            to the config's `transcription_delay_ms`.

    Returns:
        The number of audio tokens corresponding to the delay length.

    Raises:
        AssertionError: If not streaming, or no delay is set on either the
            argument or the config.
    """
    assert self.is_streaming, f"Can't call get_num_delay_tokens if {self.is_streaming=}."
    if transcription_delay_ms is None:
        transcription_delay_ms = self.transcription_delay_ms
    # streaming pad tokens
    assert transcription_delay_ms is not None, f"Can't call get_num_delay_tokens if {transcription_delay_ms=}."
    return self.num_audio_tokens(self.delay_len(transcription_delay_ms))

def n_right_pad_tokens(self, transcription_delay_ms: float | None = None) -> int:
    r"""Number of right padding tokens to add after the audio.

    Covers the induced delay, the BOS token, and a buffer for long words
    (see `OFFLINE_STREAMING_BUFFER_TOKENS`).

    Args:
        transcription_delay_ms: The delay in milliseconds. If `None`, falls back
            to the config's `transcription_delay_ms`.

    Returns:
        The number of right padding tokens.

    Raises:
        AssertionError: If not streaming, or no delay is set on either the
            argument or the config.
    """
    assert self.is_streaming, f"Can't call n_right_pad_tokens if {self.is_streaming=}."
    # we need to pad on the right to ensure the models transcribes
    # - the induced delay on the prefill step (num_delay_tokens)
    # - the BOS token (1)
    # - a heuristic that defines a max token length for a single word
    #   (OFFLINE_STREAMING_BUFFER_TOKENS)
    return (self.get_num_delay_tokens(transcription_delay_ms) + 1) + OFFLINE_STREAMING_BUFFER_TOKENS

def num_audio_tokens(self, audio_len: int) -> int:
    r"""Convert an audio length in samples into the number of audio tokens.

    Args:
        audio_len: Audio length in samples. Rounded up to a whole number of
            frames when not a multiple of the hop length.

    Returns:
        The number of audio tokens covering the given length.
    """
    if audio_len % self.encoding_config.hop_length != 0:
        audio_len = math.ceil(audio_len / self.encoding_config.hop_length - 1)
    else:
        audio_len = audio_len // self.encoding_config.hop_length

    return math.ceil(audio_len / self.audio_length_per_tok)

def __call__(self, content: AudioChunk | AudioURLChunk) -> AudioEncoding:
    r"""Call the encoder on an audio chunk or URL chunk.

    Args:
        content: Audio or URL chunk to encode.

    Returns:
        Encoded audio data and tokens.
    """
    if isinstance(content, AudioURLChunk):
        return self._encode_audio_url_chunk(content)
    elif isinstance(content, AudioChunk):
        return self._encode_audio_chunk(content)
    else:
        raise ValueError(f"Unsupported content type: {type(content)}")

def encode_audio(self, audio: Audio, transcription_delay_ms: float | None = None) -> AudioEncoding:
    r"""Encode an audio optionally with transcription delay."""
    audio.resample(self.audio_config.sampling_rate)
    audio.audio_array = self.pad(audio.audio_array, self.audio_config.sampling_rate, transcription_delay_ms)

    if self.audio_config.transcription_format == TranscriptionFormat.STREAMING:
        tokens = self.encode_streaming_tokens(transcription_delay_ms)
    else:
        tokens = self._encode_audio_tokens(audio.audio_array.shape[0])

    return AudioEncoding(
        tokens=tokens,
        audio=audio,
    )

def encode_audio_for_speech_request(self, audio: Audio | None, voice: str | None) -> AudioEncoding:
    r"""Encode audio or voice preset into an AudioEncoding for speech synthesis.

    Either `audio` (reference audio for voice cloning) or `voice` (preset name)
    must be provided. When `audio` is given it takes precedence.

    Args:
        audio: Reference audio waveform, or `None` to use a voice preset.
        voice: Preset voice name (e.g. 'Neutral Male', 'Neutral Female'), or `None` when using ref audio.

    Returns:
        AudioEncoding containing the token sequence and optional audio data.
    """
    assert audio is not None or voice is not None, (
        f"Either audio or voice must be defined to encode audio, got {audio=} and {voice=}"
    )

    if audio is not None:
        audio.resample(self.audio_config.sampling_rate)
        num_audio_tokens = self._get_num_audio_token_for_speech_request(len(audio.audio_array))
    else:
        assert self.audio_config.voice_num_audio_tokens is not None, (
            "voice_num_audio_tokens must be set in audio config to use voice-based speech requests"
        )
        assert voice is not None and voice in self.audio_config.voice_num_audio_tokens, (
            f"Unknown voice {voice!r}, expected one of {list(self.audio_config.voice_num_audio_tokens)}"
        )
        num_audio_tokens = self.audio_config.voice_num_audio_tokens[voice]
    tokens = self._encode_audio_tokens_for_speech_request(num_audio_tokens)

    return AudioEncoding(
        tokens=tokens,
        audio=audio,
    )

def encode_streaming_tokens(self, transcription_delay_ms: float | None = None) -> list[int]:
    r"""Encode the streaming tokens given a transcription delay."""
    assert isinstance(self.audio_config.encoding_config, AudioSpectrogramConfig), (
        f"Audio encoder must be spectrogram encoder, got {self.audio_config.encoding_config=}"
    )
    assert self.audio_config.transcription_delay_ms is not None

    # streaming pad tokens consist of silence we pad on left + delay tokens
    stream_pad_prefix_len = self.audio_config.n_left_pad_tokens + self.audio_config.get_num_delay_tokens(
        transcription_delay_ms
    )
    tokens = [self.streaming_pad] * stream_pad_prefix_len

    return tokens

def get_padding_audio(self, transcription_delay_ms: float | None = None) -> tuple[Audio, Audio]:
    r"""Get left and right padding for realtime audio models.

    Args:
        transcription_delay_ms: Target delay in milliseconds between the audio
            and text streams for streaming transcription. Controls the padding
            length. If `None`, falls back to the audio config's
            `transcription_delay_ms`.

    Returns:
        Tuple of (left_pad_audio, right_pad_audio) as silent `Audio` objects.

    Raises:
        AssertionError: If no delay is set on either the argument or the
            audio config.
    """

    left_pad, right_pad = self._get_streaming_pad(0, transcription_delay_ms)
    left_pad_audio = Audio(
        audio_array=np.zeros(left_pad, dtype=np.float32),
        sampling_rate=self.audio_config.sampling_rate,
        format="wav",
    )
    right_pad_audio = Audio(
        audio_array=np.zeros(right_pad, dtype=np.float32),
        sampling_rate=self.audio_config.sampling_rate,
        format="wav",
    )
    return left_pad_audio, right_pad_audio

def next_multiple_of_chunk_frames(self, audio_array_len: int, sampling_rate: int) -> int:
    r"""Calculate the next multiple of chunk frames.

    Args:
        audio_array_len: Length of the audio array.
        sampling_rate: Sampling rate of the audio.

    Returns:
        The next multiple of chunk frames.
    """
    assert sampling_rate == self.audio_config.sampling_rate, (
        f"Expected {sampling_rate=} to be {self.audio_config.sampling_rate=}"
    )
    assert self.audio_config.chunk_length_s is not None, (
        f"Can't call next_multiple_of_chunk_frames if {self.audio_config.chunk_length_s=}."
    )

    return math.ceil(audio_array_len / self.audio_config.chunk_frames) * self.audio_config.chunk_frames

def pad(
    self,
    audio_array: np.ndarray,
    sampling_rate: int,
    transcription_delay_ms: float | None = None,
    **kwargs: Any,
) -> np.ndarray:
    r"""Pad the audio array to the desired length.

    Depending on the audio config, pads the audio to a multiple of the
    chunk length, to the streaming length (both left and right), or to at
    least one spectrogram frame.

    Args:
        audio_array: Audio data as a numpy array of samples.
        sampling_rate: Sampling rate of the audio in Hz.
        transcription_delay_ms: Target delay in milliseconds between the audio
            and text streams for streaming transcription. Controls how much
            right padding is added so the model transcribes the induced delay.
            If `None`, falls back to the audio config's `transcription_delay_ms`.
        **kwargs: Ignored. Only present to swallow deprecated keyword arguments
            from callers; will be removed in 1.13.0.

    Returns:
        The padded audio array.

    Raises:
        AssertionError: If streaming and no delay is set on either the argument
            or the audio config.
    """
    # TODO(Patrick) - remove **kwargs as it's just there to swallow deprecated
    # keyword args from voxtral_realtime in vLLM. It was
    # relevant for the release. Remove in mistral_common version 1.13.0
    if self.audio_config.chunk_length_s:
        next_multiple_of_chunk_frames = self.next_multiple_of_chunk_frames(audio_array.shape[-1], sampling_rate)
        audio_array = np.pad(audio_array, (0, next_multiple_of_chunk_frames - audio_array.shape[-1]))
    elif self.audio_config.is_streaming:
        left_pad, right_pad = self._get_streaming_pad(audio_array.shape[-1], transcription_delay_ms)
        # we pad both left & right as this leads to better performance
        audio_array = np.pad(audio_array, (left_pad, right_pad))
    elif (
        isinstance(self.encoding_config, AudioSpectrogramConfig)
        and audio_array.shape[-1] < self.encoding_config.window_size
    ):
        # minimum length for audios is at least one spectrogram frame
        audio_array = np.pad(audio_array, (0, self.encoding_config.window_size - audio_array.shape[-1]))

    return audio_array