GPT Realtime Translate

# 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/openai/gpt-realtime-translate"


async def send_audio(ws, path):
    with wave.open(path, "rb") as audio:
        if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate()) != (1, 2, 24000):
            raise ValueError("Use mono, 16-bit PCM WAV audio at 24 kHz")
        while chunk := audio.readframes(4800):  # 200 ms at 24 kHz
            await ws.send(json.dumps({
                "type": "session.input_audio_buffer.append",
                "audio": base64.b64encode(chunk).decode("ascii"),
            }))
            await asyncio.sleep(len(chunk) / 48000)
    # The service flushes remaining output, then sends session.closed.
    await ws.send(json.dumps({"type": "session.close"}))


async def receive_events(ws):
    with open("translated.pcm", "wb") as output:
        async for raw in ws:
            event = json.loads(raw)
            if event["type"] == "session.output_audio.delta":
                output.write(base64.b64decode(event["delta"]))
            elif event["type"] == "session.output_transcript.delta":
                print(event["delta"], end="", flush=True)
            elif event["type"] == "error":
                raise RuntimeError(event["error"])
            elif event["type"] == "session.closed":
                print()
                return
    raise RuntimeError("Connection ended before session.closed")


async def main(path, language):
    headers = {"Authorization": f"Bearer {os.environ['MODELLIX_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers, max_size=16 * 1024 * 1024) as ws:
        # 1. Wait for session.created, then set the output language.
        created = json.loads(await ws.recv())
        if created["type"] != "session.created":
            raise RuntimeError(created)
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {"audio": {"output": {"language": language}}},
        }))

        # 2. Stream audio and read events at the same time.
        await asyncio.gather(send_audio(ws, path), receive_events(ws))


if len(sys.argv) != 3:
    raise SystemExit("Usage: python translate.py speech.wav es")

asyncio.run(main(sys.argv[1], sys.argv[2]))