Live Transcription (Websocket)

# pip install camb-sdk
import asyncio
import os
import wave

from camb.client import CambAI
from camb.live_transcription import FileAudioSource, ServerMessageType


async def main(audio_path: str = "sample.wav") -> None:
    with wave.open(audio_path, "rb") as wf:
        sample_rate, channels = wf.getframerate(), wf.getnchannels()

    client = CambAI(api_key=os.environ["CAMB_API_KEY"])
    session = await client.live_transcription.connect(
        model="boli-v5",
        language="en-us",
        encoding="linear16",
        sample_rate=sample_rate,
        channels=channels,
    )

    @session.on(ServerMessageType.READY)
    def _(_): print("Ready: streaming audio...")

    @session.on(ServerMessageType.RESULTS)
    def _(msg):
        text = msg.transcript.strip()
        if not text:
            return
        # Interim frames print as they arrive; commit the final on its own
        # line so the next utterance doesn't overwrite it.
        if not msg.is_final:
            print(f"[Interim] {text}\n", end="", flush=True)
        else:
            print(f"\r\033[K{text}\n", end="", flush=True)

    @session.on(ServerMessageType.CLOSED)
    def _(info): print(f"\nClosed: code={info.code}")

    async with session:
        await session.stream_audio(FileAudioSource(audio_path, real_time=True))


asyncio.run(main())