diff --git a/core/cache_utils.py b/core/cache_utils.py new file mode 100644 index 0000000..3ed58fe --- /dev/null +++ b/core/cache_utils.py @@ -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)) \ No newline at end of file diff --git a/core/lookup_helper.py b/core/lookup_helper.py index fcce0ed..1a45b2c 100644 --- a/core/lookup_helper.py +++ b/core/lookup_helper.py @@ -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: diff --git a/core/sig_utils.py b/core/sig_utils.py index 9eee0b0..dc4c554 100644 --- a/core/sig_utils.py +++ b/core/sig_utils.py @@ -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: diff --git a/spotproviders/gma.py b/spotproviders/gma.py index 66b706e..09a7ad5 100644 --- a/spotproviders/gma.py +++ b/spotproviders/gma.py @@ -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 != "": diff --git a/spotproviders/parksnpeaks.py b/spotproviders/parksnpeaks.py index a7d0c86..d2dc410 100644 --- a/spotproviders/parksnpeaks.py +++ b/spotproviders/parksnpeaks.py @@ -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]: diff --git a/spotproviders/pota.py b/spotproviders/pota.py index fda2fc5..c4b15da 100644 --- a/spotproviders/pota.py +++ b/spotproviders/pota.py @@ -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"] diff --git a/spotproviders/sota.py b/spotproviders/sota.py index ca600a8..e0af623 100644 --- a/spotproviders/sota.py +++ b/spotproviders/sota.py @@ -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"] diff --git a/spotproviders/wota.py b/spotproviders/wota.py index 2837cef..a4a0f7a 100644 --- a/spotproviders/wota.py +++ b/spotproviders/wota.py @@ -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"] diff --git a/spotproviders/zlota.py b/spotproviders/zlota.py index c3f7d69..9516919 100644 --- a/spotproviders/zlota.py +++ b/spotproviders/zlota.py @@ -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"]