Processing files

The Speech-to-Speech API is a real-time service: audio is consumed at the speed of speech, and you cannot push a file faster than real time (audio messages are rate-limited to ~25 packets per second, and the pipeline itself — voice activity detection, segment confirmation, endpointing — is driven by the audio timeline). Translating a file therefore takes about as long as the file's duration.

Even so, feeding pre-recorded files through the API is very useful:

• Accuracy benchmarks — run a reference corpus through the pipeline and measure WER / translation quality of different asr_model / translation_model values on your domain.
• Regression testing — verify that a settings change (glossaries, splitter options, silence thresholds) still produces the transcripts and translations you expect.
• Reproducible debugging — replay the exact same audio while investigating an issue.

The flow

1. Connect to the WebSocket endpoint and send set_task (see Translation management API). Before sending audio, wait for a task_status response with data.command: "set_task", data.event: "started", and data.task_status: "running".
2. Send the file's PCM in 320 ms chunks paced to real time via input_audio_data.
3. After the last byte is sent, send end_task. No new audio is accepted from this point on — that's why you send it only once the whole file has been streamed.
4. The server finishes processing everything it received (the phrase in progress is finalized even if the file ends abruptly, with no trailing silence), delivers the final validated_transcription / translated_transcription messages and the trailing TTS audio, then sends an end_of_stream message and closes the connection automatically.

Don't skip step 3

It can lead to the last segment missing in the translation.

Quick way: translate_file from the Python client

The palabra-ai client does all of the above for you — real-time pacing, end_task at the end, and collecting the results:

pip install palabra-ai

from palabra_ai import Palabra

palabra = Palabra()  # set PALABRA_API_KEY via ENV, or pass api_key=

results = palabra.translate_file(
    "speech_es.wav",
    source="es",
    targets="en",
    output="speech_en.wav",     # translated audio, one WAV per target
    on_transcript=print,        # partial/validated transcripts and translations
)

translate_file returns {language: pcm_s16le_24k_mono} and takes about as long as the audio itself. To collect transcripts for a benchmark, gather them in on_transcript (each Transcript has .text, .language and .is_validated).

Full example: raw WebSocket with send_audio

The same flow without the client library — useful if you need full control over the task settings or want to log every message for scoring:

import asyncio
import base64
import json
import wave

import websockets

WS_URL = "wss://<STREAMING_SERVER>.palabra.ai/streaming-api/v1/speech-to-speech/stream"
TOKEN = "<publisher JWT from the session>"  # see Session Management

RATE = 16000
CHUNK = int(RATE * 0.32) * 2  # 320 ms of pcm_s16le mono

TASK = {
    "input_stream": {
        "content_type": "audio",
        "source": {"type": "ws", "format": "pcm_s16le", "sample_rate": RATE, "channels": 1},
    },
    "output_stream": {
        "content_type": "audio",
        "target": {"type": "ws", "format": "pcm_s16le"},
    },
    "pipeline": {
        "transcription": {"source_language": "es", "asr_model": "auto"},
        "translations": [{"target_language": "en", "speech_generation": {}}],
        "allowed_message_types": [
            "partial_transcription",
            "validated_transcription",
            "translated_transcription",
        ],
    },
}


async def send(ws, message_type, data):
    await ws.send(json.dumps({"message_type": message_type, "data": data}))


async def send_audio(ws, chunk: bytes):
    await send(ws, "input_audio_data", {"data": base64.b64encode(chunk).decode()})


async def main():
    with wave.open("speech_es.wav") as w:  # mono 16 kHz s16le, matching TASK
        pcm = w.readframes(w.getnframes())

    validated, translated = [], []
    loop = asyncio.get_running_loop()
    ready = loop.create_future()

    async with websockets.connect(f"{WS_URL}?token={TOKEN}", max_size=None) as ws:

        async def reader():
            saw_eos = False
            try:
                async for raw in ws:
                    msg = json.loads(raw)
                    data = msg.get("data") or {}
                    if isinstance(data, str):
                        data = json.loads(data)
                    mt = msg.get("message_type")
                    if mt == "task_status":
                        if (data.get("command") == "set_task"
                                and data.get("event") in ("started", "updated", "unchanged", "resumed")
                                and data.get("task_status") == "running"
                                and not ready.done()):
                            ready.set_result(data)
                    elif mt == "error":
                        raise RuntimeError(f"{data.get('code')}: {data.get('desc')}")
                    elif mt == "validated_transcription":
                        validated.append(data["transcription"]["text"])
                    elif mt == "translated_transcription":
                        translated.append(data["transcription"]["text"])
                    elif mt == "end_of_stream":
                        saw_eos = True
                if not saw_eos:
                    raise RuntimeError("Connection closed before end_of_stream")
            except Exception as exc:
                if not ready.done():
                    ready.set_exception(exc)
                raise

        reader_task = asyncio.create_task(reader())
        try:
            await send(ws, "set_task", TASK)
            await asyncio.wait_for(ready, timeout=30)

            # Stream at real-time speed: one 320 ms chunk every 320 ms.
            start = loop.time()
            for i in range(0, len(pcm), CHUNK):
                if reader_task.done():
                    await reader_task  # surface server errors immediately
                await send_audio(ws, pcm[i : i + CHUNK])
                await asyncio.sleep(max(0.0, start + (i // CHUNK + 1) * 0.32 - loop.time()))

            await send(ws, "end_task", {})
            # Allow time to finalize recognition and send pending output (up to 120 seconds).
            await asyncio.wait_for(reader_task, timeout=155)
        finally:
            reader_task.cancel()
            await asyncio.gather(reader_task, return_exceptions=True)

    print("TRANSCRIPT:", " ".join(validated))
    print("TRANSLATION:", " ".join(translated))

asyncio.run(main())

Notes:

• Send end_task only after the last byte of the file — the server stops accepting audio as soon as it receives the command.
• The final validated_transcription for an abruptly-cut file typically arrives within a second or two after end_task — always wait until end_of_stream before scoring the run.
• Output TTS audio arrives as output_audio_data messages; ignore them (or drop output_stream/speech_generation from the task) if you only need text for the benchmark.