w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-06-24 05:35:10 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Merge branch 'main' into 95-send-spots-to-xota

Browse Source 
# Conflicts:
#	README.md
#	server/handlers/api/addspot.py
#	server/handlers/api/options.py
#	spotproviders/tiles.py
#	templates/about.html
#	templates/add_spot.html
#	templates/alerts.html
#	templates/api_only_home.html
#	templates/bands.html
#	templates/base.html
#	templates/conditions.html
#	templates/map.html
#	templates/spots.html
#	templates/status.html
#	webassets/css/style.css
#	webassets/js/add-spot.js
#	webassets/js/geo.js
#	webassets/js/ui-ham.js
#	webassets/js/utils.js
This commit is contained in:
Ian Renton
2026-06-19 21:48:10 +01:00
parent 88f055384d 07d0d98f3d
commit 172a31bb18
91 changed files with 1835 additions and 1261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
core/lookup_helper.py
View File

@@ -104,7 +104,7 @@ class LookupHelper:
self._hamqth_callsign_data_cache = Cache('cache/hamqth_callsign_lookup_cache')
self._clublog_api_key = config["clublog-api-key"]
self._clublog_api_key = str(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"
@@ -184,6 +184,7 @@ class LookupHelper:
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()
assert isinstance(file_content, bytes)
logging.info("Caching Clublog cty.xml...")
with open(self._clublog_xml_download_location, "wb") as f:
f.write(file_content)
@@ -384,12 +385,12 @@ class LookupHelper:
data = self._get_qrz_data_for_callsign(call, credentials)
if data and "latitude" in data and "longitude" in data and (
float(data["latitude"]) != 0 or float(data["longitude"]) != 0) and -89.9 < float(
data["latitude"]) < 89.9:
data["latitude"]) < 89.9:
return [float(data["latitude"]), float(data["longitude"])]
data = self._get_hamqth_data_for_callsign(call, credentials)
if data and "latitude" in data and "longitude" in data and (
float(data["latitude"]) != 0 or float(data["longitude"]) != 0) and -89.9 < float(
data["latitude"]) < 89.9:
data["latitude"]) < 89.9:
return [float(data["latitude"]), float(data["longitude"])]
else:
return None
@@ -446,15 +447,16 @@ class LookupHelper:
def infer_grid_from_callsign_dxcc(self, call):
"""Infer a grid locator from a callsign (using DXCC, probably very inaccurate)"""
latlon = self.infer_latlon_from_callsign_dxcc(call)
latlon = self.infer_latlon_from_callsign_dxcc(call) or []
grid = None
try:
grid = latlong_to_locator(latlon[0], latlon[1], 8)
except:
logging.debug("Invalid lat/lon received for DXCC")
if latlon:
try:
grid = latlong_to_locator(latlon[0], latlon[1], 8)
except:
logging.debug("Invalid lat/lon received for DXCC")
return grid
def _get_qrz_data_for_callsign(self, call, credentials):
def _get_qrz_data_for_callsign(self, call, credentials) -> dict | None:
"""Utility method to get QRZ.com data from cache if possible, if not get it from the API and cache it.
Returns None immediately if no credentials are provided."""
@@ -475,7 +477,7 @@ class LookupHelper:
login_data = xmltodict.parse(login_response)
session = login_data.get("QRZDatabase", {}).get("Session", {})
if "Key" in session:
session_key = session["Key"]
session_key = str(session["Key"])
else:
logging.warning("QRZ.com login details incorrect, failed to look up with QRZ.")
return None
@@ -512,7 +514,7 @@ class LookupHelper:
self._qrz_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
return None
def _get_hamqth_data_for_callsign(self, call, credentials):
def _get_hamqth_data_for_callsign(self, call, credentials) -> dict | None:
"""Utility method to get HamQTH data from cache if possible, if not get it from the API and cache it.
Returns None immediately if no credentials are provided."""
@@ -531,7 +533,7 @@ class LookupHelper:
"&p=" + urllib.parse.quote_plus(credentials.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"]
session_id = str(dict_data["HamQTH"]["session"]["session_id"])
else:
logging.warning("HamQTH login details incorrect, failed to look up with HamQTH.")
return None
@@ -566,7 +568,7 @@ class LookupHelper:
self._hamqth_callsign_data_cache.add(call, None, expire=604800) # 1 week in seconds
return None
def _get_clublog_api_data_for_callsign(self, call):
def _get_clublog_api_data_for_callsign(self, call) -> dict | None:
"""Utility method to get Clublog API data from cache if possible, if not get it from the API and cache it"""
# Fetch from cache if we can, otherwise fetch from the API and cache it
@@ -594,7 +596,7 @@ class LookupHelper:
else:
return None
def _get_clublog_xml_data_for_callsign(self, call):
def _get_clublog_xml_data_for_callsign(self, call) -> dict | None:
"""Utility method to get Clublog XML data from file"""
if self._clublog_xml_available:
@@ -608,7 +610,7 @@ class LookupHelper:
else:
return None
def _get_dxcc_data_for_callsign(self, call):
def _get_dxcc_data_for_callsign(self, call) -> dict | None:
"""Utility method to get generic DXCC data from our lookup table, if we can find it"""
for entry in self._dxcc_data.values():
@@ -627,6 +629,7 @@ class LookupHelper:
# Singleton object
lookup_helper = LookupHelper()
def infer_mode_from_comment(comment):
"""Infer a mode from the comment"""

15
core/sig_utils.py
View File

@@ -25,13 +25,13 @@ def populate_sig_ref_info(sig_ref):
if sig_ref.sig is None or sig_ref.id is None:
logging.warning("Failed to look up sig_ref info, sig or id were not set.")
sig = sig_ref.sig
sig = sig_ref.sig or ""
ref_id = sig_ref.id
try:
if sig.upper() == "POTA":
data = SEMI_STATIC_URL_DATA_CACHE.get("https://api.pota.app/park/" + ref_id, headers=HTTP_HEADERS).json()
if data:
fullname = data["name"] if "name" in data else None
fullname = str(data["name"]) if "name" in data else None
if fullname and "parktypeDesc" in data and data["parktypeDesc"] != "":
fullname = fullname + " " + data["parktypeDesc"]
sig_ref.name = fullname
@@ -81,7 +81,8 @@ def populate_sig_ref_info(sig_ref):
elif sig.upper() == "SIOTA":
siota_csv_data = SEMI_STATIC_URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv",
headers=HTTP_HEADERS)
siota_index = {row["SILO_CODE"]: row for row in csv.DictReader(siota_csv_data.content.decode().splitlines())}
siota_index = {row["SILO_CODE"]: row for row in
csv.DictReader(siota_csv_data.content.decode().splitlines())}
row = siota_index.get(ref_id)
if row:
sig_ref.name = row["NAME"] if "NAME" in row else None
@@ -129,9 +130,9 @@ def populate_sig_ref_info(sig_ref):
if data:
for ref in data:
if ref["reference_code"] == ref_id:
sig_ref.name = ref["name"]
sig_ref.name = str(ref["name"])
sig_ref.url = "https://llota.app/list/ref/" + ref_id
sig_ref.grid = ref["grid_locator"]
sig_ref.grid = str(ref["grid_locator"])
ll = locator_to_latlong(sig_ref.grid)
sig_ref.latitude = ll[0]
sig_ref.longitude = ll[1]
@@ -139,7 +140,7 @@ def populate_sig_ref_info(sig_ref):
elif sig.upper() == "WWTOTA":
if not sig_ref.name:
sig_ref.name = sig_ref.id
sig_ref.url = "https://wwtota.com/seznam/karta_rozhledny.php?ref=" + sig_ref.name
sig_ref.url = "https://wwtota.com/seznam/karta_rozhledny.php?ref=" + str(sig_ref.name)
elif sig.upper() == "TILES":
# Tiles on the Air just uses Maidenhead 6-digit squares, so ID, Name and Grid are all the same
if not sig_ref.name:
@@ -147,7 +148,7 @@ def populate_sig_ref_info(sig_ref):
if not sig_ref.grid:
sig_ref.grid = sig_ref.id
if sig_ref.grid and not sig_ref.latitude:
ll = locator_to_latlong(sig_ref.grid)
ll = locator_to_latlong(str(sig_ref.grid))
sig_ref.latitude = ll[0]
sig_ref.longitude = ll[1]
elif sig.upper() == "WAB" or sig.upper() == "WAI":

3
core/status_reporter.py
View File

@@ -89,7 +89,8 @@ class StatusReporter:
"last_page_access_time"].replace(
tzinfo=pytz.UTC).timestamp() if self._web_server.web_server_metrics[
"last_page_access_time"] else 0,
"page_access_count": self._web_server.web_server_metrics["page_access_counter"]}
"page_access_count": self._web_server.web_server_metrics[
"page_access_counter"]}
# Update Prometheus metrics
memory_use_gauge.set(psutil.Process(os.getpid()).memory_info().rss)