Stream a pre-recorded file in real time

1
import websocket
2
import json
3
import threading
4
import time
5
import wave
6
import os
7
from urllib.parse import urlencode
8
9
# --- Configuration ---
10
ASSEMBLYAI_API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
11
AUDIO_FILE = "audio.wav"
12
CHUNK_DURATION = 0.1  # Send 100ms of audio per chunk
13
SAMPLE_RATE = 16000   # Must match your audio file's sample rate
14
15
CONNECTION_PARAMS = {
16
    "speech_model": "u3-rt-pro",
17
    "sample_rate": SAMPLE_RATE,
18
}
19
API_ENDPOINT = f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS)}"
20
21
ws_app = None
22
audio_thread = None
23
stop_event = threading.Event()
24
25
26
def on_open(ws):
27
    print("Connected. Streaming audio at real-time speed...")
28
29
    def stream_file():
30
        with wave.open(AUDIO_FILE, "rb") as wf:
31
            frames_per_chunk = int(wf.getframerate() * CHUNK_DURATION)
32
            start_time = time.monotonic()
33
            chunks_sent = 0
34
35
            while not stop_event.is_set():
36
                frames = wf.readframes(frames_per_chunk)
37
                if not frames:
38
                    break
39
40
                ws.send(frames, websocket.ABNF.OPCODE_BINARY)
41
                chunks_sent += 1
42
43
                # Wall-clock pacing: sleep until the next chunk is due
44
                next_chunk_time = start_time + (chunks_sent * CHUNK_DURATION)
45
                sleep_duration = next_chunk_time - time.monotonic()
46
                if sleep_duration > 0:
47
                    time.sleep(sleep_duration)
48
49
        print("Finished sending audio. Waiting for final transcripts...")
50
        try:
51
            ws.send(json.dumps({"type": "Terminate"}))
52
        except Exception:
53
            pass
54
55
    global audio_thread
56
    audio_thread = threading.Thread(target=stream_file, daemon=True)
57
    audio_thread.start()
58
59
60
def on_message(ws, message):
61
    data = json.loads(message)
62
63
    if data["type"] == "Begin":
64
        print(f"Session ID: {data['id']}")
65
    elif data["type"] == "Turn":
66
        transcript = data.get("transcript", "")
67
        if not transcript:
68
            return
69
        if data.get("end_of_turn"):
70
            print(f"[Final]: {transcript}")
71
        else:
72
            print(f"[Partial]: {transcript}")
73
    elif data["type"] == "Termination":
74
        print(f"Done. Processed {data.get('audio_duration_seconds', 0)}s of audio.")
75
76
77
def on_error(ws, error):
78
    print(f"Error: {error}")
79
    stop_event.set()
80
81
82
def on_close(ws, status_code, msg):
83
    print(f"Disconnected (status={status_code})")
84
    stop_event.set()
85
86
87
ws_app = websocket.WebSocketApp(
88
    API_ENDPOINT,
89
    header={"Authorization": ASSEMBLYAI_API_KEY},
90
    on_open=on_open,
91
    on_message=on_message,
92
    on_error=on_error,
93
    on_close=on_close,
94
)
95
96
ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
97
ws_thread.start()
98
99
try:
100
    while ws_thread.is_alive():
101
        time.sleep(0.1)
102
except KeyboardInterrupt:
103
    print("\nStopping...")
104
    stop_event.set()
105
    if ws_app and ws_app.sock and ws_app.sock.connected:
106
        try:
107
            ws_app.send(json.dumps({"type": "Terminate"}))
108
            time.sleep(2)
109
        except Exception:
110
            pass
111
    if ws_app:
112
        ws_app.close()
113
    ws_thread.join(timeout=2.0)

1
import websocket
2
import json
3
import threading
4
import time
5
import wave
6
import os
7
from urllib.parse import urlencode
8
9
ASSEMBLYAI_API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
10
AUDIO_FILE = "audio.wav"
11
CHUNK_DURATION = 0.1  # 100ms per chunk
12
SAMPLE_RATE = 16000
13
14
CONNECTION_PARAMS = {
15
    "speech_model": "u3-rt-pro",
16
    "sample_rate": SAMPLE_RATE,
17
}
18
API_ENDPOINT = f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS)}"

1
start_time = time.monotonic()
2
chunks_sent = 0
3
4
while not stop_event.is_set():
5
    frames = wav_file.readframes(frames_per_chunk)
6
    if not frames:
7
        break
8
9
    ws.send(frames, websocket.ABNF.OPCODE_BINARY)
10
    chunks_sent += 1
11
12
    # Sleep until the next chunk is due
13
    next_chunk_time = start_time + (chunks_sent * CHUNK_DURATION)
14
    sleep_duration = next_chunk_time - time.monotonic()
15
    if sleep_duration > 0:
16
        time.sleep(sleep_duration)

1
ws.send(json.dumps({"type": "Terminate"}))