Realtime streaming

Stream PCM audio and receive partial, final, VAD, timestamp, and translation events.

Streaming uses a dedicated client and event handlers. The async client is recommended for servers already using an event loop.

Transcribe a stream

import os
import asyncio
from orbitalsai.streaming import (
    AsyncStreamingClient,
    PrintingEventHandlers,
    StreamingConfig,
)
 
async def main():
    config = StreamingConfig(
        language="hausa",
        sample_rate=16000,
        interim_results=True,
        return_timestamps=True,
    )
 
    async with AsyncStreamingClient(
        api_key=os.environ["ORBITALSAI_API_KEY"],
        config=config,
    ) as client:
        await client.connect(PrintingEventHandlers())
 
        with open("audio.pcm", "rb") as audio:
            while chunk := audio.read(16000):
                await client.send_audio(chunk)
                await asyncio.sleep(0.5)
 
        await client.flush()
 
asyncio.run(main())

Audio frames must be PCM16, mono, little-endian. The configured sample rate must match the source.

Handle final segments

Subclass StreamingEventHandlers; only override the events your application needs.

from orbitalsai.streaming import StreamingEventHandlers
 
class Handlers(StreamingEventHandlers):
    def on_transcript_partial(self, text):
        update_draft(text)
 
    def on_transcript_final(self, text, metadata):
        save_segment(
            segment_id=metadata["segment_id"],
            text=text,
            timestamps=metadata.get("timestamps"),
        )
 
    def on_speech_start(self):
        show_speaking_state()
 
    def on_error(self, error):
        report_stream_error(error)

Partial transcripts can change. Persist only final segments.

Translate finalized speech

Realtime translation is applied after a segment becomes final:

config = StreamingConfig(
    language="hausa",
    translate=True,
    target_language="Yoruba",
    translation_domain="general conversation",
)
 
class TranslationHandlers(StreamingEventHandlers):
    def on_transcript_final(self, text, metadata):
        save_source(metadata["segment_id"], text)
 
    def on_translation(self, text, metadata):
        save_translation(metadata["segment_ids"], text)
 
    def on_translation_failed(self, reason, metadata):
        mark_translation_gap(metadata["segment_ids"], reason)

segment_ids links each translation to the final transcript segment or segments that produced it. A failed translation does not invalidate the source transcript.

Inspect streaming languages

Query supported streaming input languages and realtime translation targets:

languages = client.get_streaming_languages()
print(languages["supported_languages"])
print(languages["translation_target_languages"])

Convert an audio file

Install the audio extra, then convert supported files to the required PCM format:

from orbitalsai.streaming import AudioConverter
 
audio_bytes, sample_rate = AudioConverter.from_file(
    "meeting.mp3",
    target_sample_rate=16000,
)
chunks = AudioConverter.split_chunks(audio_bytes, chunk_size=8000)

For the wire protocol and complete event sequence, see Realtime transcription, Realtime translation, and the WebSocket protocol reference.

Transcribe a stream

Handle final segments

Translate finalized speech

Inspect streaming languages

Convert an audio file

On this page