import asyncio
import json
import argparse
import numpy as np
import websockets
import librosa
from urllib.parse import urlencode
BASE_WS_URL = "wss://api.smallest.ai/waves/v1/lightning/get_text"
async def stream_audio(audio_file, api_key, language="en", encoding="linear16", sample_rate=16000, word_timestamps="true", full_transcript="false", sentence_timestamps="false", redact_pii="false", redact_pci="false", numerals="auto", diarize="false"):
params = {
"language": language,
"encoding": encoding,
"sample_rate": sample_rate,
"word_timestamps": word_timestamps,
"full_transcript": full_transcript,
"sentence_timestamps": sentence_timestamps,
"redact_pii": redact_pii,
"redact_pci": redact_pci,
"numerals": numerals,
"diarize": diarize
}
ws_url = f"{BASE_WS_URL}?{urlencode(params)}"
async with websockets.connect(ws_url, additional_headers={"Authorization": f"Bearer {api_key}"}) as ws:
print(f"Connected: {ws_url}")
async def send():
audio, _ = librosa.load(audio_file, sr=sample_rate, mono=True)
chunk_size = int(0.160 * sample_rate)
for i in range(0, len(audio), chunk_size):
chunk = audio[i:i + chunk_size]
await ws.send((chunk * 32768.0).astype(np.int16).tobytes())
await asyncio.sleep(len(chunk) / sample_rate)
await ws.send(json.dumps({"type": "finalize"}))
sender = asyncio.create_task(send())
async for message in ws:
data = json.loads(message)
print("Received:", json.dumps(data, indent=2))
if data.get("is_last"):
break
await sender
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("audio_file", nargs="?", default="path/to/audio.wav")
parser.add_argument("--api-key", "-k", default="your_api_key_here")
parser.add_argument("--language", "-l", default="en")
parser.add_argument("--encoding", "-e", default="linear16")
parser.add_argument("--sample-rate", "-sr", type=int, default=16000)
parser.add_argument("--word-timestamps", "-wt", default="true")
parser.add_argument("--full-transcript", "-ft", default="false")
parser.add_argument("--sentence-timestamps", "-st", default="false")
parser.add_argument("--redact-pii", default="false")
parser.add_argument("--redact-pci", default="false")
parser.add_argument("--numerals", default="auto")
parser.add_argument("--diarize", default="false")
args = parser.parse_args()
asyncio.run(stream_audio(
args.audio_file,
args.api_key,
args.language,
args.encoding,
args.sample_rate,
args.word_timestamps,
args.full_transcript,
args.sentence_timestamps,
args.redact_pii,
args.redact_pci,
args.numerals,
args.diarize
))
