Appearance
Python SDK API Reference
Complete API reference for the Urban Sky Python SDK.
Installation
bash
pip install ablyLoading 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 availableUrbanSkySDK Class
Initialization
python
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:
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() -> NoneDisconnects from the telemetry feed and cleans up resources.
Example:
python
sdk.disconnect()on(event, handler)
python
def on(event: str, handler: Callable) -> NoneRegisters 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:
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) -> NoneRemoves 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:
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 eventData 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:
| 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
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_temperatureRequirements
- 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