spothole/core/utils.py

import gzip
import logging
import urllib.request

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 core.config import config
from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, QRZCQ_CALLSIGN_LOOKUP_DATA

CLUBLOG_API_KEY = config["clublog-api-key"]

# 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:
        # Read the file inside the .gz archive located at url
        with urllib.request.urlopen("https://cdn.clublog.org/cty.php?api=" + CLUBLOG_API_KEY) as response:
            with gzip.GzipFile(fileobj=response) as uncompressed:
                file_content = uncompressed.read()

        # write to file in binary mode 'wb'
        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.
LOOKUP_LIB_BASIC = LookupLib(lookuptype="countryfile")
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_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 qrzcq_data["country"]:
            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 qrzcq_data["dxcc"]:
            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 qrzcq_data["continent"]:
            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 qrzcq_data["cqz"]:
            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 qrzcq_data["ituz"]:
            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__