Download ADSB Exchange data from the heatmap endpoint.

download-heatmaps.py

import argparse
import datetime
import urllib3
import os
from ratelimit import limits, sleep_and_retry
from tqdm import tqdm
import random
import time

domain = "globe.adsbexchange.com"
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
connection_pool = urllib3.HTTPSConnectionPool(domain, cert_reqs='CERT_NONE')

@sleep_and_retry
@limits(calls=1, period=5)
def actually_download(url, fn):
    directory = os.path.dirname(fn)
    os.makedirs(directory, exist_ok=True)
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
    }
    
    sleep_time = 1
    while True:
        r = connection_pool.urlopen('GET', url, headers=headers)
        with open(fn, 'wb') as f:
            if r.status == 200:
                f.write(r.data)
            elif r.status != 404:
                if args.retry:
                    sleep_time *= 1 + random.random()
                    time.sleep(sleep_time)
                    continue
                print('Error {} requesting {}'.format(r.status, url))
        return
 
def download(url, fn):
    if os.path.isfile(fn):
        return
    actually_download(url, fn)
    
def get_adbsx_url(year, month, day, index):
    return f"https://globe.adsbexchange.com/globe_history/{year}/{month:02d}/{day:02d}/heatmap/{index:02d}.bin.ttf"
    
def download_adbsx(date, index):
    url = get_adbsx_url(date.year, date.month, date.day, index)
    filename = url.split("//")[1]
    download(url, filename)
        
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Download ADS-B Exchange heatmap data')
    parser.add_argument('start_date', type=str, help='Start date in YYYY-MM-DD format')
    parser.add_argument('end_date', type=str, help='End date in YYYY-MM-DD format')
    args = parser.parse_args()

    start_date = datetime.datetime.strptime(args.start_date, '%Y-%m-%d')
    end_date = datetime.datetime.strptime(args.end_date, '%Y-%m-%d')

    current_date = start_date
    all_indices = []
    while current_date < end_date:
        for index in range(48):
            all_indices.append((current_date, index))
        current_date += datetime.timedelta(days=1)

    try:
        for date, index in tqdm(all_indices):
            download_adbsx(date, index)
    except KeyboardInterrupt:
        pass