OrbitalsAI Python SDK

Python SDK for the OrbitalsAI API. Transcribe audio files in African languages with SRT subtitle generation and real-time streaming via WebSocket.

Features

• Batch Transcription - Upload files and get transcripts
• Real-time Streaming - Live transcription via WebSocket
• Text AI - Translate, redact PII, and summarize across 19 languages
• Sync & Async - Works synchronously or asynchronously
• African Languages - Hausa, Igbo, Yoruba, Swahili, Pidgin, Kinyarwanda, English
• SRT Subtitles - Generate subtitle files
• Microphone Input - Stream from microphone
• Usage Tracking - Check balance and usage history
• Auto-Reconnect - Automatic reconnection on connection loss

Quick Start

Installation

pip install orbitalsai

For streaming with audio file support:

pip install orbitalsai[audio]

Basic Usage (Batch)

import orbitalsai

# Initialize client
client = orbitalsai.Client(api_key="your_api_key_here")

# Transcribe audio (waits automatically)
transcript = client.transcribe("audio.mp3")
print(transcript.text)

Real-time Streaming

import asyncio
from orbitalsai.streaming import AsyncStreamingClient, PrintingEventHandlers

async def main():
    async with AsyncStreamingClient(api_key="your_api_key") as client:
        await client.connect(PrintingEventHandlers())
        
        with open("audio.pcm", "rb") as f:
            while chunk := f.read(16000):
                await client.send_audio(chunk)
        
        await client.flush()

asyncio.run(main())

That's it.

Table of Contents

• Installation
• Quick Start
• Authentication
• Basic Transcription
• Real-time Streaming
• Text AI: Translate, Redact, Summarize
• Model Selection
• Async Usage
• Balance Management
• Error Handling
• API Reference
• Supported Languages
• Supported Formats
• Troubleshooting

Authentication

Get your API key from the OrbitalsAI Dashboard.

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

Basic Transcription

Simple Transcription

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

# Transcribe audio file
transcript = client.transcribe("audio.mp3")
print(transcript.text)

With Language and SRT

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

# Transcribe in Hausa with SRT subtitles
transcript = client.transcribe(
    "audio.mp3",
    language="hausa",
    generate_srt=True
)

print(transcript.text)
print(transcript.srt_content)  # SRT subtitle content

---

Real-time Streaming

Stream audio and receive transcriptions in real-time via WebSocket.

Installation

Streaming requires additional dependencies:

# Basic streaming
pip install orbitalsai

# With audio file conversion (MP3, WAV, etc.)
pip install orbitalsai[audio]

# With microphone support
pip install sounddevice

Async Streaming (Recommended)

import asyncio
from orbitalsai.streaming import (
    AsyncStreamingClient,
    StreamingConfig,
    PrintingEventHandlers,
)

async def main():
    # Configure streaming
    config = StreamingConfig(
        language="english",
        sample_rate=16000,
        interim_results=True,  # Get partial transcripts
    )
    
    async with AsyncStreamingClient(api_key="your_key", config=config) as client:
        await client.connect(PrintingEventHandlers())
        
        # Stream raw PCM audio
        with open("audio.pcm", "rb") as f:
            while chunk := f.read(16000):  # 500ms chunks
                await client.send_audio(chunk)
                await asyncio.sleep(0.1)  # Real-time pacing
        
        await client.flush()

asyncio.run(main())

Synchronous Streaming

import time
from orbitalsai.streaming import StreamingClient, StreamingEventHandlers

class MyHandlers(StreamingEventHandlers):
    def on_transcript_partial(self, text):
        print(f"Partial: {text}")
    
    def on_transcript_final(self, text, metadata):
        print(f"Final: {text}")

with StreamingClient(api_key="your_key") as client:
    client.connect(MyHandlers())
    
    with open("audio.pcm", "rb") as f:
        while chunk := f.read(16000):
            client.send_audio(chunk)
            time.sleep(0.1)
    
    client.flush()

Stream from Audio Files (MP3, WAV, etc.)

import asyncio
from orbitalsai.streaming import (
    AsyncStreamingClient,
    PrintingEventHandlers,
    AudioConverter,
)

