Python script to sort WIA Repeater Data entries by geographic location

example-chirp-output.csv

example-output.csv

sortrepeaters.py

#!/usr/bin/env python3

"""
Parse the WIA Repeater Data listing and sort the repeaters by their
approximate locations.

The actual location of the repeater is first converted to a heading/distance
measurement from the reference point (chosen as being Alice Springs, NT due to
its "central" position).  This is then "scaled" logarithmically (using scaling
factors that were chosen "to taste" rather than via any scientific method) and
quantised to fixed steps.

This quantised distance is then scaled and used to figure out the number of
angle steps; and quantise the heading according to these steps.  Thus, sites
"closer" to the reference point will have fewer angular positions than more
distant sites, leading to "sectors" which are (hopefully) approximately the
same size.

Repeaters are sorted by sector angle, then sector distance.

© 2022 Stuart Longland VK4MSL

All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:

1. Redistributions of source code must retain the above copyright notice,
   this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
   notice, this list of conditions and the following disclaimer in the
   documentation and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDER AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
"""

import csv
import argparse
import logging
import math
from geographiclib.geodesic import Geodesic

# Reference location: Alice Springs, NT.
(REF_LAT, REF_LON) = (-23.7, 133.87)

# Scaling parameters, adjust to taste
DIST_LOG = 2
DIST_SCALE = 10
DIST_STEP = 1
ANGLE_SCALE = 2

logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger("repeater-parser")

ap = argparse.ArgumentParser()
ap.add_argument("--chirp", action="store_true", help="Generate CHIRP CSV")
ap.add_argument("--start-num", type=int, default=0,
        help="Starting channel number")
ap.add_argument("--skip-num", type=int, default=0,
        help="Skip the first N repeaters")
ap.add_argument("--count-num", type=int, default=0,
        help="Limit the number of repeaters to the given count")
ap.add_argument("input")
ap.add_argument("output")

args = ap.parse_args()

# Mode whitelist; I've written this script for a Yaesu C4FM radio, but
# you might choose "FM", "DSTAR" for an Icom, "FM", "DMR" for DMR/FM
# radios, or even just plain "FM".
MODE_WL = set(["FM", "DSTAR", "P25", "DMR", "C4FM"])

# Band selection; for a tri-bander like a Yaesu VX7/VX8 or Alinco
# DJ-G7, you might choose "6m", "2m", "70cm" or "2m", "70cm",
# "23cm"; whatever floats your boat.
BAND_WL = set(["10m", "6m", "2m", "70cm", "23cm", "12cm", "6cm", "3cm"])

# Band labels; I assume a single-letter indicator here.  It's more a clue
# for radios like the FT-857/897, FT8900 or IC-706MkIIG which can do
# multiple bands that require very different antennas.
BAND_LABEL = {
    "10m": "H",  # HF
    "6m": "6",  # low VHF
    "2m": "2",  # mid VHF
    "70cm": "U",  # UHF
    "23cm": "M",  # microwave
    "12cm": "M",  # microwave
    "6cm": "M",  # microwave
    "3cm": "M",  # microwave
}

# Band edges
BAND_EDGE = (
    (135, 138, "2200m"),
    (472, 479, "630m"),
    (1800, 1875, "160m"),
    (3500, 3700, "80m"),
    (3776, 3800, "75m"),
    (5351, 5366, "60m"),
    (7000, 7300, "40m"),
    (10100, 10150, "30m"),
    (14000, 14350, "20m"),
    (18068, 18168, "17m"),
    (21000, 21450, "15m"),
    (24890, 24990, "12m"),
    (28000, 29700, "10m"),
    (50000, 54000, "6m"),
    (144000, 148000, "2m"),
    (430000, 450000, "70cm"),
    (1240000, 1300000, "23cm"),
    (2300000, 2302000, "13cm L"),
    (2400000, 2450000, "13cm"),
    (3300000, 3600000, "9cm"),
    (5650000, 5850000, "6cm"),
    (10000000, 10500000, "3cm"),
    (24000000, 24250000, "12mm"),
    (47000000, 47200000, "6mm"),
    (76000000, 81000000, "4mm"),
)

# Mnemonic length: Rather than using the pre-defined mnemonics, we
# generate our own from the shortened call-sign (minus 'VK' prefix and
# 'R' in suffix); the band label (above), and however many location
# letters will fit.
MNEMONIC_LENGTH = 16

