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

Refactor utils.py as a helper class so we have some control about when the lookup services actually start

Browse Source
This commit is contained in:
Ian Renton
2025-10-13 20:16:40 +01:00
parent 9d2b2a1f66
commit cfa3aaedb0
6 changed files with 431 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files

393
core/lookup_helper.py Normal file
View File

@@ -0,0 +1,393 @@
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).raw
with gzip.GzipFile(fileobj=response) 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()

378
core/utils.py
View File

@@ -1,378 +0,0 @@
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__

16
data/alert.py
View File

@@ -7,9 +7,7 @@ from datetime import datetime, timedelta
import pytz
from core.constants import DXCC_FLAGS
from core.utils import infer_continent_from_callsign, \
infer_country_from_callsign, infer_cq_zone_from_callsign, infer_itu_zone_from_callsign, infer_dxcc_id_from_callsign, \
infer_name_from_callsign
from core.lookup_helper import lookup_helper
# Data class that defines an alert.
@@ -89,15 +87,15 @@ class Alert:
# DX country, continent, zones etc. from callsign
if self.dx_calls and self.dx_calls[0] and not self.dx_country:
self.dx_country = infer_country_from_callsign(self.dx_calls[0])
self.dx_country = lookup_helper.infer_country_from_callsign(self.dx_calls[0])
if self.dx_calls and self.dx_calls[0] and not self.dx_continent:
self.dx_continent = infer_continent_from_callsign(self.dx_calls[0])
self.dx_continent = lookup_helper.infer_continent_from_callsign(self.dx_calls[0])
if self.dx_calls and self.dx_calls[0] and not self.dx_cq_zone:
self.dx_cq_zone = infer_cq_zone_from_callsign(self.dx_calls[0])
self.dx_cq_zone = lookup_helper.infer_cq_zone_from_callsign(self.dx_calls[0])
if self.dx_calls and self.dx_calls[0] and not self.dx_itu_zone:
self.dx_itu_zone = infer_itu_zone_from_callsign(self.dx_calls[0])
self.dx_itu_zone = lookup_helper.infer_itu_zone_from_callsign(self.dx_calls[0])
if self.dx_calls and self.dx_calls[0] and not self.dx_dxcc_id:
self.dx_dxcc_id = infer_dxcc_id_from_callsign(self.dx_calls[0])
self.dx_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.dx_calls[0])
if self.dx_dxcc_id and not self.dx_flag:
self.dx_flag = DXCC_FLAGS[self.dx_dxcc_id]
@@ -105,7 +103,7 @@ class Alert:
# the actual alertting service, e.g. we don't want to accidentally use a user's QRZ.com home lat/lon instead of
# the one from the park reference they're at.
if self.dx_calls and not self.dx_names:
self.dx_names = list(map(lambda c: infer_name_from_callsign(c), self.dx_calls))
self.dx_names = list(map(lambda c: lookup_helper.infer_name_from_callsign(c), self.dx_calls))
# Always create an ID based on a hash of every parameter *except* received_time. This is used as the index
# to a map, which as a byproduct avoids us having multiple duplicate copies of the object that are identical

40
data/spot.py
View File

