Streaming Migration Guide: Deepgram to AssemblyAI

1
import json, threading, time, websocket, pyaudio, signal
2
from urllib.parse import urlencode
3
4
DG_KEY = "YOUR_DG_API_KEY"        
5
6
PARAMS = {
7
    "model": "nova-3",
8
    "encoding": "linear16",
9
    "sample_rate": "16000",
10
    "channels": "1",
11
    "punctuate": "true",
12
    "interim_results": "true",
13
}
14
15
EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"
16
17
FRAMES      = 800           
18
CHANNELS    = 1
19
FORMAT      = pyaudio.paInt16
20
SAMPLE_RATE = int(PARAMS["sample_rate"])
21
22
audio = stream = ws_app = None
23
audio_thread = ws_thread = None
24
stop_event = threading.Event()
25
26
# ── WebSocket callbacks ───────────────────────────────────────────────
27
def on_open(ws):
28
    def mic_loop():
29
        while not stop_event.is_set():
30
            ws.send(stream.read(FRAMES, exception_on_overflow=False),
31
                    websocket.ABNF.OPCODE_BINARY)
32
    global audio_thread
33
    audio_thread = threading.Thread(target=mic_loop, daemon=True)
34
    audio_thread.start()
35
36
def on_message(ws, msg):
37
    global final_metadata
38
    d = json.loads(msg)
39
    if d.get("type") == "Results":
40
        txt = d["channel"]["alternatives"][0]["transcript"]
41
        if d["is_final"]:
42
            print(" " * 80, end="\r")
43
            print(txt)
44
        else:
45
            print(txt, end="\r")
46
    elif d.get("type") == "Metadata":
47
        final_metadata = d  # save the metadata for later
48
49
def on_error(ws, error):
50
    print(f"\nWebSocket error: {error}")
51
    stop_event.set()
52
53
def on_close(ws, *args):
54
    stop_event.set()
55
    if final_metadata:
56
        print("\nFinal Metadata from Deepgram:")
57
        print(json.dumps(final_metadata, indent=2))
58
    else:
59
        print("\nNo metadata received before close.")
60
61
62
    def graceful_shutdown(signum, frame):
63
        print("\nCtrl+C received → shutting down …")
64
        stop_event.set()
65
66
        # Send CloseStream before closing WebSocket
67
        if ws_app and ws_app.sock and ws_app.sock.connected:
68
            try:
69
                ws_app.send(json.dumps({"type": "CloseStream"}))
70
                time.sleep(0.5)  # Allow time for final metadata to come through
71
            except Exception as e:
72
                print("Error sending CloseStream:", e)
73
            ws_app.close()
74
75
        if ws_thread and ws_thread.is_alive():
76
            ws_thread.join(timeout=2.0)
77
78
signal.signal(signal.SIGINT, graceful_shutdown)
79
80
def run():
81
    global audio, stream, ws_app, ws_thread
82
83
    # 1. open microphone
84
    audio = pyaudio.PyAudio()
85
    try:
86
        stream = audio.open(format=FORMAT, channels=CHANNELS,
87
                            rate=SAMPLE_RATE, input=True,
88
                            frames_per_buffer=FRAMES)
89
        print("Microphone stream opened. Press Ctrl+C to stop.")
90
    except Exception as e:
91
        print(f"Error opening mic: {e}")
92
        audio.terminate()
93
        return
94
95
    # 2. create WebSocket
96
    ws_app = websocket.WebSocketApp(
97
        EP,
98
        header={"Authorization": f"Token {DG_KEY}"},
99
        on_open=on_open,
100
        on_message=on_message,
101
        on_error=on_error,
102
        on_close=on_close,
103
    )
104
105
    # 3. start WS thread (only once!)
106
    ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
107
    ws_thread.start()
108
109
    # 4. block main thread until WS thread ends
110
    ws_thread.join()
111
112
    # 5. cleanup
113
    if stream and stream.is_active():
114
        stream.stop_stream()
115
    if stream:
116
        stream.close()
