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.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.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