#!/usr/bin/env python3 # build_navdb.py: builds the Flight Display Pro navigation database from public sources. # Copyright (C) 2026 Tanut Apiwong # # This script is free software; you can redistribute it and/or modify it under the # terms of the GNU General Public License as published by the Free Software # Foundation, either version 2 of the License, or (at your option) any later # version. It is distributed WITHOUT ANY WARRANTY. See # https://www.gnu.org/licenses/old-licenses/gpl-2.0.html # # It is published at https://tanutapi.dev/flight-display-pro/navdata/ as the source # form of the FlightGear (GPL-2.0-or-later) data bundled in the app. """Builds the navigation database bundled with the iOS app. Reads free, redistributable aeronautical and geographic data, normalises it and writes Resources/NavData/navdata.sqlite exactly per Resources/NavData/SCHEMA.md. Downloads are cached in Tools/.navdata-cache/. Python 3.10 stdlib only. Usage: python3 Tools/build_navdb.py [--out PATH] [--cache DIR] [--basemap 10m|50m] [--flightgear] [--xplane-dir DIR] [--faa-cifp PATH] [--openaip-key KEY | --openaip-key-file PATH | OPENAIP_API_KEY=...] [--openaip-countries TH,US,... | --openaip-countries all] Sources (see Resources/NavData/LICENSES.md): OurAirports (required) airports, runways, frequencies, VOR/NDB/DME CC0 Natural Earth (required) coastlines, borders, lakes public domain --flightgear fgdata nav/fix/awy.dat (X-Plane 810 format) GPL-2.0+, stale (2013) --xplane-dir DIR nav.dat / fix.dat / awy.dat (810 .. 1150) license of the data owner --faa-cifp PATH FAA CIFP ARINC 424 (US waypoints, airways) public domain --openaip-key KEY openAIP airspaces and reporting points CC BY-NC 4.0 """ import argparse import csv import datetime as dt import gzip import io import json import math import os import re import sqlite3 import struct import sys import urllib.request import zipfile TOOL_VERSION = "1.0" SCHEMA_VERSION = "1" OURAIRPORTS_BASE = "https://davidmegginson.github.io/ourairports-data/" NATURAL_EARTH_BASE = "https://raw.githubusercontent.com/nvkelso/natural-earth-vector/master/geojson/" FLIGHTGEAR_BASE = "https://sourceforge.net/p/flightgear/fgdata/ci/next/tree/Navaids/" OPENAIP_BASE = "https://api.core.openaip.net/api/" NAVAID_TYPES = {"VOR", "VOR-DME", "VORTAC", "DME", "NDB", "NDB-DME", "TACAN"} HARD_SURFACE_PREFIXES = ("ASP", "CON", "PEM", "BIT", "ASPH", "CONC", "PAVED", "TAR", "MAC") DP_TOLERANCE_DEG = 0.001 CHUNK_POINTS = 256 # ---------------------------------------------------------------- utilities def log(msg): print(msg, flush=True) def fetch(url, cache_dir, name, headers=None, refresh=False): """Downloads url to cache_dir/name unless cached; returns the local path.""" os.makedirs(cache_dir, exist_ok=True) path = os.path.join(cache_dir, name) if os.path.exists(path) and not refresh: return path log(f" fetching {url}") req = urllib.request.Request(url, headers=headers or {"User-Agent": "flight-display-navdb/" + TOOL_VERSION}) with urllib.request.urlopen(req, timeout=120) as resp, open(path + ".part", "wb") as out: while True: chunk = resp.read(1 << 20) if not chunk: break out.write(chunk) os.replace(path + ".part", path) return path def read_text_maybe_gz(path): """Returns the text of a file that may or may not actually be gzipped.""" with open(path, "rb") as f: data = f.read() if data[:2] == b"\x1f\x8b": data = gzip.decompress(data) return data.decode("utf-8", errors="replace") def ffloat(s, default=None): try: return float(s) except (TypeError, ValueError): return default def fint(s, default=None): try: return int(float(s)) except (TypeError, ValueError): return default def pack_points(points): """Little-endian float32 (lat, lon) pairs.""" flat = [] for lat, lon in points: flat.append(lat) flat.append(lon) return struct.pack("<%df" % len(flat), *flat) def bbox(points): lats = [p[0] for p in points] lons = [p[1] for p in points] return min(lats), max(lats), min(lons), max(lons) def douglas_peucker(points, tol): """Iterative Douglas-Peucker on (lat, lon) tuples; keeps endpoints.""" n = len(points) if n < 3: return list(points) keep = [False] * n keep[0] = keep[-1] = True stack = [(0, n - 1)] tol2 = tol * tol while stack: a, b = stack.pop() if b - a < 2: continue ax, ay = points[a][1], points[a][0] bx, by = points[b][1], points[b][0] dx, dy = bx - ax, by - ay seg2 = dx * dx + dy * dy best, best_i = 0.0, -1 for i in range(a + 1, b): px, py = points[i][1], points[i][0] if seg2 == 0: d2 = (px - ax) ** 2 + (py - ay) ** 2 else: t = ((px - ax) * dx + (py - ay) * dy) / seg2 t = 0.0 if t < 0 else 1.0 if t > 1 else t cx, cy = ax + t * dx, ay + t * dy d2 = (px - cx) ** 2 + (py - cy) ** 2 if d2 > best: best, best_i = d2, i if best > tol2 and best_i > 0: keep[best_i] = True stack.append((a, best_i)) stack.append((best_i, b)) return [p for p, k in zip(points, keep) if k] def chunk_polyline(points, limit=CHUNK_POINTS): """Splits a polyline into pieces of at most `limit` points sharing boundary points.""" if len(points) <= limit: yield points return i = 0 while i < len(points) - 1: piece = points[i:i + limit] if len(piece) < 2: break yield piece i += limit - 1 # ---------------------------------------------------------------- database def create_schema(db): db.executescript(""" CREATE TABLE meta(key TEXT PRIMARY KEY, value TEXT); CREATE TABLE source(id INTEGER PRIMARY KEY, name TEXT, license TEXT, url TEXT, cycle TEXT, fetched TEXT); CREATE TABLE airport(id INTEGER PRIMARY KEY, ident TEXT, icao TEXT, iata TEXT, name TEXT, type TEXT, lat REAL, lon REAL, elev_ft REAL, country TEXT, longest_runway_ft REAL, hard_surface INTEGER, source_id INTEGER); CREATE VIRTUAL TABLE airport_rt USING rtree(id, minlat, maxlat, minlon, maxlon); CREATE TABLE runway(id INTEGER PRIMARY KEY, airport_id INTEGER, le_ident TEXT, he_ident TEXT, length_ft REAL, width_ft REAL, surface TEXT, closed INTEGER, le_lat REAL, le_lon REAL, he_lat REAL, he_lon REAL, le_heading_true REAL); CREATE TABLE frequency(id INTEGER PRIMARY KEY, airport_id INTEGER, type TEXT, description TEXT, mhz REAL); CREATE TABLE navaid(id INTEGER PRIMARY KEY, ident TEXT, name TEXT, type TEXT, lat REAL, lon REAL, elev_ft REAL, freq_khz INTEGER, mag_var REAL, range_nm REAL, country TEXT, source_id INTEGER); CREATE VIRTUAL TABLE navaid_rt USING rtree(id, minlat, maxlat, minlon, maxlon); CREATE TABLE waypoint(id INTEGER PRIMARY KEY, ident TEXT, lat REAL, lon REAL, region TEXT, country TEXT, kind TEXT, source_id INTEGER); CREATE VIRTUAL TABLE waypoint_rt USING rtree(id, minlat, maxlat, minlon, maxlon); CREATE TABLE airway(id INTEGER PRIMARY KEY, name TEXT, level TEXT, source_id INTEGER); CREATE TABLE airway_leg(id INTEGER PRIMARY KEY, airway_id INTEGER, seq INTEGER, from_ident TEXT, from_lat REAL, from_lon REAL, to_ident TEXT, to_lat REAL, to_lon REAL); CREATE VIRTUAL TABLE airway_leg_rt USING rtree(id, minlat, maxlat, minlon, maxlon); CREATE TABLE airspace(id INTEGER PRIMARY KEY, name TEXT, class TEXT, type TEXT, floor_ft INTEGER, floor_ref TEXT, ceiling_ft INTEGER, ceiling_ref TEXT, country TEXT, points BLOB, source_id INTEGER); CREATE VIRTUAL TABLE airspace_rt USING rtree(id, minlat, maxlat, minlon, maxlon); CREATE TABLE basemap(id INTEGER PRIMARY KEY, kind TEXT, points BLOB); CREATE VIRTUAL TABLE basemap_rt USING rtree(id, minlat, maxlat, minlon, maxlon); """) def create_indexes(db): db.executescript(""" CREATE INDEX airport_ident ON airport(ident); CREATE INDEX airport_icao ON airport(icao); CREATE INDEX runway_airport ON runway(airport_id); CREATE INDEX frequency_airport ON frequency(airport_id); CREATE INDEX navaid_ident ON navaid(ident); CREATE INDEX waypoint_ident ON waypoint(ident); CREATE INDEX airway_leg_airway ON airway_leg(airway_id); CREATE INDEX airway_name ON airway(name); """) class NavDB: """Thin insert helper that keeps R*Tree rows in step with the data rows.""" def __init__(self, path): if os.path.exists(path): os.remove(path) self.db = sqlite3.connect(path) self.db.execute("PRAGMA journal_mode=OFF") self.db.execute("PRAGMA synchronous=OFF") create_schema(self.db) # ident -> [(lat, lon)] of navaids already stored, for de-duplication self.navaid_index = {} self.airway_ids = {} def add_source(self, name, license_, url, cycle): cur = self.db.execute("INSERT INTO source(name, license, url, cycle, fetched) VALUES (?,?,?,?,?)", (name, license_, url, cycle, dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"))) return cur.lastrowid def add_airport(self, row): cur = self.db.execute( "INSERT INTO airport(ident, icao, iata, name, type, lat, lon, elev_ft, country, longest_runway_ft, hard_surface, source_id)" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", row) aid = cur.lastrowid lat, lon = row[5], row[6] self.db.execute("INSERT INTO airport_rt VALUES (?,?,?,?,?)", (aid, lat, lat, lon, lon)) return aid def has_navaid(self, ident, lat, lon, tol=0.05): for plat, plon in self.navaid_index.get(ident, ()): if abs(plat - lat) <= tol and abs(plon - lon) <= tol: return True return False def add_navaid(self, ident, name, type_, lat, lon, elev_ft, freq_khz, mag_var, range_nm, country, source_id): cur = self.db.execute( "INSERT INTO navaid(ident, name, type, lat, lon, elev_ft, freq_khz, mag_var, range_nm, country, source_id)" " VALUES (?,?,?,?,?,?,?,?,?,?,?)", (ident, name, type_, lat, lon, elev_ft, freq_khz, mag_var, range_nm, country, source_id)) nid = cur.lastrowid self.db.execute("INSERT INTO navaid_rt VALUES (?,?,?,?,?)", (nid, lat, lat, lon, lon)) self.navaid_index.setdefault(ident, []).append((lat, lon)) return nid def add_waypoint(self, ident, lat, lon, region, country, kind, source_id): cur = self.db.execute( "INSERT INTO waypoint(ident, lat, lon, region, country, kind, source_id) VALUES (?,?,?,?,?,?,?)", (ident, lat, lon, region, country, kind, source_id)) wid = cur.lastrowid self.db.execute("INSERT INTO waypoint_rt VALUES (?,?,?,?,?)", (wid, lat, lat, lon, lon)) return wid def airway_id(self, name, level, source_id): key = (name, level, source_id) aid = self.airway_ids.get(key) if aid is None: cur = self.db.execute("INSERT INTO airway(name, level, source_id) VALUES (?,?,?)", (name, level, source_id)) aid = cur.lastrowid self.airway_ids[key] = aid return aid def add_airway_leg(self, airway_id, seq, from_ident, from_lat, from_lon, to_ident, to_lat, to_lon): cur = self.db.execute( "INSERT INTO airway_leg(airway_id, seq, from_ident, from_lat, from_lon, to_ident, to_lat, to_lon)" " VALUES (?,?,?,?,?,?,?,?)", (airway_id, seq, from_ident, from_lat, from_lon, to_ident, to_lat, to_lon)) lid = cur.lastrowid self.db.execute("INSERT INTO airway_leg_rt VALUES (?,?,?,?,?)", (lid, min(from_lat, to_lat), max(from_lat, to_lat), min(from_lon, to_lon), max(from_lon, to_lon))) return lid # About 50 m; airspace circles from openAIP arrive with hundreds of # points and would otherwise dominate the file. AIRSPACE_TOLERANCE_DEG = 0.0005 def add_airspace(self, name, class_, type_, floor_ft, floor_ref, ceiling_ft, ceiling_ref, country, points, source_id): if len(points) < 3: return None if points[0] != points[-1]: points = list(points) + [points[0]] points = douglas_peucker(points, self.AIRSPACE_TOLERANCE_DEG) if len(points) < 4: return None cur = self.db.execute( "INSERT INTO airspace(name, class, type, floor_ft, floor_ref, ceiling_ft, ceiling_ref, country, points, source_id)" " VALUES (?,?,?,?,?,?,?,?,?,?)", (name, class_, type_, floor_ft, floor_ref, ceiling_ft, ceiling_ref, country, pack_points(points), source_id)) sid = cur.lastrowid self.db.execute("INSERT INTO airspace_rt VALUES (?,?,?,?,?)", (sid, *bbox(points))) return sid def add_basemap(self, kind, points): if len(points) < 2: return None cur = self.db.execute("INSERT INTO basemap(kind, points) VALUES (?,?)", (kind, pack_points(points))) bid = cur.lastrowid self.db.execute("INSERT INTO basemap_rt VALUES (?,?,?,?,?)", (bid, *bbox(points))) return bid def finish(self): self.db.execute("INSERT INTO meta VALUES ('schema_version', ?)", (SCHEMA_VERSION,)) self.db.execute("INSERT INTO meta VALUES ('tool_version', ?)", (TOOL_VERSION,)) self.db.execute("INSERT INTO meta VALUES ('built_at', ?)", (dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),)) create_indexes(self.db) self.db.commit() self.db.execute("VACUUM") self.db.close() # ---------------------------------------------------------------- OurAirports def import_ourairports(nav, cache): log("OurAirports") files = {} for name in ("airports", "runways", "navaids", "airport-frequencies"): files[name] = fetch(OURAIRPORTS_BASE + name + ".csv", cache, "ourairports_" + name + ".csv") cycle = dt.date.today().isoformat() sid = nav.add_source("OurAirports", "CC0 1.0 (public domain)", "https://ourairports.com/data/", cycle) # Runways first so airport rows can carry the longest runway and surface. runways_by_airport = {} with open(files["runways"], encoding="utf-8", newline="") as f: for r in csv.DictReader(f): runways_by_airport.setdefault(r["airport_ref"], []).append(r) airport_ids = {} # OurAirports id -> our id with open(files["airports"], encoding="utf-8", newline="") as f: for r in csv.DictReader(f): lat, lon = ffloat(r["latitude_deg"]), ffloat(r["longitude_deg"]) if lat is None or lon is None: continue rws = runways_by_airport.get(r["id"], []) longest = max((ffloat(x["length_ft"], 0) or 0 for x in rws if x["closed"] != "1"), default=None) hard = 1 if any((x["surface"] or "").upper().startswith(HARD_SURFACE_PREFIXES) for x in rws) else 0 aid = nav.add_airport((r["ident"], r["icao_code"] or r["gps_code"] or None, r["iata_code"] or None, r["name"], r["type"], lat, lon, ffloat(r["elevation_ft"]), r["iso_country"], longest, hard, sid)) airport_ids[r["id"]] = aid for x in rws: nav.db.execute( "INSERT INTO runway(airport_id, le_ident, he_ident, length_ft, width_ft, surface, closed," " le_lat, le_lon, he_lat, he_lon, le_heading_true) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (aid, x["le_ident"], x["he_ident"], ffloat(x["length_ft"]), ffloat(x["width_ft"]), x["surface"], 1 if x["closed"] == "1" else 0, ffloat(x["le_latitude_deg"]), ffloat(x["le_longitude_deg"]), ffloat(x["he_latitude_deg"]), ffloat(x["he_longitude_deg"]), ffloat(x["le_heading_degT"]))) with open(files["airport-frequencies"], encoding="utf-8", newline="") as f: for r in csv.DictReader(f): aid = airport_ids.get(r["airport_ref"]) mhz = ffloat(r["frequency_mhz"]) if aid is None or mhz is None: continue nav.db.execute("INSERT INTO frequency(airport_id, type, description, mhz) VALUES (?,?,?,?)", (aid, r["type"], r["description"], mhz)) with open(files["navaids"], encoding="utf-8", newline="") as f: for r in csv.DictReader(f): if r["type"] not in NAVAID_TYPES: continue lat, lon = ffloat(r["latitude_deg"]), ffloat(r["longitude_deg"]) if lat is None or lon is None: continue freq = fint(r["frequency_khz"]) if freq is None and r["type"] == "DME": freq = fint(r["dme_frequency_khz"]) nav.add_navaid(r["ident"], r["name"], r["type"], lat, lon, ffloat(r["elevation_ft"]), freq, ffloat(r["magnetic_variation_deg"]), None, r["iso_country"], sid) nav.db.commit() # ---------------------------------------------------------------- Natural Earth def import_natural_earth(nav, cache, scale): log(f"Natural Earth {scale}") sid = nav.add_source("Natural Earth", "Public domain", "https://www.naturalearthdata.com/", "v5") layers = (("coastline", f"ne_{scale}_coastline.geojson"), ("boundary", f"ne_{scale}_admin_0_boundary_lines_land.geojson"), ("lake", f"ne_{scale}_lakes.geojson")) for kind, fname in layers: path = fetch(NATURAL_EARTH_BASE + fname, cache, fname) with open(path, encoding="utf-8") as f: data = json.load(f) n_in = n_out = 0 for feat in data.get("features", []): geom = feat.get("geometry") or {} for line in geometry_polylines(geom): pts = [(float(lat), float(lon)) for lon, lat, *_ in line] n_in += len(pts) pts = douglas_peucker(pts, DP_TOLERANCE_DEG) for piece in chunk_polyline(pts): if nav.add_basemap(kind, piece) is not None: n_out += len(piece) log(f" {kind}: {n_in} -> {n_out} points") nav.db.commit() return sid def geometry_polylines(geom): t = geom.get("type") c = geom.get("coordinates", []) if t == "LineString": yield c elif t == "MultiLineString": yield from c elif t == "Polygon": yield from c elif t == "MultiPolygon": for poly in c: yield from poly elif t == "GeometryCollection": for g in geom.get("geometries", []): yield from geometry_polylines(g) # ---------------------------------------------------------------- X-Plane format def xplane_version(lines): """Returns the data-format version and the header cycle string of an X-Plane .dat.""" for line in lines[:3]: m = re.match(r"\s*(\d{3,4})\s+Version", line) if m: cyc = re.search(r"data cycle\s+([0-9.]+)", line) return int(m.group(1)), (cyc.group(1) if cyc else "unknown") return 0, "unknown" def parse_xplane_nav(text): """Yields (type, lat, lon, elev_ft, freq_khz, range_nm, mag_var, ident, region, name).""" lines = text.splitlines() version, cycle = xplane_version(lines) for line in lines: parts = line.split() if len(parts) < 8 or not parts[0].isdigit(): continue code = int(parts[0]) if code not in (2, 3, 12, 13): continue lat, lon = ffloat(parts[1]), ffloat(parts[2]) if lat is None or lon is None: continue elev = ffloat(parts[3]) freq = fint(parts[4]) rng = ffloat(parts[5]) var = ffloat(parts[6]) ident = parts[7] if version >= 1100: # ident airport_or_ENRT icao_region name... region = parts[9] if len(parts) > 9 else "" name = " ".join(parts[10:]) else: region = "" name = " ".join(parts[8:]) if code == 2: type_ = "NDB" elif code == 3: type_ = "VOR" freq = freq * 10 if freq else freq # VOR freq is given in 10 kHz units (11350 -> 113500) else: type_ = "DME" freq = freq * 10 if freq else freq uname = name.upper() if "ILS" in uname or "LOC" in uname or "LDA" in uname or "SDF" in uname: continue # DME paired with an approach aid if "VORTAC" in uname: type_ = "VORTAC" elif "TACAN" in uname: type_ = "TACAN" elif "VOR" in uname: type_ = "VOR-DME" elif "NDB" in uname: type_ = "NDB-DME" yield type_, lat, lon, elev, freq, rng, var, ident, region, name, cycle def parse_xplane_fix(text): """Yields (lat, lon, ident, region).""" lines = text.splitlines() version, cycle = xplane_version(lines) for line in lines: parts = line.split() if len(parts) < 3: continue lat, lon = ffloat(parts[0]), ffloat(parts[1]) if lat is None or lon is None or parts[2] == "99": continue ident = parts[2] region = parts[4] if version >= 1100 and len(parts) > 4 else "" yield lat, lon, ident, region, cycle def parse_xplane_awy(text, resolve): """Yields (name, level, from_ident, from_lat, from_lon, to_ident, to_lat, to_lon). `resolve(ident, region)` returns (lat, lon) or None for the 1100 format.""" lines = text.splitlines() version, cycle = xplane_version(lines) for line in lines: parts = line.split() if len(parts) < 9 or parts[0] == "99": continue if version >= 1100: if len(parts) < 11: continue f_id, f_reg, t_id, t_reg = parts[0], parts[1], parts[3], parts[4] level_code, names = parts[7], parts[10] a = resolve(f_id, f_reg) b = resolve(t_id, t_reg) if a is None or b is None: continue f_lat, f_lon = a t_lat, t_lon = b else: f_id, f_lat, f_lon, t_id, t_lat, t_lon = parts[0], ffloat(parts[1]), ffloat(parts[2]), parts[3], ffloat(parts[4]), ffloat(parts[5]) level_code, names = parts[6], parts[9] if None in (f_lat, f_lon, t_lat, t_lon): continue level = {"1": "L", "2": "H"}.get(level_code, "B") for name in names.split("-"): if name: yield name, level, f_id, f_lat, f_lon, t_id, t_lat, t_lon, cycle def import_xplane_files(nav, paths, source_name, license_, url, default_cycle): """paths: dict with optional keys nav, fix, awy -> local file path.""" cycle = default_cycle texts = {k: read_text_maybe_gz(p) for k, p in paths.items() if p and os.path.exists(p)} for t in texts.values(): v, c = xplane_version(t.splitlines()[:3]) if c != "unknown": cycle = c break sid = nav.add_source(source_name, license_, url, cycle) lookup = {} # (ident, region) and (ident, "") -> (lat, lon) n_nav = n_nav_skipped = 0 if "nav" in texts: for type_, lat, lon, elev, freq, rng, var, ident, region, name, _ in parse_xplane_nav(texts["nav"]): lookup.setdefault((ident, region), (lat, lon)) lookup.setdefault((ident, ""), (lat, lon)) if nav.has_navaid(ident, lat, lon): n_nav_skipped += 1 continue nav.add_navaid(ident, name, type_, lat, lon, elev, freq, var, rng, None, sid) n_nav += 1 n_fix = 0 if "fix" in texts: for lat, lon, ident, region, _ in parse_xplane_fix(texts["fix"]): lookup.setdefault((ident, region), (lat, lon)) lookup.setdefault((ident, ""), (lat, lon)) nav.add_waypoint(ident, lat, lon, region or None, None, "enroute", sid) n_fix += 1 n_leg = 0 if "awy" in texts: seqs = {} def resolve(ident, region): return lookup.get((ident, region)) or lookup.get((ident, "")) for name, level, f_id, f_lat, f_lon, t_id, t_lat, t_lon, _ in parse_xplane_awy(texts["awy"], resolve): aid = nav.airway_id(name, level, sid) seq = seqs.get(aid, 0) + 1 seqs[aid] = seq nav.add_airway_leg(aid, seq, f_id, f_lat, f_lon, t_id, t_lat, t_lon) n_leg += 1 nav.db.commit() log(f" {source_name}: cycle {cycle}, navaids +{n_nav} (skipped {n_nav_skipped} already present), waypoints {n_fix}, airway legs {n_leg}") def import_flightgear(nav, cache): log("FlightGear navdata (X-Plane 810 format, GPL, stale)") paths = {} for key in ("nav", "fix", "awy"): url = f"{FLIGHTGEAR_BASE}{key}.dat.gz?format=raw" try: path = fetch(url, cache, f"flightgear_{key}.dat.gz") text = read_text_maybe_gz(path) if "Version" not in text[:2000]: raise ValueError("response is not an X-Plane .dat file") paths[key] = path except Exception as e: # network or format problem: report and continue log(f" WARNING: could not fetch {url}: {e}") if not paths: log(" FlightGear data skipped") return import_xplane_files(nav, paths, "FlightGear navdata (Robin Peel / X-Plane 810)", "GPL-2.0-or-later", "https://sourceforge.net/p/flightgear/fgdata/ci/next/tree/Navaids/", "2013") def import_xplane_dir(nav, directory): log(f"X-Plane navdata from {directory}") paths = {} for key in ("nav", "fix", "awy"): for cand in (f"earth_{key}.dat", f"{key}.dat", f"earth_{key}.dat.gz", f"{key}.dat.gz"): p = os.path.join(directory, cand) if os.path.exists(p): paths[key] = p break if not paths: log(" no nav.dat / fix.dat / awy.dat found") return import_xplane_files(nav, paths, f"X-Plane format navdata ({os.path.basename(os.path.abspath(directory))})", "see data owner's license", directory, "unknown") # ---------------------------------------------------------------- FAA CIFP (ARINC 424) def arinc_lat(s): # N39514200 -> DD MM SS ss if len(s) < 9 or s[0] not in "NS": return None d, m, sec, hund = int(s[1:3]), int(s[3:5]), int(s[5:7]), int(s[7:9]) v = d + m / 60 + (sec + hund / 100) / 3600 return -v if s[0] == "S" else v def arinc_lon(s): # W104402800 -> DDD MM SS ss if len(s) < 10 or s[0] not in "EW": return None d, m, sec, hund = int(s[1:4]), int(s[4:6]), int(s[6:8]), int(s[8:10]) v = d + m / 60 + (sec + hund / 100) / 3600 return -v if s[0] == "W" else v def arinc_var(s): # E0110 / W0050 -> +11.0 / -5.0 ; T = true if len(s) < 5 or s[0] not in "EW": return None v = int(s[1:5]) / 10 return -v if s[0] == "W" else v def col(line, a, b): """1-based inclusive column slice.""" return line[a - 1:b] def import_faa_cifp(nav, path): log(f"FAA CIFP {path}") if zipfile.is_zipfile(path): with zipfile.ZipFile(path) as z: names = [n for n in z.namelist() if n.upper().startswith("FAACIFP")] if not names: log(" no FAACIFP file inside the zip") return text = z.read(names[0]).decode("ascii", errors="replace") else: text = open(path, encoding="ascii", errors="replace").read() lines = text.splitlines() cycle = "unknown" for line in lines[:10]: if line.startswith("HDR"): m = re.search(r"(\d{4})", col(line, 80, 85) or "") if m: cycle = m.group(1) if cycle == "unknown": m = re.search(r"(\d{6})", os.path.basename(path)) cycle = m.group(1) if m else "unknown" sid = nav.add_source("FAA CIFP (ARINC 424)", "Public domain (US Government)", "https://www.faa.gov/air_traffic/flight_info/aeronav/digital_products/cifp/", cycle) fixes = {} # (ident, region) -> (lat, lon) n_nav = n_wp = 0 for line in lines: if len(line) < 123 or col(line, 1, 1) != "S": continue section, sub = col(line, 5, 5), col(line, 6, 6) if section == "D" and sub == " ": if col(line, 22, 22) not in ("0", "1"): continue ident = col(line, 14, 17).strip() region = col(line, 20, 21).strip() cls = col(line, 28, 32) lat = arinc_lat(col(line, 33, 41)) lon = arinc_lon(col(line, 42, 51)) if lat is None or lon is None: lat = arinc_lat(col(line, 56, 64)) lon = arinc_lon(col(line, 65, 74)) if lat is None or lon is None: continue fixes.setdefault((ident, region), (lat, lon)) c1, c2 = cls[0], cls[1] if c2 == "I": continue # ILS DME if c1 == "V": type_ = {"D": "VOR-DME", "T": "VORTAC", "M": "VORTAC"}.get(c2, "VOR") elif c2 in ("T", "M"): type_ = "TACAN" elif c2 == "D": type_ = "DME" else: continue freq = fint(col(line, 23, 27)) freq_khz = freq * 10 if freq else None var = arinc_var(col(line, 75, 79)) name = col(line, 94, 123).strip() if not nav.has_navaid(ident, lat, lon): nav.add_navaid(ident, name, type_, lat, lon, ffloat(col(line, 80, 84)), freq_khz, var, None, "US", sid) n_nav += 1 elif section == "D" and sub == "B": if col(line, 22, 22) not in ("0", "1"): continue ident = col(line, 14, 17).strip() region = col(line, 20, 21).strip() lat = arinc_lat(col(line, 33, 41)) lon = arinc_lon(col(line, 42, 51)) if lat is None or lon is None: continue fixes.setdefault((ident, region), (lat, lon)) freq = fint(col(line, 23, 27)) freq_khz = freq // 10 if freq else None # 03620 -> 362 kHz var = arinc_var(col(line, 75, 79)) name = col(line, 94, 123).strip() if not nav.has_navaid(ident, lat, lon): nav.add_navaid(ident, name, "NDB", lat, lon, None, freq_khz, var, None, "US", sid) n_nav += 1 elif section == "E" and sub == "A": if col(line, 22, 22) not in ("0", "1"): continue ident = col(line, 14, 18).strip() region = col(line, 20, 21).strip() lat = arinc_lat(col(line, 33, 41)) lon = arinc_lon(col(line, 42, 51)) if lat is None or lon is None: continue fixes.setdefault((ident, region), (lat, lon)) nav.add_waypoint(ident, lat, lon, region or None, "US", "enroute", sid) n_wp += 1 # Airways: consecutive sequence numbers within a route form legs. routes = {} for line in lines: if len(line) < 45 or col(line, 1, 1) != "S" or col(line, 5, 5) != "E" or col(line, 6, 6) != "R": continue if col(line, 39, 39) not in ("0", "1"): continue route = col(line, 14, 18).strip() seq = fint(col(line, 26, 29)) fix_ident = col(line, 30, 34).strip() fix_region = col(line, 35, 36).strip() level = col(line, 45, 45).strip() or "B" if seq is None or not route: continue routes.setdefault((route, level), []).append((seq, fix_ident, fix_region)) n_leg = 0 for (route, level), pts in routes.items(): pts.sort() aid = None seqno = 0 for (s0, i0, r0), (s1, i1, r1) in zip(pts, pts[1:]): a = fixes.get((i0, r0)) or fixes.get((i0, "")) b = fixes.get((i1, r1)) or fixes.get((i1, "")) if a is None or b is None: continue if aid is None: aid = nav.airway_id(route, level, sid) seqno += 1 nav.add_airway_leg(aid, seqno, i0, a[0], a[1], i1, b[0], b[1]) n_leg += 1 nav.db.commit() log(f" cycle {cycle}: navaids +{n_nav}, waypoints {n_wp}, airway legs {n_leg}") # ---------------------------------------------------------------- openAIP OPENAIP_TYPES = {0: "OTHER", 1: "RESTRICTED", 2: "DANGER", 3: "PROHIBITED", 4: "CTR", 5: "TMZ", 6: "RMZ", 7: "TMA", 8: "TRA", 9: "TSA", 10: "FIR", 11: "UIR", 12: "ADIZ", 13: "ATZ", 14: "MATZ", 15: "AIRWAY", 16: "MTR", 17: "ALERT", 18: "WARNING", 19: "PROTECTED", 20: "HTZ", 21: "GLIDING", 22: "TRP", 23: "TIZ", 24: "TIA", 25: "MTA", 26: "CTA", 27: "ACC", 28: "SPORT", 29: "LOW_ALT_OVERFLIGHT_RESTRICTION"} OPENAIP_REF = {0: "GND", 1: "MSL", 2: "STD"} def openaip_limit(lim): """Returns (feet or None, reference).""" if not lim: return None, None v = lim.get("value") unit = lim.get("unit") ref = OPENAIP_REF.get(lim.get("referenceDatum"), "MSL") if v is None: return None, ref if unit == 6: ft = int(v) * 100 ref = "STD" elif unit == 0: ft = int(round(v * 3.28084)) else: ft = int(v) return ft, ref def openaip_pages(endpoint, key, country, cache): """Yields items for one country, or for the whole world when country is 'ALL'.""" page = 1 while True: filt = "" if country == "ALL" else f"country={country}&" url = f"{OPENAIP_BASE}{endpoint}?{filt}page={page}&limit=1000" path = fetch(url, cache, f"openaip_{endpoint}_{country}_{page}.json", headers={"x-openaip-api-key": key, "Accept": "application/json", "User-Agent": "flight-display-navdb/" + TOOL_VERSION}) with open(path, encoding="utf-8") as f: data = json.load(f) items = data.get("items", []) yield from items total_pages = data.get("totalPages") or 1 if page >= total_pages or not items: break page += 1 def import_openaip(nav, cache, key, countries): log(f"openAIP airspaces and reporting points for {', '.join(countries)}") sid = nav.add_source("openAIP", "CC BY-NC 4.0", "https://www.openaip.net/", dt.date.today().isoformat()) n_asp = n_rp = 0 for country in countries: try: for item in openaip_pages("airspaces", key, country, cache): geom = item.get("geometry") or {} rings = list(geometry_polylines(geom)) if not rings: continue ring = rings[0] pts = [(float(lat), float(lon)) for lon, lat, *_ in ring] icao_class = item.get("icaoClass") class_ = "ABCDEFG"[icao_class] if isinstance(icao_class, int) and 0 <= icao_class <= 6 else "" type_ = OPENAIP_TYPES.get(item.get("type"), "OTHER") floor_ft, floor_ref = openaip_limit(item.get("lowerLimit")) ceil_ft, ceil_ref = openaip_limit(item.get("upperLimit")) if nav.add_airspace(item.get("name", ""), class_, type_, floor_ft, floor_ref, ceil_ft, ceil_ref, item.get("country") or (None if country == "ALL" else country), pts, sid) is not None: n_asp += 1 for item in openaip_pages("reporting-points", key, country, cache): geom = item.get("geometry") or {} if geom.get("type") != "Point": continue lon, lat = geom["coordinates"][:2] nav.add_waypoint(item.get("name", ""), float(lat), float(lon), None, item.get("country") or (None if country == "ALL" else country), "reporting", sid) n_rp += 1 except Exception as e: log(f" WARNING: openAIP {country} failed: {e}") nav.db.commit() log(f" airspaces {n_asp}, reporting points {n_rp}") # ---------------------------------------------------------------- report def report(path): db = sqlite3.connect(path) log("\nRow counts") for table in ("airport", "runway", "frequency", "navaid", "waypoint", "airway", "airway_leg", "airspace", "basemap"): n = db.execute(f"SELECT count(*) FROM {table}").fetchone()[0] log(f" {table:12s} {n:>9,}") log("Per source") for sid, name, lic, cycle in db.execute("SELECT id, name, license, cycle FROM source"): parts = [] for table in ("airport", "navaid", "waypoint", "airway", "airspace"): n = db.execute(f"SELECT count(*) FROM {table} WHERE source_id=?", (sid,)).fetchone()[0] if n: parts.append(f"{table} {n:,}") log(f" [{sid}] {name} ({lic}, cycle {cycle}): {', '.join(parts) or 'basemap only'}") db.close() log(f"File: {path} ({os.path.getsize(path) / 1e6:.1f} MB)") # ---------------------------------------------------------------- main def main(): root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--out", default=os.path.join(root, "Resources", "NavData", "navdata.sqlite")) ap.add_argument("--cache", default=os.path.join(root, "Tools", ".navdata-cache")) ap.add_argument("--basemap", choices=("10m", "50m"), default="10m") ap.add_argument("--flightgear", action="store_true", help="add FlightGear fix/airway/navaid data (GPL, 2013)") ap.add_argument("--xplane-dir", help="directory with nav.dat, fix.dat, awy.dat (X-Plane format)") ap.add_argument("--faa-cifp", help="FAA CIFP file (FAACIFP18) or its zip") ap.add_argument("--openaip-key", help="openAIP Core API key (or set OPENAIP_API_KEY, or use --openaip-key-file)") ap.add_argument("--openaip-key-file", help="file containing the openAIP key; default Tools/.navdata-cache/openaip.key if present") ap.add_argument("--openaip-countries", default="", help="comma-separated ISO country codes for openAIP, or 'all'") args = ap.parse_args() if not args.openaip_key: key_file = args.openaip_key_file or os.path.join(args.cache, "openaip.key") if os.getenv("OPENAIP_API_KEY"): args.openaip_key = os.environ["OPENAIP_API_KEY"].strip() elif os.path.exists(key_file): with open(key_file, encoding="utf-8") as f: args.openaip_key = f.read().strip() os.makedirs(os.path.dirname(args.out), exist_ok=True) nav = NavDB(args.out) import_ourairports(nav, args.cache) import_natural_earth(nav, args.cache, args.basemap) if args.flightgear: import_flightgear(nav, args.cache) if args.xplane_dir: import_xplane_dir(nav, args.xplane_dir) if args.faa_cifp: import_faa_cifp(nav, args.faa_cifp) if args.openaip_key: countries = [c.strip().upper() for c in args.openaip_countries.split(",") if c.strip()] if countries: import_openaip(nav, args.cache, args.openaip_key, countries) else: log("openAIP: key given but no --openaip-countries; skipped") else: log("openAIP: no --openaip-key; airspace skipped") nav.finish() report(args.out) if __name__ == "__main__": main()