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__