#!/usr/bin/env python3 """ populate_summary_v2.py ======================= Standalone script -- populates Path_Profile_Summaryv2.csv. Does NOT import anything from path_profile_tool.py (avoids version-mismatch issues between scripts). For each path (row N of ATT_Network.csv -> row N of the summary), fills: - A/B Site Call Sign - A/B Site Elev AMSL (M): ground elevation at path endpoints, from output/tables/.csv (first/last row, Z_surface_elev_m) - A/B Main/Diversity Ant Height AGL (M): from fcc_antennas.csv + fcc_paths.csv (PA.dat-based exact join -- main = tallest antenna PA.dat links for this path at that site, diversity = next-tallest distinct one, if any) - A/B Main/Diversity Ant Height AMSL (M): ground elev + AGL - A/B Main/Diversity Antenna Make/Model: from fcc_antennas.csv - Path Length (Miles): converted from Path_Length (km) in ATT_Network.csv - USGS 1M / GLO 30 / Total Points: source counts from the table CSV Usage: python populate_summary_v2.py ATT_Network.csv Path_Profile_Summaryv2.csv ^ --out Path_Profile_Summaryv2_filled.csv [--all] By default fills only as many rows as the template has. Use --all to generate a row for every path in ATT_Network.csv. """ import argparse import csv import math import os def load_antennas(path): """dict (call_sign, location_number, antenna_number) -> antenna dict""" antennas = {} if not path or not os.path.exists(path): return antennas with open(path, newline="") as f: for row in csv.DictReader(f): h = row.get("antenna_height_agl_m", "").strip() if not h: continue key = (row["call_sign"].strip(), row["location_number"].strip(), row["antenna_number"].strip()) antennas[key] = { "antenna_height_agl_m": float(h), "antenna_make": row.get("antenna_make", "").strip(), "antenna_model": row.get("antenna_model", "").strip(), "polarization": row.get("polarization", "").strip(), } return antennas def load_path_records(path): """dict frozenset({call_sign_a, call_sign_b}) -> list of PA row dicts""" records = {} if not path or not os.path.exists(path): return records with open(path, newline="") as f: for row in csv.DictReader(f): a, b = row["call_sign"].strip(), row["receive_call_sign"].strip() if not a or not b: continue key = frozenset((a, b)) records.setdefault(key, []).append(row) return records def find_main_div(call_sign_a, call_sign_b, antennas, path_records): """Returns (main_a, div_a, main_b, div_b), each an antenna dict or None. Rule: for a site X in pair (X, Y), collect ALL of X's own TX antennas from PA.dat records where call_sign=X and receive_call_sign=Y (i.e. every antenna X transmits toward Y with -- typically 2 distinct physical antennas x 1-2 polarizations each = up to ~6 PA records, but only 1-2 DISTINCT HEIGHTS). Antennas with the same height (within 0.5m, i.e. dual-polarization copies of the same physical antenna) are collapsed to one. Of the resulting distinct antennas: - the TALLEST is X's MAIN antenna (main is always higher on the tower than diversity, per user) - the next-tallest (if any) is X's DIVERSITY antenna This relies ONLY on each site's own TX-side PA records (which are reliably populated), never on the RX-side references (which are frequently dangling in AN.dat -- see prior investigation).""" records = path_records.get(frozenset((call_sign_a, call_sign_b)), []) def own_antennas(cs, other_cs): """All distinct-height antennas `cs` transmits toward `other_cs`, sorted tallest-first.""" found = [] for r in records: if r["call_sign"].strip() == cs and r["receive_call_sign"].strip() == other_cs: ant = antennas.get((cs, r["tx_location_number"].strip(), r["tx_antenna_number"].strip())) if ant: found.append(ant) # dedupe by height (dual-polarization copies of the same antenna) deduped = [] for a in found: if not any(abs(a["antenna_height_agl_m"] - b["antenna_height_agl_m"]) < 0.5 for b in deduped): deduped.append(a) deduped.sort(key=lambda a: a["antenna_height_agl_m"], reverse=True) return deduped a_ants = own_antennas(call_sign_a, call_sign_b) b_ants = own_antennas(call_sign_b, call_sign_a) main_a = a_ants[0] if a_ants else None div_a = a_ants[1] if len(a_ants) > 1 else None main_b = b_ants[0] if b_ants else None div_b = b_ants[1] if len(b_ants) > 1 else None return main_a, div_a, main_b, div_b def load_raat(path): """dict Path_ID -> (a_raat_m, b_raat_m), both floats.""" raat = {} if not path or not os.path.exists(path): return raat with open(path, newline="", encoding="utf-8-sig") as f: for row in csv.DictReader(f): pid = row.get("Path_ID", "").strip() try: a = float(row["a_raat"]) b = float(row["b_raat"]) except (ValueError, KeyError): continue raat[pid] = (a, b) return raat def main(): ap = argparse.ArgumentParser() ap.add_argument("network_csv") ap.add_argument("template_csv") ap.add_argument("--out", default="Path_Profile_Summaryv2_filled.csv") ap.add_argument("--tables-dir", default="output/tables") ap.add_argument("--fcc-antennas", default="fcc_antennas.csv") ap.add_argument("--fcc-paths", default="fcc_paths.csv") ap.add_argument("--raat-csv", default="ATT_Mobility_Network_Definition_v9.csv", help="CSV with Path_ID, a_raat, b_raat columns -- used as the " "authoritative Main Antenna Height AGL for every path") ap.add_argument("--all", action="store_true", help="generate rows for ALL paths in ATT_Network.csv, " "not just as many as the template has") args = ap.parse_args() with open(args.network_csv, newline="") as f: net_rows = [r for r in csv.DictReader(f) if r.get("Ref_ID", "").strip().isdigit()] with open(args.template_csv, newline="", encoding="utf-8-sig") as f: reader = csv.reader(f) header = next(reader) template_rows = list(reader) n = len(net_rows) if args.all else len(template_rows) n = min(n, len(net_rows)) antennas = load_antennas(args.fcc_antennas) path_records = load_path_records(args.fcc_paths) raat = load_raat(args.raat_csv) print(f"Loaded {len(antennas)} antenna(s), {len(path_records)} unique site-pair(s) from FCC data, " f"{len(raat)} RAAT entries.") def fmt(x, nd=2): return "" if x is None else f"{x:.{nd}f}" def agl(ant): return ant["antenna_height_agl_m"] if ant else None def amsl(ground, ant): if ant is None or ground is None or math.isnan(ground): return None return ground + ant["antenna_height_agl_m"] def make_model(ant): if ant is None: return "", "" return ant.get("antenna_make", ""), ant.get("antenna_model", "") out_rows = [] missing_tables = [] missing_raat = [] for i in range(n): row = net_rows[i] path_id2 = row["Path_ID2"] name_a = row["Site_A_Call_Sign"] name_b = row["Site_B_Call_Sign"] table_path = os.path.join(args.tables_dir, path_id2 + ".csv") a_elev = b_elev = None n_usgs = n_glo = n_total = 0 if os.path.exists(table_path): with open(table_path, newline="") as f: trows = list(csv.DictReader(f)) n_total = len(trows) n_usgs = sum(1 for r in trows if r.get("source") == "usgs_1m") n_glo = sum(1 for r in trows if r.get("source") == "glo30") if trows: try: a_elev = float(trows[0]["Z_surface_elev_m"]) except (ValueError, KeyError): a_elev = None try: b_elev = float(trows[-1]["Z_surface_elev_m"]) except (ValueError, KeyError): b_elev = None else: missing_tables.append(path_id2) main_a, div_a, main_b, div_b = find_main_div(name_a, name_b, antennas, path_records) a_main_make, a_main_model = make_model(main_a) a_div_make, a_div_model = make_model(div_a) b_main_make, b_main_model = make_model(main_b) b_div_make, b_div_model = make_model(div_b) # If one side's MAIN antenna couldn't be resolved (no PA.dat # reference for this specific path), fall back to the OTHER # side's main antenna make/model -- height AGL/AMSL stay blank # since those are site-specific and we don't have the actual # antenna's height at this site, only its make/model as a # best-available indicator (per user instruction). if main_a is None and main_b is not None: a_main_make, a_main_model = b_main_make, b_main_model if main_b is None and main_a is not None: b_main_make, b_main_model = a_main_make, a_main_model path_length_km = row.get("Path_Length", "").strip() path_length_mi = (float(path_length_km) / 1.60934) if path_length_km else None # MAIN antenna height AGL: always sourced from the RAAT file # (complete coverage, no blanks). FCC PA.dat/AN.dat is still used # for diversity height and for main/diversity make+model. a_raat_agl, b_raat_agl = raat.get(path_id2, (None, None)) if a_raat_agl is None: missing_raat.append(path_id2) out_rows.append([ i + 1, path_id2, name_a, fmt(a_elev), fmt(a_raat_agl), fmt(agl(div_a)), fmt(amsl(a_elev, {"antenna_height_agl_m": a_raat_agl}) if a_raat_agl is not None else None), fmt(amsl(a_elev, div_a)), a_main_make, a_main_model, a_div_make, a_div_model, name_b, fmt(b_elev), fmt(b_raat_agl), fmt(agl(div_b)), fmt(amsl(b_elev, {"antenna_height_agl_m": b_raat_agl}) if b_raat_agl is not None else None), fmt(amsl(b_elev, div_b)), b_main_make, b_main_model, b_div_make, b_div_model, fmt(path_length_mi, 3), n_usgs, n_glo, n_total, ]) if out_rows and len(out_rows[0]) != len(header): print(f"WARNING: header has {len(header)} columns but generated rows have " f"{len(out_rows[0])} columns -- output may be misaligned!") with open(args.out, "w", newline="") as f: w = csv.writer(f) w.writerow(header) w.writerows(out_rows) print(f"Wrote {len(out_rows)} row(s) to {args.out}") if missing_tables: print(f" {len(missing_tables)} path(s) had no table CSV in {args.tables_dir}: " f"{', '.join(missing_tables[:10])}" + (" ..." if len(missing_tables) > 10 else "")) if missing_raat: print(f" {len(missing_raat)} path(s) had no RAAT entry in {args.raat_csv}: " f"{', '.join(missing_raat[:10])}" + (" ..." if len(missing_raat) > 10 else "")) if __name__ == "__main__": main()