Download historical data from ADS-B Exchange.

adsbx.py

import requests
import os
import argparse
import datetime
import time

parser = argparse.ArgumentParser(
    description='Download historical data from ADS-B Exchange.')
parser.add_argument('-i', '--icao', default='icao.txt', type=str,
                    help='Newline delimited text file of ICAO hex codes.')
parser.add_argument('-c', '--cache', default='cache',
                    type=str, help='Cache/download directory name.')
parser.add_argument('-d', '--days', default=365, type=int,
                    help='How many days to download.')
args = parser.parse_args()


def log(*args):
    cur_time = datetime.datetime.today().time().replace(microsecond=0)
    print(cur_time, *args)


with open(args.icao) as f:
    icao_list = f.read().splitlines()

yesterday = datetime.date.today() - datetime.timedelta(days=1)

request_rate = 5
ratelimit_wait = 30

total_start_time = time.time()
total_collected = 0
recent_start_time = time.time()
recent_collected = 0
for days in range(args.days):
    date = yesterday - datetime.timedelta(days=days)
    year, month, day = date.year, date.month, date.day

    for icao in icao_list:
        icao = icao.lower()

        directory = os.path.join(args.cache, icao)
        fn = os.path.join(directory, f'{year}-{month:02d}-{day:02d}.json')

        if os.path.exists(fn):
            # log(fn, 'cached')
            continue

        while True:
            time.sleep(request_rate)
            url = f'https://globe.adsbexchange.com/globe_history/{year}/{month:02d}/{day:02d}/traces/{icao[-2:]}/trace_full_{icao}.json'
            res = requests.get(url, headers={
                'referer': 'https://globe.adsbexchange.com/'
            })
            total_collected += 1
            recent_collected += 1

            log(fn, res.status_code)

            if res.status_code == 429:
                cur_time = time.time()
                total_duration = (cur_time - total_start_time)
                total_rps = total_collected / total_duration
                log(f'total collected {total_collected} in {total_duration:0.2f} =  {total_rps:0.2f} req/sec')
                recent_duration = (cur_time - recent_start_time)
                recent_rps = recent_collected / recent_duration
                log(f'recent collected {recent_collected} in {recent_duration:0.2f} =  {recent_rps:0.2f} req/sec')
                log('waiting and retrying', ratelimit_wait)
                time.sleep(ratelimit_wait)

                recent_start_time = time.time()
                recent_collected = 0
            else:
                break

        os.makedirs(directory, exist_ok=True)
        with open(fn, 'wb') as f:
            if res.status_code == 200:
                f.write(res.content)