orbitalsai 1.2.1
A simple and powerful Python SDK for the OrbitalsAI API with real-time streaming support
License: MIT
Requires Python: >=3.8
Keywords: ai, transcription, audio, speech, african languages, srt, subtitles, streaming, real-time, websocket

# OrbitalsAI Python SDK

[![PyPI version](https://badge.fury.io/py/orbitalsai.svg)](https://badge.fury.io/py/orbitalsai)
[![Python Support](https://img.shields.io/pypi/pyversions/orbitalsai.svg)](https://pypi.org/project/orbitalsai/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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

```bash
pip install orbitalsai
```

For streaming with audio file support:
```bash
pip install orbitalsai[audio]
```

### Basic Usage (Batch)

```python
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

```python
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](#installation)
- [Quick Start](#quick-start)
- [Authentication](#authentication)
- [Basic Transcription](#basic-transcription)
- [Real-time Streaming](#real-time-streaming)
- [Model Selection](#model-selection)
- [Async Usage](#async-usage)
- [Balance Management](#balance-management)
- [Error Handling](#error-handling)
- [API Reference](#api-reference)
- [Supported Languages](#supported-languages)
- [Supported Formats](#supported-formats)
- [Troubleshooting](#troubleshooting)

## Authentication

Get your API key from the [OrbitalsAI Dashboard](https://dashboard.orbitalsai.com).

```python
import orbitalsai

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

## Basic Transcription

### Simple Transcription

```python
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

```python
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:

```bash
# 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)

```python
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

```python
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.)

```python
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

```python
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

```python
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`:

```python
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:

```python
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

```python
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:

```python
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:

```python
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`](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

```python
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

```python
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

```python
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.

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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](https://dashboard.orbitalsai.com).

**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](#supported-formats)).

**Q: Transcription takes too long**
A: Large files take longer to process. You can increase the timeout:
```python
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:
```python
config = StreamingConfig(max_retries=10, retry_delay=2.0)
```

**Q: How do I convert audio files for streaming?**
A: Use the `AudioConverter` utility:
```python
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](https://github.com/orbitalsai/orbitalsai-python-sdk/issues)
- 📖 Docs: [Documentation](https://docs.orbitalsai.com)

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## Contributing

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

---

Made by OrbitalsAI