ADS-B Exchange APIadsbexchange.com ↗

"""
ADSB Exchange API Client using Parse

This script demonstrates how to use the ADSB Exchange API through Parse
to monitor real-time global flight data.

Get your API key from: https://parse.bot/settings
"""

import os
import json
import requests
from typing import Optional, List, Dict, Any


class ParseClient:
    """Client for interacting with ADSB Exchange API through Parse."""

    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse API client.
        
        Args:
            api_key: API key for Parse. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "2876df71-3e83-4132-beb1-769ce912574c"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key not provided and PARSE_API_KEY environment variable not set")

    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """Make a request to the Parse API.
        
        Args:
            endpoint: The endpoint name (e.g., 'get_all_aircraft')
            method: HTTP method ('GET' or 'POST')
            **params: Query parameters or request body
            
        Returns:
            JSON response from the API
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "POST":
            payload = {"api_key": self.api_key, **params}
            response = requests.post(url, headers=headers, json=payload)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")
        
        response.raise_for_status()
        return response.json()

    def get_all_aircraft(self) -> Dict[str, Any]:
        """Returns all aircraft currently being tracked globally.
        
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_all_aircraft", method="GET", api_key=self.api_key)

    def get_total_aircraft_count(self) -> Dict[str, Any]:
        """Returns only the count of aircraft currently being tracked globally.
        
        Returns:
            Dict with 'total' count and 'now' timestamp
        """
        return self._call("get_total_aircraft_count", method="GET", api_key=self.api_key)

    def filter_aircraft(self, filters: List[Dict[str, Any]], logical_operator: str = "and") -> Dict[str, Any]:
        """Filters and returns aircraft matching provided criteria.
        
        Args:
            filters: List of filter definitions, e.g., [{"property": "alt_baro", "operator": "le", "value": 5000}]
            logical_operator: 'and' or 'or' to combine filters
            
        Returns:
            Dict with filtered 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("filter_aircraft", method="POST", filters=filters, logical_operator=logical_operator)

    def get_aircraft_by_hex(self, hex_codes: str) -> Dict[str, Any]:
        """Returns real-time data for aircraft by ICAO hex code(s).
        
        Args:
            hex_codes: Single hex code or comma-separated list (e.g., 'A1B2C3' or 'A1B2C3,D4E5F6')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_by_hex", method="GET", api_key=self.api_key, hex_codes=hex_codes)

    def get_aircraft_by_callsign(self, callsigns: str) -> Dict[str, Any]:
        """Returns real-time data for aircraft by callsign(s).
        
        Args:
            callsigns: Single callsign or comma-separated list (e.g., 'UAL123' or 'UAL123,SWA456')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_by_callsign", method="GET", api_key=self.api_key, callsigns=callsigns)

    def get_aircraft_by_registration(self, registrations: str) -> Dict[str, Any]:
        """Returns real-time data for aircraft by registration (tail number).
        
        Args:
            registrations: Single registration or comma-separated list (e.g., 'N12345' or 'N12345,N67890')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_by_registration", method="GET", api_key=self.api_key, registrations=registrations)

    def get_aircraft_in_radius(self, lat: float, lon: float, dist: float) -> Dict[str, Any]:
        """Returns aircraft within a given radius of a lat/lon coordinate.
        
        Args:
            lat: Latitude of center point
            lon: Longitude of center point
            dist: Radius in nautical miles
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_in_radius", method="GET", api_key=self.api_key, 
                         lat=lat, lon=lon, dist=dist)

    def get_aircraft_near_airport(self, airport: str) -> Dict[str, Any]:
        """Returns aircraft within 5 nautical miles of a specific airport.
        
        Args:
            airport: ICAO airport code (e.g., 'KSFO')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_near_airport", method="GET", api_key=self.api_key, airport=airport)

    def get_aircraft_by_country(self, country: str) -> Dict[str, Any]:
        """Returns aircraft within the boundaries of a given country.
        
        Args:
            country: ISO 3166-1 alpha-2 country code (e.g., 'US', 'CA')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_by_country", method="GET", api_key=self.api_key, country=country)

    def get_subdivisions_by_country_code(self, country: str) -> Dict[str, Any]:
        """Returns available state/province subdivisions for a country.
        
        Args:
            country: ISO 3166-1 alpha-2 country code (e.g., 'US')
            
        Returns:
            Dict with 'country' and 'subdivisions' list
        """
        return self._call("get_subdivisions_by_country_code", method="GET", api_key=self.api_key, country=country)

    def get_aircraft_by_country_subdivision(self, country: str, subdivision: str) -> Dict[str, Any]:
        """Returns aircraft within a specific state or province.
        
        Args:
            country: ISO 3166-1 alpha-2 country code (e.g., 'US')
            subdivision: ISO 3166-2 subdivision code (e.g., 'CA' or 'US-CA')
            
        Returns:
            Dict with 'aircraft' list, 'total' count, and 'now' timestamp
        """
        return self._call("get_aircraft_by_country_subdivision", method="GET", 
                         api_key=self.api_key, country=country, subdivision=subdivision)


def main():
    """Demonstrate a practical workflow for monitoring aircraft."""
    # Initialize the client
    client = ParseClient()
    
    # Step 1: Get the total count of aircraft worldwide
    print("=" * 60)
    print("STEP 1: Checking global aircraft count")
    print("=" * 60)
    count_response = client.get_total_aircraft_count()
    total_aircraft = count_response.get("total", 0)
    print(f"Total aircraft currently tracked: {total_aircraft}")
    print(f"Data timestamp: {count_response.get('now')}\n")
    
    # Step 2: Get aircraft near a specific airport (SFO)
    print("=" * 60)
    print("STEP 2: Finding aircraft near San Francisco International (KSFO)")
    print("=" * 60)
    airport_response = client.get_aircraft_near_airport("KSFO")
    aircraft_near_sfo = airport_response.get("aircraft", [])
    print(f"Aircraft near KSFO: {airport_response.get('total', 0)}")
    
    if aircraft_near_sfo:
        # Display details of first few aircraft
        for i, aircraft in enumerate(aircraft_near_sfo[:3], 1):
            print(f"\n  Aircraft {i}:")
            print(f"    Callsign: {aircraft.get('callsign', 'N/A')}")
            print(f"    Hex: {aircraft.get('hex', 'N/A')}")
            print(f"    Altitude: {aircraft.get('alt_baro', 'N/A')} ft")
            print(f"    Ground Speed: {aircraft.get('gs', 'N/A')} knots")
            print(f"    Position: ({aircraft.get('lat', 'N/A')}, {aircraft.get('lon', 'N/A')})")
            print(f"    Heading: {aircraft.get('track', 'N/A')}°")
        
        if len(aircraft_near_sfo) > 3:
            print(f"\n  ... and {len(aircraft_near_sfo) - 3} more aircraft")
    
    # Step 3: Search for aircraft by callsign
    print("\n" + "=" * 60)
    print("STEP 3: Searching for specific flight (United Airlines UAL123)")
    print("=" * 60)
    callsign_response = client.get_aircraft_by_callsign("UAL123")
    aircraft_by_callsign = callsign_response.get("aircraft", [])
    
    if aircraft_by_callsign:
        print(f"Found {len(aircraft_by_callsign)} aircraft with callsign UAL123")
        for aircraft in aircraft_by_callsign:
            print(f"  Hex: {aircraft.get('hex', 'N/A')}")
            print(f"  Altitude: {aircraft.get('alt_baro', 'N/A')} ft")
    else:
        print("No aircraft found with callsign UAL123 (expected - using demo callsign)")
    
    # Step 4: Get aircraft within a radius (around San Francisco)
    print("\n" + "=" * 60)
    print("STEP 4: Finding aircraft within 25 nautical miles of San Francisco")
    print("=" * 60)
    # San Francisco coordinates: 37.7749, -122.4194
    radius_response = client.get_aircraft_in_radius(lat=37.7749, lon=-122.4194, dist=25)
    aircraft_in_radius = radius_response.get("aircraft", [])
    print(f"Aircraft within 25 nm radius: {radius_response.get('total', 0)}")
    
    # Step 5: Filter aircraft by altitude
    print("\n" + "=" * 60)
    print("STEP 5: Filtering aircraft at cruise altitude (above 30,000 ft)")
    print("=" * 60)
    # Using the filter endpoint to find high-altitude aircraft
    filters = [{"property": "alt_baro", "operator": "ge", "value": 30000}]
    filter_response = client.filter_aircraft(filters=filters, logical_operator="and")
    high_altitude_aircraft = filter_response.get("aircraft", [])
    print(f"Aircraft above 30,000 ft: {filter_response.get('total', 0)}")
    
    if high_altitude_aircraft:
        # Show statistics
        altitudes = [a.get('alt_baro', 0) for a in high_altitude_aircraft if a.get('alt_baro')]
        speeds = [a.get('gs', 0) for a in high_altitude_aircraft if a.get('gs')]
        
        if altitudes:
            avg_altitude = sum(altitudes) / len(altitudes)
            print(f"  Average altitude: {avg_altitude:.0f} ft")
        
        if speeds:
            avg_speed = sum(speeds) / len(speeds)
            print(f"  Average ground speed: {avg_speed:.0f} knots")
    
    # Step 6: Get subdivisions for a country and then get aircraft in a state
    print("\n" + "=" * 60)
    print("STEP 6: Getting aircraft in California (US-CA)")
    print("=" * 60)
    
    subdivisions_response = client.get_subdivisions_by_country_code("US")
    subdivisions = subdivisions_response.get("subdivisions", [])
    
    # Find California in the list
    ca_code = None
    for subdivision in subdivisions:
        if subdivision.get("name") == "California":
            ca_code = subdivision.get("iso_3166_2")
            break
    
    if ca_code:
        print(f"Found California code: {ca_code}")
        ca_response = client.get_aircraft_by_country_subdivision("US", ca_code)
        ca_aircraft_count = ca_response.get("total", 0)
        print(f"Aircraft currently in California: {ca_aircraft_count}")
    
    print("\n" + "=" * 60)
    print("Workflow complete!")
    print("=" * 60)


if __name__ == "__main__":
    main()