diff --git a/config-example.yml b/config-example.yml index 99922b1..c0ef473 100644 --- a/config-example.yml +++ b/config-example.yml @@ -123,9 +123,13 @@ max-alert-age-sec: 604800 # Login for QRZ.com to look up information. Optional. You will need an "XML Subscriber" (paid) package to retrieve all # the data for a callsign via their system. -qrz-username: "N0CALL" +qrz-username: "" qrz-password: "" +# Login for HamQTH to look up information. Optional. +hamqth-username: "" +hamqth-password: "" + # API key for Clublog to look up information. Optional. You sill need to request one via their helpdesk portal if you # want to use callsign lookups from Clublog. clublog-api-key: "" diff --git a/core/constants.py b/core/constants.py index 9a9703e..6b37570 100644 --- a/core/constants.py +++ b/core/constants.py @@ -8,6 +8,7 @@ SOFTWARE_VERSION = "0.1" # HTTP headers used for spot providers that use HTTP HTTP_HEADERS = {"User-Agent": SOFTWARE_NAME + ", v" + SOFTWARE_VERSION + " (operated by " + SERVER_OWNER_CALLSIGN + ")"} +HAMQTH_PRG = (SOFTWARE_NAME + " v" + SOFTWARE_VERSION + " operated by " + SERVER_OWNER_CALLSIGN).replace(" ", "_") # Special Interest Groups SIGS = [ diff --git a/core/lookup_helper.py b/core/lookup_helper.py index 1a45b2c..57a3d71 100644 --- a/core/lookup_helper.py +++ b/core/lookup_helper.py @@ -1,7 +1,9 @@ import gzip import logging +import urllib.parse from datetime import timedelta +import xmltodict from diskcache import Cache from pyhamtools import LookupLib, Callinfo, callinfo from pyhamtools.exceptions import APIKeyMissingError @@ -12,7 +14,8 @@ 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, \ - QRZCQ_CALLSIGN_LOOKUP_DATA, HTTP_HEADERS + QRZCQ_CALLSIGN_LOOKUP_DATA, HTTP_HEADERS, HAMQTH_PRG + # Singleton class that provides lookup functionality. class LookupHelper: @@ -32,15 +35,24 @@ class LookupHelper: 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 def start(self): - # Lookup helpers from pyhamtools. We use four (!) 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, Clublog (live API request), and - # Clublog (XML download). The lookup functions iterate through these in a sensible order, looking for suitable data. + # 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: @@ -50,12 +62,15 @@ class LookupHelper: self.LOOKUP_LIB_BASIC = LookupLib(lookuptype="countryfile") self.CALL_INFO_BASIC = Callinfo(self.LOOKUP_LIB_BASIC) - self.QRZ_AVAILABLE = config["qrz-password"] != "" + 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 != "" @@ -77,7 +92,7 @@ class LookupHelper: 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 + headers=HTTP_HEADERS).text with open(self.COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION, "w") as f: f.write(response) @@ -146,7 +161,12 @@ class LookupHelper: 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 Clublog data + # 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: @@ -164,7 +184,6 @@ class LookupHelper: # Infer a DXCC ID from a callsign def infer_dxcc_id_from_callsign(self, call): - self.get_clublog_xml_data_for_callsign("M0TRT") try: # Start with the basic country-files.com-based decoder. dxcc = self.CALL_INFO_BASIC.get_adif_id(call) @@ -175,7 +194,12 @@ class LookupHelper: 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 Clublog data + # 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: @@ -198,7 +222,12 @@ class LookupHelper: continent = self.CALL_INFO_BASIC.get_continent(call) except (KeyError, ValueError) as e: continent = None - # Couldn't get anything from basic call info database, try Clublog data + # 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: + country = 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: @@ -226,7 +255,12 @@ class LookupHelper: 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 Clublog data + # 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: @@ -254,13 +288,87 @@ class LookupHelper: 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, Clublog doesn't provide this, so try QRZCQ data + # 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 QRZCQ data if not ituz: qrzcq_data = self.get_qrzcq_data_for_callsign(call) if qrzcq_data and "ituz" in qrzcq_data: ituz = qrzcq_data["ituz"] return ituz + # Infer an operator name from a callsign (requires QRZ.com/HamQTH) + def infer_name_from_callsign(self, call): + 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 + + # Infer a latitude and longitude from a callsign (requires QRZ.com/HamQTH) + def infer_latlon_from_callsign_qrz(self, call): + data = self.get_qrz_data_for_callsign(call) + if data and "latitude" in data and "longitude" in data: + return [data["latitude"], data["longitude"]] + data = self.get_hamqth_data_for_callsign(call) + if data and "latitude" in data and "longitude" in data: + return [data["latitude"], data["longitude"]] + else: + return None + + # Infer a grid locator from a callsign (requires QRZ.com/HamQTH) + def infer_grid_from_callsign_qrz(self, call): + data = self.get_qrz_data_for_callsign(call) + if data and "locator" in data: + return data["locator"] + data = self.get_hamqth_data_for_callsign(call) + if data and "grid" in data: + return data["grid"] + else: + return None + + # Infer a latitude and longitude from a callsign (using DXCC, probably very inaccurate) + def infer_latlon_from_callsign_dxcc(self, call): + try: + data = self.CALL_INFO_BASIC.get_lat_long(call) + if data and "latitude" in data and "longitude" in data: + loc = [data["latitude"], 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 = [data["Lat"], 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 = [data["Lat"], data["Lon"]] + return loc + + # Infer a grid locator from a callsign (using DXCC, probably very inaccurate) + def infer_grid_from_callsign_dxcc(self, call): + latlon = self.infer_latlon_from_callsign_dxcc(call) + return latlong_to_locator(latlon[0], latlon[1], 8) + + # Infer a mode from the frequency (in Hz) according to the band plan. Just a guess really. + def infer_mode_from_frequency(self, freq): + try: + return freq_to_band(freq / 1000.0)["mode"] + except KeyError: + return None + # Utility method to get QRZ.com data from cache if possible, if not get it from the API and cache it def get_qrz_data_for_callsign(self, call): # Fetch from cache if we can, otherwise fetch from the API and cache it @@ -284,6 +392,49 @@ class LookupHelper: else: return None + # Utility method to get HamQTH data from cache if possible, if not get it from the API and cache it + def get_hamqth_data_for_callsign(self, call): + # 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.warn("HamQTH login details incorrect, failed to look up with HamQTH.") + except: + logging.error("Exception when looking up HamQTH data") + return None + # Utility method to get Clublog API data from cache if possible, if not get it from the API and cache it def get_clublog_api_data_for_callsign(self, call): # Fetch from cache if we can, otherwise fetch from the API and cache it @@ -332,70 +483,11 @@ class LookupHelper: return entry return None - # Infer an operator name from a callsign (requires QRZ.com) - def infer_name_from_callsign(self, call): - 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 - else: - return None - - # Infer a latitude and longitude from a callsign (requires QRZ.com) - def infer_latlon_from_callsign_qrz(self, call): - data = self.get_qrz_data_for_callsign(call) - if data and "latitude" in data and "longitude" in data: - return [data["latitude"], data["longitude"]] - else: - return None - - # Infer a grid locator from a callsign (requires QRZ.com) - def infer_grid_from_callsign_qrz(self, call): - data = self.get_qrz_data_for_callsign(call) - if data and "locator" in data: - return data["locator"] - else: - return None - - # Infer a latitude and longitude from a callsign (using DXCC, probably very inaccurate) - def infer_latlon_from_callsign_dxcc(self, call): - try: - data = self.CALL_INFO_BASIC.get_lat_long(call) - if data and "latitude" in data and "longitude" in data: - loc = [data["latitude"], 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 = [data["Lat"], 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 = [data["Lat"], data["Lon"]] - return loc - - # Infer a grid locator from a callsign (using DXCC, probably very inaccurate) - def infer_grid_from_callsign_dxcc(self, call): - latlon = self.infer_latlon_from_callsign_dxcc(call) - return latlong_to_locator(latlon[0], latlon[1], 8) - - # Infer a mode from the frequency (in Hz) according to the band plan. Just a guess really. - def infer_mode_from_frequency(self, freq): - try: - return freq_to_band(freq / 1000.0)["mode"] - except KeyError: - return None - # Shutdown method to close down any caches neatly. def stop(self): self.QRZ_CALLSIGN_DATA_CACHE.close() self.CLUBLOG_CALLSIGN_DATA_CACHE.close() + # Singleton object lookup_helper = LookupHelper()