Webhooks for streaming speech-to-text

1
import pyaudio
2
import websocket
3
import json
4
import threading
5
import time
6
from urllib.parse import urlencode
7
from datetime import datetime
8
9
# --- Configuration ---
10
YOUR_API_KEY = "<YOUR_API_KEY>"
11
12
CONNECTION_PARAMS = {
13
    "sample_rate": 16000,
14
    "format_turns": True,
15
    # Webhook parameters
16
    "webhook_url": "https://example.com/webhook",
17
    "webhook_auth_header_name": "X-Webhook-Secret",  # Optional
18
    "webhook_auth_header_value": "secret-value",  # Optional
19
}
20
API_ENDPOINT_BASE_URL = "wss://streaming.assemblyai.com/v3/ws"
21
API_ENDPOINT = f"{API_ENDPOINT_BASE_URL}?{urlencode(CONNECTION_PARAMS)}"
22
23
# Audio Configuration
24
FRAMES_PER_BUFFER = 800  # 50ms of audio (0.05s * 16000Hz)
25
SAMPLE_RATE = CONNECTION_PARAMS["sample_rate"]
26
CHANNELS = 1
27
FORMAT = pyaudio.paInt16
28
29
# Global variables
30
audio = None
31
stream = None
32
ws_app = None
33
audio_thread = None
34
stop_event = threading.Event()
35
36
37
def on_open(ws):
38
    """Called when the WebSocket connection is established."""
39
    print("WebSocket connection opened.")
40
    print(f"Connected to: {API_ENDPOINT}")
41
42
    def stream_audio():
43
        global stream
44
        print("Starting audio streaming...")
45
        while not stop_event.is_set():
46
            try:
47
                audio_data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
48
                ws.send(audio_data, websocket.ABNF.OPCODE_BINARY)
49
            except Exception as e:
50
                print(f"Error streaming audio: {e}")
51
                break
52
        print("Audio streaming stopped.")
53
54
    global audio_thread
55
    audio_thread = threading.Thread(target=stream_audio)
56
    audio_thread.daemon = True
57
    audio_thread.start()
58
59
60
def on_message(ws, message):
61
    """Called when a message is received from the WebSocket."""
62
    try:
63
        data = json.loads(message)
64
        msg_type = data.get("type")
65
66
        if msg_type == "Begin":
67
            session_id = data.get("id")
68
            expires_at = data.get("expires_at")
69
            print(f"\nSession began: ID={session_id}, ExpiresAt={datetime.fromtimestamp(expires_at)}")
70
        elif msg_type == "Turn":
71
            transcript = data.get("transcript", "")
72
            if data.get('end_of_turn'):
73
                print("\r" + " " * 80 + "\r", end="")
74
                print(transcript)
75
            else:
76
                print(f"\r{transcript}", end="")
77
        elif msg_type == "Termination":
78
            audio_duration = data.get("audio_duration_seconds", 0)
79
            session_duration = data.get("session_duration_seconds", 0)
80
            print(f"\nSession Terminated: Audio Duration={audio_duration}s, Session Duration={session_duration}s")
81
    except json.JSONDecodeError as e:
82
        print(f"Error decoding message: {e}")
83
    except Exception as e:
84
        print(f"Error handling message: {e}")
85
86
87
def on_error(ws, error):
88
    """Called when a WebSocket error occurs."""
89
    print(f"\nWebSocket Error: {error}")
90
    stop_event.set()
91
92
93
def on_close(ws, close_status_code, close_msg):
94
    """Called when the WebSocket connection is closed."""
95
    print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
96
    global stream, audio
97
    stop_event.set()
98
99
    if stream:
100
        if stream.is_active():
101
            stream.stop_stream()
102
        stream.close()
103
        stream = None
104
    if audio:
105
        audio.terminate()
106
        audio = None
107
    if audio_thread and audio_thread.is_alive():
108
        audio_thread.join(timeout=1.0)
109
110
111
def run():
112
    global audio, stream, ws_app
113
114
    audio = pyaudio.PyAudio()
115
116
    try:
117
        stream = audio.open(
118
            input=True,
119
            frames_per_buffer=FRAMES_PER_BUFFER,
120
            channels=CHANNELS,
121
            format=FORMAT,
122
            rate=SAMPLE_RATE,
123
        )
124
        print("Microphone stream opened successfully.")
125
        print("Speak into your microphone. Press Ctrl+C to stop.")
126
    except Exception as e:
127
        print(f"Error opening microphone stream: {e}")
128
        if audio:
129
            audio.terminate()
130
        return
131
132
    ws_app = websocket.WebSocketApp(
133
        API_ENDPOINT,
134
        header={"Authorization": YOUR_API_KEY},
135
        on_open=on_open,
136
        on_message=on_message,
137
        on_error=on_error,
138
        on_close=on_close,
139
    )
140
141
    ws_thread = threading.Thread(target=ws_app.run_forever)
142
    ws_thread.daemon = True
143
    ws_thread.start()
144
145
    try:
146
        while ws_thread.is_alive():
147
            time.sleep(0.1)
148
    except KeyboardInterrupt:
149
        print("\nCtrl+C received. Stopping...")
150
        stop_event.set()
151
152
        if ws_app and ws_app.sock and ws_app.sock.connected:
153
            try:
154
                terminate_message = {"type": "Terminate"}
155
                ws_app.send(json.dumps(terminate_message))
156
                time.sleep(2)
157
            except Exception as e:
158
                print(f"Error sending termination message: {e}")
159
160
        if ws_app:
161
            ws_app.close()
162
        ws_thread.join(timeout=2.0)
163
164
    finally:
165
        if stream and stream.is_active():
166
            stream.stop_stream()
167
        if stream:
168
            stream.close()
169
        if audio:
170
            audio.terminate()
171
        print("Cleanup complete.")
172
173
174
if __name__ == "__main__":
175
    run()