Real-time live captioner

# pip install pyaudio websocket-client
import pyaudio
import websocket
import json
import threading
import time
from urllib.parse import urlencode

# ── Config ────────────────────────────────────────────────────
YOUR_API_KEY = "YOUR_API_KEY"

# Add domain-specific terms to boost recognition accuracy
KEYTERMS = ["AssemblyAI", "Universal-3.5 Pro", "LLM Gateway", "speech-to-text"]

CONNECTION_PARAMS = {
    "sample_rate": 16000,
    "speech_model": "universal-3-5-pro",
    "format_turns": True,
    "keyterms_prompt": KEYTERMS,
}

API_ENDPOINT = (
    f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS, doseq=True)}"
)

# Audio settings
FRAMES_PER_BUFFER = 800
SAMPLE_RATE = 16000
stop_event = threading.Event()
caption_count = 0

def on_open(ws):
    print(f"Live captioning started — keyterms: {', '.join(KEYTERMS)}")
    print("Speak into your microphone. Press Ctrl+C to stop.\n")
    print("-" * 60)

    def stream_audio():
        audio = pyaudio.PyAudio()
        stream = audio.open(
            input=True, frames_per_buffer=FRAMES_PER_BUFFER,
            channels=1, format=pyaudio.paInt16, rate=SAMPLE_RATE,
        )
        while not stop_event.is_set():
            try:
                data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
                ws.send(data, websocket.ABNF.OPCODE_BINARY)
            except Exception:
                break
        stream.stop_stream()
        stream.close()
        audio.terminate()

    threading.Thread(target=stream_audio, daemon=True).start()

def on_message(ws, message):
    global caption_count
    data = json.loads(message)

    if data.get("type") == "Turn":
        transcript = data.get("transcript", "")
        if data.get("end_of_turn") and transcript:
            caption_count += 1
            print(f"\r[{caption_count:03d}] {transcript}")
        elif transcript:
            # Show partial (live) caption
            print(f"\r  >> {transcript[-70:]}", end="", flush=True)

    elif data.get("type") == "Termination":
        duration = data.get("audio_duration_seconds", 0)
        print(f"\n{'=' * 60}")
        print(f"Session ended — {caption_count} captions, {duration}s of audio")

def on_error(ws, error):
    print(f"\nError: {error}")
    stop_event.set()

def on_close(ws, code, msg):
    stop_event.set()

ws_app = websocket.WebSocketApp(
    API_ENDPOINT,
    header={"Authorization": YOUR_API_KEY},
    on_open=on_open, on_message=on_message,
    on_error=on_error, on_close=on_close,
)

ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
ws_thread.start()

try:
    while ws_thread.is_alive():
        time.sleep(0.1)
except KeyboardInterrupt:
    print("\n\nStopping...")
    stop_event.set()
    if ws_app.sock and ws_app.sock.connected:
        ws_app.send(json.dumps({"type": "Terminate"}))
        time.sleep(2)
    ws_app.close()