Urban Sky SDK

Appearance

Python SDK ​

Overview ​

The Urban Sky Python SDK provides real-time access to balloon telemetry data. It connects to our WebSocket service and delivers location updates for balloons in your organization.

Prerequisites ​

Requirements:

  • Python 3.8+ - For proper async/await support
  • requests - Needed to fetch the loader (pip install requests)

Note: Real-time connectivity is built into the SDK. HTTP requests use Python built-in modules (urllib.request, urllib.error), but the WebSocket transport requires centrifuge-python>=0.3.0 — a hard dependency that must be available in your environment (pip install centrifuge-python).

Installation ​

Load the SDK using our CDN loader:

python
import requests

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

# UrbanSkySDK is now available

Basic Usage ​

Initialization and Connection ​

python
import asyncio
import os

async def main():
    try:
        # Initialize and connect in one step
        sdk = await UrbanSkySDK.init({
            'apiToken': os.getenv('URBAN_SKY_API_TOKEN')
            # Most configuration options have sensible defaults
        })
        print('Connected to Urban Sky!')
    except Exception as error:
        print(f'Connection failed: {error}')

Listening for Balloon Updates ​

The SDK emits balloon:update events when balloon positions change:

The payload is a BalloonUpdate dataclass with snake_case attributes — access it with attribute syntax, not dict subscripts:

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

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

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

API Reference ​

Configuration Options ​

python
await UrbanSkySDK.init({
    'apiToken': 'your-api-token',  # Required - Your API token
    'baseUrl': 'custom-api-url'    # Optional - Custom API URL (rarely needed)
})

Note: Most configuration options have sensible defaults. You typically only need to provide your API token.

Methods ​

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

async disconnect() -> None ​

Closes the connection to Urban Sky. This is a coroutine — it must be awaited.

python
await sdk.disconnect()

on(event: str, callback: Callable) -> None ​

Registers an event listener. Every handler receives exactly one payload argument, so even handlers that ignore the payload must accept it.

python
sdk.on('balloon:update', handle_balloon_update)
sdk.on('connected', lambda _: print('SDK connected'))
sdk.on('disconnected', lambda _: print('SDK disconnected'))
sdk.on('error', lambda error: print(f'SDK error: {error}'))

off(event: str, callback: Callable = None) -> None ​

Removes event listener(s).

python
# Remove specific listener
sdk.off('balloon:update', handle_balloon_update)

# Remove all listeners for event
sdk.off('balloon:update')

Events ​

balloon:update ​

Emitted when balloon position data is updated.

Payload: a BalloonUpdate dataclass. Attributes are snake_case (the SDK converts the camelCase wire format for you before your handler sees it):

python
update.balloon_id     # str
update.mission_id     # str
update.devices        # list of device dataclasses:

device.device_id      # str
device.device_type    # str - e.g., "PLD" (Payload), "APX" (Apex), "BLS" (Ballaster)
device.lat            # float - Latitude
device.lng            # float - Longitude
device.altitude       # float - Altitude in meters (when available)
device.timestamp      # str - ISO 8601 timestamp

connected ​

Emitted when successfully connected to Urban Sky.

disconnected ​

Emitted when disconnected from Urban Sky.

error ​

Emitted when an error occurs.

Payload: a dict with two keys:

python
{
    'message': str,  # Human-readable description of what failed
    'error': str     # Error detail or code, e.g. 'RECONNECT_FAILED'
}

Complete Example ​

python
# example.py
import asyncio
import os

import requests

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

async def main():
    try:
        # Initialize and connect SDK
        sdk = await UrbanSkySDK.init({
            'apiToken': os.getenv('URBAN_SKY_API_TOKEN')
        })

        # Connection events
        sdk.on('connected', lambda _: print('✅ Connected to Urban Sky'))
        sdk.on('disconnected', lambda _: print('❌ Disconnected'))
        sdk.on('error', lambda error: print(f'❌ Error: {error}'))

        # Balloon updates
        def handle_balloon_update(update):
            print(f'🎈 Balloon {update.balloon_id} update:')
            print(f'   Mission: {update.mission_id}')

            for device in update.devices:
                print(f'   📍 Device {device.device_id}: {device.lat}, {device.lng}')
                print(f'   🕒 Time: {device.timestamp}')

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

        print('SDK initialized and connected!')

        # Keep the script running
        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print('Disconnecting...')
            await sdk.disconnect()

    except Exception as error:
        print(f'Failed to initialize SDK: {error}')

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

Testing Your Implementation ​

You can test your SDK integration using our test endpoint:

python
import requests

# Send a test balloon update to verify your implementation
def send_test_update(api_token):
    url = 'https://api.ops.atmosys.com/sdk/test/balloon'
    headers = {
        'x-api-token': api_token,
        'Content-Type': 'application/json'
    }

    try:
        response = requests.post(url, headers=headers, json={})
        if response.status_code == 200:
            print('Test message sent - you should receive it via your SDK listener')
        else:
            print(f'Test failed with status: {response.status_code}')
    except Exception as e:
        print(f'Test request failed: {e}')

# Usage
send_test_update('your-api-token')

Virtual Environment Setup ​

It's recommended to use a virtual environment:

bash
# Create virtual environment
python -m venv venv

# Activate (Linux/Mac)
source venv/bin/activate

# Activate (Windows)
venv\Scripts\activate

# Install dependencies
pip install requests centrifuge-python

# Run your script
python example.py

Dependency Installation Issues ​

Common Issues ​

"No module named 'requests'" / "No module named 'centrifuge'"

requests is needed to fetch the loader, and centrifuge-python>=0.3.0 is a required dependency of the SDK — both must be available in your environment.

bash
# Install the loader and SDK dependencies
pip install requests centrifuge-python

# Or using a virtual environment (recommended)
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install requests centrifuge-python

"ModuleNotFoundError" with async/await

  • Ensure you're using Python 3.8+ for proper async support

Error Handling ​

Common Error Codes ​

  • AUTH_INVALID_TOKEN - API token is invalid or expired
  • AUTH_INSUFFICIENT_PERMISSIONS - Token lacks SDK permissions
  • CONNECTION_FAILED - Network connection issues
  • SERVICE_UNAVAILABLE - Urban Sky service temporarily unavailable

Best Practices ​

python
import asyncio
import os

import requests

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

async def main():
    # Handle connection errors
    def handle_error(error):
        if isinstance(error, dict):
            error_code = error.get('error', 'UNKNOWN')
        else:
            error_code = 'UNKNOWN'

        if error_code == 'AUTH_INVALID_TOKEN':
            print('Invalid API token - check your credentials')
        elif error_code == 'CONNECTION_FAILED':
            print('Connection failed - retrying...')
            # SDK will automatically retry
        else:
            print(f'SDK error: {error}')

    # Initialize and connect with error handling
    try:
        sdk = await UrbanSkySDK.init({
            'apiToken': os.getenv('URBAN_SKY_API_TOKEN')
        })

        sdk.on('error', handle_error)

        # Handle reconnection
        sdk.on('disconnected', lambda _: print('Disconnected - will attempt to reconnect'))
        sdk.on('connected', lambda _: print('Connected/reconnected successfully'))

    except Exception as error:
        print(f'Initial connection failed: {error}')
        # Handle startup failure
        return

    # Your application logic here
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await sdk.disconnect()

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

Next Steps ​

Copyright © 2024 Urban Sky