117
    if audio:
118
        audio.terminate()
119
    print("Cleanup complete. Exiting.")
120
121
if __name__ == "__main__":
122
    run()

1
import json, threading, time, websocket, pyaudio, signal
2
from urllib.parse import urlencode
3
4
DG_KEY = "YOUR_DG_API_KEY"  

1
PARAMS = {
2
    "model": "nova-3",
3
    "encoding": "linear16",
4
    "sample_rate": "16000",
5
    "channels": "1",
6
    "punctuate": "true",
7
    "interim_results": "true",
8
}
9
10
EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"
11
12
FRAMES = 800          
13
CHANNELS = 1
14
FORMAT = pyaudio.paInt16
15
SAMPLE_RATE = int(PARAMS["sample_rate"])
16
17
audio = stream = ws_app = None
18
audio_thread = ws_thread = None
19
stop_event = threading.Event()

1
def on_open(ws):
2
    def mic_loop():
3
        while not stop_event.is_set():
4
            ws.send(stream.read(FRAMES, exception_on_overflow=False),
5
                    websocket.ABNF.OPCODE_BINARY)
6
    global audio_thread
7
    audio_thread = threading.Thread(target=mic_loop, daemon=True)
8
    audio_thread.start()

1
def on_message(ws, msg):
2
    global final_metadata
3
    d = json.loads(msg)
4
    if d.get("type") == "Results":
5
        txt = d["channel"]["alternatives"][0]["transcript"]
6
        if d["is_final"]:
7
            print(" " * 80, end="\r")
8
            print(txt)
9
        else:
10
            print(txt, end="\r")
11
    elif d.get("type") == "Metadata":
12
        final_metadata = d  # save the metadata for later

1
def on_error(ws, error):
2
    print(f"\nWebSocket error: {error}")
3
    stop_event.set()

1
def on_close(ws, *args):
2
    stop_event.set()
3
    if final_metadata:
4
        print("\nFinal Metadata from Deepgram:")
5
        print(json.dumps(final_metadata, indent=2))
6
    else:
7
        print("\nNo metadata received before close.")

1
global audio, stream, ws_app, ws_thread
2
3
# 1. open microphone
4
audio = pyaudio.PyAudio()
5
try:
6
    stream = audio.open(format=FORMAT, channels=CHANNELS,
7
                        rate=SAMPLE_RATE, input=True,
8
                        frames_per_buffer=FRAMES)
9
    print("Microphone stream opened. Press Ctrl+C to stop.")
10
except Exception as e:
11
    print(f"Error opening mic: {e}")
12
    audio.terminate()
13
    return
14
15
# 2. create WebSocket
16
ws_app = websocket.WebSocketApp(
17
    EP,
18
    header={"Authorization": f"Token {DG_KEY}"},
19
    on_open=on_open,
20
    on_message=on_message,
21
    on_error=on_error,
22
    on_close=on_close,
23
)
24
25
ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
26
ws_thread.start()
27
ws_thread.join()

1
def graceful_shutdown(signum, frame):
2
    print("\nCtrl+C received → shutting down …")
3
    stop_event.set()
4
5
    # Send CloseStream before closing WebSocket
6
    if ws_app and ws_app.sock and ws_app.sock.connected:
7
        try:
8
            ws_app.send(json.dumps({"type": "CloseStream"}))
9
            time.sleep(0.5)  # Allow time for final metadata to come through
10
        except Exception as e:
11
            print("Error sending CloseStream:", e)
12
        ws_app.close()
13
14
    if ws_thread and ws_thread.is_alive():
15
        ws_thread.join(timeout=2.0)
16
17
signal.signal(signal.SIGINT, graceful_shutdown)
18
19
def run():
20
    # Open microphone and create WebSocket
21
22
    # cleanup
23
    if stream and stream.is_active():
24
        stream.stop_stream()
25
    if stream:
26
        stream.close()
27
    if audio:
28
        audio.terminate()
29
    print("Cleanup complete. Exiting.")