Python SDK API Reference

Complete API reference for the Urban Sky Python SDK.

Installation

The SDK requires centrifuge-python>=0.3.0, which powers real-time connectivity, and requests for the loader snippet below. Both must be available in your environment:

pip install centrifuge-python requests

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

async def disconnect() -> None

Disconnects from the telemetry feed and cleans up resources. This is a coroutine — it must be awaited.

Example:

await 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

Handlers always receive exactly one argument — the event payload. Even for events you don't care about the payload for, the handler must accept it (lambda _: ...).

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

Handler payloads are dataclasses with snake_case attributes — access them with dot notation (update.balloon_id), not dict subscripts. The camelCase field names (balloonId, deviceId) are the wire format only; the SDK converts them before your handler sees them.

BalloonUpdate

@dataclass
class BalloonUpdate:
    balloon_id: str
    mission_id: str
    devices: List[DeviceLocation]

DeviceLocation

@dataclass
class DeviceLocation:
    device_id: str
    device_type: str
    lat: float
    lng: float
    timestamp: str
    altitude: Optional[float] = None
    battery_info: Optional[BatteryInfo] = None
    environmental: Optional[EnvironmentalInfo] = None
    status: Optional[StatusInfo] = None
    motion: Optional[MotionInfo] = None
    internal: Optional[InternalTelemetry] = None

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"
    UNASSIGNED_DEVICES = "unassigned:devices"
    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.altitude:
                print(f"    Altitude: {device.altitude}m")

        self.balloons[update.balloon_id] = update

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

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

    async def stop(self):
        if self.sdk:
            await 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...')
        await 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. init(), authenticate(), connect(), disconnect(), and reconnect() are coroutines and must be awaited; every other method (including on() and off()) is synchronous — awaiting those raises a TypeError.

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

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

asyncio.run(example_with_cleanup())

Logging Integration

The SDK integrates with Python's logging system:

import logging
import requests

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

# 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,
    DeviceLocation,
    BatteryInfo,
    EnvironmentalInfo,
    StatusInfo,
    MotionInfo,
    SdkError,
    SDKEventType
)

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

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

PayloadTelemetry (payload logs and metrics)

PayloadTelemetry runs on a payload. It sends logs and metrics to the Husky bus that the payload rides on. The bus labels them with your payload and forwards them to Urban Sky. It needs no API token, key or certificate: the bus knows your payload from its address on the payload LAN. For background, see Run on a Husky bus.

It is a single file that uses only the standard library (Python 3.8+). It does not need centrifuge-python. The CDN loader does not load it, so put the file into your payload software:

