Basic Usage Examples

This page provides practical examples for common Urban Sky SDK use cases. These examples build on the Getting Started guide and show real-world implementation patterns.

Simple Connection Example

The most basic use case - connecting to the telemetry feed and logging balloon updates:

JavaScriptPython

javascript

// Load the SDK
const response = await fetch('https://sdk.atmosys.com/runtime/js/current/loader.js')
eval(await response.text())

const sdk = await UrbanSkySDK.init({
  apiToken: 'your-api-token',
})

// Listen for balloon updates
sdk.on('balloon:update', update => {
  console.log(`Balloon ${update.balloonId} updated with ${update.devices.length} devices`)
})

// Listen for unassigned device updates
sdk.on('unassigned:devices', update => {
  console.log('Unassigned devices update received:', update)

  update.devices.forEach(device => {
    console.log(`Unassigned device ${device.deviceId} (${device.deviceType}):`)
    console.log(`  Location: ${device.lat}, ${device.lng}`)
    console.log(`  Altitude: ${device.altitude}m`)
    console.log(`  Time: ${device.timestamp}`)
  })
})

// Handle connection events
sdk.on('connected', () => console.log('Connected to Urban Sky'))
sdk.on('disconnected', () => console.log('Disconnected'))
sdk.on('error', error => console.error('Error:', error))

// Connect
await sdk.connect()

python

import requests

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

async def main():
    # Initialize and connect
    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-api-token'
    })

    def on_balloon_update(update):
    print(f"Balloon {update['balloon_id']} updated with {len(update['devices'])} devices")

def on_unassigned_devices(update):
    print(f"Unassigned devices update received: {len(update['devices'])} devices")
    for device in update['devices']:
        print(f"  {device['device_type']} {device['device_id']} at {device['lat']}, {device['lng']}")

    def on_connected():
        print("Connected to Urban Sky")

    def on_disconnected():
        print("Disconnected")

    def on_error(error):
        print(f"Error: {error}")

    # Register handlers
    sdk.on('balloon:update', on_balloon_update)
    sdk.on('unassigned:devices', on_unassigned_devices)
    sdk.on('connected', on_connected)
    sdk.on('disconnected', on_disconnected)
    sdk.on('error', on_error)

import asyncio
asyncio.run(main())

Testing Your Implementation

Before processing real data, test your implementation using our test endpoint:

JavaScriptPython

javascript

// Send a test balloon update to verify your implementation
async function sendTestUpdate() {
  const testResponse = await fetch('https://api.ops.atmosys.com/sdk/test/balloon', {
    method: 'POST',
    headers: {
      'x-api-token': 'your-api-token',
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({}),
  })

  if (testResponse.ok) {
    console.log('✅ Test message sent - you should receive it via your SDK listener')
  } else {
    console.error('❌ Test failed:', await testResponse.text())
  }
}

// Set up SDK first
const sdk = await UrbanSkySDK.init({ apiToken: 'your-api-token' })

sdk.on('balloon:update', update => {
  console.log('🎈 Received test update:', update.balloonId)
  console.log('✅ Your implementation is working!')
})

// Send test update
await sendTestUpdate()

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}')

# Set up SDK first
async def main():
    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-api-token'
    })

    def on_balloon_update(update):
    print(f"🎈 Received test update: {update['balloon_id']}")
    print('✅ Your implementation is working!')

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

    # Send test update
    send_test_update('your-api-token')

import asyncio
asyncio.run(main())

Filtering by Device Type

Filter balloon updates to only process specific device types:

JavaScriptPython

javascript

sdk.on('balloon:update', update => {
  // Filter for payload devices only
  const payloadDevices = update.devices.filter(device => device.deviceType === 'PLD')

  if (payloadDevices.length > 0) {
    console.log(`Payload update for balloon ${update.balloonId}`)
    payloadDevices.forEach(device => {
      console.log(`  Payload at: ${device.lat}, ${device.lng}`)
      if (device.altitude) {
        console.log(`  Altitude: ${device.altitude}m`)
      }
    })
  }
})

python

def on_balloon_update(update):
    # Filter for payload devices only
    payload_devices = [device for device in update['devices']
                      if device['device_type'] == 'PLD']

    if payload_devices:
        print(f"Payload update for balloon {update['balloon_id']}")
        for device in payload_devices:
            print(f"  Payload at: {device['lat']}, {device['lng']}")
            if device.get('altitude'):
                print(f"  Altitude: {device['altitude']}m")

Tracking Multiple Balloons

Keep track of multiple balloons and their latest positions:

JavaScriptPython

javascript

