orbitalsai 1.2.1

A simple and powerful Python SDK for the OrbitalsAI API with real-time streaming support

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
• 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
• 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
)

Dynamic Configuration

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

await client.configure(language="hausa")
await client.configure(sample_rate=8000)
await client.configure(return_timestamps=True)

Word-Level Timestamps

Request word-level timing information by calling configure(return_timestamps=True) after connecting. 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.

---

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 All Tasks

import orbitalsai

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

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

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()

Get all transcription tasks for the current user.

Returns: List of TranscriptTask objects

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)

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.

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

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: [e-mail]
• 🐛 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