async def stream_file(file_path: str):
    # Convert any audio format to PCM16
    audio_bytes, sample_rate = AudioConverter.from_file(
        file_path, 
        target_sample_rate=16000
    )
    
    # Split into chunks
    chunks = AudioConverter.split_chunks(audio_bytes, chunk_size=8000)
    
    async with AsyncStreamingClient(api_key="your_key") as client:
        await client.connect(PrintingEventHandlers())
        
        for chunk in chunks:
            await client.send_audio(chunk)
            await asyncio.sleep(0.1)
        
        await client.flush()

asyncio.run(stream_file("speech.mp3"))

Stream from Microphone

import asyncio
import numpy as np
import sounddevice as sd
from orbitalsai.streaming import AsyncStreamingClient, PrintingEventHandlers

async def stream_microphone(duration: int = 30):
    audio_queue = asyncio.Queue()
    
    def callback(indata, frames, time_info, status):
        audio_queue.put_nowait(indata.tobytes())
    
    async with AsyncStreamingClient(api_key="your_key") as client:
        await client.connect(PrintingEventHandlers())
        
        with sd.InputStream(
            samplerate=16000,
            channels=1,
            dtype="int16",
            blocksize=8000,
            callback=callback
        ):
            end_time = asyncio.get_event_loop().time() + duration
            
            while asyncio.get_event_loop().time() < end_time:
                audio = await asyncio.wait_for(audio_queue.get(), timeout=1.0)
                await client.send_audio(audio)
        
        await client.flush()

asyncio.run(stream_microphone(30))

Custom Event Handlers

from orbitalsai.streaming import StreamingEventHandlers

class MyHandlers(StreamingEventHandlers):
    def __init__(self):
        self.transcripts = []
        
    
    def on_open(self, session_info):
        print(f"Connected: {session_info['session_id']}")
    
    def on_transcript_partial(self, text):
        # Partial transcripts may change
        print(f"[Partial] {text}")
    
    def on_transcript_final(self, text, metadata):
        # Final transcripts are stable
        self.transcripts.append(text)
        print(f"[Final] {text}")
        print(f"  Duration: {metadata['audio_seconds']:.1f}s")
    
    def on_speech_start(self):
        print("🎤 Speech detected")
    
    def on_speech_end(self):
        print("🔇 Silence detected")
    
    def on_credits_warning(self, remaining_percent):
        print(f"⚠️ Credits low: {remaining_percent}% remaining")
    
    def on_credits_exhausted(self):
        print("❌ Credits exhausted!")
    
    def on_error(self, error):
        print(f"Error: {error}")
    
    def on_close(self, code, reason):
        print(f"Disconnected: {reason}")

Callback-style Handlers

For simpler use cases, use CallbackEventHandlers:

from orbitalsai.streaming import AsyncStreamingClient, CallbackEventHandlers

handlers = CallbackEventHandlers(
    on_final=lambda text, meta: print(f"Transcript: {text}"),
    on_error=lambda e: print(f"Error: {e}"),
)

async with AsyncStreamingClient(api_key="your_key") as client:
    await client.connect(handlers)
    # ... stream audio ...

Accumulate Transcripts

Use StreamingTranscriptAccumulator to collect all transcripts:

from orbitalsai.streaming import StreamingClient, StreamingTranscriptAccumulator

accumulator = StreamingTranscriptAccumulator()

with StreamingClient(api_key="your_key") as client:
    client.connect(accumulator)
    # ... stream audio ...
    client.flush()

# Get results
print(accumulator.get_full_transcript())
print(f"Total duration: {accumulator.total_seconds:.1f}s")

Streaming Configuration

from orbitalsai.streaming import StreamingConfig

config = StreamingConfig(
    # Audio settings
    sample_rate=16000,        # 8000-48000 Hz (16kHz recommended)
    chunk_size=8000,          # Samples per chunk (500ms at 16kHz)
    
    # Language
    language="english",       # english, hausa, igbo, yoruba
    
    # Connection settings
    max_retries=5,            # Reconnection attempts
    retry_delay=1.0,          # Initial retry delay (exponential backoff)
    connection_timeout=30.0,  # Connection timeout in seconds
    
    # Processing
    interim_results=True,     # Receive partial transcripts

    # Realtime translation (finals only -- see "Realtime Translation" below)
    translate=False,
    target_language=None,     # any of the 19 text-API languages; required if translate=True
    translation_domain=None,  # optional vertical hint, e.g. "banking"
    translation_window_ms=1200,
)