class BalloonTracker {
  constructor() {
    this.balloons = new Map()
  }

  updateBalloon(update) {
    const balloonId = update.balloonId

    // Get or create balloon record
    if (!this.balloons.has(balloonId)) {
      this.balloons.set(balloonId, {
        id: balloonId,
        missionId: update.missionId,
        devices: new Map(),
        lastUpdate: null,
      })
    }

    const balloon = this.balloons.get(balloonId)

    // Update device positions
    update.devices.forEach(device => {
      balloon.devices.set(device.deviceId, {
        ...device,
        receivedAt: new Date(),
      })
    })

    balloon.lastUpdate = new Date()

    console.log(`Tracking ${this.balloons.size} balloons`)
    console.log(`Balloon ${balloonId} has ${balloon.devices.size} devices`)
  }

  getBalloonStatus(balloonId) {
    return this.balloons.get(balloonId)
  }

  getAllBalloons() {
    return Array.from(this.balloons.values())
  }
}

const tracker = new BalloonTracker()

sdk.on('balloon:update', update => {
  tracker.updateBalloon(update)
})

python

from datetime import datetime
import requests

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

class BalloonTracker:
    def __init__(self):
        self.balloons = {}

    def update_balloon(self, update):
        balloon_id = update['balloon_id']

        # Get or create balloon record
        if balloon_id not in self.balloons:
            self.balloons[balloon_id] = {
                'id': balloon_id,
                'mission_id': update['mission_id'],
                'devices': {},
                'last_update': None
            }

        balloon = self.balloons[balloon_id]

        # Update device positions
        for device in update['devices']:
            balloon['devices'][device['device_id']] = {
                **device,
                'received_at': datetime.now()
            }

        balloon['last_update'] = datetime.now()

        print(f"Tracking {len(self.balloons)} balloons")
        print(f"Balloon {balloon_id} has {len(balloon['devices'])} devices")

    def get_balloon_status(self, balloon_id):
        return self.balloons.get(balloon_id)

    def get_all_balloons(self):
        return list(self.balloons.values())

async def main():
    tracker = BalloonTracker()

    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-api-token'
    })

    def on_balloon_update(update):
        tracker.update_balloon(update)

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

import asyncio
asyncio.run(main())

Distance and Speed Calculations

Calculate distance traveled and speed between updates:

JavaScriptPython

javascript

class LocationTracker {
  constructor() {
    this.lastPositions = new Map()
  }

  // Haversine formula for distance calculation
  calculateDistance(lat1, lon1, lat2, lon2) {
    const R = 6371000 // Earth's radius in meters
    const φ1 = (lat1 * Math.PI) / 180
    const φ2 = (lat2 * Math.PI) / 180
    const Δφ = ((lat2 - lat1) * Math.PI) / 180
    const Δλ = ((lon2 - lon1) * Math.PI) / 180

    const a =
      Math.sin(Δφ / 2) * Math.sin(Δφ / 2) +
      Math.cos(φ1) * Math.cos(φ2) * Math.sin(Δλ / 2) * Math.sin(Δλ / 2)
    const c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a))

    return R * c // Distance in meters
  }

  processUpdate(update) {
    update.devices.forEach(device => {
      const deviceKey = `${update.balloonId}-${device.deviceId}`
      const lastPos = this.lastPositions.get(deviceKey)

      if (lastPos) {
        const distance = this.calculateDistance(lastPos.lat, lastPos.lng, device.lat, device.lng)

        const timeDiff = (new Date(device.timestamp) - new Date(lastPos.timestamp)) / 1000
        const speed = distance / timeDiff // m/s

        console.log(`Device ${device.deviceId}:`)
        console.log(`  Distance traveled: ${distance.toFixed(1)}m`)
        console.log(`  Speed: ${speed.toFixed(2)} m/s (${(speed * 3.6).toFixed(1)} km/h)`)

        if (device.altitude && lastPos.altitude) {
          const altitudeChange = device.altitude - lastPos.altitude
          const verticalSpeed = altitudeChange / timeDiff
          console.log(`  Vertical speed: ${verticalSpeed.toFixed(2)} m/s`)
        }
      }

      // Store current position
      this.lastPositions.set(deviceKey, {
        lat: device.lat,
        lng: device.lng,
        altitude: device.altitude,
        timestamp: device.timestamp,
      })
    })
  }
}

const locationTracker = new LocationTracker()

sdk.on('balloon:update', update => {
  locationTracker.processUpdate(update)
})

python

import math
import requests
from datetime import datetime

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

