Python SDK API Reference

Complete API reference for the Urban Sky Python SDK.

Installation

The SDK requires centrifuge-python>=0.3.0, which powers real-time connectivity, and requests for the loader snippet below. Both must be available in your environment:

pip install centrifuge-python requests

Loading the SDK

Load the SDK using our CDN loader:

import requests

# Load the SDK dynamically
exec(requests.get('https://sdk.atmosys.com/runtime/py/current/loader.py').text)

# UrbanSkySDK is now available

UrbanSkySDK Class

Initialization

await UrbanSkySDK.init(config)

Initializes and connects to the Urban Sky SDK with the provided configuration.

Parameters:

Parameter	Type	Description
config	dict	SDK configuration dictionary

Configuration Options:

Key	Type	Default	Description
'apiToken'	str	required	Your Urban Sky API token
'baseUrl'	str	'https://api.ops.atmosys.com'	API base URL

Example:

# UrbanSkySDK is available after loading
sdk = await UrbanSkySDK.init({
    'apiToken': 'your-api-token',
    'baseUrl': 'https://api.ops.atmosys.com'
})

Methods

Note: The init() method automatically handles both initialization and connection, so you typically don't need to call connect() separately.

disconnect()

async def disconnect() -> None

Disconnects from the telemetry feed and cleans up resources. This is a coroutine — it must be awaited.

Example:

await sdk.disconnect()

on(event, handler)

def on(event: str, handler: Callable) -> None

Registers an event handler for SDK events.

Parameters:

Parameter	Type	Description
event	str	Event type to listen for
handler	Callable	Function to call when event occurs

Handlers always receive exactly one argument — the event payload. Even for events you don't care about the payload for, the handler must accept it (lambda _: ...).

Events:

• 'balloon:update' - Balloon telemetry update received
• 'unassigned:devices' - Unassigned device telemetry received
• 'connected' - Successfully connected to telemetry feed
• 'disconnected' - Disconnected from telemetry feed
• 'connecting' - Attempting to connect
• 'failed' - Connection failed
• 'error' - Error occurred

Example:

def on_balloon_update(update):
    print(f"Balloon update: {update.balloon_id}")

def on_error(error):
    print(f'SDK error: {error}')

sdk.on('balloon:update', on_balloon_update)
sdk.on('error', on_error)

off(event, handler)

def off(event: str, handler: Optional[Callable] = None) -> None

Removes an event handler. If no handler is provided, removes all handlers for the event.

Parameters:

Parameter	Type	Description
event	str	Event type
handler	Callable?	Specific handler to remove (optional)

Example:

def handler(update):
    print(update)

sdk.on('balloon:update', handler)
sdk.off('balloon:update', handler)  # Remove specific handler
sdk.off('balloon:update')  # Remove all handlers for this event

Data Structures

Handler payloads are dataclasses with snake_case attributes — access them with dot notation (update.balloon_id), not dict subscripts. The camelCase field names (balloonId, deviceId) are the wire format only; the SDK converts them before your handler sees them.

BalloonUpdate

@dataclass
class BalloonUpdate:
    balloon_id: str
    mission_id: str
    devices: List[DeviceLocation]

DeviceLocation

@dataclass
class DeviceLocation:
    device_id: str
    device_type: str
    lat: float
    lng: float
    timestamp: str
    altitude: Optional[float] = None
    battery_info: Optional[BatteryInfo] = None
    environmental: Optional[EnvironmentalInfo] = None
    status: Optional[StatusInfo] = None
    motion: Optional[MotionInfo] = None
    internal: Optional[InternalTelemetry] = None

Error Handling

Errors are emitted as events with the following structure:

{
    'message': str,
    'error': str  # Error code or additional details
}

Common Error Codes:

Code	Description
'INVALID_TOKEN'	API token is invalid or expired
'AUTHENTICATION_FAILED'	Authentication failed
'CONNECTION_LOST'	Connection to server lost
'NETWORK_ERROR'	Network connectivity issue
'TIMEOUT'	Operation timed out
'RATE_LIMITED'	Rate limit exceeded

Enums

SDKEventType

class SDKEventType(Enum):
    BALLOON_UPDATE = "balloon:update"
    UNASSIGNED_DEVICES = "unassigned:devices"
    CONNECTED = "connected"
    DISCONNECTED = "disconnected"
    CONNECTING = "connecting"
    FAILED = "failed"
    ERROR = "error"

Complete Example

import asyncio
import requests
from typing import Dict

# Load the SDK
exec(requests.get('https://sdk.atmosys.com/runtime/py/current/loader.py').text)

