Ona SDK — Python

Python SDK for live energy asset data — inverter telemetry, OODA terminal alerts, Partner API snapshots, battery warranty intelligence, and ODS-E schema validation.

Installation

pip install asoba

Or from source:

git clone https://github.com/AsobaCloud/sdk.git
cd sdk/python
pip install -e .

Quick start

from asoba import OnaClient

client = OnaClient()

# Query historical inverter data
records = client.inverter_telemetry.get_inverter_telemetry(
    asset_id='INV-1000000054495190',
    site_id='Sibaya',
    time_range={'start': '2025-11-01T00:00:00', 'end': '2025-11-01T12:00:00'},
    resolution='5min',
    limit=100,
)

# Stream live data
for record in client.inverter_telemetry.stream_inverter(
    asset_id='INV-1000000054495190',
    site_id='Sibaya',
    polling_interval=30,
):
    print(f"{record.timestamp}: {record.power} kW")

Set environment variables before running:

export INVERTER_TELEMETRY_ENDPOINT=https://af5jy5ob3e.execute-api.af-south-1.amazonaws.com/prod
export OODA_TERMINAL_ENDPOINT=https://3lpq00xevg.execute-api.af-south-1.amazonaws.com/prod
export PARTNER_API_ENDPOINT=https://8el3o25tc1.execute-api.af-south-1.amazonaws.com/prod
export INVERTER_TELEMETRY_API_KEY=<your_api_key>
export OODA_TERMINAL_API_KEY=<your_api_key>
export PARTNER_API_KEY=<your_api_key>

The same API key value works for all three variables.

Inverter Telemetry

from asoba import OnaClient

client = OnaClient()

# Historical data for one inverter
records = client.inverter_telemetry.get_inverter_telemetry(
    asset_id='INV-1000000054495190',
    site_id='Sibaya',
    time_range={'start': '2025-11-01T00:00:00', 'end': '2025-11-01T12:00:00'},
    resolution='5min',
    limit=100,
)

# Historical data for all inverters at a site
site_records = client.inverter_telemetry.get_site_telemetry(
    site_id='Sibaya',
    time_range={'start': '2025-11-01T00:00:00', 'end': '2025-11-01T12:00:00'},
)

# Discover available data range
period = client.inverter_telemetry.get_data_period(site_id='Sibaya')
print(f"Data from {period.first_record} to {period.last_record}")

# Stream live data (cursor-resumable)
for record in client.inverter_telemetry.stream_inverter(
    asset_id='INV-1000000054495190',
    site_id='Sibaya',
    polling_interval=30,
):
    print(f"{record.timestamp}: {record.power} kW")

OODA Terminal Alerts

from asoba.models.ooda import TimeRange

# Historical alerts for one terminal device
alerts = client.ooda_terminal.get_terminal_alerts(
    terminal_device_id='TERM-1000000054495190',
    site_id='Sibaya',
    time_range=TimeRange(start='2025-11-01T00:00:00', end='2025-11-01T12:00:00'),
    limit=100,
)

# All terminal devices at a site
site_alerts = client.ooda_terminal.get_site_alerts(
    site_id='Sibaya',
    time_range=TimeRange(start='2025-11-01T00:00:00', end='2025-11-01T12:00:00'),
)

# Discover available data range
period = client.ooda_terminal.get_data_period(site_id='Sibaya')
print(f"Data from {period.first_record} to {period.last_record}")

# Stream live alerts
for alert in client.ooda_terminal.stream_terminal(
    terminal_device_id='TERM-1000000054495190',
    site_id='Sibaya',
    polling_interval=30,
):
    print(f"{alert.timestamp}: [{alert.alert_severity}] {alert.message}")

Partner API

Pre-computed snapshots with ETag caching — sub-100ms on repeat calls.

partner_api_endpoint must use https:// — OnaConfig raises ConfigurationError on init otherwise.

# KPI rollup (first call: full fetch; second call: returns cached if ETag matches)
kpis = client.partner.get_kpi_rollup(site_id='Sibaya')
cached = client.partner.get_kpi_rollup(site_id='Sibaya')

# Maintenance signals (detected anomalies) — optional since/severity filters
signals = client.partner.get_maintenance_signals(
    site_id='Sibaya',
    since='2025-11-01T00:00:00',
    severity='high',
)

# 24h solar forecast snapshot
forecast = client.partner.get_forecast_snapshot(site_id='Sibaya')
print(f"{forecast['horizon_hours']}h, {len(forecast['intervals'])} intervals")

# 90-day preventive maintenance schedule (SEP-062)
schedule = client.partner.get_maintenance_schedule(site_id='Sibaya')
for task in schedule['tasks']:
    print(f"{task['recommended_date']} — {task['asset_id']} — {task['task_type']} ({task['priority']})")

Battery & Site Intelligence

# Site summary with battery health KPIs
summary = client.terminal.get_site_summary(site_id='Sibaya')
print(f"Fleet PR: {summary['fleet_pr_pct']}%")