@@ -8,11 +8,7 @@ import pytz
from pyhamtools.locator import locator_to_latlong, latlong_to_locator
from core.constants import DXCC_FLAGS
from core.utils import infer_mode_type_from_mode, infer_band_from_freq, infer_continent_from_callsign, \
infer_country_from_callsign, infer_cq_zone_from_callsign, infer_itu_zone_from_callsign, infer_dxcc_id_from_callsign, \
infer_mode_from_comment, infer_name_from_callsign, infer_latlon_from_callsign_dxcc, infer_grid_from_callsign_dxcc, \
infer_latlon_from_callsign_qrz, infer_grid_from_callsign_qrz, infer_mode_from_frequency
from core.lookup_helper import lookup_helper
# Data class that defines a spot.
@dataclass
@@ -119,15 +115,15 @@ class Spot:
# DX country, continent, zones etc. from callsign
if self.dx_call and not self.dx_country:
self.dx_country = infer_country_from_callsign(self.dx_call)
self.dx_country = lookup_helper.infer_country_from_callsign(self.dx_call)
if self.dx_call and not self.dx_continent:
self.dx_continent = infer_continent_from_callsign(self.dx_call)
self.dx_continent = lookup_helper.infer_continent_from_callsign(self.dx_call)
if self.dx_call and not self.dx_cq_zone:
self.dx_cq_zone = infer_cq_zone_from_callsign(self.dx_call)
self.dx_cq_zone = lookup_helper.infer_cq_zone_from_callsign(self.dx_call)
if self.dx_call and not self.dx_itu_zone:
self.dx_itu_zone = infer_itu_zone_from_callsign(self.dx_call)
self.dx_itu_zone = lookup_helper.infer_itu_zone_from_callsign(self.dx_call)
if self.dx_call and not self.dx_dxcc_id:
self.dx_dxcc_id = infer_dxcc_id_from_callsign(self.dx_call)
self.dx_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.dx_call)
if self.dx_dxcc_id and DXCC_FLAGS[self.dx_dxcc_id] and not self.dx_flag:
self.dx_flag = DXCC_FLAGS[self.dx_dxcc_id]
@@ -137,27 +133,27 @@ class Spot:
# Spotter country, continent, zones etc. from callsign
if self.de_call and not self.de_country:
self.de_country = infer_country_from_callsign(self.de_call)
self.de_country = lookup_helper.infer_country_from_callsign(self.de_call)
if self.de_call and not self.de_continent:
self.de_continent = infer_continent_from_callsign(self.de_call)
self.de_continent = lookup_helper.infer_continent_from_callsign(self.de_call)
if self.de_call and not self.de_dxcc_id:
self.de_dxcc_id = infer_dxcc_id_from_callsign(self.de_call)
self.de_dxcc_id = lookup_helper.infer_dxcc_id_from_callsign(self.de_call)
if self.de_dxcc_id and not self.de_flag:
self.de_flag = DXCC_FLAGS[self.de_dxcc_id]
# Band from frequency
if self.freq and not self.band:
band = infer_band_from_freq(self.freq)
band = lookup_helper.infer_band_from_freq(self.freq)
self.band = band.name
# Mode from comments or bandplan
if self.mode:
self.mode_source = "SPOT"
if self.comment and not self.mode:
self.mode = infer_mode_from_comment(self.comment)
self.mode = lookup_helper.infer_mode_from_comment(self.comment)
self.mode_source = "COMMENT"
if self.freq and not self.mode:
self.mode = infer_mode_from_frequency(self.freq)
self.mode = lookup_helper.infer_mode_from_frequency(self.freq)
self.mode_source = "BANDPLAN"
# Normalise "generic digital" modes. "DIGITAL", "DIGI" and "DATA" are just the same thing with no extra
@@ -167,7 +163,7 @@ class Spot:
# Mode type from mode
if self.mode and not self.mode_type:
self.mode_type = infer_mode_type_from_mode(self.mode)
self.mode_type = lookup_helper.infer_mode_type_from_mode(self.mode)
# Grid to lat/lon and vice versa
if self.grid and not self.latitude:
@@ -187,22 +183,22 @@ class Spot:
# the actual spotting service, e.g. we don't want to accidentally use a user's QRZ.com home lat/lon instead of
# the one from the park reference they're at.
if self.dx_call and not self.dx_name:
self.dx_name = infer_name_from_callsign(self.dx_call)
self.dx_name = lookup_helper.infer_name_from_callsign(self.dx_call)
if self.dx_call and not self.latitude:
latlon = infer_latlon_from_callsign_qrz(self.dx_call)
latlon = lookup_helper.infer_latlon_from_callsign_qrz(self.dx_call)
if latlon:
self.latitude = latlon[0]
self.longitude = latlon[1]
self.grid = infer_grid_from_callsign_qrz(self.dx_call)
self.grid = lookup_helper.infer_grid_from_callsign_qrz(self.dx_call)
self.location_source = "QRZ"
# Last resort for getting a position, use the DXCC entity.
if self.dx_call and not self.latitude:
latlon = infer_latlon_from_callsign_dxcc(self.dx_call)
latlon = lookup_helper.infer_latlon_from_callsign_dxcc(self.dx_call)
if latlon:
self.latitude = latlon[0]
self.longitude = latlon[1]
self.grid = infer_grid_from_callsign_dxcc(self.dx_call)
self.grid = lookup_helper.infer_grid_from_callsign_dxcc(self.dx_call)
self.location_source = "DXCC"
# Location is "good" if it is from a spot, or from QRZ if the callsign doesn't contain a slash, so the operator

8
server/webserver.py
View File

@@ -9,7 +9,6 @@ from bottle import run, request, response, template
from core.config import MAX_SPOT_AGE, ALLOW_SPOTTING
from core.constants import BANDS, ALL_MODES, MODE_TYPES, SIGS, CONTINENTS
from core.utils import serialize_everything
from data.spot import Spot
@@ -252,3 +251,10 @@ class WebServer:
options["spot_sources"].append("API")
return options
# 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__

8
spothole.py
View File

@@ -8,8 +8,8 @@ from diskcache import Cache
from core.cleanup import CleanupTimer
from core.config import config, WEB_SERVER_PORT
from core.lookup_helper import lookup_helper
from core.status_reporter import StatusReporter
from core.utils import QRZ_CALLSIGN_DATA_CACHE
from server.webserver import WebServer
# Globals
@@ -31,8 +31,9 @@ def shutdown(sig, frame):
if p.enabled:
p.stop()
cleanup_timer.stop()
QRZ_CALLSIGN_DATA_CACHE.close()
lookup_helper.stop()
spots.close()
alerts.close()
# Utility method to get a spot provider based on the class specified in its config entry.
@@ -65,6 +66,9 @@ if __name__ == '__main__':
# Shut down gracefully on SIGINT
signal.signal(signal.SIGINT, shutdown)
# Set up lookup helper
lookup_helper.start()
# Fetch, set up and start spot providers
for entry in config["spot-providers"]:
spot_providers.append(get_spot_provider_from_config(entry))