Migration guide: Speechmatics to AssemblyAI

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

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