w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2025-10-27 08:49:27 +00:00
Code Issues 11 Packages Projects Releases Wiki Activity
Files
587d3b4cf1f4a2f8a668b5401348e6ec3175821c
spothole/core/lookup_helper.py

404 lines
19 KiB
Python
Raw Blame History

import gzip
import logging
from datetime import timedelta
from diskcache import Cache
from pyhamtools import LookupLib, Callinfo, 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
if call in self.QRZ_CALLSIGN_DATA_CACHE:
return self.QRZ_CALLSIGN_DATA_CACHE.get(call)
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, but maybe it had prefixes or suffixes. Try again with the base call.
try:
data = self.LOOKUP_LIB_QRZ.lookup_callsign(callsign=callinfo.Callinfo.get_homecall(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
if call in self.CLUBLOG_CALLSIGN_DATA_CACHE:
return self.CLUBLOG_CALLSIGN_DATA_CACHE.get(call)
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, but maybe it had prefixes or suffixes. Try again with the base call.
try:
data = self.LOOKUP_LIB_CLUBLOG_API.lookup_callsign(callsign=callinfo.Callinfo.get_homecall(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()
Reference in New Issue View Git Blame Copy Permalink