mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2025-12-15 08:33:38 +00:00
Use HamQTH as a lookup provider. Closes #73
This commit is contained in:
@@ -123,9 +123,13 @@ max-alert-age-sec: 604800
|
||||
|
||||
# Login for QRZ.com to look up information. Optional. You will need an "XML Subscriber" (paid) package to retrieve all
|
||||
# the data for a callsign via their system.
|
||||
qrz-username: "N0CALL"
|
||||
qrz-username: ""
|
||||
qrz-password: ""
|
||||
|
||||
# Login for HamQTH to look up information. Optional.
|
||||
hamqth-username: ""
|
||||
hamqth-password: ""
|
||||
|
||||
# API key for Clublog to look up information. Optional. You sill need to request one via their helpdesk portal if you
|
||||
# want to use callsign lookups from Clublog.
|
||||
clublog-api-key: ""
|
||||
|
||||
@@ -8,6 +8,7 @@ SOFTWARE_VERSION = "0.1"
|
||||
|
||||
# HTTP headers used for spot providers that use HTTP
|
||||
HTTP_HEADERS = {"User-Agent": SOFTWARE_NAME + ", v" + SOFTWARE_VERSION + " (operated by " + SERVER_OWNER_CALLSIGN + ")"}
|
||||
HAMQTH_PRG = (SOFTWARE_NAME + " v" + SOFTWARE_VERSION + " operated by " + SERVER_OWNER_CALLSIGN).replace(" ", "_")
|
||||
|
||||
# Special Interest Groups
|
||||
SIGS = [
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import gzip
|
||||
import logging
|
||||
import urllib.parse
|
||||
from datetime import timedelta
|
||||
|
||||
import xmltodict
|
||||
from diskcache import Cache
|
||||
from pyhamtools import LookupLib, Callinfo, callinfo
|
||||
from pyhamtools.exceptions import APIKeyMissingError
|
||||
@@ -12,7 +14,8 @@ from requests_cache import CachedSession
|
||||
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
|
||||
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
|
||||
QRZCQ_CALLSIGN_LOOKUP_DATA, HTTP_HEADERS, HAMQTH_PRG
|
||||
|
||||
|
||||
# Singleton class that provides lookup functionality.
|
||||
class LookupHelper:
|
||||
@@ -32,15 +35,24 @@ class LookupHelper:
|
||||
self.QRZ_CALLSIGN_DATA_CACHE = None
|
||||
self.LOOKUP_LIB_QRZ = None
|
||||
self.QRZ_AVAILABLE = None
|
||||
self.HAMQTH_AVAILABLE = None
|
||||
self.HAMQTH_CALLSIGN_DATA_CACHE = None
|
||||
self.HAMQTH_BASE_URL = "https://www.hamqth.com/xml.php"
|
||||
# HamQTH session keys expire after an hour. Rather than working out how much time has passed manually, we cheat
|
||||
# and cache the HTTP response for 55 minutes, so when the login URL is queried within 55 minutes of the previous
|
||||
# time, you just get the cached response.
|
||||
self.HAMQTH_SESSION_LOOKUP_CACHE = CachedSession("cache/hamqth_session_cache",
|
||||
expire_after=timedelta(minutes=55))
|
||||
self.CALL_INFO_BASIC = None
|
||||
self.LOOKUP_LIB_BASIC = None
|
||||
self.COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION = 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.
|
||||
# Lookup helpers from pyhamtools. We use five (!) 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, HamQTH, 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_DOWNLOAD_LOCATION = "cache/cty.plist"
|
||||
success = self.download_country_files_cty_plist()
|
||||
if success:
|
||||
@@ -50,12 +62,15 @@ class LookupHelper:
|
||||
self.LOOKUP_LIB_BASIC = LookupLib(lookuptype="countryfile")
|
||||
self.CALL_INFO_BASIC = Callinfo(self.LOOKUP_LIB_BASIC)
|
||||
|
||||
self.QRZ_AVAILABLE = config["qrz-password"] != ""
|
||||
self.QRZ_AVAILABLE = config["qrz-username"] != "" and 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.HAMQTH_AVAILABLE = config["hamqth-username"] != "" and config["hamqth-password"] != ""
|
||||
self.HAMQTH_CALLSIGN_DATA_CACHE = Cache('cache/hamqth_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 != ""
|
||||
@@ -77,7 +92,7 @@ class LookupHelper:
|
||||
try:
|
||||
logging.info("Downloading Country-files.com cty.plist...")
|
||||
response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.country-files.com/cty/cty.plist",
|
||||
headers=HTTP_HEADERS).text
|
||||
headers=HTTP_HEADERS).text
|
||||
|
||||
with open(self.COUNTRY_FILES_CTY_PLIST_DOWNLOAD_LOCATION, "w") as f:
|
||||
f.write(response)
|
||||
@@ -146,7 +161,12 @@ class LookupHelper:
|
||||
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
|
||||
# Couldn't get anything from QRZ.com database, try HamQTH
|
||||
if not country:
|
||||
hamqth_data = self.get_hamqth_data_for_callsign(call)
|
||||
if hamqth_data and "country" in hamqth_data:
|
||||
country = hamqth_data["country"]
|
||||
# Couldn't get anything from HamQTH 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:
|
||||
@@ -164,7 +184,6 @@ class LookupHelper:
|
||||
|
||||
# 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)
|
||||
@@ -175,7 +194,12 @@ class LookupHelper:
|
||||
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
|
||||
# Couldn't get anything from QRZ.com database, try HamQTH
|
||||
if not dxcc:
|
||||
hamqth_data = self.get_hamqth_data_for_callsign(call)
|
||||
if hamqth_data and "adif" in hamqth_data:
|
||||
dxcc = hamqth_data["adif"]
|
||||
# Couldn't get anything from HamQTH 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:
|
||||
@@ -198,7 +222,12 @@ class LookupHelper:
|
||||
continent = self.CALL_INFO_BASIC.get_continent(call)
|
||||
except (KeyError, ValueError) as e:
|
||||
continent = None
|
||||
# Couldn't get anything from basic call info database, try Clublog data
|
||||
# Couldn't get anything from basic call info database, try HamQTH
|
||||
if not continent:
|
||||
hamqth_data = self.get_hamqth_data_for_callsign(call)
|
||||
if hamqth_data and "continent" in hamqth_data:
|
||||
country = hamqth_data["continent"]
|
||||
# Couldn't get anything from HamQTH 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:
|
||||
@@ -226,7 +255,12 @@ class LookupHelper:
|
||||
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
|
||||
# Couldn't get anything from QRZ.com database, try HamQTH
|
||||
if not cqz:
|
||||
hamqth_data = self.get_hamqth_data_for_callsign(call)
|
||||
if hamqth_data and "cq" in hamqth_data:
|
||||
cqz = hamqth_data["cq"]
|
||||
# Couldn't get anything from HamQTH 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:
|
||||
@@ -254,13 +288,87 @@ class LookupHelper:
|
||||
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
|
||||
# Couldn't get anything from QRZ.com database, try HamQTH
|
||||
if not ituz:
|
||||
hamqth_data = self.get_hamqth_data_for_callsign(call)
|
||||
if hamqth_data and "itu" in hamqth_data:
|
||||
ituz = hamqth_data["itu"]
|
||||
# Couldn't get anything from HamQTH 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
|
||||
|
||||
# Infer an operator name from a callsign (requires QRZ.com/HamQTH)
|
||||
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
|
||||
data = self.get_hamqth_data_for_callsign(call)
|
||||
if data and "nick" in data:
|
||||
return data["nick"]
|
||||
else:
|
||||
return None
|
||||
|
||||
# Infer a latitude and longitude from a callsign (requires QRZ.com/HamQTH)
|
||||
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"]]
|
||||
data = self.get_hamqth_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/HamQTH)
|
||||
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"]
|
||||
data = self.get_hamqth_data_for_callsign(call)
|
||||
if data and "grid" in data:
|
||||
return data["grid"]
|
||||
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
|
||||
|
||||
# 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
|
||||
@@ -284,6 +392,49 @@ class LookupHelper:
|
||||
else:
|
||||
return None
|
||||
|
||||
# Utility method to get HamQTH data from cache if possible, if not get it from the API and cache it
|
||||
def get_hamqth_data_for_callsign(self, call):
|
||||
# Fetch from cache if we can, otherwise fetch from the API and cache it
|
||||
if call in self.HAMQTH_CALLSIGN_DATA_CACHE:
|
||||
return self.HAMQTH_CALLSIGN_DATA_CACHE.get(call)
|
||||
elif self.HAMQTH_AVAILABLE:
|
||||
try:
|
||||
# First we need to log in and get a session token.
|
||||
session_data = self.HAMQTH_SESSION_LOOKUP_CACHE.get(
|
||||
self.HAMQTH_BASE_URL + "?u=" + urllib.parse.quote_plus(config["hamqth-username"]) +
|
||||
"&p=" + urllib.parse.quote_plus(config["hamqth-password"]), headers=HTTP_HEADERS).content
|
||||
dict_data = xmltodict.parse(session_data)
|
||||
if "session_id" in dict_data["HamQTH"]["session"]:
|
||||
session_id = dict_data["HamQTH"]["session"]["session_id"]
|
||||
|
||||
# Now look up the actual data.
|
||||
try:
|
||||
lookup_data = SEMI_STATIC_URL_DATA_CACHE.get(
|
||||
self.HAMQTH_BASE_URL + "?id=" + session_id + "&callsign=" + urllib.parse.quote_plus(
|
||||
call) + "&prg=" + HAMQTH_PRG, headers=HTTP_HEADERS).content
|
||||
data = xmltodict.parse(lookup_data)["HamQTH"]["search"]
|
||||
self.HAMQTH_CALLSIGN_DATA_CACHE.add(call, data, expire=604800) # 1 week in seconds
|
||||
return data
|
||||
except (KeyError, ValueError):
|
||||
# HamQTH had no info for the call, but maybe it had prefixes or suffixes. Try again with the base call.
|
||||
try:
|
||||
lookup_data = SEMI_STATIC_URL_DATA_CACHE.get(
|
||||
self.HAMQTH_BASE_URL + "?id=" + session_id + "&callsign=" + urllib.parse.quote_plus(
|
||||
callinfo.Callinfo.get_homecall(call)) + "&prg=" + HAMQTH_PRG, headers=HTTP_HEADERS).content
|
||||
data = xmltodict.parse(lookup_data)["HamQTH"]["search"]
|
||||
self.HAMQTH_CALLSIGN_DATA_CACHE.add(call, data, expire=604800) # 1 week in seconds
|
||||
return data
|
||||
except (KeyError, ValueError):
|
||||
# HamQTH had no info for the call, that's OK. Cache a None so we don't try to look this up again
|
||||
self.HAMQTH_CALLSIGN_DATA_CACHE.add(call, None, expire=604800) # 1 week in seconds
|
||||
return None
|
||||
|
||||
else:
|
||||
logging.warn("HamQTH login details incorrect, failed to look up with HamQTH.")
|
||||
except:
|
||||
logging.error("Exception when looking up HamQTH data")
|
||||
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
|
||||
@@ -332,70 +483,11 @@ class LookupHelper:
|
||||
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
Block a user