Streaming

1
import asyncio
2
import json
3
import base64
4
import wave
5
import os
6
import websockets
7
8
API_KEY = os.environ["SMALLEST_API_KEY"]
9
WS_URL = "wss://api.smallest.ai/waves/v1/lightning-v3.1/get_speech/stream"
10
11
async def stream_tts(text):
12
    audio_chunks = []
13
14
    async with websockets.connect(
15
        WS_URL,
16
        extra_headers={"Authorization": f"Bearer {API_KEY}"},
17
    ) as ws:
18
        await ws.send(json.dumps({
19
            "text": text,
20
            "voice_id": "magnus",
21
            "sample_rate": 24000,
22
        }))
23
24
        while True:
25
            response = await ws.recv()
26
            data = json.loads(response)
27
28
            if data["status"] == "chunk":
29
                audio = base64.b64decode(data["data"]["audio"])
30
                audio_chunks.append(audio)
31
            elif data["status"] == "complete":
32
                break
33
34
    # Save as WAV
35
    raw = b"".join(audio_chunks)
36
    with wave.open("streamed.wav", "wb") as wf:
37
        wf.setnchannels(1)
38
        wf.setsampwidth(2)
39
        wf.setframerate(24000)
40
        wf.writeframes(raw)
41
42
    print(f"Saved streamed.wav ({len(audio_chunks)} chunks)")
43
44
asyncio.run(stream_tts("Streaming delivers audio in real-time for voice assistants and chatbots."))

1
import os
2
import json
3
import base64
4
import wave
5
import requests
6
7
API_KEY = os.environ["SMALLEST_API_KEY"]
8
9
response = requests.post(
10
    "https://api.smallest.ai/waves/v1/lightning-v3.1/stream",
11
    headers={
12
        "Authorization": f"Bearer {API_KEY}",
13
        "Content-Type": "application/json",
14
        "Accept": "text/event-stream",
15
    },
16
    json={
17
        "text": "SSE streaming is simpler to set up than WebSocket.",
18
        "voice_id": "magnus",
19
        "sample_rate": 24000,
20
    },
21
    stream=True,
22
)
23
24
audio_chunks = []
25
for line in response.iter_lines():
26
    if not line:
27
        continue
28
    line = line.decode()
29
    if not line.startswith("data: "):
30
        continue
31
32
    data = json.loads(line[6:])
33
    if data["status"] == "chunk":
34
        audio_chunks.append(base64.b64decode(data["data"]["audio"]))
35
    elif data["status"] == "complete":
36
        break
37
38
raw = b"".join(audio_chunks)
39
with wave.open("sse_output.wav", "wb") as wf:
40
    wf.setnchannels(1)
41
    wf.setsampwidth(2)
42
    wf.setframerate(24000)
43
    wf.writeframes(raw)