Quickstart: transcribe speech in realtime

import asyncio
import os
import wave

from mistralai.client import Mistral
from mistralai.client.models import (
    AudioFormat,
    RealtimeTranscriptionError,
    TranscriptionStreamDone,
    TranscriptionStreamTextDelta,
)

AUDIO_FILE = "support_call.wav"  # 16-bit mono PCM WAV

async def read_chunks(path):
    with wave.open(path, "rb") as wav:
        while chunk := wav.readframes(wav.getframerate() // 10):
            yield chunk
            await asyncio.sleep(0.1)  # send 100 ms of audio every 100 ms, like a live source

async def main():
    with wave.open(AUDIO_FILE, "rb") as wav:
        sample_rate = wav.getframerate()

    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
    async for event in client.audio.realtime.transcribe_stream(
        read_chunks(AUDIO_FILE),
        model="voxtral-mini-transcribe-realtime-3",
        audio_format=AudioFormat(encoding="pcm_s16le", sample_rate=sample_rate),
    ):
        if isinstance(event, TranscriptionStreamTextDelta):
            print(event.text, end="", flush=True)
        elif isinstance(event, TranscriptionStreamDone):
            print(f"\n\nFinal transcript: {event.text}")
        elif isinstance(event, RealtimeTranscriptionError):
            print(f"\nError: {event.error.message}")

asyncio.run(main())