Transcription (STT) via WebSocket

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")
    stt_config = {"language": "en", "delay_in_frames": 16}

    async with client.stt_realtime(
        model_name="default",
        input_format="pcm",
        json_config=stt_config,
    ) as stt:
        # Producer: push 80 ms PCM frames as they're captured.
        async def producer():
            async for chunk in microphone_chunks():
                await stt.send_audio(chunk)
            await stt.send_eos()

        # Consumer: print transcripts as they arrive.
        async def consumer():
            async for msg in stt:
                if msg["type"] == "text":
                    print(msg["text"], end=" ", flush=True)
                elif msg["type"] == "end_of_stream":
                    return

        await asyncio.gather(producer(), consumer())

asyncio.run(main())

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    async def audio_generator(audio_data, chunk_size=1920):
        for i in range(0, len(audio_data), chunk_size):
            yield audio_data[i : i + chunk_size]

    setup = {
        "model_name": "default",
        "input_format": "pcm",
        "json_config": {"language": "en", "delay_in_frames": 16},
    }

    stream = await client.stt_stream(setup, audio_generator(audio_data))

    async for segment in stream.iter_text():
        print(f"[{segment.start_s:.2f}s] {segment.text}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    async with client.stt_realtime(
        model_name="default",
        input_format="pcm",
        json_config={"language": "en", "delay_in_frames": 16},
    ) as stt:
        async def producer():
            async for chunk in microphone_chunks():  # 80 ms PCM frames
                await stt.send_audio(chunk)
            await stt.send_eos()

        async def consumer():
            transcript = []
            async for msg in stt:
                if msg["type"] == "text":
                    transcript.append(msg["text"])
                elif msg["type"] == "step":
                    if msg["vad"][2]["inactivity_prob"] > 0.5:
                        print(f"[turn] {' '.join(transcript)}")
                        transcript = []
                elif msg["type"] == "end_of_stream":
                    return

        await asyncio.gather(producer(), consumer())

if __name__ == "__main__":
    asyncio.run(main())

async with client.stt_realtime(
    model_name="default",
    input_format="pcm",
    json_config={"language": "en"},
) as stt:
    # ... send audio in a producer task ...

    async for msg in stt:
        if msg["type"] == "text":
            print(f"Transcription: {msg['text']}")

        elif msg["type"] == "step":
            # VAD steps arrive every 80 ms.
            inactivity_prob = msg["vad"][2]["inactivity_prob"]
            print(f"Inactivity probability: {inactivity_prob}")

        elif msg["type"] == "flushed":
            print(f"Flush complete: {msg['flush_id']}")

{"type": "text",     "text": "Hello world", "start_s": 0.5, "stream_id": 0}
{"type": "end_text", "stop_s": 2.5, "stream_id": 0}

{
  "type": "step",
  "vad": [
    {"horizon_s": 0.5, "inactivity_prob": 0.9140625},
    {"horizon_s": 1.0, "inactivity_prob": 0.859375},
    {"horizon_s": 2.0, "inactivity_prob": 0.53515625},
    {"horizon_s": 3.0, "inactivity_prob": 0.40625}
  ],
  "step_idx": 5,
  "step_duration_s": 0.08,
  "total_duration_s": 0.4
}

# vad is ordered by horizon: 0.5 s, 1.0 s, 2.0 s, 3.0 s
turn_ended = msg["vad"][3]["inactivity_prob"] > 0.5  # 3 s horizon

async with client.stt_realtime(
    model_name="default",
    input_format="pcm",
    json_config={"language": "en", "delay_in_frames": 10},
    wait_for_ready_on_start=True,
) as stt:
    print(stt.ready["delay_in_frames"])

async with client.stt_realtime(
    model_name="default", input_format="pcm", json_config={"language": "en"}
) as stt:
    # Send some audio
    for i in range(0, len(audio_chunk), 1920):
        await stt.send_audio(audio_chunk[i:i + 1920])

    # Force processing of all buffered audio
    await stt.send_flush(flush_id=1)

    # Continue receiving - text results arrive, then a flushed confirmation
    async for msg in stt:
        if msg["type"] == "text":
            print(f"Transcription: {msg['text']}")
        elif msg["type"] == "flushed":
            print(f"All audio up to flush {msg['flush_id']} has been processed")
            break

wscat -c "wss://api.gradium.ai/api/speech/asr" \
  -H "x-api-key: your_api_key"
# After connection, type:
# {"type":"setup","model_name":"default","input_format":"pcm","json_config":{"language":"en","delay_in_frames":16}}

import asyncio
import base64
import json

import websockets

CHUNK_BYTES = 1920 * 2  # 80 ms at 24 kHz, 16-bit mono.


async def transcribe(api_key: str, pcm_audio: bytes):
    setup = {
        "type": "setup",
        "model_name": "default",
        "input_format": "pcm",
        "json_config": {
            "language": "en",
            "delay_in_frames": 16,
        },
    }

    async with websockets.connect(
        "wss://api.gradium.ai/api/speech/asr",
        additional_headers={"x-api-key": api_key},
    ) as ws:
        await ws.send(json.dumps(setup))
        ready = json.loads(await ws.recv())
        assert ready["type"] == "ready"

        async def producer():
            for off in range(0, len(pcm_audio), CHUNK_BYTES):
                chunk = pcm_audio[off : off + CHUNK_BYTES]
                await ws.send(json.dumps({
                    "type": "audio",
                    "audio": base64.b64encode(chunk).decode(),
                }))
            await ws.send(json.dumps({"type": "end_of_stream"}))

        async def consumer():
            while True:
                msg = json.loads(await ws.recv())
                if msg["type"] == "text":
                    print(msg["text"])
                elif msg["type"] == "end_of_stream":
                    return
                elif msg["type"] == "error":
                    raise RuntimeError(msg["message"])

        await asyncio.gather(producer(), consumer())

asyncio.run(transcribe("your_api_key", open("input.pcm", "rb").read()))