Stream a local audio file

uv run blynt_websocket_example.py --api-key <YOUR_API_KEY> path/to/audio.wav

# /// script
# dependencies = [
#   "websockets",
#   "scipy",
#   "numpy",
#   "soundfile",
# ]
# ///
"""
Minimal Blynt WebSocket client example.

Connects to the Blynt API, streams an audio file through a single manual turn,
and prints the results.

Usage:
    uv run blynt_websocket_example.py --api-key <YOUR_API_KEY> path/to/audio.wav
"""

import asyncio
import json
import sys
from pathlib import Path

import numpy as np
import soundfile as sf
import websockets
from scipy.signal import resample

# ── Configuration ─────────────────────────────────────────────────────────────

DEPLOYMENT_ID = "<your-deployment-id>"
BASE_URL = "wss://api.blynt.ai"

SAMPLE_RATE = 16_000  # Hz. Blynt expects 16 kHz mono PCM
CHUNK_SIZE = 512  # samples per chunk  →  1 024 bytes


# ── Audio helpers ──────────────────────────────────────────────────────────────


def load_audio(path: Path) -> np.ndarray:
    """Load any audio file and return 16 kHz mono int16 PCM."""
    data, sr = sf.read(path, dtype="float32", always_2d=True)
    data = data.mean(axis=1)  # stereo → mono
    if sr != SAMPLE_RATE:
        data = resample(data, int(len(data) * SAMPLE_RATE / sr)).astype(np.float32)
    data = np.clip(data, -1.0, 1.0)
    return (data * 32767).astype(np.int16)


# ── WebSocket session ──────────────────────────────────────────────────────────


async def run(api_key: str, audio_path: Path) -> None:
    uri = f"{BASE_URL}/api/v1/deployments/{DEPLOYMENT_ID}/ws"
    audio = load_audio(audio_path)
    chunk_duration = CHUNK_SIZE / SAMPLE_RATE  # seconds per chunk

    print(f"Connecting to {uri} …")
    async with websockets.connect(
        uri, additional_headers={"Authorization": f"Bearer {api_key}"}
    ) as ws:
        print("Connected.\n")

        # 1. Start the session (French + manual turn-taking)
        await ws.send(
            json.dumps(
                {
                    "type": "start_session",
                    "language": "fr",
                    "turn_taking_mode": "manual",
                }
            )
        )

        # 2. Start a turn, biasing recognition toward the expected value(s)
        await ws.send(
            json.dumps(
                {
                    "type": "start_turn",
                    "turnContext": {"hints": [{"type": "values", "values": ["huit"]}]},
                }
            )
        )

        start_time = asyncio.get_event_loop().time()

        def ts() -> str:
            return f"[+{asyncio.get_event_loop().time() - start_time:.2f}s]"

        # 3. Receiver task, runs concurrently with audio sending
        async def receive() -> None:
            while True:
                msg = await asyncio.wait_for(ws.recv(), timeout=30.0)
                if isinstance(msg, bytes):
                    continue
                event = json.loads(msg)
                event_type = event.get("type")

                if event_type == "turn_partial":
                    print(f"{ts()} [partial]  {event.get('transcript', '')!r}")
                elif event_type == "turn_ended":
                    print(
                        f"{ts()} [final]    {event.get('transcript', '')!r}  "
                        f"kind={event.get('kind')}  "
                        f"duration={event.get('turnAudioDuration', 0):.2f}s"
                    )
                    break
                elif event_type == "error":
                    print(
                        f"{ts()} [error]    {event.get('message', '')}", file=sys.stderr
                    )
                    break
                else:
                    print(f"{ts()} [{event_type}]")

        receiver = asyncio.create_task(receive())

        # 4. Stream the audio in real-time chunks
        print("Sending audio …")
        for i in range(0, len(audio), CHUNK_SIZE):
            chunk = audio[i : i + CHUNK_SIZE]
            if len(chunk) < CHUNK_SIZE:  # pad the last chunk
                chunk = np.pad(chunk, (0, CHUNK_SIZE - len(chunk)))
            await ws.send(chunk.tobytes())
            await asyncio.sleep(chunk_duration)  # simulate real-time pacing

        # 5. Signal the end of the turn and wait for the final result
        await ws.send(json.dumps({"type": "end_turn"}))
        print(f"{ts()} Audio sent, waiting for final result …\n")

        await asyncio.wait_for(receiver, timeout=60.0)


# ── Entry point ────────────────────────────────────────────────────────────────


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(
        description="Send an audio file to the Blynt WebSocket API and print the transcript."
    )
    parser.add_argument("--api-key", required=True, help="Your Blynt API key")
    parser.add_argument("audio", type=Path, help="Path to the audio file (WAV, FLAC, …)")
    args = parser.parse_args()

    if not args.audio.is_file():
        print(f"Error: file not found: {args.audio}", file=sys.stderr)
        sys.exit(1)

    asyncio.run(run(args.api_key, args.audio))


if __name__ == "__main__":
    main()

Connecting to wss://api.blynt.ai/api/v1/deployments/<your-deployment-id>/ws …
Connected.

Sending audio …
[+0.52s] [partial]  'hui'
[+0.98s] [partial]  'huit'
[+1.21s] Audio sent, waiting for final result …
[+1.36s] [final]    'huit'  kind=end_of_utterance  duration=1.30s