Skip to content

Instantly share code, notes, and snippets.

@kylemcdonald
Created November 6, 2023 22:26
Show Gist options
  • Save kylemcdonald/3cdde0e534238b022b2c3a622dabda52 to your computer and use it in GitHub Desktop.
Save kylemcdonald/3cdde0e534238b022b2c3a622dabda52 to your computer and use it in GitHub Desktop.
Download ADSB Exchange data from the heatmap endpoint.
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment