Limits

Generate text is 2001 characters, over the 2000-character limit. Split it across multiple Generate messages (and Flush at sentence boundaries) instead of sending it all at once.

import asyncio
import base64
import json
import os
import re
import struct

import websockets

URL = "wss://tts.us.assemblyai.com/v1/ws?voice_id=49c443a4-8a1d-5f7e-b70b-98bdbd9823c3&language=english"
HEADERS = {"Authorization": os.environ["ASSEMBLYAI_API_KEY"]}

# Flushes sent and not yet answered with FlushDone. Two or three keeps the
# server busy without queuing a deep backlog behind the request being spoken.
MAX_IN_FLIGHT = 3

TEXT = """
The lighthouse at the end of the point was built in 1874, and for a hundred
years a keeper lived in the small stone house beside it. Every evening he
climbed the spiral stairs, trimmed the wick, and wound the clockwork that
turned the lens. On clear nights the beam could be seen from twenty miles out
at sea, and fishing crews set their course by it on the way home.

The last keeper retired in 1978, when the light was automated. The clockwork
was replaced by a motor, the oil lamp by a bulb, and the logbook by a sensor
that reports to the coast guard twice a day. The house stood empty for a
decade before the town bought it and turned it into a small museum.

Visitors today can climb the same stairs and stand beside the lens. The guides
like to point out that the glass has never been replaced. It was cut in Paris,
shipped around the cape, and carried up the cliff by hand, and it still throws
the same beam it did on the first night. Is it worth the climb? Ask anyone who
has been up there at dusk.
"""


def split_sentences(text):
    """Minimal splitter: break after . ! or ? followed by whitespace.

    Each sentence gets its own Flush below, so the whitespace this drops is
    not needed as a separator. A sentence over 2000 characters would still
    need splitting; this passage has none.
    """
    return [s for s in re.split(r"(?<=[.!?])\s+", text.strip()) if s]


def wav_header(n_bytes, rate, channels=1, bits=16):
    byte_rate = rate * channels * bits // 8
    return (
        b"RIFF" + struct.pack("<I", 36 + n_bytes) + b"WAVEfmt "
        + struct.pack("<IHHIIHH", 16, 1, channels, rate, byte_rate,
                      channels * bits // 8, bits)
        + b"data" + struct.pack("<I", n_bytes)
    )


async def read_frame(ws, pcm):
    """Read until a frame that is not Audio. Audio is appended to pcm and an
    Error ends the program; anything else is returned to the caller."""
    while True:
        message = json.loads(await ws.recv())
        kind = message.get("type")
        if kind == "Audio":
            pcm += base64.b64decode(message["audio"])
            continue
        if kind == "Error":
            raise SystemExit(f"{message['error_code']}: {message['error']}")
        if kind == "FlushDone":
            print(f"flush {message['flush_id']}: "
                  f"{message['audio_duration_ms']} ms delivered")
        # FlushDone, WordBoundaries, Termination, or a type you do not handle.
        return message


async def speak(ws, text, pcm):
    """Send text one sentence per request, topping up on FlushDone."""
    in_flight = 0
    for sentence in split_sentences(text):
        while in_flight >= MAX_IN_FLIGHT:
            if (await read_frame(ws, pcm))["type"] == "FlushDone":
                in_flight -= 1
        await ws.send(json.dumps({"type": "Generate", "text": sentence}))
        await ws.send(json.dumps({"type": "Flush"}))
        in_flight += 1
    # The last flushes are still in flight here. Terminate drains them.


async def main():
    pcm = bytearray()
    async with websockets.connect(URL, additional_headers=HEADERS) as ws:
        first = json.loads(await ws.recv())
        if first["type"] == "Error":
            raise SystemExit(f"{first['error_code']}: {first['error']}")
        rate = first["configuration"]["sample_rate"]

        await speak(ws, TEXT, pcm)

        await ws.send(json.dumps({"type": "Terminate"}))
        while True:
            message = await read_frame(ws, pcm)
            if message["type"] == "Termination":
                break

    with open("out.wav", "wb") as f:
        f.write(wav_header(len(pcm), rate))
        f.write(pcm)
    print(f"wrote out.wav: {message['audio_duration_seconds']} s of audio "
          f"from {message['total_input_char_length']} characters")


asyncio.run(main())

flush 0: 7680 ms delivered
flush 1: 5760 ms delivered
flush 2: 6880 ms delivered
flush 3: 4560 ms delivered
flush 4: 7920 ms delivered
flush 5: 5280 ms delivered
flush 6: 4400 ms delivered
flush 7: 3920 ms delivered
flush 8: 7440 ms delivered
flush 9: 1520 ms delivered
flush 10: 2400 ms delivered
wrote out.wav: 57.76 s of audio from 995 characters

{"type": "Error", "error_code": 3010, "error": "Input text rate exceeded: 8541 characters sent but not yet synthesized (limit 8000). The sustained budget is 100 chars/sec; send text at the pace it is spoken.", "error_type": "input_rate_exceeded", "retryable": false, "param": null, "retry_after_seconds": null}

{"type": "Error", "error_code": 3010, "error": "Control frame rate exceeded: more than 20 Flush/KeepAlive/Cancel frames or empty Generates per second sustained. Send control frames only as the conversation needs them.", "error_type": "input_rate_exceeded", "retryable": false, "param": null, "retry_after_seconds": null}

Session reached its maximum duration; open a new connection to continue