spothole/core/lookup_helper.py

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