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(): 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