Real-Time Voice Quickstart

import asyncio
import base64
import json
import signal

import requests
import sounddevice
import websockets

AUTH_KEY = "YOUR_AUTH_KEY"
SESSION_ENDPOINT = "https://api.deepl.com/v3/voice/realtime"
TARGET_LANGUAGES = ["de", "fr"]
SAMPLE_RATE = 16000  # Must match the rate declared in source_media_content_type
CHUNK_FRAMES = 3200  # 200 ms per chunk at 16 kHz
RECORD_SECONDS = 30  # Safety cap; Ctrl+C stops recording earlier


def create_session() -> dict:
    response = requests.post(
        SESSION_ENDPOINT,
        headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        json={
            "source_media_content_type": "audio/pcm;encoding=s16le;rate=16000",
            "target_languages": TARGET_LANGUAGES,
        },
    )
    response.raise_for_status()
    return response.json()


async def send_microphone_audio(ws, stop: asyncio.Event) -> None:
    stream = sounddevice.RawInputStream(
        samplerate=SAMPLE_RATE, channels=1, dtype="int16"
    )
    stream.start()
    try:
        for _ in range(RECORD_SECONDS * SAMPLE_RATE // CHUNK_FRAMES):
            if stop.is_set():
                break
            # Read in a worker thread so receiving continues while we block
            data, _overflowed = await asyncio.to_thread(stream.read, CHUNK_FRAMES)
            encoded = base64.b64encode(bytes(data)).decode("ascii")
            await ws.send(json.dumps({"source_media_chunk": {"data": encoded}}))
    finally:
        stream.stop()
        stream.close()
    print("Finalizing transcripts...")
    await ws.send(json.dumps({"end_of_source_media": {}}))


async def receive_results(ws) -> None:
    pending = {}  # Concluded text per language, buffered until a sentence ends

    def handle_update(label: str, segments: list) -> None:
        text = pending.get(label, "") + "".join(s["text"] for s in segments)
        if text.rstrip().endswith((".", "!", "?")):
            print(f"[{label}] {text.strip()}")
            text = ""
        pending[label] = text

    async for message in ws:
        data = json.loads(message)
        if "source_transcript_update" in data:
            handle_update("source", data["source_transcript_update"]["concluded"])
        elif "target_transcript_update" in data:
            update = data["target_transcript_update"]
            handle_update(update["language"], update["concluded"])
        elif "end_of_stream" in data:
            # The very last message: flush any unfinished sentences and exit
            for label, text in pending.items():
                if text.strip():
                    print(f"[{label}] {text.strip()}")
            return
        elif "error" in data:
            raise RuntimeError(f"Voice API error: {data['error']['error_message']}")


async def main() -> None:
    stop = asyncio.Event()
    try:
        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, stop.set)
    except NotImplementedError:
        pass  # Windows: Ctrl+C raises KeyboardInterrupt instead

    session = create_session()
    url = f"{session['streaming_url']}?token={session['token']}"
    async with websockets.connect(url) as ws:
        print("Connected. Speak now (Ctrl+C to stop)...")
        await asyncio.gather(send_microphone_audio(ws, stop), receive_results(ws))
    print("Done. All transcripts are final.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass