spothole/core/sig_utils.py

import csv
import logging

from pyhamtools.locator import latlong_to_locator, locator_to_latlong

from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import SIGS, HTTP_HEADERS
from core.geo_utils import wab_wai_square_to_lat_lon

# Load Spanish municipality data for the DME programme. There's no convenient lookup API for this, so we embed the data
# file in Spothole and load it on startup.
with open("datafiles/MUNICIPIOS.csv", encoding="latin-1") as _f:
    _DME_INDEX = {row["COD_INE"][:5]: row for row in csv.DictReader(_f, delimiter=";")}


def get_ref_regex_for_sig(sig):
    """Utility function to get the regex string for a SIG reference for a named SIG. If no match is found, None will be returned."""

    for s in SIGS:
        if s.name.upper() == sig.upper():
            return s.ref_regex
    return None


def get_sig_name_from_comment_name(sig):
    """Utility function to get the name of a SIG from its "comment name". Generally these will be the same but there are
    some cases (e.g. is "TOTA" Towers, Tiles or Toilets?) where we need to transform one to the other."""

    for s in SIGS:
        if any(n.upper() == sig.upper() for n in s.comment_names):
            return s.name
    return None


def populate_sig_ref_info(sig_ref):
    """Look up details of a SIG reference (e.g. POTA park) such as name, lat/lon, and grid. Takes in a sig_ref object which
    must at minimum have a "sig" and an "id". The rest of the object will be populated and returned.
    Note there is currently no support for KRMNPA location lookup, see issue #61."""

    if sig_ref.sig is None or sig_ref.id is None:
        logging.warning("Failed to look up sig_ref info, sig or id were not set.")

    sig = sig_ref.sig or ""
    ref_id = sig_ref.id
    try:
        if sig.upper() == "POTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://api.pota.app/park/" + ref_id, headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if data:
                fullname = str(data["name"]) if "name" in data else None
                if fullname and "parktypeDesc" in data and data["parktypeDesc"] != "":
                    fullname = fullname + " " + data["parktypeDesc"]
                sig_ref.name = fullname
                sig_ref.url = "https://pota.app/#/park/" + ref_id
                sig_ref.grid = data["grid6"] if "grid6" in data else None
                sig_ref.latitude = data["latitude"] if "latitude" in data else None
                sig_ref.longitude = data["longitude"] if "longitude" in data else None
        elif sig.upper() == "SOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + ref_id,
                                                      headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if data:
                sig_ref.name = data["name"] if "name" in data else None
                sig_ref.url = "https://www.sotadata.org.uk/en/summit/" + ref_id
                sig_ref.grid = data["locator"] if "locator" in data else None
                sig_ref.latitude = data["latitude"] if "latitude" in data else None
                sig_ref.longitude = data["longitude"] if "longitude" in data else None
                sig_ref.activation_score = data["points"] if "points" in data else None
        elif sig.upper() == "WWBOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + ref_id,
                                                      headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if data:
                sig_ref.name = data["name"] if "name" in data else None
                sig_ref.url = "https://bunkerwiki.org/?s=" + ref_id if ref_id.startswith("B/G") else None
                sig_ref.grid = data["locator"] if "locator" in data else None
                sig_ref.latitude = data["lat"] if "lat" in data else None
                sig_ref.longitude = data["long"] if "long" in data else None
        elif sig.upper() == "GMA" or sig.upper() == "ARLHS" or sig.upper() == "ILLW" or sig.upper() == "WCA" or sig.upper() == "MOTA" or sig.upper() == "IOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + ref_id,
                                                      headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if data:
                sig_ref.name = data["name"] if "name" in data else None
                sig_ref.url = "https://www.cqgma.org/zinfo.php?ref=" + ref_id
                sig_ref.grid = data["locator"] if "locator" in data else None
                sig_ref.latitude = data["latitude"] if "latitude" in data else None
                sig_ref.longitude = data["longitude"] if "longitude" in data else None
        elif sig.upper() == "WWFF":
            wwff_response = SEMI_STATIC_URL_DATA_CACHE.get("https://wwff.co/wwff-data/wwff_directory.csv",
                                                           headers=HTTP_HEADERS)
            if not wwff_response.ok:
                logging.warning("HTTP %d looking up %s ref %s", wwff_response.status_code, sig, ref_id)
                return sig_ref
            wwff_index = {row["reference"]: row for row in csv.DictReader(wwff_response.content.decode().splitlines())}
            row = wwff_index.get(ref_id)
            if row:
                sig_ref.name = row["name"] if "name" in row else None
                sig_ref.url = "https://wwff.co/directory/?showRef=" + ref_id
                sig_ref.grid = row["iaruLocator"] if "iaruLocator" in row and row["iaruLocator"] != "-" else None
                sig_ref.latitude = float(row["latitude"]) if "latitude" in row and row["latitude"] != "-" else None
                sig_ref.longitude = float(row["longitude"]) if "longitude" in row and row["longitude"] != "-" else None
        elif sig.upper() == "SIOTA":
            siota_response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv",
                                                            headers=HTTP_HEADERS)
            if not siota_response.ok:
                logging.warning("HTTP %d looking up %s ref %s", siota_response.status_code, sig, ref_id)
                return sig_ref
            siota_index = {row["SILO_CODE"]: row for row in
                           csv.DictReader(siota_response.content.decode().splitlines())}
            row = siota_index.get(ref_id)
            if row:
                sig_ref.name = row["NAME"] if "NAME" in row else None
                sig_ref.grid = row["LOCATOR"] if "LOCATOR" in row else None
                sig_ref.latitude = float(row["LAT"]) if "LAT" in row else None
                sig_ref.longitude = float(row["LNG"]) if "LNG" in row else None
        elif sig.upper() == "WOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json",
                                                      headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if data:
                for feature in data.get("features", []):
                    if feature["properties"]["wotaId"] == ref_id:
                        sig_ref.name = feature["properties"]["title"]
                        # Fudge WOTA URLs. Outlying fell (LDO) URLs don't match their ID numbers but require 214 to be
                        # added to them
                        sig_ref.url = "https://www.wota.org.uk/MM_" + ref_id
                        if ref_id.upper().startswith("LDO-"):
                            number = int(ref_id.upper().replace("LDO-", ""))
                            sig_ref.url = "https://www.wota.org.uk/MM_LDO-" + str(number + 214)
                        sig_ref.grid = feature["properties"]["qthLocator"]
                        sig_ref.latitude = feature["geometry"]["coordinates"][1]
                        sig_ref.longitude = feature["geometry"]["coordinates"][0]
                        break
        elif sig.upper() == "ZLOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if isinstance(data, list):
                for asset in data:
                    if asset["code"] == ref_id:
                        sig_ref.name = asset["name"]
                        sig_ref.url = "https://ontheair.nz/assets/ZLI_OT-030" + ref_id.replace("/", "_")
                        try:
                            sig_ref.grid = latlong_to_locator(asset["y"], asset["x"], 6)
                        except:
                            logging.debug("Invalid lat/lon received for reference")
                            sig_ref.latitude = asset["y"]
                            sig_ref.longitude = asset["x"]
                        break
        elif sig.upper() == "BOTA":
            if not sig_ref.name:
                sig_ref.name = sig_ref.id
            sig_ref.url = "https://www.beachesontheair.com/beaches/" + sig_ref.name.lower().replace(" ", "-")
        elif sig.upper() == "LLOTA":
            response = SEMI_STATIC_URL_DATA_CACHE.get("https://llota.app/api/public/references",
                                                      headers=HTTP_HEADERS)
            if not response.ok:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)
            data = response.json() if response.ok else None
            if isinstance(data, list):
                for ref in data:
                    if ref["reference_code"] == ref_id:
                        sig_ref.name = str(ref["name"])
                        sig_ref.url = "https://llota.app/list/ref/" + ref_id
                        sig_ref.grid = str(ref["grid_locator"])
                        ll = locator_to_latlong(sig_ref.grid)
                        sig_ref.latitude = ll[0]
                        sig_ref.longitude = ll[1]
                        break
        elif sig.upper() == "WWTOTA":
            if not sig_ref.name:
                sig_ref.name = sig_ref.id
            sig_ref.url = "https://wwtota.com/seznam/karta_rozhledny.php?ref=" + str(sig_ref.name)
        elif sig.upper() == "TILES":
            # Tiles on the Air just uses Maidenhead 6-digit squares, so ID, Name and Grid are all the same
            if not sig_ref.name:
                sig_ref.name = sig_ref.id
            if not sig_ref.grid:
                sig_ref.grid = sig_ref.id
            if sig_ref.grid and not sig_ref.latitude:
                ll = locator_to_latlong(str(sig_ref.grid))
                sig_ref.latitude = ll[0]
                sig_ref.longitude = ll[1]
        elif sig.upper() == "WAB" or sig.upper() == "WAI":
            ll = wab_wai_square_to_lat_lon(ref_id)
            if ll:
                sig_ref.name = ref_id
                try:
                    sig_ref.grid = latlong_to_locator(ll[0], ll[1], 6)
                    sig_ref.latitude = ll[0]
                    sig_ref.longitude = ll[1]
                except:
                    logging.debug("Invalid lat/lon received for reference")
        elif sig.upper() == "DME":
            # Zero-pad to 5 digits to match our source data
            row = _DME_INDEX.get(ref_id.zfill(5))
            if row:
                sig_ref.name = row["NOMBRE_ACTUAL"] + ", " + row["PROVINCIA"]
                sig_ref.latitude = float(row["LATITUD_ETRS89_REGCAN95"].replace(",", ".")) if row.get("LATITUD_ETRS89_REGCAN95") else None
                sig_ref.longitude = float(row["LONGITUD_ETRS89_REGCAN95"].replace(",", ".")) if row.get("LONGITUD_ETRS89_REGCAN95") else None
                if sig_ref.latitude and sig_ref.longitude:
                    try:
                        sig_ref.grid = latlong_to_locator(sig_ref.latitude, sig_ref.longitude, 6)
                    except Exception:
                        logging.debug("Invalid lat/lon received for reference")
    except Exception:
        logging.warning("Failed to look up sig_ref info for " + sig + " ref " + ref_id, exc_info=True)
    return sig_ref


# Regex matching any SIG's "comment name", i.e. how it may be referred to in spot comments
ANY_SIG_REGEX = r"(" + r"|".join(n for s in SIGS for n in s.comment_names) + r")"