import csv import logging from pyhamtools.locator import latlong_to_locator, locator_to_latlong from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout 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 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 = SEMI_STATIC_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 = SEMI_STATIC_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 = SEMI_STATIC_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 "latitude" in data and "longitude" in data and "lat2" in data and "lng2" in data: 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 "latitude" in data else None sig_ref.longitude = float(data["longitude"]) if "longitude" in data 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": response = SEMI_STATIC_URL_DATA_CACHE.get("https://wwff.co/wwff-data/wwff_directory.csv", headers=HTTP_HEADERS) if response.ok: wwff_index = {row["reference"]: row for row in csv.DictReader(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 not response.from_cache: logging.warning("WWFF database did not contain data for ref %s", 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() == "SIOTA": response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", headers=HTTP_HEADERS) if response.ok: siota_index = {row["SILO_CODE"]: row for row in csv.DictReader(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 not response.from_cache: logging.warning("SIOTA database did not contain data for ref %s", 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() == "WOTA": response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", headers=HTTP_HEADERS) if response.ok: data = response.json() 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 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() == "ZLOTA": response = SEMI_STATIC_URL_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS) if response.ok: data = response.json() 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/" + 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 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() == "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 response.ok: data = response.json() 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 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() == "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") 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.warning("Invalid lat/lon received for DME reference") else: logging.warning("DME database did not contain data for ref %s", ref_id) 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")"