spothole/core/sig_utils.py

import logging

from pyhamtools.locator import latlong_to_locator, locator_to_latlong
from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout

from core.constants import SIGS, HTTP_HEADERS
from core.data_store import DATA_STORE
from core.geo_utils import wab_wai_square_to_lat_lon
from core.url_data_cache import URL_DATA_CACHE


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. This makes
    use of SIG ref data in the data store, live lookups from the web, or just automatic calculation depending on which
    SIG we are getting data for.
    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 or sig_ref.id == "":
        logging.debug("Failed to look up sig_ref info, sig or id were not set.")
        return sig_ref

    sig = sig_ref.sig or ""
    ref_id = sig_ref.id
    try:
        if sig.upper() == "POTA":
            response = URL_DATA_CACHE.get("https://api.pota.app/park/" + ref_id, headers=HTTP_HEADERS)
            if response.ok:
                data = response.json()
                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 not response.from_cache:
                    logging.warning("Malformed response looking up %s ref %s", sig, ref_id)
            elif not response.from_cache:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)

        elif sig.upper() == "SOTA":
            response = URL_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + ref_id,
                                                      headers=HTTP_HEADERS)
            if response.ok:
                data = response.json()
                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 not response.from_cache:
                    logging.warning("Malformed response looking up %s ref %s", sig, ref_id)
            elif not response.from_cache:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)

        elif sig.upper() == "WWBOTA":
            response = URL_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + ref_id,
                                                      headers=HTTP_HEADERS)
            if response.ok:
                data = response.json()
                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 not response.from_cache:
                    logging.warning("Malformed response looking up %s ref %s", sig, ref_id)
            elif not response.from_cache:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)

        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 = URL_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + ref_id,
                                                      headers=HTTP_HEADERS)
            if response.ok:
                data = response.json()
                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

                    # For some things (just IOTA?) the GMA actually returns a box where "latitude" and "longitude" are
                    # the zeroest corner of the box, then "lat2" and "lng2" provide the other corner. We detect this
                    # and provide a single lat/lon for the centre. Otherwise if we don't have these extra parameters,
                    # just use the single point we have.
                    if data.get("latitude") is not None and data.get("longitude") is not None and data.get(
                            "lat2") is not None and data.get("lng2") is not None:
                        sig_ref.latitude = (float(data["latitude"]) + float(data["lat2"])) / 2.0
                        sig_ref.longitude = (float(data["longitude"]) + float(data["lng2"])) / 2.0
                    else:
                        sig_ref.latitude = float(data["latitude"]) if data.get("latitude") is not None else None
                        sig_ref.longitude = float(data["longitude"]) if data.get("longitude") is not None else None
                elif not response.from_cache:
                    logging.warning("Malformed response looking up %s ref %s via GMA", sig, ref_id)
            elif not response.from_cache:
                logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id)

        elif sig.upper() == "WWFF":
            lookup_data = DATA_STORE.sigrefs["WWFF"][ref_id] if ref_id in DATA_STORE.sigrefs["WWFF"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("WWFF database did not contain data for ref %s", ref_id)

        elif sig.upper() == "SIOTA":
            lookup_data = DATA_STORE.sigrefs["SIOTA"][ref_id] if ref_id in DATA_STORE.sigrefs["SIOTA"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("SIOTA database did not contain data for ref %s", ref_id)

        elif sig.upper() == "WOTA":
            lookup_data = DATA_STORE.sigrefs["WOTA"][ref_id] if ref_id in DATA_STORE.sigrefs["WOTA"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("WOTA database did not contain data for ref %s", ref_id)

        elif sig.upper() == "ZLOTA":
            lookup_data = DATA_STORE.sigrefs["ZLOTA"][ref_id] if ref_id in DATA_STORE.sigrefs["ZLOTA"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("ZLOTA database did not contain data for ref %s", ref_id)

        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":
            lookup_data = DATA_STORE.sigrefs["LLOTA"][ref_id] if ref_id in DATA_STORE.sigrefs["LLOTA"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("LLOTA database did not contain data for ref %s", ref_id)

        elif sig.upper() == "DME":
            # Zero-pad to 5 digits to match our source data
            lookup_data = DATA_STORE.sigrefs["DME"][ref_id.zfill(5)] if ref_id.zfill(5) in DATA_STORE.sigrefs["DME"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("DME database did not contain data for ref %s", ref_id)

        elif sig.upper() == "TOTA":
            lookup_data = DATA_STORE.sigrefs["TOTA"][ref_id] if ref_id in DATA_STORE.sigrefs["TOTA"] else None
            if lookup_data:
                # Copy new sig ref data into existing object
                sig_ref.__dict__.update(lookup_data.__dict__)
            else:
                logging.warning("TOTA database did not contain data for ref %s", ref_id)

        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.warning("Invalid lat/lon received for WAB/WAI reference")

    except ConnectionError:
        logging.warning("Connection error when looking up sig_ref info for " + sig + " ref " + ref_id)
    except (ConnectTimeout, ReadTimeout):
        logging.warning(f"Timeout when looking up sig_ref info for " + sig + " ref " + ref_id)
    except Exception:
        logging.error("Exception when looking 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")"