# These repeaters do not have a location recorded in the WIA Repeater
# Database, so this is a "guessed" location based on the reported location or
# service area.
GUESSED_LOCATION = dict(
    VK2RMO=(-30.8164901, 150.5661923),
    VK2RNG=(-33.7587262, 150.7330947),
    VK2RRR=(-33.892734, 150.3767496),
    VK2RSH=(-36.2481666, 149.0426176),
    VK3RCR=(-37.8362545, 145.3335526),
    VK4RBS=(-27.5307788, 153.2388942),
    VK4RCC=(-27.1479971, 152.8361802),
    VK4RMA=(-24.0947758, 150.7291737),
    VK6RAK=(-30.753071, 121.3945294),
    VK6RTP=(-22.6960047, 117.7271023),
)

# Some repeaters are "portable", for the sake of argument, I've picked a
# random spot roughly in the "centre" of each state/territory.
STATE_CENTRE = dict(
    VK1=(-35.2757342, 147.9060907),
    VK2=(-32.8150779, 142.3023483),
    VK3=(-36.7165983, 139.6767246),
    VK4=(-20.832548, 139.0722132),
    VK5=(-29.6556117, 133.8800449),
    VK6=(-24.3415112, 129.9050152),
    VK7=(-41.8696616, 145.5029302),
    VK8=(-23.8441145, 137.1081804),
)

# Output CSV field headers
FIELDS = [
    "Channel",
    "Sector",
    "Band",
    "Region",
    "Mode",
    "Output",
    "Offset",
    "Call",
    "mNemonic",
    "Location",
    "Service Area",
    "Latitude",
    "Longitude",
    "Angle",
    "Distance",
    "S",
    "ERP",
    "HASL",
    "T/O",
    "Sp",
    "Tone",
    "Notes",
]

# Chirp headers
CHIRP_HEADERS = [
        "Location", "Name", "Frequency",
        "Duplex", "Offset", "Tone",
        "rToneFreq", "cToneFreq", "DtcsCode",
        "DtcsPolarity", "RxDtcsCode",
        "CrossMode",
        "Mode", "TStep",
        "Skip", "Comment",
        "URCALL", "RPT1CALL", "RPT2CALL", "DVCODE"
]

# Default CHIRP tone frequency
CHIRP_DEFAULT_TONE = 88.5

# Defaults for other Chirp settings
CHIRP_DEFAULTS = {
        "DtcsCode": 23,
        "RxDtcsCode": 23,
        "DtcsPolarity": "NN",
        "CrossMode": "Tone->Tone",
        "Skip": "",
        "TStep": 25.0,
        "Comment": "",
        "URCALL": "",
        "RPT1CALL": "",
        "RPT2CALL": "",
        "DVCODE": 0,
}

# Sort order for repeaters
SORT_ORDER = ("Sector", "Call", "Output", "Mode")

# Sector start letter
A = ord("A")

# Computed values
FIELD_WL = set(FIELDS)
repeaters = []

# Chirp value conversion
CHIRP_CONV = {
        "Location": lambda r : r["Channel"],
        "Name": lambda r : r["mNemonic"],
        "Mode": lambda r : r["Mode"],
        "Frequency": lambda r : r["Output"] / 1000,
        "Duplex": lambda r : (
            ("+" if (r["Offset"] > 0) else "-")
            if (r["Offset"] != 0) else ""
        ),
        "Offset": lambda r : abs(r["Offset"]/1000),
        "Tone": lambda r : "Tone" if r["Tone"] else "",
        "rToneFreq": lambda r : r["Tone"] or CHIRP_DEFAULT_TONE,
        "cToneFreq": lambda r : r["Tone"] or CHIRP_DEFAULT_TONE,
}

def sector(heading, distance):
    """
    Return the sector for the given heading and distance.

    The sector is the heading mapped to the letters AA (-180°) through to
    ZZ (+180°); followed by a 3-digit distance.
    """
    logdistance = (
        int(((DIST_SCALE * math.log(distance, DIST_LOG)) / DIST_STEP) + 0.5)
        * DIST_STEP
    )

    # Map heading from -180° - +180°
    heading = 180 + heading

    # Quantise heading
    heading_step = int((360 / (ANGLE_SCALE * logdistance)) + 0.5)
    qheading = int((heading / heading_step) + 0.5) * heading_step

    # Scale 0-360 to 0-26², then split apart to two 0-25 "digits"
    (first, second) = divmod(int(((26**2) * qheading) / 360), 26)
    return "%s%s%03d" % (
        chr(A + min(first, 25)),
        chr(A + second),
        int(logdistance),
    )