if 'soiling' in summary:
    print(f"Soiling rate: {summary['soiling']['soiling_rate_pct_day']}%/day")

if 'battery' in summary:
    print(f"Avg SOH: {summary['battery']['avg_soh']}%")

# Asset detail (includes battery capacity and warranty)
asset = client.terminal.get_asset(customer_id='cust123', asset_id='BAT-001')

# Calculate remaining warranty life
status = client.terminal.calculate_remaining_warranty_life(
    warranty_expiry_date='2030-12-31',
    warranty_throughput_kwh=10000.0,
    current_throughput_kwh=8500.0,
)
print(f"{status['warranty_status']} — {status['throughput_remaining_pct']}% throughput remaining")

ODS-E Data Validation

Validate records locally against the full 65-field energy-timeseries schema before uploading.

from asoba.utils.validation import validate_batch, validate_with_profile

records = [
    {'timestamp': '2025-01-01T00:00:00Z', 'kWh': 100.5, 'error_type': 'normal'},
    {'timestamp': 'invalid-date', 'kWh': 'not-a-number', 'error_type': 'unknown'},
]

result = validate_batch(records)
print(f"Valid: {result['summary']['valid']}/{result['summary']['total']}")

for item in result['invalid_records']:
    print(f"Errors: {item['errors']}")

Conformance profile validation

# Bilateral trade settlement
is_valid, errors, normalized = validate_with_profile({
    'timestamp': '2026-06-27T14:00:00+02:00',
    'kWh': 87.3,
    'error_type': 'normal',
    'seller_party_id': 'nersa:gen:SOLARPK-001',
    'buyer_party_id': 'nersa:offtaker:MUN042',
    'settlement_period_start': '2026-06-27T14:00:00+02:00',
    'settlement_period_end': '2026-06-27T14:30:00+02:00',
    'contract_reference': 'PPA-SOLARPK-MUN042-2025-003',
    'settlement_type': 'bilateral',
}, 'bilateral')

# BESS dispatch
is_valid, errors, normalized = validate_with_profile({
    'timestamp': '2026-06-27T10:00:00Z',
    'kWh': 50.0,
    'error_type': 'normal',
    'dispatch_mode': 'charging',
    'soc': 75.0,
}, 'bess_dispatch')

# Wind SCADA
is_valid, errors, normalized = validate_with_profile({
    'timestamp': '2026-06-27T10:00:00Z',
    'kWh': 320.0,
    'error_type': 'normal',
    'wind_speed_ms': 8.5,
}, 'wind_scada')

Available profiles: bilateral, wheeling, sawem_brp, municipal_recon, bess_dispatch, wind_scada.

Advanced ML Services

# Trigger a training job
client.training.trigger_training(customer_id='Sibaya', promote=True)

# Check status
status = client.training.get_training_status(customer_id='Sibaya')
print(f"Training status: {status['status']}")

# Gap detection
result = client.gap_detection.detect_gaps(customer_id='Sibaya')
if result['needs_backfill']:
    print(f"Missing intervals: {result['total_missing_intervals']}")

Error handling

from asoba import OnaClient
from asoba.exceptions import ConfigurationError, ValidationError

try:
    records = client.inverter_telemetry.get_inverter_telemetry(...)
except ValidationError as e:
    print(f"Bad params: {e}")
except ConfigurationError as e:
    print(f"Config error: {e}")

API reference

Inverter Telemetry

OODA Terminal Alerts

Partner API

Shared parameters

Rate limits

• 60 requests per minute per API key
• Max 1000 records per query
• Max 31-day time range per query
• Min 5-second polling interval for streaming

Migrating from ona_platform

The package was renamed from ona_platform to asoba. The old name still works during the migration window but emits a DeprecationWarning:

# Before
from ona_platform import OnaClient   # DeprecationWarning

# After
from asoba import OnaClient

Architecture

asoba/
├── client.py          — OnaClient
├── config.py          — OnaConfig dataclass
├── exceptions.py      — ConfigurationError, ValidationError, …
├── services/          — per-API clients
│   ├── inverter_telemetry.py
│   ├── ooda_terminal.py
│   ├── partner_api.py
│   ├── terminal.py
│   ├── training.py
│   └── …
├── models/            — typed dataclasses
│   ├── odse.py        — ODS-E schema constants and enums
│   ├── ooda.py        — TimeRange, alert models
│   ├── snapshots.py   — KpiRollupSnapshot, EarKpis, FinancialKpis
│   └── …
└── utils/
    ├── validation.py  — validate_batch, validate_with_profile
    └── …
ona_platform/          — deprecation shim (re-exports asoba)

Requirements

• Python >= 3.8
• boto3 >= 1.28.0
• requests >= 2.31.0

Examples

python3 examples/inverter_telemetry_example.py
python3 examples/ooda_terminal_example.py
python3 examples/partner_api_example.py

Development

pip install -e ".[dev]"
pytest

License

MIT © Asoba