Migration guide: Speechmatics to AssemblyAI

1
import pyaudio
2
import websocket
3
import json
4
import threading
5
import time
6
7
# --- Configuration ---
8
9
YOUR_API_KEY = "YOUR-API-KEY" # Replace with your actual API key
10
11
CONNECTION_PARAMS = {
12
    "language": "en",
13
    "enable_partials": True,
14
    "max_delay": 2.0
15
}
16
API_ENDPOINT = "wss://eu2.rt.speechmatics.com/v2/en"
17
18
# Audio Configuration
19
20
FRAMES_PER_BUFFER = 1024 # Chunk size
21
SAMPLE_RATE = None # Will be set based on device capabilities
22
CHANNELS = 1
23
FORMAT = pyaudio.paFloat32 # Speechmatics uses float32 format
24
25
# Global variables for audio stream and websocket
26
27
audio = None
28
stream = None
29
ws_app = None
30
audio_thread = None
31
stop_event = threading.Event() # To signal the audio thread to stop
32
audio_seq_no = 0 # Track number of audio chunks sent
33
34
# --- WebSocket Event Handlers ---
35
36
def on_open(ws):
37
    """Called when the WebSocket connection is established."""
38
    print("WebSocket connection opened.")
39
    print(f"Connected to: {API_ENDPOINT}")
40
41
    # Send StartRecognition message
42
    start_message = {
43
        "message": "StartRecognition",
44
        "audio_format": {
45
            "type": "raw",
46
            "encoding": "pcm_f32le",
47
            "sample_rate": SAMPLE_RATE
48
        },
49
        "transcription_config": {
50
            "language": CONNECTION_PARAMS["language"],
51
            "enable_partials": CONNECTION_PARAMS["enable_partials"],
52
            "max_delay": CONNECTION_PARAMS["max_delay"]
53
        }
54
    }
55
    ws.send(json.dumps(start_message))
56
57
def on_message(ws, message):
58
    global audio_seq_no
59
60
    try:
61
        data = json.loads(message)
62
        msg_type = data.get('message')
63
64
        if msg_type == "RecognitionStarted":
65
            session_id = data.get('id')
66
            print(f"\nSession began: ID={session_id}")
67
68
            # Start sending audio data in a separate thread
69
            def stream_audio():
70
                global audio_seq_no, stream
71
                print("Starting audio streaming...")
72
                while not stop_event.is_set():
73
                    try:
74
                        audio_data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
75
                        # Send audio data as binary message
76
                        ws.send(audio_data, websocket.ABNF.OPCODE_BINARY)
77
                        audio_seq_no += 1
78
                    except Exception as e:
79
                        print(f"Error streaming audio: {e}")
80
                        # If stream read fails, likely means it's closed, stop the loop
81
                        break
82
                print("Audio streaming stopped.")
83
84
            global audio_thread
85
            audio_thread = threading.Thread(target=stream_audio)
86
            audio_thread.daemon = (
87
                True  # Allow main thread to exit even if this thread is running
88
            )
89
            audio_thread.start()
90
91
        elif msg_type == "AddPartialTranscript":
92
            transcript = data.get('metadata', {}).get('transcript', '')
93
            if transcript:
94
                print(f"\r{transcript}", end='')
95
96
        elif msg_type == "AddTranscript":
97
            transcript = data.get('metadata', {}).get('transcript', '')
98
            if transcript:
99
                # Clear previous line for final messages
100
                print('\r' + ' ' * 80 + '\r', end='')
101
                print(transcript)
102
103
        elif msg_type == "EndOfTranscript":
104
            print("\nSession Terminated: Transcription complete")
105
106
        elif msg_type == "Error":
107
            error_type = data.get('type')
108
            reason = data.get('reason')
109
            print(f"\nWebSocket Error: {error_type} - {reason}")
110
            stop_event.set()
111
112
    except json.JSONDecodeError as e:
113
        print(f"Error decoding message: {e}")
114
    except Exception as e:
115
        print(f"Error handling message: {e}")
116
117
def on_error(ws, error):
118
    """Called when a WebSocket error occurs."""
119
    print(f"\nWebSocket Error: {error}")  # Attempt to signal stop on error