class LocationTracker:
    def __init__(self):
        self.last_positions = {}

    def calculate_distance(self, lat1, lon1, lat2, lon2):
        """Haversine formula for distance calculation"""
        R = 6371000  # Earth's radius in meters
        φ1 = math.radians(lat1)
        φ2 = math.radians(lat2)
        Δφ = math.radians(lat2 - lat1)
        Δλ = math.radians(lon2 - lon1)

        a = (math.sin(Δφ/2) ** 2 +
             math.cos(φ1) * math.cos(φ2) * math.sin(Δλ/2) ** 2)
        c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))

        return R * c  # Distance in meters

    def process_update(self, update):
        for device in update['devices']:
            device_key = f"{update['balloon_id']}-{device['device_id']}"
            last_pos = self.last_positions.get(device_key)

            if last_pos:
                distance = self.calculate_distance(
                    last_pos['lat'], last_pos['lng'],
                    device['lat'], device['lng']
                )

                time_diff = (datetime.fromisoformat(device['timestamp'].replace('Z', '+00:00')) -
                           datetime.fromisoformat(last_pos['timestamp'].replace('Z', '+00:00'))).total_seconds()

                if time_diff > 0:
                    speed = distance / time_diff  # m/s

                    print(f"Device {device['device_id']}:")
                    print(f"  Distance traveled: {distance:.1f}m")
                    print(f"  Speed: {speed:.2f} m/s ({speed * 3.6:.1f} km/h)")

                    if device.get('altitude') and last_pos.get('altitude'):
                        altitude_change = device['altitude'] - last_pos['altitude']
                        vertical_speed = altitude_change / time_diff
                        print(f"  Vertical speed: {vertical_speed:.2f} m/s")

            # Store current position
            self.last_positions[device_key] = {
                'lat': device['lat'],
                'lng': device['lng'],
                'altitude': device.get('altitude'),
                'timestamp': device['timestamp']
            }

async def main():
    location_tracker = LocationTracker()

    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-api-token'
    })

    def on_balloon_update(update):
        location_tracker.process_update(update)

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

import asyncio
asyncio.run(main())

Data Logging and Storage

Log balloon data for later analysis:

JavaScriptPython

javascript

class DataLogger {
  constructor() {
    this.data = []
    this.maxEntries = 10000 // Keep last 10,000 entries
  }

  logUpdate(update) {
    const timestamp = new Date().toISOString()

    const logEntry = {
      timestamp,
      balloonId: update.balloonId,
      missionId: update.missionId,
      devices: update.devices.map(device => ({
        deviceId: device.deviceId,
        deviceType: device.deviceType,
        lat: device.lat,
        lng: device.lng,
        altitude: device.altitude,
      })),
    }

    this.data.push(logEntry)

    // Keep only recent entries
    if (this.data.length > this.maxEntries) {
      this.data = this.data.slice(-this.maxEntries)
    }

    console.log(`Logged update for balloon ${update.balloonId}`)
  }

  getRecentUpdates(count = 100) {
    return this.data.slice(-count)
  }

  getBalloonUpdates(balloonId) {
    return this.data.filter(entry => entry.balloonId === balloonId)
  }

  getStats() {
    const balloonIds = new Set(this.data.map(entry => entry.balloonId))
    return {
      totalUpdates: this.data.length,
      uniqueBalloons: balloonIds.size,
      oldestUpdate: this.data[0]?.timestamp,
      newestUpdate: this.data[this.data.length - 1]?.timestamp,
    }
  }
}

const logger = new DataLogger()

sdk.on('balloon:update', update => {
  logger.logUpdate(update)

  // Display stats periodically
  if (logger.data.length % 100 === 0) {
    const stats = logger.getStats()
    console.log('📊 Stats:', stats)
  }
})

python

import requests
from datetime import datetime

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

class DataLogger:
    def __init__(self):
        self.data = []
        self.max_entries = 10000  # Keep last 10,000 entries

    def log_update(self, update):
        timestamp = datetime.now().isoformat()

        log_entry = {
            'timestamp': timestamp,
            'balloon_id': update['balloon_id'],
            'mission_id': update['mission_id'],
            'devices': [{
                'device_id': device['device_id'],
                'device_type': device['device_type'],
                'lat': device['lat'],
                'lng': device['lng'],
                'altitude': device.get('altitude')
            } for device in update['devices']]
        }

        self.data.append(log_entry)

        # Keep only recent entries
        if len(self.data) > self.max_entries:
            self.data = self.data[-self.max_entries:]

        print(f"Logged update for balloon {update['balloon_id']}")

    def get_recent_updates(self, count=100):
        return self.data[-count:]

    def get_balloon_updates(self, balloon_id):
        return [entry for entry in self.data if entry['balloon_id'] == balloon_id]

    def get_stats(self):
        balloon_ids = set(entry['balloon_id'] for entry in self.data)
        return {
            'total_updates': len(self.data),
            'unique_balloons': len(balloon_ids),
            'oldest_update': self.data[0]['timestamp'] if self.data else None,
            'newest_update': self.data[-1]['timestamp'] if self.data else None
        }

