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