SDK_VERSION=$(curl -fsSL https://sdk.atmosys.com/runtime/py/current/version.txt)
curl -fsSL "https://sdk.atmosys.com/runtime/py/${SDK_VERSION}/urbansky_sdk/payload.py" -o urbansky_payload.py

from urbansky_payload import PayloadLogHandler, PayloadTelemetry

Constructor

PayloadTelemetry(
    bus_url: Optional[str] = None,
    flush_interval: float = 2.0,
    metric_interval: float = 10.0,
    max_queued_lines: int = 2000,
    timeout: float = 3.0,
    opener: Optional[Callable] = None,
    start: bool = True,
)

Parameter	Default	Meaning
bus_url	USKY_BUS_INGEST_URL, else the bus with a LAN route (see below)	The bus's ingest address on the payload LAN.
flush_interval	2.0	Seconds between log sends.
metric_interval	10.0	Seconds between metric sends.
max_queued_lines	2000	Lines kept in memory while the bus does not take them. The oldest go first.
timeout	3.0	Seconds for one request to the bus.
opener	urllib.request.urlopen	Replaces the HTTP call, for tests.
start	True	Starts the background sender thread. With False, nothing is sent until you call flush().

The sender is a daemon thread. Call close() before your process exits, or use the client as a context manager (with PayloadTelemetry() as telemetry:).

Log methods

log(level, message, event=None, attrs=None) -> None

Queues one log line and returns at once.

• level: "debug", "info", "warn" or "error". "warning" counts as "warn". Any other value is sent as "info".
• message: the text. The bus replaces digits in it with #, so put numbers in attrs.
• event: an optional snake_case event name, such as "camera_frame_drop". A line with an event is forwarded at any level.
• attrs: an optional flat dict of strings, numbers and booleans. Only the first 32 keys are kept.

The client stamps each line with the payload's UTC time. A line over 8 KiB is sent with its message cut to 1,024 characters and with no attrs.

debug(message, **kwargs), info(...), warn(...), error(...)

These call log() with that level. kwargs are event and attrs.

telemetry.warn("frame drop", event="camera_frame_drop", attrs={"dropped": 12, "stream": "eo"})
telemetry.error("gimbal stall", event="gimbal_stall", attrs={"axis": "az", "request_id": request_id})

Metric methods

Declare each metric for your payload first. The bus drops undeclared names, attribute keys and attribute values. Use the name without a prefix: the bus adds payload_<kind>_.

declare_metrics(metrics)

For a payload with no :8090 capability document. metrics is the same list as that document's metrics: each entry has name, type (GAUGE or COUNTER), and optional unit, description and attributes ([{"key": ..., "values": [...]}]). The client sends it with PUT /api/v1/payload/metrics/declarations before the next metrics, and again whenever the bus answers undeclared, because the bus keeps the list in memory only.

resolve_bus_url(bus_url=None, route_table=None) -> str

The address the client uses: the argument, else USKY_BUS_INGEST_URL, else the bus address (192.168.8.1, then 10.55.0.1) that the payload has a LAN route to in /proc/net/route, else http://192.168.8.1:4318.

gauge(name, value, **attrs) -> None

Sets a GAUGE's current value for one series (name plus attributes). Only the latest value is sent.

count(name, by=1, **attrs) -> None

Adds by to a COUNTER. The client keeps the running total and sends the total, because the bus expects a cumulative value. The total starts again at 0 when your process restarts.

telemetry.gauge("temp_c", 54.5, sensor="cpu")
telemetry.count("commands_failed")
telemetry.count("frames_dropped", 12, stream="eo")

Attribute keyword names must not be name, value or by. Attribute values are sent as strings.

Sending and status

Member	Meaning
flush() -> None	Sends the queued lines and the current metric values now, on your thread. It still obeys the pacing and back-off below.
close(timeout=5.0) -> None	Stops the sender after one last flush. Without the thread (start=False), it flushes.
status() -> Optional[dict]	GET /api/v1/payload/status: your payload.id, payload.kind, admitted and refused metrics, and drop counts. None if the bus does not answer.
dropped: int	Lines the client dropped: queue full, rate drops the bus reported, or a body the bus refused.
last_result: Optional[dict]	The bus's last reply: {"accepted": n, "dropped": n, "reasons": {...}}.
bus_url: str	The address in use.

Queue, pacing and drops

Situation	What the client does
Normal	Sends logs every flush_interval, at most 64 KiB per request. Sends metric values every metric_interval.
A backlog	Paces itself under the bus's log cap: 448 KiB at once, then 60 KiB per minute. The rest waits in the queue.
429 with Retry-After	Keeps the lines and sends nothing until Retry-After has passed (1 to 60 seconds).
200 with rate or bus_rate drops	The bus took part of the batch. The client counts the rest in dropped and waits 10 seconds.
Any other HTTP error (400, 403, 413)	Counts the lines in dropped. Does not retry them.
Bus unreachable	Keeps the lines and tries again after max(flush_interval, 5) seconds.
Queue full	Drops the oldest line and counts it in dropped.

A send error never raises in your code.

The bus's own limits are in Limits. All payloads on one bus share 256 KiB/min of logs, so another payload can use part of your budget. The bus_rate back-off covers that.

PayloadLogHandler

PayloadLogHandler(telemetry, level=logging.INFO) is a logging.Handler that sends your existing log calls through a PayloadTelemetry. It maps WARNING to "warn", and ERROR and CRITICAL to "error". It adds the logger name as the logger attribute. Set an event and attributes with extra:

import logging

logging.getLogger().addHandler(PayloadLogHandler(telemetry))
logging.warning("gimbal stall", extra={"event": "gimbal_stall", "attrs": {"axis": "az"}})

See Send payload logs and metrics for a complete example.

Requirements

• Python 3.8+
• centrifuge-python>=0.3.0 — a required dependency; it must be available in your environment (pip install centrifuge-python)

Note: HTTP requests use Python built-in modules (urllib.request, urllib.error), so no additional HTTP client is required.

PayloadTelemetry (urbansky_payload.py) needs only the standard library. It does not need centrifuge-python.

---

Next Steps:

• Imagery SDK Guide - Access mission imagery and photos
• JavaScript SDK Reference - JavaScript SDK documentation
• Type Definitions - Complete type reference
• Send payload logs and metrics - PayloadTelemetry on a payload
• Examples - Practical usage examples