Self-Hosted Streaming

1
"""
2
Example script for streaming audio to AssemblyAI's self-hosted streaming transcription API.
3
This is a minimal reference implementation for demonstration purposes only.
4
For production use cases, best practices, and the complete API specification, please visit https://www.assemblyai.com/docs
5
"""
6
7
import argparse
8
import json
9
import logging
10
import math
11
import os
12
import time
13
import wave
14
from concurrent.futures import ThreadPoolExecutor
15
from dataclasses import dataclass
16
from datetime import datetime, timedelta
17
from typing import List, Optional
18
from urllib.parse import urlencode
19
20
from websockets.sync.client import ClientConnection, connect
21
22
LOGGER = logging.getLogger(__name__)
23
24
25
@dataclass(frozen=True)
26
class AudioChunk:
27
    data: bytes
28
    duration_ms: int
29
30
31
def _validate_and_get_pcm16_raw_bytes(
32
    wav_file_path: str, expected_sample_rate: int
33
) -> bytes:
34
    """
35
    Validate that the WAV file is PCM16 encoded with the expected sample rate and extract raw audio data.
36
37
    :param wav_file_path: Path to the WAV file.
38
    :param expected_sample_rate: Expected sample rate (e.g., 16000).
39
    :return: Raw audio content as bytes.
40
    :raises ValueError: If the file is not PCM16 or doesn't match expected sample rate.
41
    """
42
    with wave.open(wav_file_path, "rb") as wav_file:
43
        # Check if it's PCM16
44
        if wav_file.getsampwidth() != 2:
45
            raise ValueError(
46
                f"Audio file must be 16-bit PCM. Found sample width: {wav_file.getsampwidth() * 8}-bit"
47
            )
48
49
        if wav_file.getcomptype() != "NONE":
50
            raise ValueError(
51
                f"Audio file must be uncompressed PCM. Found compression type: {wav_file.getcomptype()}"
52
            )
53
54
        # Check sample rate
55
        actual_sample_rate = wav_file.getframerate()
56
        if actual_sample_rate != expected_sample_rate:
57
            raise ValueError(
58
                f"Audio file must have sample rate of {expected_sample_rate} Hz. "
59
                f"Found: {actual_sample_rate} Hz"
60
            )
61
62
        # Check if mono
63
        if wav_file.getnchannels() != 1:
64
            raise ValueError(
65
                f"Audio file must be mono (1 channel). Found: {wav_file.getnchannels()} channels"
66
            )
67
68
        raw_audio = wav_file.readframes(wav_file.getnframes())
69
70
    return raw_audio
71
72
73
def _get_chunks_from_file(
74
    filepath: str,
75
    sample_rate: int,
76
    chunk_size_ms: int,
77
) -> List[AudioChunk]:
78
    """
79
    Read a PCM16 WAV file and split it into chunks.
80
81
    :param filepath: Path to the PCM16 WAV file.
82
    :param sample_rate: Expected sample rate of the audio file.
83
    :param chunk_size_ms: Duration of each chunk in milliseconds.
84
    :return: List of AudioChunk objects.
85
    :raises ValueError: If the file is not in the correct format.
86
    """
87
    chunks = []
88
    audio_bytes: bytes = _validate_and_get_pcm16_raw_bytes(filepath, sample_rate)
89
90
    read_bytes = 0
91
    while read_bytes < len(audio_bytes):
92
        frame_size = 2  # 16-bit PCM (2 bytes per sample)