async def main():
    logger = DataLogger()

    sdk = await UrbanSkySDK.init({
        'apiToken': 'your-api-token'
    })

    def on_balloon_update(update):
        logger.log_update(update)

        # Display stats periodically
        if len(logger.data) % 100 == 0:
            stats = logger.get_stats()
            print('📊 Stats:', stats)

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

import asyncio
asyncio.run(main())

Real-time Alerts

Set up alerts for specific conditions:

JavaScriptPython

javascript

class AlertSystem {
  constructor() {
    this.alertRules = []
  }

  addRule(name, condition, action) {
    this.alertRules.push({ name, condition, action })
  }

  checkAlerts(update) {
    update.devices.forEach(device => {
      this.alertRules.forEach(rule => {
        if (rule.condition(device, update)) {
          rule.action(device, update, rule.name)
        }
      })
    })
  }
}

const alerts = new AlertSystem()

// Add alert rules
alerts.addRule(
  'High Altitude',
  device => device.altitude && device.altitude > 30000,
  (device, update, ruleName) => {
    console.warn(`🚨 ${ruleName}: Device ${device.deviceId} at ${device.altitude}m altitude`)
  },
)

alerts.addRule(
  'Rapid Descent',
  device => device.altitude && device.altitude < 5000,
  (device, update, ruleName) => {
    console.warn(
      `🚨 ${ruleName}: Device ${device.deviceId} descending - altitude ${device.altitude}m`,
    )
  },
)

sdk.on('balloon:update', update => {
  alerts.checkAlerts(update)
})

python

class AlertSystem:
    def __init__(self):
        self.alert_rules = []

    def add_rule(self, name, condition, action):
        self.alert_rules.append({
            'name': name,
            'condition': condition,
            'action': action
        })

    def check_alerts(self, update):
        for device in update['devices']:
            for rule in self.alert_rules:
                if rule['condition'](device, update):
                    rule['action'](device, update, rule['name'])

alerts = AlertSystem()

# Add alert rules
def high_altitude_condition(device, update):
    return device.get('altitude') and device['altitude'] > 30000

def high_altitude_action(device, update, rule_name):
    print(f"🚨 {rule_name}: Device {device['device_id']} at {device['altitude']}m altitude")

def rapid_descent_condition(device, update):
    return device.get('altitude') and device['altitude'] < 5000

def rapid_descent_action(device, update, rule_name):
    print(f"🚨 {rule_name}: Device {device['device_id']} descending - altitude {device['altitude']}m")

alerts.add_rule('High Altitude', high_altitude_condition, high_altitude_action)
alerts.add_rule('Rapid Descent', rapid_descent_condition, rapid_descent_action)

def on_balloon_update(update):
    alerts.check_alerts(update)

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

Error Handling and Reconnection

Robust error handling for production applications:

JavaScriptPython

javascript

class RobustSDKWrapper {
  constructor(config) {
    this.config = config
    this.sdk = null
    this.reconnectAttempts = 0
    this.maxReconnectAttempts = 10
    this.reconnectDelay = 5000
    this.isConnected = false
    this.eventHandlers = new Map()
  }

  async connect() {
    try {
      this.sdk = await UrbanSkySDK.init(this.config)
      this.setupEventHandlers()

      await this.sdk.connect()
      this.isConnected = true
      this.reconnectAttempts = 0

      console.log('✅ Connected to Urban Sky')
    } catch (error) {
      console.error('❌ Connection failed:', error.message)
      await this.handleReconnection()
    }
  }

  setupEventHandlers() {
    this.sdk.on('connected', () => {
      this.isConnected = true
      console.log('🔗 SDK connected')
    })

    this.sdk.on('disconnected', () => {
      this.isConnected = false
      console.log('🔌 SDK disconnected')
      this.handleReconnection()
    })

    this.sdk.on('error', error => {
      console.error('🚨 SDK error:', error)

      if (error.code === 'AUTH_INVALID_TOKEN') {
        console.error('❌ Authentication failed - check your API token')
        return // Don't retry on auth errors
      }

      this.handleReconnection()
    })

    // Forward balloon updates to registered handlers
    this.sdk.on('balloon:update', update => {
      this.eventHandlers.get('balloon:update')?.forEach(handler => {
        try {
          handler(update)
        } catch (error) {
          console.error('Error in balloon update handler:', error)
        }
      })
    })
  }

