import asyncio
import websockets
import json
import os
import pathlib
from urllib.parse import urlencode
BASE_WS_URL = "wss://waves-api.smallest.ai/api/v1/lightning/get_text"
params = {
"language": "en",
"encoding": "linear16",
"sample_rate": 16000,
"word_timestamps": "true"
}
WS_URL = f"{BASE_WS_URL}?{urlencode(params)}"
API_KEY = "YOUR_API_KEY"
AUDIO_FILE = "path/to/audio.wav"
async def stream_audio():
headers = {
"Authorization": f"Bearer {API_KEY}"
}
async with websockets.connect(WS_URL, additional_headers=headers) as ws:
print("Connected to ASR WebSocket")
audio_bytes = pathlib.Path(AUDIO_FILE).read_bytes()
chunk_size = 4096
offset = 0
print(f"Streaming {len(audio_bytes)} bytes from {os.path.basename(AUDIO_FILE)}")
async def send_chunks():
nonlocal offset
while offset < len(audio_bytes):
chunk = audio_bytes[offset: offset + chunk_size]
await ws.send(chunk)
offset += chunk_size
await asyncio.sleep(0.05)
print("Finished sending audio, sending end signal...")
await ws.send(json.dumps({"type": "end"}))
sender = asyncio.create_task(send_chunks())
try:
async for message in ws:
try:
data = json.loads(message)
print("Received:", json.dumps(data, indent=2))
except json.JSONDecodeError:
print("Received raw:", message)
except websockets.ConnectionClosed as e:
print(f"Connection closed: {e.code} - {e.reason}")
await sender
if __name__ == "__main__":
asyncio.run(stream_audio())