Python SDK API Reference

Complete API reference for the Urban Sky Python SDK.

Installation

No additional dependencies required — real-time connectivity is built into the SDK.

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

def disconnect() -> None

Disconnects from the telemetry feed and cleans up resources.

Example:

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

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

BalloonUpdate

{
    'balloon_id': str,
    'mission_id': str,
    'devices': List[Device]
}

Device

{
    'device_id': str,
    'device_type': str,
    'lat': float,
    'lng': float,
    'altitude': Optional[float],
    'timestamp': str
}

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"
    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.get('altitude'):
                print(f"    Altitude: {device['altitude']}m")

        self.balloons[update['balloon_id']] = update

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

    def handle_disconnected(self):
        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}")

    def stop(self):
        if self.sdk:
            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...')
        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:

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

    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...")
        sdk.disconnect()

asyncio.run(example_with_cleanup())

Logging Integration

The SDK integrates with Python's logging system:

import logging
from urbansky_sdk import UrbanSkySDK

# 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,
    Device,
    BatteryInfo,
    EnvironmentalData,
    StatusInfo,
    MotionData,
    SDKError,
    SDKEventType
)

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

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

Requirements

• Python 3.8+
• ably (install with pip install ably)

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