Gemini 3.5 Transcribe Live

# pip install "websockets>=14"
import asyncio
import base64
import json
import os
import sys
import wave

import websockets

URL = "wss://api.modellix.ai/api/v1/live/google/gemini-3.5-transcribe-live"


async def send_audio(ws, path):
    with wave.open(path, "rb") as audio:
        if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate()) != (1, 2, 16000):
            raise ValueError("Use mono, 16-bit PCM WAV audio at 16 kHz")
        await ws.send(json.dumps({"realtimeInput": {"activityStart": {}}}))
        while chunk := audio.readframes(1600):
            await ws.send(json.dumps({
                "realtimeInput": {
                    "audio": {
                        "mimeType": "audio/pcm;rate=16000",
                        "data": base64.b64encode(chunk).decode("ascii"),
                    }
                }
            }))
            await asyncio.sleep(len(chunk) / 32000)
        await ws.send(json.dumps({"realtimeInput": {"activityEnd": {}}}))


async def receive_utterance(ws):
    async for raw in ws:
        message = json.loads(raw)
        if message.get("error"):
            raise RuntimeError(message["error"])
        content = message.get("serverContent") or {}
        if content.get("interimInputTranscription"):
            print("Interim:", content["interimInputTranscription"]["text"])
        if content.get("inputTranscription"):
            print("Final:", content["inputTranscription"]["text"])
        if message.get("usageMetadata"):
            print("Usage:", message["usageMetadata"])
        if content.get("generationComplete"):
            return
    raise RuntimeError("Connection ended before utterance completion")


async def main():
    headers = {"Authorization": f"Bearer {os.environ['MODELLIX_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        await ws.send(json.dumps({
            "setup": {
                "generationConfig": {"responseModalities": ["TEXT"]},
                "inputAudioTranscription": {"languageCodes": []},
                "realtimeInputConfig": {
                    "automaticActivityDetection": {"disabled": True}
                },
            }
        }))
        reply = json.loads(await ws.recv())
        if "setupComplete" not in reply:
            raise RuntimeError(reply)
        for path in sys.argv[1:]:
            await asyncio.wait_for(
                asyncio.gather(send_audio(ws, path), receive_utterance(ws)),
                timeout=600,
            )


if len(sys.argv) < 2:
    raise SystemExit("Usage: python transcribe_live.py first.wav [second.wav]")

asyncio.run(main())