smoke_test.py

#!/usr/bin/env python3
import asyncio, json, base64, time
import websockets
import requests

WS = "ws://localhost:5001/ws/stream/test-sess"
REST = "http://localhost:5001"

async def send_chunks():
    async with websockets.connect(WS) as ws:
        start = int(time.time() * 1000)
        for i in range(3):
            t0 = start + i*500
            t1 = t0 + 500
            # send tiny fake payload
            payload = {
                "type": "audio.chunk",
                "session_id": "test-sess",
                "seq": i,
                "t0_ms": t0,
                "t1_ms": t1,
                "mime": "audio/webm;codecs=opus",
                "b64": base64.b64encode(b"fake").decode("ascii"),
            }
            await ws.send(json.dumps(payload))
            ack = await ws.recv()
            print("ACK:", ack)

def request_transcribe():
    r = requests.post(f"{REST}/transcribe_slice", json={
        "session_id": "test-sess",
        "slice_id": "slice-1",
        "offset_ms": 1500,
        "requested_tier": "A"
    }, timeout=5)
    r.raise_for_status()
    job_id = r.json()["job_id"]
    print("job_id:", job_id)
    for _ in range(40):
        jr = requests.get(f"{REST}/job/{job_id}", timeout=5)
        if jr.status_code != 200:
            time.sleep(0.1)
            continue
        d = jr.json()
        if d.get("status") == "done":
            print("RESULT:", json.dumps(d["result"], ensure_ascii=False))
            return
        time.sleep(0.1)
    print("Timed out waiting for result")

if __name__ == "__main__":
    asyncio.run(send_chunks())
    request_transcribe()