Dynamic Configuration

Change language, sample rate, enable timestamps, or enable translation during streaming:

await client.configure(language="hausa")
await client.configure(sample_rate=8000)
await client.configure(return_timestamps=True)
await client.configure(translate=True, target_language="English")

Word-Level Timestamps

Request word-level timing information either on the config (applied automatically on connect, and re-applied after a reconnect):

config = StreamingConfig(language="hausa", return_timestamps=True)
client = AsyncStreamingClient(api_key="...", config=config)

…or by calling configure(return_timestamps=True) any time after connecting, which also lets you turn timings back off mid-session.

When enabled, on_transcript_final receives metadata["timestamps"] -- a list of dicts with per-word start/end times:

from orbitalsai.streaming import AsyncStreamingClient, StreamingEventHandlers

class TimestampHandler(StreamingEventHandlers):
    def on_transcript_final(self, transcript, metadata):
        print(f"Final: {transcript}")
        for word in metadata.get("timestamps", []):
            start = word.get("start", 0)
            end = word.get("end", 0)
            text = word.get("text", "")
            print(f"  [{start:.2f}s - {end:.2f}s] \"{text}\"")

async def main():
    async with AsyncStreamingClient(api_key="your_key") as client:
        await client.connect(TimestampHandler())
        await client.configure(return_timestamps=True)
        # ... stream audio ...
        await client.flush()

Each timestamp object has the shape {"start": float, "end": float, "text": str} where times are in seconds. The key is only present in metadata when the server includes it; callers that don't enable timestamps see no change.

See examples/streaming_with_timestamps.py for a complete runnable example.

---

Realtime Translation

Pair live transcription with a running translation of it. Enable on the config (re-applied automatically after a reconnect, same as return_timestamps):

config = StreamingConfig(
    language="hausa",
    translate=True,
    target_language="English",   # any of the 19 languages the text API supports
)
client = AsyncStreamingClient(api_key="...", config=config)

…or call configure(translate=True, target_language="English") any time after connecting; configure(translate=False) turns it off (flushing anything buffered first).

