From 34df1d2ad80a4c1254c589e924b22818277ec61a Mon Sep 17 00:00:00 2001 From: TheFGFSEagle Date: Sat, 30 Jul 2022 17:39:21 +0200 Subject: [PATCH] Added FG tile span and index generator functions and script to pull apt.dat and apt.txt files from the XPlane gateway --- scenery/pull_xplane_aptdat.py | 131 ++++++++++++++++++++++++++++++++++ utils/__init__.py | 39 ++++++++++ 2 files changed, 170 insertions(+) create mode 100755 scenery/pull_xplane_aptdat.py diff --git a/scenery/pull_xplane_aptdat.py b/scenery/pull_xplane_aptdat.py new file mode 100755 index 0000000..8ca661b --- /dev/null +++ b/scenery/pull_xplane_aptdat.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python +#-*- coding:utf-8 -*- + +import os +import sys +import argparse +import json +import io +import zipfile +import base64 +import requests +import appdirs + +def get_airports_list(): + os.makedirs(appdirs.user_cache_dir("fgtools"), exist_ok=True) + airports_json_path = os.path.join(appdirs.user_cache_dir("fgtools"), "airports.json") + if not os.path.isfile(airports_json_path): + airports_json = requests.get("https://gateway.x-plane.com/apiv1/airports").json() + with open(airports_json_path, "w") as f: + json.dump(airports_json, f) + else: + with open(airports_json_path, "r") as f: + airports_json = json.load(f) + + return airports_json["total"], airports_json["airports"] + +def filter_airports_list(airports_count, airports_list, icaos, bbox): + airports_list_filtered = [] + i = 0 + for airport in airports_list: + print(f"Filtering airports … {i / airports_count * 100:.1f}% ({i} of {airports_count})\r", end="") + airport_metadata = airport.get("metadata", {}) or {} + if icaos and any(icao in (airport.get("AirportCode", None), airport_metadata.get("icao_code", None)) for icao in icaos): + airports_list_filtered.append(airport) + elif bbox: + if bbox[0] < airport["Latitude"] < bbox[2] and bbox[1] < airport["Longitude"] < bbox[3]: + airports_list_filtered.append(airport) + i += 1 + print("Filtering airports … done ") + + return airports_list_filtered + +def write_aptdat_files(airports_list, output, txt_output): + airports_count = len(airports_list) + i = 0 + for airport in airports_list: + print(f"Downloading and writing airports … {i / airports_count * 100:.1f}% ({i} of {airports_count})\r", end="") + airport_metadata = airport.get("metadata", {}) or {} + icao_code = airport_metadata.get("icao_code", None) or airport["AirportCode"] + + scenery_id = airport.get("RecommendedSceneryId", None) + if not scenery_id: + print("Airport", icao_code, "has no scenery - skipping ") + continue + + scenery_json = requests.get("https://gateway.x-plane.com/apiv1/scenery/" + str(scenery_id)).json() + scenery_blob = io.BytesIO(base64.b64decode(scenery_json["scenery"]["masterZipBlob"])) + with zipfile.ZipFile(scenery_blob, "r") as scenery_zip: + try: + scenery_dat = scenery_zip.open(icao_code + ".dat") + except KeyError: + pass + else: + with open(os.path.join(output, icao_code + ".dat"), "wb") as aptdat: + aptdat.write(scenery_dat.read()) + + if txt_output and os.path.isdir(txt_output): + try: + scenery_txt = scenery_zip.open(icao_code + ".txt") + except KeyError: + pass + else: + with open(os.path.join(txt_output, icao_code + ".txt"), "wb") as apttxt: + apttxt.write(scenery_txt.read()) + + i += 1 + print("Downloading and writing airports … done ") + +if __name__ == "__main__": + argp = argparse.ArgumentParser(description="Pulls apt.dat files from the XPlane Gateway selected either by a bounding box or ICAO codes") + + argp.add_argument( + "-i", "--icao", + help="ICAO code(s) of the apt.dat files to be pulled", + nargs="+" + ) + + argp.add_argument( + "-b", "--bbox", + help="All airports in the given bounding box will be pulled", + nargs=4, + type=float + ) + + argp.add_argument( + "-o", "--output", + help="Directory to put apt.dat files in", + required=True + ) + + argp.add_argument( + "-q", "--query", + help="Don't write apt.dat files, only output a list of found airports", + action="store_true" + ) + + argp.add_argument( + "-t", "--txt-output", + help="Where to write .txt files to (.txt files won't be written if this option is not specified or the directory doesn't exist)", + default="" + ) + + args = argp.parse_args() + + if not args.icao and not args.bbox: + argp.error("At least one of -i/--icao and -b/--bbox must be given !") + + if not os.path.isdir(args.output): + argp.error(f"Output folder {args.output} does not exist - exiting") + + + airports_count, airports_list = get_airports_list() + airports_list_filtered = filter_airports_list(airports_count, airports_list, args.icao, args.bbox) + + if args.query: + print("Airport code ICAO code") + for airport in airports_list_filtered: + print(airport["AirportCode"] + "\t", (airport.get("metadata", {}) or {}).get("icao_code", "\t"), sep="\t") + + write_aptdat_files(airports_list_filtered, args.output, args.txt_output) + diff --git a/utils/__init__.py b/utils/__init__.py index e69de29..a43258d 100755 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python +#-*- coding:utf-8 -*- + +import math + +def get_fg_tile_span(lat): + if lat >= 89: + return 12 + elif lat >= 86: + return 4 + elif lat >= 83: + return 2 + elif lat >= 76: + return 1 + elif lat >= 62: + return 0.5 + elif lat >= 22: + return 0.25 + elif lat >= -22: + return 0.125 + elif lat >= -62: + return 0.25 + elif lat >= -76: + return 0.5 + elif lat >= -83: + return 1 + elif lat >= -86 + return 2 + elif lat >= -89 + return 4 + else: + return 12 + +def get_fg_tile_index(lon, lat): + tile_width = get_fg_tile_span(lat) + x = math.floor((lon - math.floor(math.floor(lon / tile_width) * tile_width)) / tile_width) + y = trunc((lat - math.floor(lat)) + 8) + return ((lon + 180) << 14) + ((lat + 90) << 6) + (y << 3) + x +