import asyncio
import base64
import json
import wave
import websockets
API_KEY = "YOUR_API_KEY"
WS_URL = "wss://api.kugelaudio.com"
async def generate_speech(text: str, voice_id: int = 268):
"""Generate speech via WebSocket and save to WAV file."""
ws_url = f"{WS_URL}/ws/tts?api_key={API_KEY}"
audio_chunks = []
async with websockets.connect(ws_url) as ws:
# Send TTS request
await ws.send(json.dumps({
"text": text,
"model_id": "kugel-1-turbo",
"voice_id": voice_id,
"cfg_scale": 2.0,
"sample_rate": 24000,
}))
# Receive audio chunks
async for msg in ws:
data = json.loads(msg)
if data.get("error"):
raise Exception(data["error"])
if data.get("audio"):
audio_chunks.append(base64.b64decode(data["audio"]))
print(f"Chunk {data['idx']}: {data['samples']} samples")
if data.get("final"):
print(f"Done: {data['dur_ms']:.0f}ms audio, generated in {data['gen_ms']:.0f}ms")
break
# Save to WAV
with wave.open("output.wav", "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 16-bit
wf.setframerate(24000)
wf.writeframes(b"".join(audio_chunks))
print("Saved to output.wav")
# Run
asyncio.run(generate_speech("Hello, this is a test of the raw API."))
