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

Note: Real-time connectivity is built into the SDK. HTTP requests use Python built-in modules (urllib.request, urllib.error), so no additional dependencies are required.

Installation ​

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

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:

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.

disconnect() -> None ​

Closes the connection to Urban Sky.

python
sdk.disconnect()

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

Registers an event listener.

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:

python
{
    'balloonId': str,
    'missionId': str,
    'devices': [
        {
            'deviceId': str,
            'deviceType': str,    # e.g., "PLD" (Payload), "APX" (Apex), "BLS" (Ballaster)
            'lat': float,         # Latitude
            'lng': float,         # Longitude
            'altitude': float,    # Altitude in meters (when available)
            '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:

python
{
    'code': str,
    'message': str,
    'details': Any  # Optional additional error details
}

Complete Example ​

python
# example.py
import asyncio
import os
import requests

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

        # 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...')
            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

# Run your script
python example.py

Dependency Installation Issues ​

Common Issues ​

"No module named 'requests'"

bash
# Install the required dependency
pip install requests

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

"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 requests
import os

# Load the SDK
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:
        sdk.disconnect()

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

Next Steps ​

Copyright © 2024 Urban Sky