What actually gets translated, and when. Only finalised transcripts are translated — never partials, since translating a prefix that is still being revised would just show a translation that rewrites itself moments later. Short finals are also coalesced into a brief window (translation_window_ms, default 1200ms) before being sent for translation, both for quality and to bound how many calls a chatty session makes against a shared model server. The practical effect: a translation typically lands 1–3 seconds after the on_transcript_final it covers, not instantly, and it arrives on on_translation naming every segment_id (from on_transcript_final's metadata) that window covered.

from orbitalsai.streaming import AsyncStreamingClient, StreamingEventHandlers

class TranslationHandler(StreamingEventHandlers):
    def on_transcript_final(self, transcript, metadata):
        print(f"[{metadata['segment_id']}] {transcript}")

    def on_translation(self, text, metadata):
        print(f"[{metadata['segment_ids']}] -> {text}  (${metadata['cost']:.4f})")

    def on_translation_failed(self, reason, metadata):
        # The transcript is still valid -- only the translation is missing.
        # Always reported, never silently dropped.
        print(f"[{metadata['segment_ids']}] translation failed: {reason}")

async def main():
    config = StreamingConfig(language="hausa", translate=True, target_language="English")
    async with AsyncStreamingClient(api_key="your_key", config=config) as client:
        await client.connect(TranslationHandler())
        # ... stream audio ...
        await client.flush()

Pivoted directions cost roughly double. A language pair where neither side is English (e.g. Hausa → Yoruba) is served as two model calls through English rather than one direct call — about twice the latency and cost of a direct pair. on_translation_configured reports which route will run ("direct" or "pivot") and a rough estimated_latency_ms before you send any audio, so you can decide whether to warn the user up front.

Failures are always reported, never silent. If a segment can't be translated — the model's output looked fabricated, it hit a timeout, the server was overloaded — on_translation_failed fires with a reason and the transcript for that segment stands on its own. A failed attempt may still carry a nonzero cost: the server bills what it generated even when the result wasn't usable, since the GPU time was real.

Billing is one shared balance. Transcription (per audio-second) and translation (per token) draw on the same balance and the same on_credits_warning / on_credits_critical / on_credits_exhausted thresholds — there's no separate "translation credits."

StreamingTranslationAccumulator collects aligned (segment_ids, source, translation) rows for you, the way StreamingTranscriptAccumulator collects finals:

from orbitalsai.streaming import StreamingClient, StreamingTranslationAccumulator

accumulator = StreamingTranslationAccumulator()
with StreamingClient(api_key="...", config=config) as client:
    client.connect(accumulator)
    # ... send audio ...
    client.flush()

for row in accumulator.get_rows():
    print(f"{row['source']} -> {row['translation']}")
print(f"Translation cost: ${accumulator.total_cost:.4f}")

See examples/streaming_translate.py and examples/streaming_translate_microphone.py for complete runnable examples.

---

Text AI: Translate, Redact, Summarize

Three text tasks run on a self-hosted, fine-tuned model. All take plain text — a transcript from transcribe(), or any text you already have.

19 languages: Afrikaans, Amharic, Arabic, English, French, German, Hausa, Igbo, Isixhosa, Italian, Kinyarwanda, Sesotho, Setswana, Shona, Spanish, Swahili, Twi, Yoruba, Zulu.

Translate

Any of the 19 languages to any other — 342 directions.

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

result = client.translate(
    "Agent: Good morning, how can I help you today?",
    source_language="English",
    target_language="Hausa",
)

print(result.text)          # "Agent: Ina kwana, yaya zan iya taimaka maka a yau?"
print(result.usage.cost)     # credits charged

result stringifies to the translation, so f"{result}" works directly.

Directions where neither side is English run as two legs through English, because most of the training data is English-paired — two strong legs beat one rare direct pair. This is automatic; it costs roughly twice as much and takes about twice as long:

result = client.translate("Sannu da zuwa", "Hausa", "Yoruba")
print(result.was_pivoted)   # True
print(result.route)         # "pivot"
print(result.usage.calls)   # 2

Redact PII

Replaces personal information with placeholders and leaves everything else alone. Output stays in the input language.

result = client.redact(
    "Agent: May I have your name?\n"
    "Customer: My name is Amina Yusuf, phone 08012345678.",
    language="English",
)
print(result.text)
# Agent: May I have your name?
# Customer: My name is [PERSON_NAME], phone [PHONE_NUMBER].

Only these placeholders are ever emitted (orbitalsai.REDACTION_PLACEHOLDERS):

[PERSON_NAME] [PHONE_NUMBER] [EMAIL] [ADDRESS] [ACCOUNT_NUMBER] [TRANSACTION_ID] [GOVERNMENT_ID] [CARD_NUMBER] [BVN] [DATE_OF_BIRTH] [CUSTOMER_ID] [EMPLOYEE_ID] [STUDENT_ID] [POLICY_NUMBER] [MEDICAL_ID]

Give it a real transcript. The model was fine-tuned on multi-turn conversations. On a very short fragment it can invent content that was not in your input; the server detects that and raises TextGenerationError rather than returning fabricated text.

Summarize

Summaries are always in English, whatever the input language.

result = client.summarize(transcript, language="Hausa", style="short")
print(result.text)

Three styles: "short" (2–4 sentences), "detailed" (5–8), and "structured" — which returns six parsed fields and is the one to build UI on:

result = client.summarize(transcript, "English", style="structured")
s = result.structured

print(s.summary)
print(s.customer_issue)
print(s.sentiment)
print(s.resolution_status)
print(s.next_action)
print(s.important_entities)   # list of strings

important_entities is model-generated and, on short inputs, may include plausible-looking values that were not in your transcript. Do not treat it as extracted data without checking it against the source.

Domain (optional)

Passing your vertical nudges the model toward that kind of conversation. Omit it if you are unsure — the default handles general conversation.

result = client.summarize(transcript, "English", domain="banking")

Valid values are in orbitalsai.DOMAINS: banking, customer support, ecommerce, education, fintech, general conversation, government services, healthcare, human resources, insurance, legal admin, logistics, telecom, travel hospitality, utilities.

Long documents: use a job

An hour-long transcript is minutes of GPU time, which no HTTP request should hold open. Jobs chunk the input automatically and survive a cold model:

job = client.submit_text_job("summarize", long_transcript, "Hausa",
                             style="structured")
print(job.job_id, job.status)          # 42 pending

done = client.wait_for_text_job(job.job_id)   # blocks until finished
print(done.text)
print(done.result.structured.next_action)

Poll manually instead if you prefer:

job = client.get_text_job(42)
if job.is_finished:
    print(job.text)

Link a job to the transcript it came from:

transcript = client.transcribe("call.mp3", language="hausa")
job = client.submit_text_job(
    "summarize", transcript.text, "Hausa",
    source_audio_id=transcript.task_id,
)

List past jobs:

jobs = client.list_text_jobs(page=1, page_size=20, job_type="translate")
for job in jobs:
    print(job.job_id, job.status, job.source_language, "->", job.target_language)

Partial results

On a long document a chunk can fail. The result then says so, and names the gap rather than leaving a silent hole — always check is_complete before treating output as the whole document:

result = client.translate(long_text, "English", "Hausa")

if not result.is_complete:
    print(f"{result.failed_units} of {result.total_units} sections failed")
    for gap in result.missing_ranges:
        print("missing:", gap)

Cold starts

The language model takes 10–15 minutes to load if it has been idle. The client retries for up to 2 minutes by default and then raises ModelWarmingError:

from orbitalsai import ModelWarmingError

try:
    result = client.translate(text, "English", "Hausa")
except ModelWarmingError as e:
    print(f"Model is loading; try again in {e.retry_after}s")

Tune or disable that:

client = orbitalsai.Client(api_key="...", warming_timeout=300)
client = orbitalsai.Client(api_key="...", retry_on_warming=False)

For work that can wait out a full cold start, submit a job — the server keeps retrying those until the model is ready.

Async

Every text method has an async twin with the same signature:

import asyncio
import orbitalsai

async def main():
    async with orbitalsai.AsyncClient(api_key="your_api_key_here") as client:
        translated, summary = await asyncio.gather(
            client.translate(text, "English", "Hausa"),
            client.summarize(text, "English", style="structured"),
        )
        print(translated.text)
        print(summary.structured.next_action)

asyncio.run(main())

Discovering what is supported

support = client.get_text_languages()
print(support["languages"])
print(support["summary_styles"])
print(support["domains"])

Model Selection

Choose which model to use for transcription. Different models have different pricing.

List Available Models

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

# Get all available models
models = client.get_models()

for model in models:
    print(f"{model.model_name}: ${model.transcription_rate_per_hour:.2f}/hour")

Transcribe with Specific Model

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

# Transcribe with Perigee-1 model
transcript = client.transcribe(
    "audio.mp3",
    language="hausa",
    model_name="Perigee-1"  # Specify the model
)

print(transcript.text)

Choose Model Based on Budget

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

# Get the cheapest available model
models = client.get_models()
cheapest_model = min(models, key=lambda m: m.transcription_rate_per_hour)

print(f"Using {cheapest_model.model_name} at ${cheapest_model.transcription_rate_per_hour:.2f}/hour")

transcript = client.transcribe(
    "audio.mp3",
    language="english",
    model_name=cheapest_model.model_name
)

Async Usage

For processing multiple files or use in async applications.

import asyncio
import orbitalsai

async def main():
    async with orbitalsai.AsyncClient(api_key="your_api_key_here") as client:
        # List available models
        models = await client.get_models()
        print(f"Available models: {[m.model_name for m in models]}")
        
        # Transcribe multiple files concurrently
        tasks = await asyncio.gather(
            client.transcribe("audio1.mp3", model_name="Perigee-1"),
            client.transcribe("audio2.wav", model_name="Perigee-1"),
            client.transcribe("audio3.m4a", model_name="Perigee-1")
        )
        
        for transcript in tasks:
            print(transcript.text)

asyncio.run(main())

Balance Management

Check Balance

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

balance = client.get_balance()
print(f"Current balance: ${balance.balance:.2f}")
print(f"Last updated: {balance.last_updated}")

Usage History

import orbitalsai
from datetime import date, timedelta

client = orbitalsai.Client(api_key="your_api_key_here")

# Get last 7 days of usage
end_date = date.today()
start_date = end_date - timedelta(days=7)

usage = client.get_daily_usage(start_date=start_date, end_date=end_date)
print(f"Total cost: ${usage.total_cost:.2f}")
print(f"Total audio processed: {usage.total_audio_seconds:.1f} seconds")

for day in usage.daily_records:
    print(f"{day.date}: ${day.total_cost:.4f} ({day.transcription_usage:.1f}s transcription)")

Other Features

List Past Tasks

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

tasks = client.list_tasks(page=1, page_size=20)
for task in tasks:
    print(f"Task {task.task_id}: {task.status} - {task.original_filename}")

print(f"Page {tasks.page} of {tasks.total_pages} ({tasks.total_items} tasks)")

Results are newest first. The returned TaskList is a normal list, and also carries the server's pagination metadata so you can walk the history:

page = 1
while True:
    tasks = client.list_tasks(page=page, page_size=50)
    for task in tasks:
        print(task.task_id, task.status)
    if not tasks.has_next:
        break
    page += 1

Get User Information

import orbitalsai

client = orbitalsai.Client(api_key="your_api_key_here")

user = client.get_user()
print(f"User: {user.first_name} {user.last_name} ({user.email})")
print(f"Verified: {user.is_verified}")

Error Handling

Batch Transcription Errors

import orbitalsai
from orbitalsai.exceptions import (
    AuthenticationError, InsufficientBalanceError, 
    UnsupportedFileError, UnsupportedLanguageError,
    TranscriptionError, TimeoutError
)

client = orbitalsai.Client(api_key="your_api_key_here")

try:
    transcript = client.transcribe("audio.mp3", language="hausa")
    print(transcript.text)
    
except UnsupportedFileError:
    print("File format not supported")
except UnsupportedLanguageError:
    print("Language not supported")
except InsufficientBalanceError:
    print("Not enough credits")
except AuthenticationError:
    print("Invalid API key")
except TranscriptionError as e:
    print(f"Transcription failed: {e}")
except TimeoutError:
    print("Transcription timed out")

Streaming Errors

from orbitalsai.streaming import AsyncStreamingClient, StreamingEventHandlers
from orbitalsai.streaming.exceptions import (
    ConnectionError,
    AuthenticationError,
    InsufficientCreditsError,
    ReconnectionFailedError,
    SessionClosedError,
)

class MyHandlers(StreamingEventHandlers):
    def on_error(self, error):
        if isinstance(error, AuthenticationError):
            print("Invalid API key")
        elif isinstance(error, InsufficientCreditsError):
            print("Credits exhausted - please top up")
        elif isinstance(error, ReconnectionFailedError):
            print(f"Failed to reconnect after {error.attempts} attempts")
        else:
            print(f"Error: {error}")

try:
    async with AsyncStreamingClient(api_key="your_key") as client:
        await client.connect(MyHandlers())
        # ... stream audio ...
except ConnectionError as e:
    print(f"Connection failed: {e}")
except SessionClosedError:
    print("Session was closed")

API Reference

Batch Client Methods

get_models()

Get all available AI models with their pricing information.

Returns: List of Model objects

transcribe(file_path, language="english", generate_srt=False, model_name="Perigee-1", wait=True, timeout=300, poll_interval=5)

Transcribe an audio file.

Parameters:

• file_path (str): Path to the audio file
• language (str): Language code (default: "english")
• generate_srt (bool): Generate SRT subtitles (default: False)
• model_name (str): AI model to use (default: "Perigee-1")
• wait (bool): Wait for completion (default: True)
• timeout (int): Maximum wait time in seconds (default: 300)
• poll_interval (int): Seconds between status checks (default: 5)

Returns: Transcript object (if wait=True) or TranscriptTask object (if wait=False)

get_task(task_id)

Get the status of a transcription task.

Returns: TranscriptTask object

wait_for_task(task_id, timeout=300, poll_interval=5)

Wait for a task to complete.

Returns: Transcript object

list_tasks(page=1, page_size=20)

Get transcription tasks for the current user, newest first.

Parameters:

• page (int): Page number, 1-indexed (default: 1)
• page_size (int): Items per page, 1-100 (default: 20)

Returns: TaskList — a list of TranscriptTask that also exposes page, page_size, total_items, total_pages, has_next, has_previous

Raises: ValueError if page or page_size is out of range

get_tasks() is a deprecated alias for this method and emits a DeprecationWarning. Use list_tasks().

translate(text, source_language, target_language, domain=None)

Translate text between any two of the 19 supported languages.

Parameters:

• text (str): Text or transcript to translate
• source_language (str): Input language, e.g. "Hausa" (any casing accepted)
• target_language (str): Output language, e.g. "Yoruba"
• domain (str, optional): Vertical from orbitalsai.DOMAINS

Returns: TextResult

Raises: UnsupportedLanguageError (unknown language, or source == target), ContentTooLongError, ModelWarmingError, TextGenerationError

redact(text, language, domain=None)

Replace personal information with placeholders. Output stays in language.

Returns: TextResult

summarize(text, language, style="short", domain=None)

Summarize a transcript. Output is always English.

Parameters:

• style (str): "short", "detailed" or "structured"

Returns: TextResult — with .structured populated for "structured"

submit_text_job(task, text, language, target_language=None, style="short", domain=None, source_audio_id=None)

Queue a text task. Use for transcript-length input.

Parameters:

• task (str): "translate", "redact" or "summarize"
• target_language (str): Required for "translate"
• source_audio_id (int, optional): Link to the transcript it came from

Returns: TextJob with status == "pending"

get_text_job(job_id)

Fetch a text job's current state. Returns: TextJob

wait_for_text_job(job_id, timeout=7200, poll_interval=5)

Block until a text job finishes. Returns: TextJob

Raises: TimeoutError (job keeps running server-side), TextGenerationError

list_text_jobs(page=1, page_size=20, job_type=None)

List text jobs, newest first. Returns: TextJobList

get_text_languages()

Fetch supported languages, styles and domains from the server. Returns: dict

TextResult

• text (str): The output. str(result) returns this.
• structured (StructuredSummary | None): Six fields, for style="structured"
• status (str): "complete" or "partial"
• route (str): single_pass | pivot | hierarchical | sectional
• is_complete (bool): False if any part failed — check this
• was_pivoted (bool): Translation ran through English as two legs
• was_chunked (bool): Input was split across multiple model calls
• missing_ranges (list[str]): Named gaps when status == "partial"
• sections (list[TextSection]): Per-chunk summaries with timestamps
• usage (TextUsage): prompt_tokens, completion_tokens, total_tokens, calls, cost
• job_id (int): The stored record

StructuredSummary

• summary, customer_issue, sentiment, resolution_status, next_action (str)
• important_entities (list[str]) — model-generated; verify against the source

TextJob

• job_id, status, job_type, source_language, target_language, style, domain
• result (TextResult | None): Populated once complete
• text (str): Shortcut for result.text, or "" while running
• is_finished (bool), is_failed (bool), error (str | None)

get_balance()

Get the current user's balance.

Returns: Balance object

get_daily_usage(start_date=None, end_date=None, page=1, page_size=30)

Get daily usage history for the current user.

Returns: DailyUsage object

get_user()

Get current user details.

Returns: User object

Streaming Client Methods

connect(handlers)

Establish WebSocket connection and start receiving events.

Parameters:

• handlers (StreamingEventHandlers): Event handler instance

send_audio(audio_data)

Send PCM16 audio chunk.

Parameters:

• audio_data (bytes): Raw PCM16 mono little-endian bytes

configure(language=None, sample_rate=None, return_timestamps=None, translate=None, target_language=None, translation_domain=None, translation_window_ms=None)

Update session configuration dynamically.

Parameters:

• language (str, optional): New transcription language
• sample_rate (int, optional): New sample rate in Hz
• return_timestamps (bool, optional): Enable (True) or disable (False) word-level timestamps. When enabled, on_transcript_final metadata will include a "timestamps" key.
• translate (bool, optional): Enable (True) or disable (False) realtime translation of finalised speech. Enabling requires target_language; disabling flushes any buffered finals and translates them first. See Realtime Translation.
• target_language (str, optional): Translation target, required when translate=True
• translation_domain (str, optional): Vertical hint for the translation prompt, e.g. "banking"
• translation_window_ms (int, optional): How long short finals are buffered before being translated (default: 1200)

flush()

Force transcription of remaining audio buffer.

disconnect()

Close connection gracefully.

Data Models

Transcript

• text (str): Transcribed text
• srt_content (str, optional): SRT subtitle content
• task_id (int): Task ID
• original_filename (str): Original filename
• audio_url (str, optional): URL to processed audio

TranscriptTask

• task_id (int): Task ID
• status (str): Task status ("pending", "processing", "completed", "failed")
• original_filename (str): Original filename
• audio_url (str, optional): URL to processed audio
• srt_requested (bool): Whether SRT was requested
• result_text (str, optional): Transcribed text
• srt_content (str, optional): SRT subtitle content
• error (str, optional): Error message if failed

Balance

• balance (float): Current balance in credits
• last_updated (datetime): Last update timestamp

Model

• id (int): Model ID
• model_name (str): Name of the model (e.g., "Perigee-1")
• transcription_rate_per_second (float): Cost per second of audio
• transcription_rate_per_hour (float): Cost per hour of audio
• is_active (bool): Whether the model is currently available

StreamingConfig

• sample_rate (int): Audio sample rate (8000-48000 Hz)
• chunk_size (int): Samples per chunk
• language (str): Transcription language
• max_retries (int): Maximum reconnection attempts
• retry_delay (float): Initial retry delay
• connection_timeout (float): Connection timeout
• interim_results (bool): Whether to receive partial transcripts
• return_timestamps (bool): Request word-level timings (default: False). Applied on connect and re-applied after a reconnect.

Supported Languages

Language	Code
English	english
Hausa	hausa
Igbo	igbo
Yoruba	yoruba
Swahili	swahili
Pidgin	pidgin
Kinyarwanda	kinyarwanda

Supported Formats

Batch Transcription

• WAV (.wav, .wave)
• MP3 (.mp3, .mpeg)
• OGG (.ogg, .oga)
• FLAC (.flac)
• AAC (.aac)
• M4A (.m4a)
• WMA (.wma)
• AMR (.amr)
• 3GP (.3gp)

Streaming

• Input: PCM16 mono little-endian (raw bytes)
• Conversion supported: All batch formats via AudioConverter

Maximum File Size (Batch)

• 200 MB per file

Troubleshooting

Common Issues

Q: I get "Invalid API key" error A: Make sure your API key is correct. Get it from the OrbitalsAI Dashboard.

Q: I get "Insufficient balance" error A: Add credits to your account through the dashboard.

Q: I get "Unsupported file format" error A: Make sure your audio file is in a supported format (see Supported Formats).

Q: Transcription takes too long A: Large files take longer to process. You can increase the timeout:

transcript = client.transcribe("large_file.mp3", timeout=600)  # 10 minutes

Q: Streaming connection keeps dropping A: The SDK auto-reconnects with exponential backoff. You can configure this:

config = StreamingConfig(max_retries=10, retry_delay=2.0)

Q: How do I convert audio files for streaming? A: Use the AudioConverter utility:

from orbitalsai.streaming import AudioConverter
audio_bytes, sample_rate = AudioConverter.from_file("speech.mp3")

Q: Streaming shows "Credits exhausted" A: Your account ran out of credits during streaming. The connection will close automatically. Top up your credits and reconnect.

Getting Help

• 📧 Email: support@orbitalsai.com
• 🐛 Issues: GitHub Issues
• 📖 Docs: Documentation

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

---

Made by OrbitalsAI