Skip to content

Python SDK API Reference

Complete API reference for the Urban Sky Python SDK.

Installation

bash
pip install ably

Loading the SDK

Load the SDK using our CDN loader:

python
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

python
await UrbanSkySDK.init(config)

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

Parameters:

ParameterTypeDescription
configdictSDK configuration dictionary

Configuration Options:

KeyTypeDefaultDescription
'apiToken'strrequiredYour Urban Sky API token
'baseUrl'str'https://api.ops.atmosys.com'API base URL

Example:

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

python
def disconnect() -> None

Disconnects from the telemetry feed and cleans up resources.

Example:

python
sdk.disconnect()

on(event, handler)

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

Registers an event handler for SDK events.

Parameters:

ParameterTypeDescription
eventstrEvent type to listen for
handlerCallableFunction 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:

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

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

ParameterTypeDescription
eventstrEvent type
handlerCallable?Specific handler to remove (optional)

Example:

python
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

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

Device

python
{
    '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:

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

Common Error Codes:

CodeDescription
'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

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

Complete Example

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

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

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

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

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