120
    stop_event.set()
121
122
def on_close(ws, close_status_code, close_msg):
123
    """Called when the WebSocket connection is closed."""
124
    print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
125
    # Ensure audio resources are released
126
    global stream, audio
127
    stop_event.set()  # Signal audio thread just in case it's still running
128
129
    if stream:
130
        if stream.is_active():
131
            stream.stop_stream()
132
        stream.close()
133
        stream = None
134
    if audio:
135
        audio.terminate()
136
        audio = None
137
    # Try to join the audio thread to ensure clean exit
138
    if audio_thread and audio_thread.is_alive():
139
        audio_thread.join(timeout=1.0)
140
141
# --- Main Execution ---
142
143
def run():
144
    global audio, stream, ws_app, SAMPLE_RATE
145
146
    # Initialize PyAudio
147
    audio = pyaudio.PyAudio()
148
149
    # Get default input device (can alter to specify specific device)
150
    default_device = audio.get_default_input_device_info()
151
    device_index = default_device['index']
152
    SAMPLE_RATE = int(audio.get_device_info_by_index(device_index)['defaultSampleRate'])
153
154
    print(f"Using microphone: {default_device['name']}")
155
156
    # Open microphone stream
157
    try:
158
        stream = audio.open(
159
            input=True,
160
            frames_per_buffer=FRAMES_PER_BUFFER,
161
            channels=CHANNELS,
162
            format=FORMAT,
163
            rate=SAMPLE_RATE,
164
            input_device_index=device_index
165
        )
166
        print("Microphone stream opened successfully.")
167
        print("Speak into your microphone. Press Ctrl+C to stop.")
168
    except Exception as e:
169
        print(f"Error opening microphone stream: {e}")
170
        if audio:
171
            audio.terminate()
172
        return  # Exit if microphone cannot be opened
173
174
    # Create WebSocketApp
175
    ws_app = websocket.WebSocketApp(
176
        API_ENDPOINT,
177
        header={"Authorization": f"Bearer {YOUR_API_KEY}"},  # Speechmatics uses Bearer token
178
        on_open=on_open,
179
        on_message=on_message,
180
        on_error=on_error,
181
        on_close=on_close,
182
    )
183
184
    # Run WebSocketApp in a separate thread to allow main thread to catch KeyboardInterrupt
185
    ws_thread = threading.Thread(target=lambda: ws_app.run_forever(ping_interval=30, ping_timeout=10))
186
    ws_thread.daemon = True
187
    ws_thread.start()
188
189
    try:
190
        # Keep main thread alive until interrupted
191
        while ws_thread.is_alive():
192
            time.sleep(0.1)
193
    except KeyboardInterrupt:
194
        print("\nCtrl+C received. Stopping...")
195
        stop_event.set()  # Signal audio thread to stop
196
197
        # Send EndOfStream message to the server
198
        if ws_app and ws_app.sock and ws_app.sock.connected:
199
            try:
200
                end_message = {
201
                    "message": "EndOfStream",
202
                    "last_seq_no": audio_seq_no
203
                }
204
                print(f"Sending termination message: {json.dumps(end_message)}")
205
                ws_app.send(json.dumps(end_message))
206
                # Give a moment for messages to process before forceful close
207
                time.sleep(1)
208
            except Exception as e:
209
                print(f"Error sending termination message: {e}")
210
211
        # Close the WebSocket connection (will trigger on_close)
212
        if ws_app:
213
            ws_app.close()
214
215
        # Wait for WebSocket thread to finish
216
        ws_thread.join(timeout=2.0)
217
218
    except Exception as e:
219
        print(f"\nAn unexpected error occurred: {e}")
220
        stop_event.set()
221
        if ws_app:
222
            ws_app.close()
223
        ws_thread.join(timeout=2.0)
224
225
    finally:
226
        # Final cleanup (already handled in on_close, but good as a fallback)
227
        if stream and stream.is_active():
228
            stream.stop_stream()
229
        if stream:
230
            stream.close()
231
        if audio:
232
            audio.terminate()
233
        print("Cleanup complete. Exiting.")
234
235
if __name__ == "__main__":
236
    run()

1
if __name__ == "__main__":
2
    run()