mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-03-15 12:24:29 +00:00
616 lines
29 KiB
Python
616 lines
29 KiB
Python
import gzip
|
|
import json
|
|
import logging
|
|
import re
|
|
import urllib.parse
|
|
from datetime import timedelta
|
|
|
|
import xmltodict
|
|
from diskcache import Cache
|
|
from pyhamtools import LookupLib, Callinfo, callinfo
|
|
from pyhamtools.exceptions import APIKeyMissingError
|
|
from pyhamtools.frequency import freq_to_band
|
|
from pyhamtools.locator import latlong_to_locator
|
|
from requests_cache import CachedSession
|
|
|
|
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
|
|
from core.config import config
|
|
from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, \
|
|
HTTP_HEADERS, HAMQTH_PRG, MODE_ALIASES
|
|
|
|
|
|
class LookupHelper:
|
|
"""Singleton class that provides lookup functionality."""
|
|
|
|
def __init__(self):
|
|
"""Create the lookup helper. Note that nothing actually happens until the start() method is called, and that all
|
|
lookup methods will fail if start() has not yet been called. This therefore needs starting before any spot or
|
|
alert handlers are created."""
|
|
|
|
self._clublog_callsign_data_cache = None
|
|
self._lookup_lib_clublog_xml = None
|
|
self._clublog_xml_available = None
|
|
self._lookup_lib_clublog_api = None
|
|
self._clublog_xml_download_location = None
|
|
self._clublog_api_available = None
|
|
self._clublog_cty_xml_cache = None
|
|
self._clublog_api_key = None
|
|
self._qrz_callsign_data_cache = None
|
|
self._lookup_lib_qrz = None
|
|
self._qrz_available = None
|
|
self._hamqth_available = None
|
|
self._hamqth_callsign_data_cache = None
|
|
self._hamqth_base_url = "https://www.hamqth.com/xml.php"
|
|
# HamQTH session keys expire after an hour. Rather than working out how much time has passed manually, we cheat
|
|
# and cache the HTTP response for 55 minutes, so when the login URL is queried within 55 minutes of the previous
|
|
# time, you just get the cached response.
|
|
self._hamqth_session_lookup_cache = CachedSession("cache/hamqth_session_cache",
|
|
expire_after=timedelta(minutes=55))
|
|
self._call_info_basic = None
|
|
self._lookup_lib_basic = None
|
|
self._country_files_cty_plist_download_location = None
|
|
self._dxcc_json_download_location = None
|
|
self._dxcc_data = None
|
|
|
|
def start(self):
|
|
# Lookup helpers from pyhamtools. We use five (!) of these. The simplest is country-files.com, which downloads
|
|
# the data once on startup, and requires no login/key, but does not have the best coverage.
|
|
# If the user provides login details/API keys, we also set up helpers for QRZ.com, HamQTH, Clublog (live API
|
|
# request), and Clublog (XML download). The lookup functions iterate through these in a sensible order, looking
|
|
# for suitable data.
|
|
self._country_files_cty_plist_download_location = "cache/cty.plist"
|
|
success = self._download_country_files_cty_plist()
|
|
if success:
|
|
self._lookup_lib_basic = LookupLib(lookuptype="countryfile",
|
|
filename=self._country_files_cty_plist_download_location)
|
|
else:
|
|
self._lookup_lib_basic = LookupLib(lookuptype="countryfile")
|
|
self._call_info_basic = Callinfo(self._lookup_lib_basic)
|
|
|
|
self._qrz_available = config["qrz-username"] != "" and config["qrz-password"] != ""
|
|
if self._qrz_available:
|
|
self._lookup_lib_qrz = LookupLib(lookuptype="qrz", username=config["qrz-username"],
|
|
pwd=config["qrz-password"])
|
|
self._qrz_callsign_data_cache = Cache('cache/qrz_callsign_lookup_cache')
|
|
|
|
self._hamqth_available = config["hamqth-username"] != "" and config["hamqth-password"] != ""
|
|
self._hamqth_callsign_data_cache = Cache('cache/hamqth_callsign_lookup_cache')
|
|
|
|
self._clublog_api_key = config["clublog-api-key"]
|
|
self._clublog_cty_xml_cache = CachedSession("cache/clublog_cty_xml_cache", expire_after=timedelta(days=10))
|
|
self._clublog_api_available = self._clublog_api_key != ""
|
|
self._clublog_xml_download_location = "cache/cty.xml"
|
|
if self._clublog_api_available:
|
|
self._lookup_lib_clublog_api = LookupLib(lookuptype="clublogapi", apikey=self._clublog_api_key)
|
|
success = self._download_clublog_ctyxml()
|
|
self._clublog_xml_available = success
|
|
if success:
|
|
self._lookup_lib_clublog_xml = LookupLib(lookuptype="clublogxml",
|
|
filename=self._clublog_xml_download_location)
|
|
self._clublog_callsign_data_cache = Cache('cache/clublog_callsign_lookup_cache')
|
|
|
|
# We also get a lookup of DXCC data from K0SWE to use for additional lookups of e.g. flags.
|
|
self._dxcc_json_download_location = "cache/dxcc.json"
|
|
success = self._download_dxcc_json()
|
|
if success:
|
|
with open(self._dxcc_json_download_location) as f:
|
|
tmp_dxcc_data = json.load(f)["dxcc"]
|
|
# Reformat as a map for faster lookup
|
|
self._dxcc_data = {}
|
|
for dxcc in tmp_dxcc_data:
|
|
self._dxcc_data[dxcc["entityCode"]] = dxcc
|
|
else:
|
|
logging.error("Could not download DXCC data, flags and similar data may be missing!")
|
|
|
|
# Precompile regex matches for DXCCs to improve efficiency when iterating through them
|
|
for dxcc in (self._dxcc_data.values() if self._dxcc_data else []):
|
|
dxcc["_prefixRegexCompiled"] = re.compile(dxcc["prefixRegex"])
|
|
|
|
def _download_country_files_cty_plist(self):
|
|
"""Download the cty.plist file from country-files.com on first startup. The pyhamtools lib can actually download and use
|
|
this itself, but it's occasionally offline which causes it to throw an error. By downloading it separately, we can
|
|
catch errors and handle them, falling back to a previous copy of the file in the cache, and we can use the
|
|
requests_cache library to prevent re-downloading too quickly if the software keeps restarting."""
|
|
|
|
try:
|
|
logging.info("Downloading Country-files.com cty.plist...")
|
|
response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.country-files.com/cty/cty.plist",
|
|
headers=HTTP_HEADERS).text
|
|
|
|
with open(self._country_files_cty_plist_download_location, "w") as f:
|
|
f.write(response)
|
|
f.flush()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error("Exception when downloading Clublog cty.xml", e)
|
|
return False
|
|
|
|
def _download_dxcc_json(self):
|
|
"""Download the dxcc.json file on first startup."""
|
|
|
|
try:
|
|
logging.info("Downloading dxcc.json...")
|
|
response = SEMI_STATIC_URL_DATA_CACHE.get(
|
|
"https://raw.githubusercontent.com/k0swe/dxcc-json/refs/heads/main/dxcc.json",
|
|
headers=HTTP_HEADERS).text
|
|
|
|
with open(self._dxcc_json_download_location, "w") as f:
|
|
f.write(response)
|
|
f.flush()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error("Exception when downloading dxcc.json", e)
|
|
return False
|
|
|
|
def _download_clublog_ctyxml(self):
|
|
"""Download the cty.xml (gzipped) file from Clublog on first startup, so we can use it in preference to querying the
|
|
database live if possible."""
|
|
|
|
try:
|
|
logging.info("Downloading Clublog cty.xml.gz...")
|
|
response = self._clublog_cty_xml_cache.get("https://cdn.clublog.org/cty.php?api=" + self._clublog_api_key,
|
|
headers=HTTP_HEADERS)
|
|
logging.info("Caching Clublog cty.xml.gz...")
|
|
open(self._clublog_xml_download_location + ".gz", 'wb').write(response.content)
|
|
with gzip.open(self._clublog_xml_download_location + ".gz", "rb") as uncompressed:
|
|
file_content = uncompressed.read()
|
|
logging.info("Caching Clublog cty.xml...")
|
|
with open(self._clublog_xml_download_location, "wb") as f:
|
|
f.write(file_content)
|
|
f.flush()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error("Exception when downloading Clublog cty.xml", e)
|
|
return False
|
|
|
|
def infer_country_from_callsign(self, call):
|
|
"""Infer a country name from a callsign"""
|
|
|
|
try:
|
|
# Start with the basic country-files.com-based decoder.
|
|
country = self._call_info_basic.get_country_name(call)
|
|
except (KeyError, ValueError):
|
|
country = None
|
|
# Couldn't get anything from basic call info database, try QRZ.com
|
|
if not country:
|
|
qrz_data = self._get_qrz_data_for_callsign(call)
|
|
if qrz_data and "country" in qrz_data:
|
|
country = qrz_data["country"]
|
|
# Couldn't get anything from QRZ.com database, try HamQTH
|
|
if not country:
|
|
hamqth_data = self._get_hamqth_data_for_callsign(call)
|
|
if hamqth_data and "country" in hamqth_data:
|
|
country = hamqth_data["country"]
|
|
# Couldn't get anything from HamQTH database, try Clublog data
|
|
if not country:
|
|
clublog_data = self._get_clublog_xml_data_for_callsign(call)
|
|
if clublog_data and "Name" in clublog_data:
|
|
country = clublog_data["Name"]
|
|
if not country:
|
|
clublog_data = self._get_clublog_api_data_for_callsign(call)
|
|
if clublog_data and "Name" in clublog_data:
|
|
country = clublog_data["Name"]
|
|
# Couldn't get anything from Clublog database, try DXCC data
|
|
if not country:
|
|
dxcc_data = self._get_dxcc_data_for_callsign(call)
|
|
if dxcc_data and "name" in dxcc_data:
|
|
country = dxcc_data["name"]
|
|
return country
|
|
|
|
def infer_dxcc_id_from_callsign(self, call):
|
|
"""Infer a DXCC ID from a callsign"""
|
|
|
|
try:
|
|
# Start with the basic country-files.com-based decoder.
|
|
dxcc = self._call_info_basic.get_adif_id(call)
|
|
except (KeyError, ValueError):
|
|
dxcc = None
|
|
# Couldn't get anything from basic call info database, try QRZ.com
|
|
if not dxcc:
|
|
qrz_data = self._get_qrz_data_for_callsign(call)
|
|
if qrz_data and "adif" in qrz_data:
|
|
dxcc = qrz_data["adif"]
|
|
# Couldn't get anything from QRZ.com database, try HamQTH
|
|
if not dxcc:
|
|
hamqth_data = self._get_hamqth_data_for_callsign(call)
|
|
if hamqth_data and "adif" in hamqth_data:
|
|
dxcc = hamqth_data["adif"]
|
|
# Couldn't get anything from HamQTH database, try Clublog data
|
|
if not dxcc:
|
|
clublog_data = self._get_clublog_xml_data_for_callsign(call)
|
|
if clublog_data and "DXCC" in clublog_data:
|
|
dxcc = clublog_data["DXCC"]
|
|
if not dxcc:
|
|
clublog_data = self._get_clublog_api_data_for_callsign(call)
|
|
if clublog_data and "DXCC" in clublog_data:
|
|
dxcc = clublog_data["DXCC"]
|
|
# Couldn't get anything from Clublog database, try DXCC data
|
|
if not dxcc:
|
|
dxcc_data = self._get_dxcc_data_for_callsign(call)
|
|
if dxcc_data and "entityCode" in dxcc_data:
|
|
dxcc = dxcc_data["entityCode"]
|
|
return dxcc
|
|
|
|
def infer_continent_from_callsign(self, call):
|
|
"""Infer a continent shortcode from a callsign"""
|
|
|
|
try:
|
|
# Start with the basic country-files.com-based decoder.
|
|
continent = self._call_info_basic.get_continent(call)
|
|
except (KeyError, ValueError):
|
|
continent = None
|
|
# Couldn't get anything from basic call info database, try HamQTH
|
|
if not continent:
|
|
hamqth_data = self._get_hamqth_data_for_callsign(call)
|
|
if hamqth_data and "continent" in hamqth_data:
|
|
continent = hamqth_data["continent"]
|
|
# Couldn't get anything from HamQTH database, try Clublog data
|
|
if not continent:
|
|
clublog_data = self._get_clublog_xml_data_for_callsign(call)
|
|
if clublog_data and "Continent" in clublog_data:
|
|
continent = clublog_data["Continent"]
|
|
if not continent:
|
|
clublog_data = self._get_clublog_api_data_for_callsign(call)
|
|
if clublog_data and "Continent" in clublog_data:
|
|
continent = clublog_data["Continent"]
|
|
# Couldn't get anything from Clublog database, try DXCC data
|
|
if not continent:
|
|
dxcc_data = self._get_dxcc_data_for_callsign(call)
|
|
# Some DXCCs are in two continents, if so don't use the continent data as we can't be sure
|
|
if dxcc_data and "continent" in dxcc_data and len(dxcc_data["continent"]) == 1:
|
|
continent = dxcc_data["continent"][0]
|
|
return continent
|
|
|
|
def infer_cq_zone_from_callsign(self, call):
|
|
"""Infer a CQ zone from a callsign"""
|
|
|
|
try:
|
|
# Start with the basic country-files.com-based decoder.
|
|
cqz = self._call_info_basic.get_cqz(call)
|
|
except (KeyError, ValueError):
|
|
cqz = None
|
|
# Couldn't get anything from basic call info database, try QRZ.com
|
|
if not cqz:
|
|
qrz_data = self._get_qrz_data_for_callsign(call)
|
|
if qrz_data and "cqz" in qrz_data:
|
|
cqz = qrz_data["cqz"]
|
|
# Couldn't get anything from QRZ.com database, try HamQTH
|
|
if not cqz:
|
|
hamqth_data = self._get_hamqth_data_for_callsign(call)
|
|
if hamqth_data and "cq" in hamqth_data:
|
|
cqz = hamqth_data["cq"]
|
|
# Couldn't get anything from HamQTH database, try Clublog data
|
|
if not cqz:
|
|
clublog_data = self._get_clublog_xml_data_for_callsign(call)
|
|
if clublog_data and "CQZ" in clublog_data:
|
|
cqz = clublog_data["CQZ"]
|
|
if not cqz:
|
|
clublog_data = self._get_clublog_api_data_for_callsign(call)
|
|
if clublog_data and "CQZ" in clublog_data:
|
|
cqz = clublog_data["CQZ"]
|
|
# Couldn't get anything from Clublog database, try DXCC data
|
|
if not cqz:
|
|
dxcc_data = self._get_dxcc_data_for_callsign(call)
|
|
# Some DXCCs are in multiple zones, if so don't use the zone data as we can't be sure
|
|
if dxcc_data and "cq" in dxcc_data and len(dxcc_data["cq"]) == 1:
|
|
cqz = dxcc_data["cq"][0]
|
|
return cqz
|
|
|
|
def infer_itu_zone_from_callsign(self, call):
|
|
"""Infer a ITU zone from a callsign"""
|
|
|
|
try:
|
|
# Start with the basic country-files.com-based decoder.
|
|
ituz = self._call_info_basic.get_ituz(call)
|
|
except (KeyError, ValueError):
|
|
ituz = None
|
|
# Couldn't get anything from basic call info database, try QRZ.com
|
|
if not ituz:
|
|
qrz_data = self._get_qrz_data_for_callsign(call)
|
|
if qrz_data and "ituz" in qrz_data:
|
|
ituz = qrz_data["ituz"]
|
|
# Couldn't get anything from QRZ.com database, try HamQTH
|
|
if not ituz:
|
|
hamqth_data = self._get_hamqth_data_for_callsign(call)
|
|
if hamqth_data and "itu" in hamqth_data:
|
|
ituz = hamqth_data["itu"]
|
|
# Couldn't get anything from HamQTH database, Clublog doesn't provide this, so try DXCC data
|
|
if not ituz:
|
|
dxcc_data = self._get_dxcc_data_for_callsign(call)
|
|
# Some DXCCs are in multiple zones, if so don't use the zone data as we can't be sure
|
|
if dxcc_data and "itu" in dxcc_data and len(dxcc_data["itu"]) == 1:
|
|
ituz = dxcc_data["itu"]
|
|
return ituz
|
|
|
|
def get_flag_for_dxcc(self, dxcc):
|
|
"""Get an emoji flag for a given DXCC entity ID"""
|
|
|
|
return self._dxcc_data[dxcc]["flag"] if dxcc in self._dxcc_data else None
|
|
|
|
def infer_name_from_callsign_online_lookup(self, call):
|
|
"""Infer an operator name from a callsign (requires QRZ.com/HamQTH)"""
|
|
|
|
data = self._get_qrz_data_for_callsign(call)
|
|
if data and "fname" in data:
|
|
name = data["fname"]
|
|
if "name" in data:
|
|
name = name + " " + data["name"]
|
|
return name
|
|
data = self._get_hamqth_data_for_callsign(call)
|
|
if data and "nick" in data:
|
|
return data["nick"]
|
|
else:
|
|
return None
|
|
|
|
def infer_latlon_from_callsign_online_lookup(self, call):
|
|
"""Infer a latitude and longitude from a callsign (requires QRZ.com/HamQTH)
|
|
Coordinates that look default are rejected (apologies if your position really is 0,0, enjoy your voyage)"""
|
|
|
|
data = self._get_qrz_data_for_callsign(call)
|
|
if data and "latitude" in data and "longitude" in data and (
|
|
float(data["latitude"]) != 0 or float(data["longitude"]) != 0) and -89.9 < float(
|
|
data["latitude"]) < 89.9:
|
|
return [float(data["latitude"]), float(data["longitude"])]
|
|
data = self._get_hamqth_data_for_callsign(call)
|
|
if data and "latitude" in data and "longitude" in data and (
|
|
float(data["latitude"]) != 0 or float(data["longitude"]) != 0) and -89.9 < float(
|
|
data["latitude"]) < 89.9:
|
|
return [float(data["latitude"]), float(data["longitude"])]
|
|
else:
|
|
return None
|
|
|
|
def infer_grid_from_callsign_online_lookup(self, call):
|
|
"""Infer a grid locator from a callsign (requires QRZ.com/HamQTH).
|
|
Grids that look default are rejected (apologies if your grid really is AA00aa, enjoy your research)"""
|
|
|
|
data = self._get_qrz_data_for_callsign(call)
|
|
if data and "locator" in data and data["locator"].upper() != "AA00" and data["locator"].upper() != "AA00AA" and \
|
|
data["locator"].upper() != "AA00AA00":
|
|
return data["locator"]
|
|
data = self._get_hamqth_data_for_callsign(call)
|
|
if data and "grid" in data and data["grid"].upper() != "AA00" and data["grid"].upper() != "AA00AA" and data[
|
|
"grid"].upper() != "AA00AA00":
|
|
return data["grid"]
|
|
else:
|
|
return None
|
|
|
|
def infer_qth_from_callsign_online_lookup(self, call):
|
|
"""Infer a textual QTH from a callsign (requires QRZ.com/HamQTH)"""
|
|
|
|
data = self._get_qrz_data_for_callsign(call)
|
|
if data and "addr2" in data:
|
|
return data["addr2"]
|
|
data = self._get_hamqth_data_for_callsign(call)
|
|
if data and "qth" in data:
|
|
return data["qth"]
|
|
else:
|
|
return None
|
|
|
|
def infer_latlon_from_callsign_dxcc(self, call):
|
|
"""Infer a latitude and longitude from a callsign (using DXCC, probably very inaccurate)"""
|
|
|
|
try:
|
|
data = self._call_info_basic.get_lat_long(call)
|
|
if data and "latitude" in data and "longitude" in data:
|
|
loc = [float(data["latitude"]), float(data["longitude"])]
|
|
else:
|
|
loc = None
|
|
except KeyError:
|
|
loc = None
|
|
# Couldn't get anything from basic call info database, try Clublog data
|
|
if not loc:
|
|
data = self._get_clublog_xml_data_for_callsign(call)
|
|
if data and "Lat" in data and "Lon" in data:
|
|
loc = [float(data["Lat"]), float(data["Lon"])]
|
|
if not loc:
|
|
data = self._get_clublog_api_data_for_callsign(call)
|
|
if data and "Lat" in data and "Lon" in data:
|
|
loc = [float(data["Lat"]), float(data["Lon"])]
|
|
return loc
|
|
|
|
def infer_grid_from_callsign_dxcc(self, call):
|
|
"""Infer a grid locator from a callsign (using DXCC, probably very inaccurate)"""
|
|
|
|
latlon = self.infer_latlon_from_callsign_dxcc(call)
|
|
grid = None
|
|
try:
|
|
grid = latlong_to_locator(latlon[0], latlon[1], 8)
|
|
except:
|
|
logging.debug("Invalid lat/lon received for DXCC")
|
|
return grid
|
|
|
|
def _get_qrz_data_for_callsign(self, call):
|
|
"""Utility method to get QRZ.com data from cache if possible, if not get it from the API and cache it"""
|
|
|
|
# Fetch from cache if we can, otherwise fetch from the API and cache it
|
|
if call in self._qrz_callsign_data_cache:
|
|
return self._qrz_callsign_data_cache.get(call)
|
|
elif self._qrz_available:
|
|
try:
|
|
data = self._lookup_lib_qrz.lookup_callsign(callsign=call)
|
|
self._qrz_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# QRZ had no info for the call, but maybe it had prefixes or suffixes. Try again with the base call.
|
|
try:
|
|
data = self._lookup_lib_qrz.lookup_callsign(callsign=callinfo.Callinfo.get_homecall(call))
|
|
self._qrz_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# QRZ had no info for the call, that's OK. Cache a None so we don't try to look this up again
|
|
self._qrz_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
|
|
return None
|
|
except Exception:
|
|
# General exception like a timeout when communicating with QRZ. Return None this time, but don't cache
|
|
# that, so we can try again next time.
|
|
logging.error("Exception when looking up QRZ data")
|
|
return None
|
|
else:
|
|
return None
|
|
|
|
def _get_hamqth_data_for_callsign(self, call):
|
|
"""Utility method to get HamQTH data from cache if possible, if not get it from the API and cache it"""
|
|
|
|
# Fetch from cache if we can, otherwise fetch from the API and cache it
|
|
if call in self._hamqth_callsign_data_cache:
|
|
return self._hamqth_callsign_data_cache.get(call)
|
|
elif self._hamqth_available:
|
|
try:
|
|
# First we need to log in and get a session token.
|
|
session_data = self._hamqth_session_lookup_cache.get(
|
|
self._hamqth_base_url + "?u=" + urllib.parse.quote_plus(config["hamqth-username"]) +
|
|
"&p=" + urllib.parse.quote_plus(config["hamqth-password"]), headers=HTTP_HEADERS).content
|
|
dict_data = xmltodict.parse(session_data)
|
|
if "session_id" in dict_data["HamQTH"]["session"]:
|
|
session_id = dict_data["HamQTH"]["session"]["session_id"]
|
|
|
|
# Now look up the actual data.
|
|
try:
|
|
lookup_data = SEMI_STATIC_URL_DATA_CACHE.get(
|
|
self._hamqth_base_url + "?id=" + session_id + "&callsign=" + urllib.parse.quote_plus(
|
|
call) + "&prg=" + HAMQTH_PRG, headers=HTTP_HEADERS).content
|
|
data = xmltodict.parse(lookup_data)["HamQTH"]["search"]
|
|
self._hamqth_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# HamQTH had no info for the call, but maybe it had prefixes or suffixes. Try again with the base call.
|
|
try:
|
|
lookup_data = SEMI_STATIC_URL_DATA_CACHE.get(
|
|
self._hamqth_base_url + "?id=" + session_id + "&callsign=" + urllib.parse.quote_plus(
|
|
callinfo.Callinfo.get_homecall(call)) + "&prg=" + HAMQTH_PRG,
|
|
headers=HTTP_HEADERS).content
|
|
data = xmltodict.parse(lookup_data)["HamQTH"]["search"]
|
|
self._hamqth_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# HamQTH had no info for the call, that's OK. Cache a None so we don't try to look this up again
|
|
self._hamqth_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
|
|
return None
|
|
|
|
else:
|
|
logging.warning("HamQTH login details incorrect, failed to look up with HamQTH.")
|
|
except:
|
|
logging.error("Exception when looking up HamQTH data")
|
|
return None
|
|
return None
|
|
|
|
def _get_clublog_api_data_for_callsign(self, call):
|
|
"""Utility method to get Clublog API data from cache if possible, if not get it from the API and cache it"""
|
|
|
|
# Fetch from cache if we can, otherwise fetch from the API and cache it
|
|
if call in self._clublog_callsign_data_cache:
|
|
return self._clublog_callsign_data_cache.get(call)
|
|
elif self._clublog_api_available:
|
|
try:
|
|
data = self._lookup_lib_clublog_api.lookup_callsign(callsign=call)
|
|
self._clublog_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# Clublog had no info for the call, but maybe it had prefixes or suffixes. Try again with the base call.
|
|
try:
|
|
data = self._lookup_lib_clublog_api.lookup_callsign(callsign=callinfo.Callinfo.get_homecall(call))
|
|
self._clublog_callsign_data_cache.add(call, data, expire=604800) # 1 week in seconds
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# Clublog had no info for the call, that's OK. Cache a None so we don't try to look this up again
|
|
self._clublog_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
|
|
return None
|
|
except APIKeyMissingError:
|
|
# User API key was wrong, warn
|
|
logging.error("Could not look up via Clublog API, key " + self._clublog_api_key + " was rejected.")
|
|
return None
|
|
else:
|
|
return None
|
|
|
|
def _get_clublog_xml_data_for_callsign(self, call):
|
|
"""Utility method to get Clublog XML data from file"""
|
|
|
|
if self._clublog_xml_available:
|
|
try:
|
|
data = self._lookup_lib_clublog_xml.lookup_callsign(callsign=call)
|
|
return data
|
|
except (KeyError, ValueError):
|
|
# Clublog had no info for the call, that's OK. Cache a None so we don't try to look this up again
|
|
self._clublog_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
|
|
return None
|
|
else:
|
|
return None
|
|
|
|
def _get_dxcc_data_for_callsign(self, call):
|
|
"""Utility method to get generic DXCC data from our lookup table, if we can find it"""
|
|
|
|
for entry in self._dxcc_data.values():
|
|
if entry["_prefixRegexCompiled"].match(call):
|
|
return entry
|
|
return None
|
|
|
|
def stop(self):
|
|
"""Shutdown method to close down any caches neatly."""
|
|
|
|
self._qrz_callsign_data_cache.close()
|
|
self._clublog_callsign_data_cache.close()
|
|
|
|
|
|
# Singleton object
|
|
lookup_helper = LookupHelper()
|
|
|
|
def infer_mode_from_comment(comment):
|
|
"""Infer a mode from the comment"""
|
|
|
|
for mode in ALL_MODES:
|
|
if mode in comment.upper():
|
|
return mode
|
|
for mode in MODE_ALIASES.keys():
|
|
if mode in comment.upper():
|
|
return MODE_ALIASES[mode]
|
|
return None
|
|
|
|
|
|
def infer_mode_type_from_mode(mode):
|
|
"""Infer a "mode family" from a mode."""
|
|
|
|
if mode.upper() in CW_MODES:
|
|
return "CW"
|
|
elif mode.upper() in PHONE_MODES:
|
|
return "PHONE"
|
|
elif mode.upper() in DATA_MODES:
|
|
return "DATA"
|
|
else:
|
|
if mode.upper() != "OTHER":
|
|
logging.warning("Found an unrecognised mode: " + mode + ". Developer should categorise this.")
|
|
return None
|
|
|
|
|
|
def infer_band_from_freq(freq):
|
|
"""Infer a band from a frequency in Hz"""
|
|
|
|
for b in BANDS:
|
|
if b.start_freq <= freq <= b.end_freq:
|
|
return b
|
|
return UNKNOWN_BAND
|
|
|
|
|
|
def infer_mode_from_frequency(freq):
|
|
"""Infer a mode from the frequency (in Hz) according to the band plan. Just a guess really."""
|
|
|
|
try:
|
|
khz = freq / 1000.0
|
|
mode = freq_to_band(khz)["mode"]
|
|
# Some additional common digimode ranges in addition to what the 3rd-party freq_to_band function returns.
|
|
# This is mostly here just because freq_to_band is very specific about things like FT8 frequencies, and e.g.
|
|
# a spot at 7074.5 kHz will be indicated as LSB, even though it's clearly in the FT8 range. Future updates
|
|
# might include other common digimode centres of activity here, but this achieves the main goal of keeping
|
|
# large numbers of clearly-FT* spots off the list of people filtering out digimodes.
|
|
if (7074 <= khz < 7077) or (10136 <= khz < 10139) or (14074 <= khz < 14077) or (18100 <= khz < 18103) or (
|
|
21074 <= khz < 21077) or (24915 <= khz < 24918) or (28074 <= khz < 28077):
|
|
mode = "FT8"
|
|
if (7047.5 <= khz < 7050.5) or (10140 <= khz < 10143) or (14080 <= khz < 14083) or (
|
|
18104 <= khz < 18107) or (21140 <= khz < 21143) or (24919 <= khz < 24922) or (28180 <= khz < 28183):
|
|
mode = "FT4"
|
|
return mode
|
|
except KeyError:
|
|
return None
|