diff --git a/core/lookup_helper.py b/core/lookup_helper.py new file mode 100644 index 0000000..552b87d --- /dev/null +++ b/core/lookup_helper.py @@ -0,0 +1,393 @@ +import gzip +import logging +from datetime import timedelta + +from diskcache import Cache +from pyhamtools import LookupLib, 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.config import config +from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, \ + QRZCQ_CALLSIGN_LOOKUP_DATA, HTTP_HEADERS + +# Singleton class that provides lookup functionality. +class LookupHelper: + + # 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. + def __init__(self): + 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.CALL_INFO_BASIC = None + self.LOOKUP_LIB_BASIC = None + self.COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION = None + self.COUNTRY_FILES_CTY_PLIST_CACHE = 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. + self.COUNTRY_FILES_CTY_PLIST_CACHE = CachedSession("cache/country_files_city_plist_cache", + expire_after=timedelta(days=10)) + 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-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.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') + + # 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. + def download_country_files_cty_plist(self): + try: + logging.info("Downloading Country-files.com cty.plist...") + response = self.COUNTRY_FILES_CTY_PLIST_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 + + # 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. + def download_clublog_ctyxml(self): + try: + logging.info("Downloading Clublog cty.xml...") + response = self.CLUBLOG_CTY_XML_CACHE.get("https://cdn.clublog.org/cty.php?api=" + self.CLUBLOG_API_KEY, + headers=HTTP_HEADERS).raw + with gzip.GzipFile(fileobj=response) as uncompressed: + file_content = uncompressed.read() + + 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 + + # Infer a mode from the comment + def infer_mode_from_comment(self, comment): + for mode in ALL_MODES: + if mode in comment.upper(): + return mode + return None + + # Infer a "mode family" from a mode. + def infer_mode_type_from_mode(self, 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.warn("Found an unrecognised mode: " + mode + ". Developer should categorise this.") + return None + + # Infer a band from a frequency in Hz + def infer_band_from_freq(self, freq): + for b in BANDS: + if b.start_freq <= freq <= b.end_freq: + return b + return UNKNOWN_BAND + + # Infer a country name from a callsign + def infer_country_from_callsign(self, call): + try: + # Start with the basic country-files.com-based decoder. + country = self.CALL_INFO_BASIC.get_country_name(call) + except KeyError as e: + 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 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 QRZCQ data + if not country: + qrzcq_data = self.get_qrzcq_data_for_callsign(call) + if qrzcq_data and "country" in qrzcq_data: + country = qrzcq_data["country"] + return country + + # 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) + except KeyError as e: + 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 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 QRZCQ data + if not dxcc: + qrzcq_data = self.get_qrzcq_data_for_callsign(call) + if qrzcq_data and "dxcc" in qrzcq_data: + dxcc = qrzcq_data["dxcc"] + return dxcc + + # Infer a continent shortcode from a callsign + def infer_continent_from_callsign(self, call): + try: + # Start with the basic country-files.com-based decoder. + continent = self.CALL_INFO_BASIC.get_continent(call) + except KeyError as e: + continent = None + # Couldn't get anything from basic call info 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 QRZCQ data + if not continent: + qrzcq_data = self.get_qrzcq_data_for_callsign(call) + if qrzcq_data and "continent" in qrzcq_data: + continent = qrzcq_data["continent"] + return continent + + # Infer a CQ zone from a callsign + def infer_cq_zone_from_callsign(self, call): + try: + # Start with the basic country-files.com-based decoder. + cqz = self.CALL_INFO_BASIC.get_cqz(call) + except KeyError as e: + 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 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 QRZCQ data + if not cqz: + qrzcq_data = self.get_qrzcq_data_for_callsign(call) + if qrzcq_data and "cqz" in qrzcq_data: + cqz = qrzcq_data["cqz"] + return cqz + + # Infer a ITU zone from a callsign + def infer_itu_zone_from_callsign(self, call): + try: + # Start with the basic country-files.com-based decoder. + ituz = self.CALL_INFO_BASIC.get_ituz(call) + except KeyError as e: + 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, 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 + + # 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 + qrz_data = self.QRZ_CALLSIGN_DATA_CACHE.get(call) + if qrz_data: + return qrz_data + 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: + # 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 + else: + 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 + clublog_data = self.CLUBLOG_CALLSIGN_DATA_CACHE.get(call) + if clublog_data: + return clublog_data + 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: + # 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 + + # Utility method to get Clublog XML data from file + def get_clublog_xml_data_for_callsign(self, call): + if self.CLUBLOG_XML_AVAILABLE: + try: + data = self.LOOKUP_LIB_CLUBLOG_XML.lookup_callsign(callsign=call) + return data + except KeyError: + # 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 + + # Utility method to get QRZCQ data from our constants table, if we can find it + def get_qrzcq_data_for_callsign(self, call): + # Iterate in reverse order - see comments on the data structure itself + for entry in reversed(QRZCQ_CALLSIGN_LOOKUP_DATA): + if call.startswith(entry["prefix"]): + 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() diff --git a/core/utils.py b/core/utils.py deleted file mode 100644 index c3a9ac2..0000000 --- a/core/utils.py +++ /dev/null @@ -1,378 +0,0 @@ -import gzip -import logging -from datetime import timedelta - -from diskcache import Cache -from pyhamtools import LookupLib, 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.config import config -from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, \ - QRZCQ_CALLSIGN_LOOKUP_DATA, HTTP_HEADERS - - -# 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. -def download_country_files_cty_plist(): - try: - logging.info("Downloading Country-files.com cty.plist...") - response = COUNTRY_FILES_CTY_PLIST_CACHE.get("https://www.country-files.com/cty/cty.plist", - headers=HTTP_HEADERS).text - - with open(COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION, "w") as f: - f.write(response) - return True - - except Exception as e: - logging.error("Exception when downloading Clublog cty.xml", e) - return False - -# 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. -def download_clublog_ctyxml(): - try: - logging.info("Downloading Clublog cty.xml...") - response = CLUBLOG_CTY_XML_CACHE.get("https://cdn.clublog.org/cty.php?api=" + CLUBLOG_API_KEY, - headers=HTTP_HEADERS).raw - with gzip.GzipFile(fileobj=response) as uncompressed: - file_content = uncompressed.read() - - with open(CLUBLOG_XML_DOWNLOAD_LOCATION, "wb") as f: - f.write(file_content) - return True - - except Exception as e: - logging.error("Exception when downloading Clublog cty.xml", e) - return False - -# 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. -COUNTRY_FILES_CTY_PLIST_CACHE = CachedSession("cache/country_files_city_plist_cache", expire_after=timedelta(days=10)) -COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION = "cache/cty.plist" -download_country_files_cty_plist() -LOOKUP_LIB_BASIC = LookupLib(lookuptype="countryfile", filename=COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION) -CALL_INFO_BASIC = Callinfo(LOOKUP_LIB_BASIC) - -QRZ_AVAILABLE = config["qrz-password"] != "" -if QRZ_AVAILABLE: - LOOKUP_LIB_QRZ = LookupLib(lookuptype="qrz", username=config["qrz-username"], pwd=config["qrz-password"]) -QRZ_CALLSIGN_DATA_CACHE = Cache('cache/qrz_callsign_lookup_cache') - -CLUBLOG_API_KEY = config["clublog-api-key"] -CLUBLOG_CTY_XML_CACHE = CachedSession("cache/clublog_cty_xml_cache", expire_after=timedelta(days=10)) -CLUBLOG_API_AVAILABLE = CLUBLOG_API_KEY != "" -CLUBLOG_XML_DOWNLOAD_LOCATION = "cache/cty.xml" -if CLUBLOG_API_AVAILABLE: - LOOKUP_LIB_CLUBLOG_API = LookupLib(lookuptype="clublogapi", apikey=CLUBLOG_API_KEY) - success = download_clublog_ctyxml() - CLUBLOG_XML_AVAILABLE = success - if success: - LOOKUP_LIB_CLUBLOG_XML = LookupLib(lookuptype="clublogxml", filename=CLUBLOG_XML_DOWNLOAD_LOCATION) -CLUBLOG_CALLSIGN_DATA_CACHE = Cache('cache/clublog_callsign_lookup_cache') - - -# Infer a mode from the comment -def infer_mode_from_comment(comment): - for mode in ALL_MODES: - if mode in comment.upper(): - return mode - return None - - -# Infer a "mode family" from a mode. -def infer_mode_type_from_mode(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.warn("Found an unrecognised mode: " + mode + ". Developer should categorise this.") - return None - - -# Infer a band from a frequency in Hz -def infer_band_from_freq(freq): - for b in BANDS: - if b.start_freq <= freq <= b.end_freq: - return b - return UNKNOWN_BAND - - -# Infer a country name from a callsign -def infer_country_from_callsign(call): - try: - # Start with the basic country-files.com-based decoder. - country = CALL_INFO_BASIC.get_country_name(call) - except KeyError as e: - country = None - # Couldn't get anything from basic call info database, try QRZ.com - if not country: - qrz_data = 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 - if not country: - clublog_data = get_clublog_xml_data_for_callsign(call) - if clublog_data and "Name" in clublog_data: - country = clublog_data["Name"] - if not country: - clublog_data = 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 QRZCQ data - if not country: - qrzcq_data = get_qrzcq_data_for_callsign(call) - if qrzcq_data and "country" in qrzcq_data: - country = qrzcq_data["country"] - return country - - -# Infer a DXCC ID from a callsign -def infer_dxcc_id_from_callsign(call): - get_clublog_xml_data_for_callsign("M0TRT") - try: - # Start with the basic country-files.com-based decoder. - dxcc = CALL_INFO_BASIC.get_adif_id(call) - except KeyError as e: - dxcc = None - # Couldn't get anything from basic call info database, try QRZ.com - if not dxcc: - qrz_data = 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 - if not dxcc: - clublog_data = get_clublog_xml_data_for_callsign(call) - if clublog_data and "DXCC" in clublog_data: - dxcc = clublog_data["DXCC"] - if not dxcc: - clublog_data = 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 QRZCQ data - if not dxcc: - qrzcq_data = get_qrzcq_data_for_callsign(call) - if qrzcq_data and "dxcc" in qrzcq_data: - dxcc = qrzcq_data["dxcc"] - return dxcc - - -# Infer a continent shortcode from a callsign -def infer_continent_from_callsign(call): - try: - # Start with the basic country-files.com-based decoder. - continent = CALL_INFO_BASIC.get_continent(call) - except KeyError as e: - continent = None - # Couldn't get anything from basic call info database, try Clublog data - if not continent: - clublog_data = get_clublog_xml_data_for_callsign(call) - if clublog_data and "Continent" in clublog_data: - continent = clublog_data["Continent"] - if not continent: - clublog_data = 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 QRZCQ data - if not continent: - qrzcq_data = get_qrzcq_data_for_callsign(call) - if qrzcq_data and "continent" in qrzcq_data: - continent = qrzcq_data["continent"] - return continent - - -# Infer a CQ zone from a callsign -def infer_cq_zone_from_callsign(call): - try: - # Start with the basic country-files.com-based decoder. - cqz = CALL_INFO_BASIC.get_cqz(call) - except KeyError as e: - cqz = None - # Couldn't get anything from basic call info database, try QRZ.com - if not cqz: - qrz_data = 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 - if not cqz: - clublog_data = get_clublog_xml_data_for_callsign(call) - if clublog_data and "CQZ" in clublog_data: - cqz = clublog_data["CQZ"] - if not cqz: - clublog_data = 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 QRZCQ data - if not cqz: - qrzcq_data = get_qrzcq_data_for_callsign(call) - if qrzcq_data and "cqz" in qrzcq_data: - cqz = qrzcq_data["cqz"] - return cqz - - -# Infer a ITU zone from a callsign -def infer_itu_zone_from_callsign(call): - try: - # Start with the basic country-files.com-based decoder. - ituz = CALL_INFO_BASIC.get_ituz(call) - except KeyError as e: - ituz = None - # Couldn't get anything from basic call info database, try QRZ.com - if not ituz: - qrz_data = 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 - if not ituz: - qrzcq_data = get_qrzcq_data_for_callsign(call) - if qrzcq_data and "ituz" in qrzcq_data: - ituz = qrzcq_data["ituz"] - return ituz - - -# 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(call): - # Fetch from cache if we can, otherwise fetch from the API and cache it - qrz_data = QRZ_CALLSIGN_DATA_CACHE.get(call) - if qrz_data: - return qrz_data - elif QRZ_AVAILABLE: - try: - data = LOOKUP_LIB_QRZ.lookup_callsign(callsign=call) - QRZ_CALLSIGN_DATA_CACHE.add(call, data, expire=604800) # 1 week in seconds - return data - except KeyError: - # QRZ had no info for the call, that's OK. Cache a None so we don't try to look this up again - QRZ_CALLSIGN_DATA_CACHE.add(call, None, expire=604800) # 1 week in seconds - return None - else: - 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(call): - # Fetch from cache if we can, otherwise fetch from the API and cache it - clublog_data = CLUBLOG_CALLSIGN_DATA_CACHE.get(call) - if clublog_data: - return clublog_data - elif CLUBLOG_API_AVAILABLE: - try: - data = LOOKUP_LIB_CLUBLOG_API.lookup_callsign(callsign=call) - CLUBLOG_CALLSIGN_DATA_CACHE.add(call, data, expire=604800) # 1 week in seconds - return data - except KeyError: - # Clublog had no info for the call, that's OK. Cache a None so we don't try to look this up again - 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 " + CLUBLOG_API_KEY + " was rejected.") - return None - else: - return None - - -# Utility method to get Clublog XML data from file -def get_clublog_xml_data_for_callsign(call): - if CLUBLOG_XML_AVAILABLE: - try: - data = LOOKUP_LIB_CLUBLOG_XML.lookup_callsign(callsign=call) - return data - except KeyError: - # Clublog had no info for the call, that's OK. Cache a None so we don't try to look this up again - CLUBLOG_CALLSIGN_DATA_CACHE.add(call, None, expire=604800) # 1 week in seconds - return None - else: - return None - - -# Utility method to get QRZCQ data from our constants table, if we can find it -def get_qrzcq_data_for_callsign(call): - # Iterate in reverse order - see comments on the data structure itself - for entry in reversed(QRZCQ_CALLSIGN_LOOKUP_DATA): - if call.startswith(entry["prefix"]): - return entry - return None - - -# Infer an operator name from a callsign (requires QRZ.com) -def infer_name_from_callsign(call): - data = 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(call): - data = 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(call): - data = 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(call): - try: - data = 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 = 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 = 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(call): - latlon = 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(freq): - try: - return freq_to_band(freq / 1000.0)["mode"] - except KeyError: - return None - - -# Convert objects to serialisable things. Used by JSON serialiser as a default when it encounters unserializable things. -# Just converts objects to dict. Try to avoid doing anything clever here when serialising spots, because we also need -# to receive spots without complex handling. -def serialize_everything(obj): - return obj.__dict__ diff --git a/data/alert.py b/data/alert.py index ec66a3b..8a7f474 100644 --- a/data/alert.py +++ b/data/alert.py @@ -7,9 +7,7 @@ from datetime import datetime, timedelta import pytz from core.constants import DXCC_FLAGS -from core.utils import infer_continent_from_callsign, \ - infer_country_from_callsign, infer_cq_zone_from_callsign, infer_itu_zone_from_callsign, infer_dxcc_id_from_callsign, \ - infer_name_from_callsign +from core.lookup_helper import lookup_helper # Data class that defines an alert. @@ -89,15 +87,15 @@ class Alert: # DX country, continent, zones etc. from callsign if self.dx_calls and self.dx_calls[0] and not self.dx_country: - self.dx_country = infer_country_from_callsign(self.dx_calls[0]) + self.dx_country = lookup_helper.infer_country_from_callsign(self.dx_calls[0]) if self.dx_calls and self.dx_calls[0] and not self.dx_continent: - self.dx_continent = infer_continent_from_callsign(self.dx_calls[0]) + self.dx_continent = lookup_helper.infer_continent_from_callsign(self.dx_calls[0]) if self.dx_calls and self.dx_calls[0] and not self.dx_cq_zone: - self.dx_cq_zone = infer_cq_zone_from_callsign(self.dx_calls[0]) + self.dx_cq_zone = lookup_helper.infer_cq_zone_from_callsign(self.dx_calls[0]) if self.dx_calls and self.dx_calls[0] and not self.dx_itu_zone: - self.dx_itu_zone = infer_itu_zone_from_callsign(self.dx_calls[0]) + self.dx_itu_zone = lookup_helper.infer_itu_zone_from_callsign(self.dx_calls[0]) if self.dx_calls and self.dx_calls[0] and not self.dx_dxcc_id: - self.dx_dxcc_id = infer_dxcc_id_from_callsign(self.dx_calls[0]) + self.dx_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.dx_calls[0]) if self.dx_dxcc_id and not self.dx_flag: self.dx_flag = DXCC_FLAGS[self.dx_dxcc_id] @@ -105,7 +103,7 @@ class Alert: # the actual alertting service, e.g. we don't want to accidentally use a user's QRZ.com home lat/lon instead of # the one from the park reference they're at. if self.dx_calls and not self.dx_names: - self.dx_names = list(map(lambda c: infer_name_from_callsign(c), self.dx_calls)) + self.dx_names = list(map(lambda c: lookup_helper.infer_name_from_callsign(c), self.dx_calls)) # Always create an ID based on a hash of every parameter *except* received_time. This is used as the index # to a map, which as a byproduct avoids us having multiple duplicate copies of the object that are identical diff --git a/data/spot.py b/data/spot.py index 9e897d0..07dae61 100644 --- a/data/spot.py +++ b/data/spot.py @@ -8,11 +8,7 @@ import pytz from pyhamtools.locator import locator_to_latlong, latlong_to_locator from core.constants import DXCC_FLAGS -from core.utils import infer_mode_type_from_mode, infer_band_from_freq, infer_continent_from_callsign, \ - infer_country_from_callsign, infer_cq_zone_from_callsign, infer_itu_zone_from_callsign, infer_dxcc_id_from_callsign, \ - infer_mode_from_comment, infer_name_from_callsign, infer_latlon_from_callsign_dxcc, infer_grid_from_callsign_dxcc, \ - infer_latlon_from_callsign_qrz, infer_grid_from_callsign_qrz, infer_mode_from_frequency - +from core.lookup_helper import lookup_helper # Data class that defines a spot. @dataclass @@ -119,15 +115,15 @@ class Spot: # DX country, continent, zones etc. from callsign if self.dx_call and not self.dx_country: - self.dx_country = infer_country_from_callsign(self.dx_call) + self.dx_country = lookup_helper.infer_country_from_callsign(self.dx_call) if self.dx_call and not self.dx_continent: - self.dx_continent = infer_continent_from_callsign(self.dx_call) + self.dx_continent = lookup_helper.infer_continent_from_callsign(self.dx_call) if self.dx_call and not self.dx_cq_zone: - self.dx_cq_zone = infer_cq_zone_from_callsign(self.dx_call) + self.dx_cq_zone = lookup_helper.infer_cq_zone_from_callsign(self.dx_call) if self.dx_call and not self.dx_itu_zone: - self.dx_itu_zone = infer_itu_zone_from_callsign(self.dx_call) + self.dx_itu_zone = lookup_helper.infer_itu_zone_from_callsign(self.dx_call) if self.dx_call and not self.dx_dxcc_id: - self.dx_dxcc_id = infer_dxcc_id_from_callsign(self.dx_call) + self.dx_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.dx_call) if self.dx_dxcc_id and DXCC_FLAGS[self.dx_dxcc_id] and not self.dx_flag: self.dx_flag = DXCC_FLAGS[self.dx_dxcc_id] @@ -137,27 +133,27 @@ class Spot: # Spotter country, continent, zones etc. from callsign if self.de_call and not self.de_country: - self.de_country = infer_country_from_callsign(self.de_call) + self.de_country = lookup_helper.infer_country_from_callsign(self.de_call) if self.de_call and not self.de_continent: - self.de_continent = infer_continent_from_callsign(self.de_call) + self.de_continent = lookup_helper.infer_continent_from_callsign(self.de_call) if self.de_call and not self.de_dxcc_id: - self.de_dxcc_id = infer_dxcc_id_from_callsign(self.de_call) + self.de_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.de_call) if self.de_dxcc_id and not self.de_flag: self.de_flag = DXCC_FLAGS[self.de_dxcc_id] # Band from frequency if self.freq and not self.band: - band = infer_band_from_freq(self.freq) + band = lookup_helper.infer_band_from_freq(self.freq) self.band = band.name # Mode from comments or bandplan if self.mode: self.mode_source = "SPOT" if self.comment and not self.mode: - self.mode = infer_mode_from_comment(self.comment) + self.mode = lookup_helper.infer_mode_from_comment(self.comment) self.mode_source = "COMMENT" if self.freq and not self.mode: - self.mode = infer_mode_from_frequency(self.freq) + self.mode = lookup_helper.infer_mode_from_frequency(self.freq) self.mode_source = "BANDPLAN" # Normalise "generic digital" modes. "DIGITAL", "DIGI" and "DATA" are just the same thing with no extra @@ -167,7 +163,7 @@ class Spot: # Mode type from mode if self.mode and not self.mode_type: - self.mode_type = infer_mode_type_from_mode(self.mode) + self.mode_type = lookup_helper.infer_mode_type_from_mode(self.mode) # Grid to lat/lon and vice versa if self.grid and not self.latitude: @@ -187,22 +183,22 @@ class Spot: # the actual spotting service, e.g. we don't want to accidentally use a user's QRZ.com home lat/lon instead of # the one from the park reference they're at. if self.dx_call and not self.dx_name: - self.dx_name = infer_name_from_callsign(self.dx_call) + self.dx_name = lookup_helper.infer_name_from_callsign(self.dx_call) if self.dx_call and not self.latitude: - latlon = infer_latlon_from_callsign_qrz(self.dx_call) + latlon = lookup_helper.infer_latlon_from_callsign_qrz(self.dx_call) if latlon: self.latitude = latlon[0] self.longitude = latlon[1] - self.grid = infer_grid_from_callsign_qrz(self.dx_call) + self.grid = lookup_helper.infer_grid_from_callsign_qrz(self.dx_call) self.location_source = "QRZ" # Last resort for getting a position, use the DXCC entity. if self.dx_call and not self.latitude: - latlon = infer_latlon_from_callsign_dxcc(self.dx_call) + latlon = lookup_helper.infer_latlon_from_callsign_dxcc(self.dx_call) if latlon: self.latitude = latlon[0] self.longitude = latlon[1] - self.grid = infer_grid_from_callsign_dxcc(self.dx_call) + self.grid = lookup_helper.infer_grid_from_callsign_dxcc(self.dx_call) self.location_source = "DXCC" # Location is "good" if it is from a spot, or from QRZ if the callsign doesn't contain a slash, so the operator diff --git a/server/webserver.py b/server/webserver.py index cdf15d6..6505a37 100644 --- a/server/webserver.py +++ b/server/webserver.py @@ -9,7 +9,6 @@ from bottle import run, request, response, template from core.config import MAX_SPOT_AGE, ALLOW_SPOTTING from core.constants import BANDS, ALL_MODES, MODE_TYPES, SIGS, CONTINENTS -from core.utils import serialize_everything from data.spot import Spot @@ -252,3 +251,10 @@ class WebServer: options["spot_sources"].append("API") return options + + +# Convert objects to serialisable things. Used by JSON serialiser as a default when it encounters unserializable things. +# Just converts objects to dict. Try to avoid doing anything clever here when serialising spots, because we also need +# to receive spots without complex handling. +def serialize_everything(obj): + return obj.__dict__ \ No newline at end of file diff --git a/spothole.py b/spothole.py index 0bbe949..71b1259 100644 --- a/spothole.py +++ b/spothole.py @@ -8,8 +8,8 @@ from diskcache import Cache from core.cleanup import CleanupTimer from core.config import config, WEB_SERVER_PORT +from core.lookup_helper import lookup_helper from core.status_reporter import StatusReporter -from core.utils import QRZ_CALLSIGN_DATA_CACHE from server.webserver import WebServer # Globals @@ -31,8 +31,9 @@ def shutdown(sig, frame): if p.enabled: p.stop() cleanup_timer.stop() - QRZ_CALLSIGN_DATA_CACHE.close() + lookup_helper.stop() spots.close() + alerts.close() # Utility method to get a spot provider based on the class specified in its config entry. @@ -65,6 +66,9 @@ if __name__ == '__main__': # Shut down gracefully on SIGINT signal.signal(signal.SIGINT, shutdown) + # Set up lookup helper + lookup_helper.start() + # Fetch, set up and start spot providers for entry in config["spot-providers"]: spot_providers.append(get_spot_provider_from_config(entry))