w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2025-12-15 08:33:38 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Single common URL cache for semi-static lookups #74

Browse Source
This commit is contained in:
Ian Renton
2025-11-02 14:22:15 +00:00
parent 0e8c7873d8
commit 28010a68ae
9 changed files with 37 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
core/cache_utils.py Normal file
View File

@@ -0,0 +1,10 @@
from datetime import timedelta
from requests_cache import CachedSession
# Cache for "semi-static" data such as the locations of parks, CSVs of reference lists, etc.
# This has an expiry time of 30 days, so will re-request from the source after that amount
# of time has passed. This is used throughout Spothole to cache data that does not change
# rapidly.
SEMI_STATIC_URL_DATA_CACHE = CachedSession("cache/semi_static_url_data_cache",
expire_after=timedelta(days=30))

6
core/lookup_helper.py
View File

@@ -9,6 +9,7 @@ from pyhamtools.frequency import freq_to_band
from pyhamtools.locator import latlong_to_locator
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
@@ -34,15 +35,12 @@ class LookupHelper:
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:
@@ -78,7 +76,7 @@ class LookupHelper:
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",
response = SEMI_STATIC_URL_DATA_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:

22
core/sig_utils.py
View File

@@ -1,9 +1,8 @@
import csv
from datetime import timedelta
from pyhamtools.locator import latlong_to_locator
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import SIGS, HTTP_HEADERS
from core.geo_utils import wab_wai_square_to_lat_lon
@@ -22,43 +21,38 @@ def get_ref_regex_for_sig(sig):
return s.ref_regex
return None
# Cache for SIG ref lookups
SIG_REF_DATA_CACHE_TIME_DAYS = 30
SIG_REF_DATA_CACHE = CachedSession("cache/sig_ref_lookup_cache",
expire_after=timedelta(days=SIG_REF_DATA_CACHE_TIME_DAYS))
# Look up details of a SIG reference (e.g. POTA park) such as name, lat/lon, and grid.
def get_sig_ref_info(sig, sig_ref_id):
if sig.upper() == "POTA":
data = SIG_REF_DATA_CACHE.get("https://api.pota.app/park/" + sig_ref_id, headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://api.pota.app/park/" + sig_ref_id, headers=HTTP_HEADERS).json()
if data:
return {"name": data["name"] if "name" in data else None,
"grid": data["grid6"] if "grid6" in data else None,
"latitude": data["latitude"] if "latitude" in data else None,
"longitude": data["longitude"] if "longitude" in data else None}
elif sig.upper() == "SOTA":
data = SIG_REF_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + sig_ref_id, headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + sig_ref_id, headers=HTTP_HEADERS).json()
if data:
return {"name": data["name"] if "name" in data else None,
"grid": data["locator"] if "locator" in data else None,
"latitude": data["latitude"] if "latitude" in data else None,
"longitude": data["longitude"] if "longitude" in data else None}
elif sig.upper() == "WWBOTA":
data = SIG_REF_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + sig_ref_id, headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + sig_ref_id, headers=HTTP_HEADERS).json()
if data:
return {"name": data["name"] if "name" in data else None,
"grid": data["locator"] if "locator" in data else None,
"latitude": data["lat"] if "lat" in data else None,
"longitude": data["long"] if "long" in data else None}
elif sig.upper() == "GMA" or sig.upper() == "ARLHS" or sig.upper() == "ILLW" or sig.upper() == "WCA" or sig.upper() == "MOTA" or sig.upper() == "IOTA":
data = SIG_REF_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + sig_ref_id, headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + sig_ref_id, headers=HTTP_HEADERS).json()
if data:
return {"name": data["name"] if "name" in data else None,
"grid": data["locator"] if "locator" in data else None,
"latitude": data["latitude"] if "latitude" in data else None,
"longitude": data["longitude"] if "longitude" in data else None}
elif sig.upper() == "SIOTA":
siota_csv_data = SIG_REF_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", headers=HTTP_HEADERS)
siota_csv_data = SEMI_STATIC_URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", headers=HTTP_HEADERS)
siota_dr = csv.DictReader(siota_csv_data.content.decode().splitlines())
for row in siota_dr:
if row["SILO_CODE"] == sig_ref_id:
@@ -67,7 +61,7 @@ def get_sig_ref_info(sig, sig_ref_id):
"latitude": float(row["LAT"]) if "LAT" in row else None,
"longitude": float(row["LNG"]) if "LNG" in row else None}
elif sig.upper() == "WOTA":
data = SIG_REF_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", headers=HTTP_HEADERS).json()
if data:
for feature in data["features"]:
if feature["properties"]["wotaId"] == sig_ref_id:
@@ -76,7 +70,7 @@ def get_sig_ref_info(sig, sig_ref_id):
"latitude": feature["geometry"]["coordinates"][1],
"longitude": feature["geometry"]["coordinates"][0]}
elif sig.upper() == "ZLOTA":
data = SIG_REF_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS).json()
data = SEMI_STATIC_URL_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS).json()
if data:
for asset in data:
if asset["code"] == sig_ref_id:

8
spotproviders/gma.py
View File

@@ -1,9 +1,9 @@
import logging
from datetime import datetime, timedelta
from datetime import datetime
import pytz
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig
from data.sig_ref import SIGRef
@@ -17,8 +17,6 @@ class GMA(HTTPSpotProvider):
SPOTS_URL = "https://www.cqgma.org/api/spots/25/"
# GMA spots don't contain the details of the programme they are for, we need a separate lookup for that
REF_INFO_URL_ROOT = "https://www.cqgma.org/api/ref/?"
REF_INFO_CACHE_TIME_DAYS = 30
REF_INFO_CACHE = CachedSession("cache/gma_ref_info_cache", expire_after=timedelta(days=REF_INFO_CACHE_TIME_DAYS))
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
@@ -44,7 +42,7 @@ class GMA(HTTPSpotProvider):
dx_longitude=float(source_spot["LON"]) if (source_spot["LON"] and source_spot["LON"] != "") else None)
# GMA doesn't give what programme (SIG) the reference is for until we separately look it up.
ref_response = self.REF_INFO_CACHE.get(self.REF_INFO_URL_ROOT + source_spot["REF"],
ref_response = SEMI_STATIC_URL_DATA_CACHE.get(self.REF_INFO_URL_ROOT + source_spot["REF"],
headers=HTTP_HEADERS)
# Sometimes this is blank, so handle that
if ref_response.text is not None and ref_response.text != "":

8
spotproviders/parksnpeaks.py
View File

@@ -1,11 +1,11 @@
import csv
import logging
import re
from datetime import datetime, timedelta
from datetime import datetime
import pytz
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig
from data.sig_ref import SIGRef
@@ -18,8 +18,6 @@ class ParksNPeaks(HTTPSpotProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://www.parksnpeaks.org/api/ALL"
SIOTA_LIST_URL = "https://www.silosontheair.com/data/silos.csv"
SIOTA_LIST_CACHE_TIME_DAYS = 30
SIOTA_LIST_CACHE = CachedSession("cache/siota_data_cache", expire_after=timedelta(days=SIOTA_LIST_CACHE_TIME_DAYS))
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
@@ -59,7 +57,7 @@ class ParksNPeaks(HTTPSpotProvider):
# SiOTA lat/lon/grid lookup
if spot.sig == "SIOTA":
siota_csv_data = self.SIOTA_LIST_CACHE.get(self.SIOTA_LIST_URL, headers=HTTP_HEADERS)
siota_csv_data = SEMI_STATIC_URL_DATA_CACHE.get(self.SIOTA_LIST_URL, headers=HTTP_HEADERS)
siota_dr = csv.DictReader(siota_csv_data.content.decode().splitlines())
for row in siota_dr:
if row["SILO_CODE"] == spot.sig_refs[0]:

9
spotproviders/pota.py
View File

@@ -1,9 +1,9 @@
import re
from datetime import datetime, timedelta
from datetime import datetime
import pytz
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig, get_ref_regex_for_sig
from data.sig_ref import SIGRef
@@ -17,9 +17,6 @@ class POTA(HTTPSpotProvider):
SPOTS_URL = "https://api.pota.app/spot/activator"
# Might need to look up extra park data
PARK_URL_ROOT = "https://api.pota.app/park/"
PARK_DATA_CACHE_TIME_DAYS = 30
PARK_DATA_CACHE = CachedSession("cache/pota_park_data_cache",
expire_after=timedelta(days=PARK_DATA_CACHE_TIME_DAYS))
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
@@ -52,7 +49,7 @@ class POTA(HTTPSpotProvider):
ref = SIGRef(id=r.upper(), url="https://pota.app/#/park/" + r.upper())
# Now we need to look up the name of that reference from the API, because the comment won't have it
park_response = self.PARK_DATA_CACHE.get(self.PARK_URL_ROOT + r.upper(), headers=HTTP_HEADERS)
park_response = SEMI_STATIC_URL_DATA_CACHE.get(self.PARK_URL_ROOT + r.upper(), headers=HTTP_HEADERS)
park_data = park_response.json()
if park_data and "name" in park_data:
ref.name = park_data["name"]

5
spotproviders/sota.py
View File

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta
import requests
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig
from data.sig_ref import SIGRef
@@ -21,8 +22,6 @@ class SOTA(HTTPSpotProvider):
SPOTS_URL = "https://api-db2.sota.org.uk/api/spots/60/all/all"
# SOTA spots don't contain lat/lon, we need a separate lookup for that
SUMMIT_URL_ROOT = "https://api-db2.sota.org.uk/api/summits/"
SUMMIT_DATA_CACHE_TIME_DAYS = 30
SUMMIT_DATA_CACHE = CachedSession("cache/sota_summit_data_cache", expire_after=timedelta(days=SUMMIT_DATA_CACHE_TIME_DAYS))
def __init__(self, provider_config):
super().__init__(provider_config, self.EPOCH_URL, self.POLL_INTERVAL_SEC)
@@ -57,7 +56,7 @@ class SOTA(HTTPSpotProvider):
# SOTA doesn't give summit lat/lon/grid in the main call, so we need another separate call for this
try:
summit_response = self.SUMMIT_DATA_CACHE.get(self.SUMMIT_URL_ROOT + source_spot["summitCode"], headers=HTTP_HEADERS)
summit_response = SEMI_STATIC_URL_DATA_CACHE.get(self.SUMMIT_URL_ROOT + source_spot["summitCode"], headers=HTTP_HEADERS)
summit_data = summit_response.json()
spot.dx_grid = summit_data["locator"]
spot.dx_latitude = summit_data["latitude"]

8
spotproviders/wota.py
View File

@@ -1,9 +1,9 @@
from datetime import timedelta, datetime
from datetime import datetime
import pytz
from requests_cache import CachedSession
from rss_parser import RSSParser
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig
from data.sig_ref import SIGRef
@@ -16,8 +16,6 @@ class WOTA(HTTPSpotProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://www.wota.org.uk/spots_rss.php"
LIST_URL = "https://www.wota.org.uk/mapping/data/summits.json"
LIST_CACHE_TIME_DAYS = 30
LIST_CACHE = CachedSession("cache/wota_data_cache", expire_after=timedelta(days=LIST_CACHE_TIME_DAYS))
RSS_DATE_TIME_FORMAT = "%a, %d %b %Y %H:%M:%S %z"
def __init__(self, provider_config):
@@ -75,7 +73,7 @@ class WOTA(HTTPSpotProvider):
# WOTA name/grid/lat/lon lookup
if ref:
wota_data = self.LIST_CACHE.get(self.LIST_URL, headers=HTTP_HEADERS).json()
wota_data = SEMI_STATIC_URL_DATA_CACHE.get(self.LIST_URL, headers=HTTP_HEADERS).json()
for feature in wota_data["features"]:
if feature["properties"]["wotaId"] == ref:
spot.sig_refs[0].name = feature["properties"]["title"]

11
spotproviders/zlota.py
View File

@@ -1,11 +1,8 @@
import csv
import logging
import re
from datetime import datetime, timedelta
from datetime import datetime
import pytz
from requests_cache import CachedSession
from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE
from core.constants import HTTP_HEADERS
from core.sig_utils import get_icon_for_sig
from data.sig_ref import SIGRef
@@ -18,8 +15,6 @@ class ZLOTA(HTTPSpotProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://ontheair.nz/api/spots?zlota_only=true"
LIST_URL = "https://ontheair.nz/assets/assets.json"
LIST_CACHE_TIME_DAYS = 30
LIST_CACHE = CachedSession("cache/zlota_data_cache", expire_after=timedelta(days=LIST_CACHE_TIME_DAYS))
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
@@ -47,7 +42,7 @@ class ZLOTA(HTTPSpotProvider):
time=datetime.fromisoformat(source_spot["referenced_time"]).astimezone(pytz.UTC).timestamp())
# ZLOTA lat/lon lookup
zlota_data = self.LIST_CACHE.get(self.LIST_URL, headers=HTTP_HEADERS).json()
zlota_data = SEMI_STATIC_URL_DATA_CACHE.get(self.LIST_URL, headers=HTTP_HEADERS).json()
for asset in zlota_data:
if asset["code"] == spot.sig_refs[0]:
spot.dx_latitude = asset["y"]