93
        chunk_bytes_len = int(sample_rate * chunk_size_ms * frame_size // 1000)
94
        data = audio_bytes[read_bytes : read_bytes + chunk_bytes_len]
95
        read_bytes += len(data)
96
        actual_chunk_ms = math.ceil(len(data) * 1000 / (sample_rate * frame_size))
97
        chunks.append(AudioChunk(data=data, duration_ms=actual_chunk_ms))
98
99
    return chunks
100
101
102
def _write_to_ws(ws: ClientConnection, audio_chunks: List[AudioChunk]) -> None:
103
    """
104
    Write audio chunks to the WebSocket connection.
105
106
    :param ws: WebSocket connection.
107
    :param audio_chunks: List of audio chunks to send.
108
    """
109
    try:
110
        for chunk in audio_chunks:
111
            # Sleep for the chunk duration to send chunks with realtime rate
112
            time.sleep(chunk.duration_ms / 1000)
113
            ws.send(chunk.data)
114
        ws.send('{"type": "Terminate"}')
115
    except Exception as e:
116
        LOGGER.error(
117
            f"Exception occurred while writing to websocket: {e}", exc_info=True
118
        )
119
        ws.close()
120
        raise
121
122
123
def _read_from_ws(ws: ClientConnection) -> None:
124
    """
125
    Read and process messages from the WebSocket connection.
126
127
    :param ws: WebSocket connection.
128
    """
129
    try:
130
        for message in ws:
131
            data = json.loads(message)
132
            if "type" not in data:
133
                raise Exception(f"Unknown message received: {data}")
134
            elif data["type"] == "Turn":
135
                if data["words"]:
136
                    text = " ".join([word["text"] for word in data["words"]])
137
                    audio_start = data["words"][0]["start"]
138
                    audio_end = data["words"][-1]["end"]
139
                    end_of_turn = "True " if data["end_of_turn"] else "False"
140
                    LOGGER.info(
141
                        f"{timedelta(milliseconds=audio_start)}-"
142
                        f"{timedelta(milliseconds=audio_end)}, end-of-turn: {end_of_turn}: {text}",
143
                    )
144
            elif data["type"] == "Begin":
145
                expires_at = datetime.fromtimestamp(int(data["expires_at"]))
146
                LOGGER.info(
147
                    f"Session started. Session id: {data['id']}, expires at: {expires_at}",
148
                )
149
            elif data["type"] == "Termination":
150
                LOGGER.info(
151
                    f"Session completed with session duration: {data['session_duration_seconds']} sec.",
152
                )
153
            else:
154
                LOGGER.error(f"Unknown message type: {data}")
155
    except Exception as e:
156
        LOGGER.error(
157
            f"Exception occurred while reading from the websocket: {e}", exc_info=True
158
        )
159
        ws.close()
160
        raise
161
162
163
def run_session(
164
    api_endpoint: str,
165
    audio_chunks: List[AudioChunk],
166
    sample_rate: int,
167
    keyterms_prompt: Optional[List[str]] = None,
168
    language: Optional[str] = None,
169
) -> None:
170
    """
171
    Run a WebSocket session to stream audio and receive transcriptions.
172
173
    :param api_endpoint: WebSocket endpoint URL.
174
    :param audio_chunks: List of audio chunks to send.
175
    :param sample_rate: Sample rate of the audio.
176
    :param keyterms_prompt: Optional list of key terms for the transcription.
177
    :param language: Optional language code for transcription.
178
    """
179
    try:
180
        params = {
181
            "sample_rate": sample_rate,
182
        }
183
        if keyterms_prompt:
184
            params["keyterms"] = json.dumps(keyterms_prompt)
185
        if language:
186
            params["language"] = language
187
188
        endpoint_str = f"{api_endpoint}?{urlencode(params)}"
189
        headers = {"Authorization": "self-hosted"}
190
        LOGGER.info(f"Endpoint: {endpoint_str}")
191
        with ThreadPoolExecutor(max_workers=2) as executor:
192
            with connect(endpoint_str, additional_headers=headers) as websocket:
193
                write_future = executor.submit(
194
                    _write_to_ws,
195
                    websocket,
196
                    audio_chunks,
197
                )
198
                read_future = executor.submit(
199
                    _read_from_ws,
200
                    websocket,
201
                )
202
                write_future.result()
203
                read_future.result()
204
    except Exception as e:
205
        LOGGER.error(
206
            f"Exception occurred: {e}",
207
            exc_info=True,
208
        )
209
        raise
210
211
212
def parse_args():
213
    """Parse command line arguments."""
214
    parser = argparse.ArgumentParser(
215
        description="Stream audio to AssemblyAI self-hosted real-time transcription service",
216
        formatter_class=argparse.RawDescriptionHelpFormatter,
217
        epilog="""
218
Examples:
219
  # Basic usage with default endpoint
220
  python example_with_prerecorded_audio_file.py --audio-file example_audio_file.wav
221
222
  # Specify custom endpoint and language
223
  python example_with_prerecorded_audio_file.py --audio-file example_audio_file.wav --endpoint ws://localhost:8080 --language multi
224
225
Note: Audio file must be PCM 16-bit WAV format, mono channel, 16kHz sample rate.
226
        """,
227
    )
228
    parser.add_argument(
229
        "--audio-file",
230
        type=str,
231
        default=os.path.dirname(__file__) + os.path.sep + "example_audio_file.wav",
232
        help="Path to the audio file to transcribe (must be PCM 16-bit WAV, mono, 16kHz)",
233
    )
234
    parser.add_argument(
235
        "--endpoint",
236
        type=str,
237
        default="ws://localhost:8080",
238
        help="WebSocket endpoint URL (default: ws://localhost:8080)",
239
    )
240
    parser.add_argument(
241
        "--language",
242
        type=str,
243
        default="",
244
        help="Language code for transcription (e.g., 'multi')",
245
    )
246
    return parser.parse_args()
247
248
249
if __name__ == "__main__":
250
    try:
251
        args = parse_args()
252
        logging.basicConfig(level=logging.INFO, format="%(message)s")
253
        sample_rate = 16_000
254
255
        audio_chunks = _get_chunks_from_file(
256
            args.audio_file,
257
            sample_rate=sample_rate,
258
            chunk_size_ms=100,
259
        )
260
        run_session(
261
            api_endpoint=args.endpoint,
262
            audio_chunks=audio_chunks,
263
            sample_rate=sample_rate,
264
            language=args.language if args.language else None,
265
        )
266
    except KeyboardInterrupt:
267
        LOGGER.info("Interrupted by user, exiting.")
268
        exit(0)
269
    except ValueError as e:
270
        LOGGER.error(f"Audio file validation error: {e}")
271
        exit(1)

1
import asyncio
2
import websockets
3
import pyaudio
4
import json
5
6
# Replace with your server's IP address or use 'localhost' for local testing
7
SERVER_IP = "your.server.ip.address"
8
9
async def stream_audio(language="en"):
10
    # Build WebSocket URL with query parameters
11
    params = f"sample_rate=16000&language={language}"
12
    WS_URL = f"ws://{SERVER_IP}:8080/v3/ws?{params}"
13
    
14
    # Add authorization header (required for self-hosted)
15
    headers = {"Authorization": "self-hosted"}
16
    
17
    print(f"Connecting to {WS_URL}...")
18
    
19
    async with websockets.connect(WS_URL, extra_headers=headers) as ws:
20
        print("Connected! Starting to stream audio...")
21
        
22
        # Set up audio stream from microphone
23
        p = pyaudio.PyAudio()
24
        stream = p.open(
25
            format=pyaudio.paInt16,
26
            channels=1,
27
            rate=16000,
28
            input=True,
29
            frames_per_buffer=3200  # 100ms chunks at 16kHz
30
        )
31
        
32
        print(f"\n🎤 Listening with language={language}... speak into your microphone!")
33
        print("Press Ctrl+C to stop\n")
34
        
35
        # Function to send audio
36
        async def send_audio():
37
            try:
38
                while True:
39
                    data = stream.read(3200, exception_on_overflow=False)
40
                    await ws.send(data)
41
                    await asyncio.sleep(0.1)  # 100ms chunks
42
            except KeyboardInterrupt:
43
                await ws.send(json.dumps({"type": "Terminate"}))
44
                print("\nStopping...")
45
            finally:
46
                stream.stop_stream()
47
                stream.close()
48
                p.terminate()
49
        
50
        # Function to receive transcripts
51
        async def receive_transcripts():
52
            try:
53
                async for message in ws:
54
                    data = json.loads(message)
55
                    
56
                    if data.get("type") == "Begin":
57
                        print(f"✅ Session started! ID: {data.get('id')}")
58
                    
59
                    elif data.get("type") == "Turn":
60
                        if data.get("words"):
61
                            text = " ".join([word["text"] for word in data["words"]])
62
                            end_of_turn = "[FINAL]" if data.get("end_of_turn") else ""
63
                            print(f"📝 {text} {end_of_turn}")
64
                    
65
                    elif data.get("type") == "Termination":
66
                        print(f"✅ Session completed. Duration: {data.get('session_duration_seconds')}s")
67
                        break
68
                    
69
            except Exception as e:
70
                print(f"Error receiving: {e}")
71
        
72
        # Run both tasks concurrently
73
        await asyncio.gather(send_audio(), receive_transcripts())
74
75
if __name__ == "__main__":
76
    import sys
77
    
78
    # Usage: python live_microphone_streaming.py [language]
79
    # Examples:
80
    #   python live_microphone_streaming.py en       # English
81
    #   python live_microphone_streaming.py multi    # Multilingual with auto-detect
82
    #   python live_microphone_streaming.py es       # Spanish
83
    language = sys.argv[1] if len(sys.argv) > 1 else "en"
84
    
85
    try:
86
        asyncio.run(stream_audio(language))
87
    except KeyboardInterrupt:
88
        print("\nStopped by user")