feat: repeater CSV ingest, load, and rank-ordered search
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+117
@@ -0,0 +1,117 @@
|
||||
"""
|
||||
Repeater directory: ingest a RepeaterBook CSV export into a normalized
|
||||
repeaters.json, then load / text-search / rank it. Pure logic + small file I/O;
|
||||
no radio, no network.
|
||||
|
||||
CSV is RepeaterBook's proximity export (nearest-first), so file order IS the
|
||||
distance ranking -> each record carries an integer `rank`.
|
||||
|
||||
Normalized record schema:
|
||||
callsign, location, output_mhz, offset_mhz, duplex ("+"/"-"/"simplex"),
|
||||
tone_hz (or None), tone_mode ("ctcss"/"none"), mode ("FM"/"DV"),
|
||||
rank (int), notes
|
||||
|
||||
Usage:
|
||||
python3 repeaters.py ingest repeaterbook/RB_xxxx.csv [repeaters.json]
|
||||
"""
|
||||
import csv
|
||||
import json
|
||||
import sys
|
||||
|
||||
# US state -> USPS abbreviation (only the ones this export touches; extend freely)
|
||||
_STATE_ABBR = {
|
||||
"Virginia": "VA", "Maryland": "MD", "District of Columbia": "DC",
|
||||
"Pennsylvania": "PA", "West Virginia": "WV", "Delaware": "DE",
|
||||
"New Jersey": "NJ", "North Carolina": "NC",
|
||||
}
|
||||
|
||||
|
||||
def _f(val):
|
||||
"""Parse a possibly-empty numeric string to float, else None."""
|
||||
try:
|
||||
return float(val)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def normalize_csv(row, rank):
|
||||
"""Map a RepeaterBook CSV row (dict) + its file rank to the internal schema."""
|
||||
out = _f(row.get("Output Freq"))
|
||||
inp = _f(row.get("Input Freq"))
|
||||
sign = (row.get("Offset") or "").strip()
|
||||
if out is not None and inp is not None and sign in ("+", "-"):
|
||||
duplex = sign
|
||||
offset = round(abs(inp - out), 5)
|
||||
else:
|
||||
duplex, offset = "simplex", 0.0
|
||||
|
||||
tone = _f(row.get("Uplink Tone"))
|
||||
city = (row.get("Location") or "").strip()
|
||||
state = (row.get("State") or "").strip()
|
||||
location = ", ".join(p for p in (city, _STATE_ABBR.get(state, state)) if p)
|
||||
|
||||
modes = (row.get("Modes") or "").upper()
|
||||
mode = "FM" if "FM" in modes else ("DV" if "DSTAR" in modes else "FM")
|
||||
|
||||
return {
|
||||
"callsign": (row.get("Call") or "").strip(),
|
||||
"location": location,
|
||||
"output_mhz": out,
|
||||
"offset_mhz": offset,
|
||||
"duplex": duplex,
|
||||
"tone_hz": tone,
|
||||
"tone_mode": "ctcss" if tone else "none",
|
||||
"mode": mode,
|
||||
"rank": rank,
|
||||
"notes": "",
|
||||
}
|
||||
|
||||
|
||||
def load(path):
|
||||
"""Read a normalized repeaters.json (list of records). Missing file -> []."""
|
||||
try:
|
||||
with open(path) as f:
|
||||
data = json.load(f)
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
return data if isinstance(data, list) else data.get("repeaters", [])
|
||||
|
||||
|
||||
def search(records, q, limit=25):
|
||||
"""Filter by case-insensitive substring (callsign/location), keep rank order."""
|
||||
q = (q or "").strip().lower()
|
||||
out = [r for r in records
|
||||
if not q or q in f"{r.get('callsign','')} {r.get('location','')}".lower()]
|
||||
out.sort(key=lambda r: r.get("rank", 1_000_000))
|
||||
return out[:limit]
|
||||
|
||||
|
||||
def ingest_csv(csv_path, out_path):
|
||||
"""Read a RepeaterBook CSV, filter to FM/DSTAR on 2m/70cm, write repeaters.json.
|
||||
Returns the number of records written. File order is preserved as `rank`."""
|
||||
recs = []
|
||||
with open(csv_path, newline="") as f:
|
||||
for i, row in enumerate(csv.DictReader(f)):
|
||||
rec = normalize_csv(row, i)
|
||||
f_mhz = rec["output_mhz"]
|
||||
if f_mhz is None:
|
||||
continue
|
||||
modes = (row.get("Modes") or "").upper()
|
||||
if "FM" not in modes and "DSTAR" not in modes:
|
||||
continue
|
||||
if not (144 <= f_mhz < 148 or 420 <= f_mhz < 450):
|
||||
continue
|
||||
recs.append(rec)
|
||||
with open(out_path, "w") as f:
|
||||
json.dump(recs, f, indent=1)
|
||||
return len(recs)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) >= 3 and sys.argv[1] == "ingest":
|
||||
out = sys.argv[3] if len(sys.argv) > 3 else "repeaters.json"
|
||||
n = ingest_csv(sys.argv[2], out)
|
||||
print(f"wrote {n} repeaters -> {out}")
|
||||
else:
|
||||
print("usage: python3 repeaters.py ingest <csv> [out.json]", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
Reference in New Issue
Block a user