def band(freq):
    """
    Determine the band given the frequency in kHz.
    """
    for (low, high, band) in BAND_EDGE:
        if freq < low:
            raise ValueError("Out of band")
        if freq < high:
            return band

    raise ValueError("Out of band")


# Read the input repeaters and compute location
with open(args.input, "r") as infile:
    incsv = csv.DictReader(infile)
    group = None

    for row in incsv:
        log.debug("Parsing %r (group=%r)", row, group)

        if row["Notes"] and not row["Output"]:
            # Group header
            notes = row["Notes"]
            if notes.startswith("{{") and notes.endswith("}}"):
                (_, mode, region) = notes[2:-2].split(";", 2)
                group = dict(Mode=mode, Region=region)
                log.debug("New group: %r", group)
        elif group and row["Output"]:
            if group["Mode"] not in MODE_WL:
                continue

            repeater = row.copy()
            coord = True
            try:
                for field in ("Input", "Output", "Latitude", "Longitude"):
                    try:
                        repeater[field] = float(repeater[field])
                    except ValueError:
                        # Tolerate for Latitude/Longitude only
                        if field in ("Latitude", "Longitude"):
                            coord = False
                            pass
                        else:
                            raise

                for field in ("Input", "Output"):
                    # Convert to kHz
                    repeater[field] *= 1000
            except ValueError:
                log.debug("Ignoring malformed repeater %r", row)
                continue

            # Figure out the band -- use the input frequency since
            # that antenna *MUST* be tuned to the transmitter.
            repeater["Band"] = band(repeater["Input"])
            if repeater["Band"] not in BAND_WL:
                continue

            if not coord and (repeater["Call"] in GUESSED_LOCATION):
                (
                    repeater["Latitude"],
                    repeater["Longitude"],
                ) = GUESSED_LOCATION[repeater["Call"]]
                coord = True

            region = repeater["Call"][:3]
            if not coord and (region in STATE_CENTRE):
                (repeater["Latitude"], repeater["Longitude"]) = STATE_CENTRE[
                    region
                ]
                coord = True

            if coord:
                geodesic = Geodesic.WGS84.Inverse(
                    REF_LAT,
                    REF_LON,
                    repeater["Latitude"],
                    repeater["Longitude"],
                )
                heading = geodesic["azi1"]
                distance = geodesic["s12"] / 1000.0
                repeater["Angle"] = heading
                repeater["Distance"] = distance
                repeater["Sector"] = sector(heading, distance)
            else:
                repeater["Angle"] = -180
                repeater["Distance"] = -1
                repeater["Sector"] = "AA000"

            repeater["Offset"] = repeater["Input"] - repeater["Output"]
            repeater.update(group)

            # Yaesu allow 16 characters
            # - 3 for shortened callsign
            # - 1 for band label
            # = 12 characters for location
            shortcall = repeater["Call"][2] + repeater["Call"][4:6]
            bandlabel = BAND_LABEL[repeater["Band"]]
            locationlen = MNEMONIC_LENGTH - len(shortcall + bandlabel)

            repeater["mNemonic"] = (
                shortcall
                + repeater["Location"][:locationlen].ljust(locationlen)
                + bandlabel
            )

            # Sanitise the tone field
            if repeater["Tone"] == "-":
                repeater["Tone"] = None
            else:
                repeater["Tone"] = float(repeater["Tone"])

            repeaters.append(repeater)

# Sort the repeater by the criteria
log.info("Parsed %d repeaters", len(repeaters))
repeaters.sort(key=lambda r: tuple([r[f] for f in SORT_ORDER]))

# Skip the first N repeaters if given
if args.skip_num:
    repeaters = repeaters[args.skip_num:]

# Truncate the list to the number of repeaters specified
if args.count_num:
    repeaters = repeaters[:args.count_num]

# Number the repeaters
for idx, repeater in enumerate(repeaters, start=args.start_num):
    repeater["Channel"] = idx

# Write out the sorted list
with open(args.output, "w") as outfile:
    if args.chirp:
        outcsv = csv.DictWriter(outfile, CHIRP_HEADERS)
        outcsv.writeheader()
        for repeater in repeaters:
            row = CHIRP_DEFAULTS.copy()
            for label, conv in CHIRP_CONV.items():
                row[label] = conv(repeater)

            outcsv.writerow(row)
    else:
        outcsv = csv.DictWriter(outfile, FIELDS)
        outcsv.writeheader()

        for repeater in repeaters:
            for field in set(repeater.keys()) - FIELD_WL:
                del repeater[field]

            outcsv.writerow(repeater)