#!/usr/bin/env python3 """ fcc_antenna_heights.py ======================= Downloads the FCC ULS Microwave service bulk database (l_micro.zip), extracts: - AN.dat (Antenna): antenna centerline height AGL per (call_sign, location_number, antenna_number) - PA.dat (Path): for each licensed point-to-point path, the exact transmit/receive (location_number, antenna_number) on each end, and the call sign at the other end -- this is the precise join table, avoiding any azimuth-matching guesswork. Produces two files: fcc_antennas.csv: call_sign, location_number, antenna_number, antenna_height_agl_m, polarization fcc_paths.csv: call_sign, receive_call_sign, path_number, tx_location_number, tx_antenna_number, rx_location_number, rx_antenna_number Source: https://data.fcc.gov/download/pub/uls/complete/l_micro.zip (~1-2 GB; this is the full Microwave service license database, updated daily by FCC) Usage: python fcc_antenna_heights.py ATT_Network.csv """ import argparse import csv import io import os import zipfile import requests L_MICRO_URL = "https://data.fcc.gov/download/pub/uls/complete/l_micro.zip" # 1-based column positions, confirmed against real l_micro.zip rows. AN_COLS = { "call_sign": 5, "location_number": 8, "antenna_number": 7, "antenna_height_agl_m": 12, # "Height to Center RAD" "antenna_make": 13, "antenna_model": 14, "polarization": 16, } PA_COLS = { "call_sign": 5, "path_number": 7, "tx_location_number": 8, "tx_antenna_number": 9, "rx_location_number": 10, "rx_antenna_number": 11, "receive_call_sign": 17, } def col(fields, idx_1based): i = idx_1based - 1 if 0 <= i < len(fields): return fields[i].strip() return "" def to_float(s): try: return float(s) except (ValueError, TypeError): return None def load_call_signs(csv_file): sites = set() with open(csv_file, newline="") as f: for row in csv.DictReader(f): if not row.get("Ref_ID", "").strip().isdigit(): continue sites.add(row["Site_A_Call_Sign"].strip()) sites.add(row["Site_B_Call_Sign"].strip()) return sites def main(): ap = argparse.ArgumentParser() ap.add_argument("csv_file") ap.add_argument("--antennas-out", default="fcc_antennas.csv") ap.add_argument("--paths-out", default="fcc_paths.csv") ap.add_argument("--zip-cache", default="l_micro.zip", help="local path to cache the downloaded zip (re-used if present)") ap.add_argument("--limit", type=int, default=None, help="only match the first N unique call signs (for quick testing -- " "note this does NOT reduce the l_micro.zip download size)") args = ap.parse_args() call_signs = load_call_signs(args.csv_file) if args.limit is not None: call_signs = set(sorted(call_signs)[:args.limit]) print(f"{len(call_signs)} unique call signs to match.") if not os.path.exists(args.zip_cache): print(f"Downloading {L_MICRO_URL} -> {args.zip_cache} (this is large, ~1-2 GB)...") with requests.get(L_MICRO_URL, stream=True, timeout=120) as r: r.raise_for_status() total = int(r.headers.get("content-length", 0)) done = 0 with open(args.zip_cache, "wb") as f: for chunk in r.iter_content(chunk_size=1 << 20): f.write(chunk) done += len(chunk) if total: print(f"\r {done/1e6:.0f} / {total/1e6:.0f} MB", end="") print() else: print(f"Reusing cached {args.zip_cache}") antennas = [] # list of dicts paths = [] # list of dicts with zipfile.ZipFile(args.zip_cache) as z: # --- AN.dat: every antenna record ----------------------------- print("Parsing AN.dat ...") with z.open("AN.dat") as raw: for line in io.TextIOWrapper(raw, encoding="latin-1"): fields = line.rstrip("\n").split("|") cs = col(fields, AN_COLS["call_sign"]) if cs not in call_signs: continue h = to_float(col(fields, AN_COLS["antenna_height_agl_m"])) if h is None: continue antennas.append({ "call_sign": cs, "location_number": col(fields, AN_COLS["location_number"]), "antenna_number": col(fields, AN_COLS["antenna_number"]), "antenna_height_agl_m": h, "antenna_make": col(fields, AN_COLS["antenna_make"]), "antenna_model": col(fields, AN_COLS["antenna_model"]), "polarization": col(fields, AN_COLS["polarization"]), }) # --- PA.dat: every path record --------------------------------- print("Parsing PA.dat ...") with z.open("PA.dat") as raw: for line in io.TextIOWrapper(raw, encoding="latin-1"): fields = line.rstrip("\n").split("|") cs = col(fields, PA_COLS["call_sign"]) rx_cs = col(fields, PA_COLS["receive_call_sign"]) if cs not in call_signs and rx_cs not in call_signs: continue paths.append({ "call_sign": cs, "receive_call_sign": rx_cs, "path_number": col(fields, PA_COLS["path_number"]), "tx_location_number": col(fields, PA_COLS["tx_location_number"]), "tx_antenna_number": col(fields, PA_COLS["tx_antenna_number"]), "rx_location_number": col(fields, PA_COLS["rx_location_number"]), "rx_antenna_number": col(fields, PA_COLS["rx_antenna_number"]), }) print(f"Matched {len(antennas)} antenna record(s), {len(paths)} path record(s).") with open(args.antennas_out, "w", newline="") as f: w = csv.writer(f) w.writerow(["call_sign", "location_number", "antenna_number", "antenna_height_agl_m", "antenna_make", "antenna_model", "polarization"]) for a in antennas: w.writerow([a["call_sign"], a["location_number"], a["antenna_number"], f"{a['antenna_height_agl_m']:.2f}", a["antenna_make"], a["antenna_model"], a["polarization"]]) with open(args.paths_out, "w", newline="") as f: w = csv.writer(f) w.writerow(["call_sign", "receive_call_sign", "path_number", "tx_location_number", "tx_antenna_number", "rx_location_number", "rx_antenna_number"]) for p in paths: w.writerow([p["call_sign"], p["receive_call_sign"], p["path_number"], p["tx_location_number"], p["tx_antenna_number"], p["rx_location_number"], p["rx_antenna_number"]]) found_signs = {a["call_sign"] for a in antennas} | {p["call_sign"] for p in paths} | \ {p["receive_call_sign"] for p in paths} missing = call_signs - found_signs print(f"Wrote {args.antennas_out} ({len(antennas)} rows) and " f"{args.paths_out} ({len(paths)} rows).") print(f" {len(found_signs & call_signs)}/{len(call_signs)} call signs appear in " f"AN.dat/PA.dat; {len(missing)} not found at all.") if missing: with open("antenna_heights_missing.txt", "w") as f: f.write("\n".join(sorted(missing))) print(" (missing call signs written to antenna_heights_missing.txt)") if __name__ == "__main__": main()