  async handleReconnection() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('❌ Max reconnection attempts reached')
      return
    }

    this.reconnectAttempts++
    const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1) // Exponential backoff

    console.log(
      `🔄 Reconnection attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms`,
    )

    setTimeout(async () => {
      await this.connect()
    }, delay)
  }

  on(event, handler) {
    if (!this.eventHandlers.has(event)) {
      this.eventHandlers.set(event, [])
    }
    this.eventHandlers.get(event).push(handler)
  }

  disconnect() {
    if (this.sdk) {
      this.sdk.disconnect()
      this.isConnected = false
    }
  }
}

// Usage
const robustSDK = new RobustSDKWrapper({
  apiToken: 'your-api-token',
})

robustSDK.on('balloon:update', update => {
  console.log(`Balloon update: ${update.balloonId}`)
})

await robustSDK.connect()

python

import asyncio
from typing import Dict, List, Callable

class RobustSDKWrapper:
    def __init__(self, config):
        self.config = config
        self.sdk = None
        self.reconnect_attempts = 0
        self.max_reconnect_attempts = 10
        self.reconnect_delay = 5
        self.is_connected = False
        self.event_handlers = {}

    async def connect(self):
        try:
            self.sdk = UrbanSkySDK(**self.config)
            self.setup_event_handlers()

            await self.sdk.connect()
            self.is_connected = True
            self.reconnect_attempts = 0

            print("✅ Connected to Urban Sky")

        except Exception as error:
            print(f"❌ Connection failed: {error}")
            await self.handle_reconnection()

    def setup_event_handlers(self):
        def on_connected():
            self.is_connected = True
            print("🔗 SDK connected")

        def on_disconnected():
            self.is_connected = False
            print("🔌 SDK disconnected")
            asyncio.create_task(self.handle_reconnection())

        def on_error(error):
            print(f"🚨 SDK error: {error}")

            if hasattr(error, 'code') and error.code == 'AUTH_INVALID_TOKEN':
                print("❌ Authentication failed - check your API token")
                return  # Don't retry on auth errors

            asyncio.create_task(self.handle_reconnection())

        def on_balloon_update(update):
            handlers = self.event_handlers.get('balloon:update', [])
            for handler in handlers:
                try:
                    handler(update)
                except Exception as error:
                    print(f"Error in balloon update handler: {error}")

        self.sdk.on('connected', on_connected)
        self.sdk.on('disconnected', on_disconnected)
        self.sdk.on('error', on_error)
        self.sdk.on('balloon:update', on_balloon_update)

    async def handle_reconnection(self):
        if self.reconnect_attempts >= self.max_reconnect_attempts:
            print("❌ Max reconnection attempts reached")
            return

        self.reconnect_attempts += 1
        delay = self.reconnect_delay * (2 ** (self.reconnect_attempts - 1))  # Exponential backoff

        print(f"🔄 Reconnection attempt {self.reconnect_attempts}/{self.max_reconnect_attempts} in {delay}s")

        await asyncio.sleep(delay)
        await self.connect()

    def on(self, event, handler):
        if event not in self.event_handlers:
            self.event_handlers[event] = []
        self.event_handlers[event].append(handler)

    def disconnect(self):
        if self.sdk:
            self.sdk.disconnect()
            self.is_connected = False

# Usage
robust_sdk = RobustSDKWrapper({
    'api_token': 'your-api-token'
})

def on_balloon_update(update):
    print(f"Balloon update: {update['balloon_id']}")

robust_sdk.on('balloon:update', on_balloon_update)

await robust_sdk.connect()

Next Steps

These examples provide a foundation for building robust balloon tracking applications. For more advanced use cases, see:

Error Handling Guide - Comprehensive error handling strategies
API Reference - Complete SDK API documentation
Python SDK Reference - Python SDK documentation

Need help? Check our troubleshooting guide or contact support.

Basic Usage Examples ​

Simple Connection Example ​

Testing Your Implementation ​

Filtering by Device Type ​

Tracking Multiple Balloons ​

Distance and Speed Calculations ​

Data Logging and Storage ​

Real-time Alerts ​

Error Handling and Reconnection ​

Next Steps ​

Basic Usage Examples

Simple Connection Example

Testing Your Implementation

Filtering by Device Type

Tracking Multiple Balloons

Distance and Speed Calculations

Data Logging and Storage

Real-time Alerts

Error Handling and Reconnection

Next Steps