class BalloonTracker:
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.sdk = None
        self.balloons: Dict[str, dict] = {}

    async def start(self):
        try:
            # Initialize and connect SDK
            self.sdk = await UrbanSkySDK.init({
                'apiToken': self.api_token,
                'baseUrl': 'https://api.ops.atmosys.com'
            })

            self.setup_event_handlers()

        except Exception as error:
            print(f"Failed to start tracker: {error}")
            raise

    def setup_event_handlers(self):
        self.sdk.on('balloon:update', self.handle_balloon_update)
        self.sdk.on('connected', self.handle_connected)
        self.sdk.on('disconnected', self.handle_disconnected)
        self.sdk.on('error', self.handle_error)

    def handle_balloon_update(self, update):
        print(f"Balloon {update.balloon_id} update:")

        for device in update.devices:
            print(f"  Device {device.device_id} ({device.device_type}):")
            print(f"    Location: {device.lat}, {device.lng}")

            if device.altitude:
                print(f"    Altitude: {device.altitude}m")

        self.balloons[update.balloon_id] = update

    def handle_connected(self, _payload):
        print("✅ Connected to Urban Sky telemetry feed")

    def handle_disconnected(self, _payload):
        print("🔌 Disconnected from telemetry feed")

    def handle_error(self, error):
        print(f"🚨 SDK Error: {error}")

        if isinstance(error, dict) and error.get('error') == 'INVALID_TOKEN':
            print("Please check your API token")
        elif isinstance(error, dict) and error.get('error') == 'NETWORK_ERROR':
            print("Network connectivity issue")
        else:
            print(f"Unexpected error: {error}")

    async def stop(self):
        if self.sdk:
            await self.sdk.disconnect()

    def get_balloon(self, balloon_id: str):
        return self.balloons.get(balloon_id)

    def get_all_balloons(self):
        return list(self.balloons.values())

# Usage
async def main():
    tracker = BalloonTracker('your-api-token')

    try:
        await tracker.start()
        print('Balloon tracker started')

        # Keep running
        while True:
            await asyncio.sleep(1)

    except KeyboardInterrupt:
        print('Shutting down...')
        await tracker.stop()
    except Exception as error:
        print(f'Failed to start tracker: {error}')

if __name__ == '__main__':
    asyncio.run(main())

Async/Await Support

The Python SDK is built with async/await support. init(), authenticate(), connect(), disconnect(), and reconnect() are coroutines and must be awaited; every other method (including on() and off()) is synchronous — awaiting those raises a TypeError.

import asyncio
import requests

# Load the SDK
exec(requests.get('https://sdk.atmosys.com/runtime/py/current/loader.py').text)

async def example():
    # Initialize and connect
    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-token'
    })

    # Event handlers can be regular functions
    def on_balloon_update(update):
        print(f"Balloon: {update.balloon_id}")

    sdk.on('balloon:update', on_balloon_update)

    # Keep running
    await asyncio.sleep(3600)  # Run for 1 hour

    await sdk.disconnect()

# Run the example
asyncio.run(example())

Graceful Shutdown

Handle cleanup properly in your applications:

async def example_with_cleanup():
    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-token'
    })

    def on_balloon_update(update):
        print(f"Balloon: {update.balloon_id}")

    sdk.on('balloon:update', on_balloon_update)

    try:
        # Keep running
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        print("Shutting down...")
        await sdk.disconnect()

asyncio.run(example_with_cleanup())

Logging Integration

The SDK integrates with Python's logging system:

import logging
import requests

# Load the SDK
exec(requests.get('https://sdk.atmosys.com/runtime/py/current/loader.py').text)

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('urbansky_sdk')

# Enable debug logging
sdk = UrbanSkySDK(
    api_token='your-token',
    debug=True  # This will enable debug logs
)

Type Hints

The SDK includes comprehensive type hints for better IDE support:

from typing import Optional, List, Callable
from urbansky_sdk import (
    UrbanSkySDK,
    BalloonUpdate,
    DeviceLocation,
    BatteryInfo,
    EnvironmentalInfo,
    StatusInfo,
    MotionInfo,
    SdkError,
    SDKEventType
)

def process_device(device: DeviceLocation) -> None:
    if device.battery_info:
        battery: BatteryInfo = device.battery_info
        soc: Optional[float] = battery.soc

    if device.environmental:
        env: EnvironmentalInfo = device.environmental
        temp: Optional[float] = env.air_temperature

Requirements

• Python 3.8+
• centrifuge-python>=0.3.0 — a required dependency; it must be available in your environment (pip install centrifuge-python)

Note: HTTP requests use Python built-in modules (urllib.request, urllib.error), so no additional HTTP client is required.

---

Next Steps:

• Imagery SDK Guide - Access mission imagery and photos
• JavaScript SDK Reference - JavaScript SDK documentation
• Type Definitions - Complete type reference
• Examples - Practical usage examples