From d26ddff7d168f867e2fe7520b8553cf44ff1f838 Mon Sep 17 00:00:00 2001 From: Ian Renton Date: Fri, 31 Jul 2026 14:12:55 +0100 Subject: [PATCH] Refactor of caching & data storage part 1 --- alertproviders/alert_provider.py | 2 +- core/cache_utils.py | 27 ------ core/cleanup.py | 73 --------------- core/data_store.py | 69 +++++++++++++++ core/live_data_cache.py | 72 +++++++++++++++ core/lookup_helper.py | 12 +-- core/sig_utils.py | 88 ++++++++----------- core/status_reporter.py | 85 ++++++++---------- core/url_data_cache.py | 23 +++++ requirements.txt | 3 +- server/handlers/api/addspot.py | 2 +- server/webserver.py | 21 ++--- solarconditionsproviders/giroionosonde.py | 4 +- .../solar_conditions_provider.py | 7 +- spothole.py | 34 ++----- spotproviders/gma.py | 6 +- spotproviders/spot_provider.py | 2 +- 17 files changed, 277 insertions(+), 253 deletions(-) delete mode 100644 core/cache_utils.py delete mode 100644 core/cleanup.py create mode 100644 core/data_store.py create mode 100644 core/live_data_cache.py create mode 100644 core/url_data_cache.py diff --git a/alertproviders/alert_provider.py b/alertproviders/alert_provider.py index 9645e97..4498679 100644 --- a/alertproviders/alert_provider.py +++ b/alertproviders/alert_provider.py @@ -44,7 +44,7 @@ class AlertProvider: def _add_alert(self, alert): if not alert.expired(): - self._alerts.add(alert.id, alert, expire=MAX_ALERT_AGE) + self._alerts.set(alert.id, alert) # Ping the web server in case we have any SSE connections that need to see this immediately if self._web_server: self._web_server.notify_new_alert(alert) diff --git a/core/cache_utils.py b/core/cache_utils.py deleted file mode 100644 index 68e521f..0000000 --- a/core/cache_utils.py +++ /dev/null @@ -1,27 +0,0 @@ -import threading -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. The ThreadSafeSession construct here protects it against some multithreading -# contention weirdness we sometimes used to see on startup where the cache was hammered -# pretty hard. The expanded list of allowable_codes ensures we also cache and return 400-type -# responses, e.g "this SOTA summit ref doesn't actually exist", to avoid hammering remote -# servers for data they've told us they can't provide. -_session = CachedSession("cache/semi_static_url_data_cache", expire_after=timedelta(days=30), - allowable_codes=(200, 400, 401, 403, 404)) -_lock = threading.Lock() - - -class _ThreadSafeSession: - """Wraps CachedSession with a lock to prevent concurrent SQLite access across threads.""" - - def get(self, *args, **kwargs): - with _lock: - return _session.get(*args, **kwargs) - - -SEMI_STATIC_URL_DATA_CACHE = _ThreadSafeSession() diff --git a/core/cleanup.py b/core/cleanup.py deleted file mode 100644 index 0592a7f..0000000 --- a/core/cleanup.py +++ /dev/null @@ -1,73 +0,0 @@ -import logging -from datetime import datetime -from threading import Event, Thread - -import pytz - - -class CleanupTimer: - """Provides a timed cleanup of the spot list.""" - - def __init__(self, spots, alerts, web_server, cleanup_interval): - """Constructor""" - - self._spots = spots - self._alerts = alerts - self._web_server = web_server - self._cleanup_interval = cleanup_interval - self.last_cleanup_time = datetime.min.replace(tzinfo=pytz.UTC) - self.status = "Starting" - self._thread = None - self._stop_event = Event() - - def start(self): - """Start the cleanup timer""" - - self._thread = Thread(target=self._run, daemon=True) - self._thread.start() - - def stop(self): - """Stop any threads and prepare for application shutdown""" - - self._stop_event.set() - - def _run(self): - while not self._stop_event.wait(timeout=self._cleanup_interval): - self._cleanup() - - def _cleanup(self): - """Perform cleanup and reschedule next timer""" - - try: - # Perform cleanup via letting the data expire - self._spots.expire() - self._alerts.expire() - - # Explicitly clean up any spots and alerts that have expired - for i in list(self._spots.iterkeys()): - try: - spot = self._spots[i] - if spot.expired(): - self._spots.delete(i) - except KeyError: - # Must have already been deleted, OK with that - pass - for i in list(self._alerts.iterkeys()): - try: - alert = self._alerts[i] - if alert.expired(): - self._alerts.delete(i) - except KeyError: - # Must have already been deleted, OK with that - pass - - # Clean up web server SSE spot/alert queues - self._web_server.clean_up_sse_queues() - - self.status = "OK" - self.last_cleanup_time = datetime.now(pytz.UTC) - - except Exception: - self.status = "Error" - logging.exception("Exception in Cleanup thread") - self._stop_event.wait(timeout=1) diff --git a/core/data_store.py b/core/data_store.py new file mode 100644 index 0000000..b78affd --- /dev/null +++ b/core/data_store.py @@ -0,0 +1,69 @@ +from pathlib import Path + +import diskcache + +from core.config import MAX_SPOT_AGE, MAX_ALERT_AGE +from core.live_data_cache import LiveDataCache +from data.solar_conditions import SolarConditions + + +class DataStore: + """Data caching/storage object. Handles storage of spots, alerts, solar conditions, SIG reference data, and callsign + lookup data using different caching strategies for each.""" + + def __init__(self): + cache_dir = "./cache" + self.MAX_SPOT_COUNT = 10000 + self.MAX_ALERT_COUNT = 10000 + self.SPOT_ALERT_SNAPSHOT_INTERVAL_SEC = 300 + self.CALLSIGN_DATA_TTL_SEC = 30 * 24 * 60 * 60 + + Path(cache_dir).mkdir(parents=True, exist_ok=True) + + # Special caches for spots and alerts, which have TTL and write snapshots to disk at an interval + self.spots = LiveDataCache(maxsize=self.MAX_SPOT_COUNT, ttl=MAX_SPOT_AGE, + snapshot_dir=cache_dir + "/spots", + snapshot_interval_sec=self.SPOT_ALERT_SNAPSHOT_INTERVAL_SEC) + self.alerts = LiveDataCache(maxsize=self.MAX_ALERT_COUNT, ttl=MAX_ALERT_AGE, + snapshot_dir=cache_dir + "/alerts", + snapshot_interval_sec=self.SPOT_ALERT_SNAPSHOT_INTERVAL_SEC) + + # Standard disk cache for solar data and status data, but each cache contains only a single object which we + # expose to the wider application + self.solar = diskcache.Cache(cache_dir + "/solar") + if "solar_conditions" not in self.solar: + self.solar.add("solar_conditions", SolarConditions()) + self.solar_conditions = self.solar.get("solar_conditions") + self.status = diskcache.Cache(cache_dir + "/status") + if "status_data" not in self.status: + self.status.add("status_data", {}) + self.status_data = self.status.get("status_data") + + # Standard disk caches for SIG ref data. Separate provider threads will repopulate these on a regular basis + # but there's no need for a TTL since old data is better than no data. + self.sigrefs_wwff = diskcache.Cache(cache_dir + "/sigrefs/wwff") + self.sigrefs_siota = diskcache.Cache(cache_dir + "/sigrefs/siota") + self.sigrefs_wota = diskcache.Cache(cache_dir + "/sigrefs/wota") + self.sigrefs_zlota = diskcache.Cache(cache_dir + "/sigrefs/zlota") + self.sigrefs_llota = diskcache.Cache(cache_dir + "/sigrefs/llota") + self.sigrefs_dme = diskcache.Cache(cache_dir + "/sigrefs/dme") + + # Standard disk cache for callsign data. This data does have a TTL to trigger an occasional re-lookup. + # Old data *is* better than no data, but we can't have a background thread re-looking-up every callsign + # we've seen, so we rely on them timing out and this triggering another lookup. + self.callsigns = diskcache.Cache(cache_dir + "/callsigns") + + def close(self): + self.spots.close() + self.alerts.close() + self.solar.close() + self.status.close() + self.sigrefs_wwff.close() + self.sigrefs_siota.close() + self.sigrefs_wota.close() + self.sigrefs_zlota.close() + self.sigrefs_llota.close() + self.callsigns.close() + +# Global object +DATA_STORE = DataStore() \ No newline at end of file diff --git a/core/live_data_cache.py b/core/live_data_cache.py new file mode 100644 index 0000000..9528ae9 --- /dev/null +++ b/core/live_data_cache.py @@ -0,0 +1,72 @@ +import logging +import threading +import time + +import diskcache +from cachetools import TTLCache + + +class LiveDataCache: + """Cache for spots and alerts. Uses the faster in-memory TTLCache for normal data I/O, including the TTL to enforce + maximum lifetime, and adds a separate diskcache to which we can save and load the TTLCache to provide persistence. + Also adds thread safety which TTLCache doesn't do.""" + + def __init__(self, maxsize, ttl, snapshot_dir, snapshot_interval_sec): + self._cache = TTLCache(maxsize=maxsize, ttl=ttl) + self._lock = threading.Lock() + self._ttl = ttl + self._snapshot_dir = snapshot_dir + self._disk_cache = diskcache.Cache(str(snapshot_dir)) + self._load_snapshot() + self._start_periodic_snapshot(snapshot_interval_sec) + + def set(self, key, value): + with self._lock: + self._cache[key] = value + + def get(self, key, default=None): + with self._lock: + return self._cache.get(key, default) + + def delete(self, key): + with self._lock: + self._cache.pop(key, None) + + def values(self): + with self._lock: + return list(self._cache.values()) + + def save_snapshot(self): + with self._lock: + # Store the time with the data so we can avoid loading anything nxt time that's older than TTL + data = [(k, v, time.time()) for k, v in self._cache.items()] + try: + self._disk_cache.set("snapshot", data) + except Exception as e: + logging.error("Failed to write snapshot to %s", self._snapshot_dir, e, exc_info=True) + + def _load_snapshot(self): + data = self._disk_cache.get("snapshot") + if not data: + return + + now = time.time() + with self._lock: + for key, value, saved_at in data: + # Only restore entries that would still be within TTL + if now - saved_at < self._ttl: + self._cache[key] = value + logging.info("Loaded snapshot from %s", self._snapshot_dir) + + def _start_periodic_snapshot(self, interval): + def loop(): + while True: + time.sleep(interval) + self.save_snapshot() + + t = threading.Thread(target=loop, daemon=True, name=f"snapshot-{self._snapshot_dir}") + t.start() + + def close(self): + self.save_snapshot() + self._disk_cache.close() diff --git a/core/lookup_helper.py b/core/lookup_helper.py index 8d8dec4..c0fdbaf 100644 --- a/core/lookup_helper.py +++ b/core/lookup_helper.py @@ -14,7 +14,7 @@ from pyhamtools.locator import latlong_to_locator from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout from requests_cache import CachedSession -from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE +from core.url_data_cache import URL_DATA_CACHE from core.config import config from core.constants import BANDS, UNKNOWN_BAND, CW_MODES, PHONE_MODES, DATA_MODES, ALL_MODES, \ HTTP_HEADERS, HAMQTH_PRG, MODE_ALIASES @@ -142,8 +142,8 @@ 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) + response = URL_DATA_CACHE.get("https://www.country-files.com/cty/cty.plist", + headers=HTTP_HEADERS) if response.ok: with open(self._country_files_cty_plist_download_location, "w") as f: @@ -167,7 +167,7 @@ class LookupHelper: try: logging.info("Downloading dxcc.json...") - response = SEMI_STATIC_URL_DATA_CACHE.get( + response = URL_DATA_CACHE.get( "https://raw.githubusercontent.com/k0swe/dxcc-json/refs/heads/main/dxcc.json", headers=HTTP_HEADERS) @@ -515,7 +515,7 @@ class LookupHelper: for lookup_call in calls_to_try: try: - response = SEMI_STATIC_URL_DATA_CACHE.get( + response = URL_DATA_CACHE.get( self._qrz_base_url + "?s=" + session_key + "&callsign=" + urllib.parse.quote_plus(lookup_call), headers=HTTP_HEADERS, timeout=10) if response.ok: @@ -593,7 +593,7 @@ class LookupHelper: for lookup_call in calls_to_try: try: - response = SEMI_STATIC_URL_DATA_CACHE.get( + response = URL_DATA_CACHE.get( self._hamqth_base_url + "?id=" + session_id + "&callsign=" + urllib.parse.quote_plus( lookup_call) + "&prg=" + HAMQTH_PRG, headers=HTTP_HEADERS) if response.ok: diff --git a/core/sig_utils.py b/core/sig_utils.py index 258d5ac..82f7809 100644 --- a/core/sig_utils.py +++ b/core/sig_utils.py @@ -4,23 +4,16 @@ import logging from pyhamtools.locator import latlong_to_locator, locator_to_latlong from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout -from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE from core.constants import SIGS, HTTP_HEADERS +from core.data_store import DATA_STORE from core.geo_utils import wab_wai_square_to_lat_lon +from core.url_data_cache import URL_DATA_CACHE # Load Spanish municipality data for the DME programme. There's no convenient lookup API for this, so we embed the data # file in Spothole and load it on startup. with open("datafiles/MUNICIPIOS.csv", encoding="latin-1") as _f: - _DME_INDEX = {row["COD_INE"][:5]: row for row in csv.DictReader(_f, delimiter=";")} -# Caches for data for the SIGs where we have to download a whole global reference list, rather than looking up a single -# reference. These get populated from the SEMI_STATIC_URL_DATA_CACHE only if the data actually came -# live from the internet, to avoid repopulating them every time we pull the same data from the cache. -_WWFF_INDEX_CACHE = {} -_SIOTA_INDEX_CACHE = {} -_WOTA_INDEX_CACHE = {} -_ZLOTA_INDEX_CACHE = {} -_LLOTA_INDEX_CACHE = {} - + for row in csv.DictReader(_f, delimiter=";"): + DATA_STORE.sigrefs_dme.add(row["COD_INE"][:5], row) def get_ref_regex_for_sig(sig): """Utility function to get the regex string for a SIG reference for a named SIG. If no match is found, None will be returned.""" @@ -53,7 +46,7 @@ def populate_sig_ref_info(sig_ref): ref_id = sig_ref.id try: if sig.upper() == "POTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://api.pota.app/park/" + ref_id, headers=HTTP_HEADERS) + response = URL_DATA_CACHE.get("https://api.pota.app/park/" + ref_id, headers=HTTP_HEADERS) if response.ok: data = response.json() if data: @@ -71,7 +64,7 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "SOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + ref_id, + response = URL_DATA_CACHE.get("https://api-db2.sota.org.uk/api/summits/" + ref_id, headers=HTTP_HEADERS) if response.ok: data = response.json() @@ -88,7 +81,7 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "WWBOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + ref_id, + response = URL_DATA_CACHE.get("https://api.wwbota.org/bunkers/" + ref_id, headers=HTTP_HEADERS) if response.ok: data = response.json() @@ -104,7 +97,7 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "GMA" or sig.upper() == "ARLHS" or sig.upper() == "ILLW" or sig.upper() == "WCA" or sig.upper() == "MOTA" or sig.upper() == "IOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + ref_id, + response = URL_DATA_CACHE.get("https://www.cqgma.org/api/ref/?" + ref_id, headers=HTTP_HEADERS) if response.ok: data = response.json() @@ -130,15 +123,14 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "WWFF": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://wwff.co/wwff-data/wwff_directory.csv", + response = URL_DATA_CACHE.get("https://wwff.co/wwff-data/wwff_directory.csv", headers=HTTP_HEADERS) if response.ok: - global _WWFF_INDEX_CACHE - if not bool(_WWFF_INDEX_CACHE) or not response.from_cache: + if not bool(DATA_STORE.sigrefs_wwff) or not response.from_cache: # New data from WWFF, update our internal map - _WWFF_INDEX_CACHE = {row["reference"]: row for row in - csv.DictReader(response.content.decode().splitlines())} - row = _WWFF_INDEX_CACHE.get(ref_id) + for row in csv.DictReader(response.content.decode().splitlines()): + DATA_STORE.sigrefs_wwff.add(row["reference"], row) + row = DATA_STORE.sigrefs_wwff.get(ref_id) if row: sig_ref.name = row["name"] if "name" in row else None sig_ref.url = "https://wwff.co/directory/?showRef=" + ref_id @@ -152,15 +144,14 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "SIOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", + response = URL_DATA_CACHE.get("https://www.silosontheair.com/data/silos.csv", headers=HTTP_HEADERS) if response.ok: - global _SIOTA_INDEX_CACHE - if not bool(_SIOTA_INDEX_CACHE) or not response.from_cache: + if not bool(DATA_STORE.sigrefs_siota) or not response.from_cache: # New data from SIOTA, update our internal map - _SIOTA_INDEX_CACHE = {row["SILO_CODE"]: row for row in - csv.DictReader(response.content.decode().splitlines())} - row = _SIOTA_INDEX_CACHE.get(ref_id) + for row in csv.DictReader(response.content.decode().splitlines()): + DATA_STORE.sigrefs_siota.add(row["SILO_CODE"], row) + row = DATA_STORE.sigrefs_siota.get(ref_id) if row: sig_ref.name = row["NAME"] if "NAME" in row else None sig_ref.grid = row["LOCATOR"] if "LOCATOR" in row else None @@ -172,17 +163,16 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "WOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", + response = URL_DATA_CACHE.get("https://www.wota.org.uk/mapping/data/summits.json", headers=HTTP_HEADERS) if response.ok: data = response.json() if data: - global _WOTA_INDEX_CACHE - if not bool(_WOTA_INDEX_CACHE) or not response.from_cache: + if not bool(DATA_STORE.sigrefs_wota) or not response.from_cache: # New data from WOTA, update our internal map - _WOTA_INDEX_CACHE = {feature["properties"]["wotaId"]: feature for feature in - data.get("features", [])} - feature = _WOTA_INDEX_CACHE.get(ref_id) + for feature in data.get("features", []): + DATA_STORE.sigrefs_wota.add(feature["properties"]["wotaId"], feature) + feature = DATA_STORE.sigrefs_wota.get(ref_id) if feature: sig_ref.name = feature["properties"]["title"] # Fudge WOTA URLs. Outlying fell (LDO) URLs don't match their ID numbers but require 214 to be @@ -200,24 +190,24 @@ def populate_sig_ref_info(sig_ref): logging.warning("HTTP %d looking up %s ref %s", response.status_code, sig, ref_id) elif sig.upper() == "ZLOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS) + response = URL_DATA_CACHE.get("https://ontheair.nz/assets/assets.json", headers=HTTP_HEADERS) if response.ok: data = response.json() if isinstance(data, list): - global _ZLOTA_INDEX_CACHE - if not bool(_ZLOTA_INDEX_CACHE) or not response.from_cache: + if not bool(DATA_STORE.sigrefs_zlota) or not response.from_cache: # New data from ZLOTA, update our internal map - _ZLOTA_INDEX_CACHE = {asset["code"]: asset for asset in data} - asset = _ZLOTA_INDEX_CACHE.get(ref_id) - if asset: - sig_ref.name = asset["name"] + for ref in data: + DATA_STORE.sigrefs_zlota.add(ref["code"], ref) + ref = DATA_STORE.sigrefs_zlota.get(ref_id) + if ref: + sig_ref.name = ref["name"] sig_ref.url = "https://ontheair.nz/assets/" + ref_id.replace("/", "_") try: - sig_ref.grid = latlong_to_locator(asset["y"], asset["x"], 6) + sig_ref.grid = latlong_to_locator(ref["y"], ref["x"], 6) except: logging.debug("Invalid lat/lon received for reference") - sig_ref.latitude = asset["y"] - sig_ref.longitude = asset["x"] + sig_ref.latitude = ref["y"] + sig_ref.longitude = ref["x"] elif not response.from_cache: logging.warning("Malformed response looking up %s ref %s", sig, ref_id) elif not response.from_cache: @@ -229,16 +219,16 @@ def populate_sig_ref_info(sig_ref): sig_ref.url = "https://www.beachesontheair.com/beaches/" + sig_ref.name.lower().replace(" ", "-") elif sig.upper() == "LLOTA": - response = SEMI_STATIC_URL_DATA_CACHE.get("https://llota.app/api/public/references", + response = URL_DATA_CACHE.get("https://llota.app/api/public/references", headers=HTTP_HEADERS) if response.ok: data = response.json() if isinstance(data, list): - global _LLOTA_INDEX_CACHE - if not bool(_LLOTA_INDEX_CACHE) or not response.from_cache: + if not bool(DATA_STORE.sigrefs_llota) or not response.from_cache: # New data from LLOTA, update our internal map - _LLOTA_INDEX_CACHE = {ref["reference_code"]: ref for ref in data} - ref = _LLOTA_INDEX_CACHE.get(ref_id) + for ref in data: + DATA_STORE.sigrefs_llota.add(ref["reference_code"], ref) + ref = DATA_STORE.sigrefs_llota.get(ref_id) if ref: sig_ref.name = str(ref["name"]) sig_ref.url = "https://llota.app/list/ref/" + ref_id @@ -280,7 +270,7 @@ def populate_sig_ref_info(sig_ref): elif sig.upper() == "DME": # Zero-pad to 5 digits to match our source data - row = _DME_INDEX.get(ref_id.zfill(5)) + row = DATA_STORE.sigrefs_dme.get(ref_id.zfill(5)) if row: sig_ref.name = row["NOMBRE_ACTUAL"] + ", " + row["PROVINCIA"] sig_ref.latitude = float(row["LATITUD_ETRS89_REGCAN95"].replace(",", ".")) if row.get( diff --git a/core/status_reporter.py b/core/status_reporter.py index ff6e39a..5b42de1 100644 --- a/core/status_reporter.py +++ b/core/status_reporter.py @@ -13,25 +13,21 @@ from core.prometheus_metrics_handler import memory_use_gauge, spots_gauge, alert class StatusReporter: """Provides a timed update of the application's status data.""" - def __init__(self, status_data, run_interval, web_server, cleanup_timer, spots, spot_providers, alerts, - alert_providers, solar_condition_providers): + def __init__(self, data_store, run_interval, web_server,spot_providers, alert_providers, solar_condition_providers): """Constructor""" - self._status_data = status_data + self._data_store = data_store self._run_interval = run_interval self._web_server = web_server - self._cleanup_timer = cleanup_timer - self._spots = spots self._spot_providers = spot_providers - self._alerts = alerts self._alert_providers = alert_providers self._solar_condition_providers = solar_condition_providers self._thread = None self._stop_event = Event() self._startup_time = datetime.now(pytz.UTC) - self._status_data["software-version"] = SOFTWARE_VERSION - self._status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN + self._data_store.status_data["software-version"] = SOFTWARE_VERSION + self._data_store.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN def start(self): """Start the reporter thread""" @@ -55,44 +51,41 @@ class StatusReporter: def _report(self): """Write status information""" - self._status_data["uptime"] = (datetime.now(pytz.UTC) - self._startup_time).total_seconds() - self._status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3) - self._status_data["num_spots"] = len(self._spots) - self._status_data["num_alerts"] = len(self._alerts) - self._status_data["spot_providers"] = list( - map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, - "last_updated": p.last_update_time.replace( - tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0, - "last_spot": p.last_spot_time.replace( - tzinfo=pytz.UTC).timestamp() if p.last_spot_time.year > 2000 else 0}, - self._spot_providers)) - self._status_data["alert_providers"] = list( - map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, - "last_updated": p.last_update_time.replace( - tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0}, - self._alert_providers)) - self._status_data["solar_condition_providers"] = list( - map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, - "last_updated": p.last_update_time.replace( - tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0}, - self._solar_condition_providers)) - self._status_data["cleanup"] = {"status": self._cleanup_timer.status, - "last_ran": self._cleanup_timer.last_cleanup_time.replace( - tzinfo=pytz.UTC).timestamp() if self._cleanup_timer.last_cleanup_time else 0} - self._status_data["webserver"] = {"status": self._web_server.web_server_metrics["status"], - "last_api_access": self._web_server.web_server_metrics[ - "last_api_access_time"].replace( - tzinfo=pytz.UTC).timestamp() if self._web_server.web_server_metrics[ - "last_api_access_time"] else 0, - "api_access_count": self._web_server.web_server_metrics["api_access_counter"], - "last_page_access": self._web_server.web_server_metrics[ - "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"]} + self._data_store.status_data["uptime"] = (datetime.now(pytz.UTC) - self._startup_time).total_seconds() + self._data_store.status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3) + self._data_store.status_data["num_spots"] = len(self._data_store.spots.values()) + self._data_store.status_data["num_alerts"] = len(self._data_store.alerts.values()) + self._data_store.status_data["spot_providers"] = list( + map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, + "last_updated": p.last_update_time.replace( + tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0, + "last_spot": p.last_spot_time.replace( + tzinfo=pytz.UTC).timestamp() if p.last_spot_time.year > 2000 else 0}, + self._spot_providers)) + self._data_store.status_data["alert_providers"] = list( + map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, + "last_updated": p.last_update_time.replace( + tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0}, + self._alert_providers)) + self._data_store.status_data["solar_condition_providers"] = list( + map(lambda p: {"name": p.name, "enabled": p.enabled, "status": p.status, + "last_updated": p.last_update_time.replace( + tzinfo=pytz.UTC).timestamp() if p.last_update_time.year > 2000 else 0}, + self._solar_condition_providers)) + self._data_store.status_data["webserver"] = {"status": self._web_server.web_server_metrics["status"], + "last_api_access": self._web_server.web_server_metrics[ + "last_api_access_time"].replace( + tzinfo=pytz.UTC).timestamp() if self._web_server.web_server_metrics[ + "last_api_access_time"] else 0, + "api_access_count": self._web_server.web_server_metrics["api_access_counter"], + "last_page_access": self._web_server.web_server_metrics[ + "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"]} # Update Prometheus metrics memory_use_gauge.set(psutil.Process(os.getpid()).memory_info().rss) - spots_gauge.set(len(self._spots)) - alerts_gauge.set(len(self._alerts)) + spots_gauge.set(len(self._data_store.spots.values())) + alerts_gauge.set(len(self._data_store.alerts.values())) diff --git a/core/url_data_cache.py b/core/url_data_cache.py new file mode 100644 index 0000000..7119e96 --- /dev/null +++ b/core/url_data_cache.py @@ -0,0 +1,23 @@ +import threading +from datetime import timedelta + +from requests_cache import CachedSession + +# Cache for "semi-static" data retrieved from a URL. This is a layet of caching in addition to the normal caching of +# spots, alerts and other data in the DataStore class. Its purpose is to avoid hitting remote endpoints frequently when +# e.g. restarting Spothole many times during testing. +_session = CachedSession("cache/semi_static_urls", expire_after=timedelta(days=1), + allowable_codes=(200, 400, 401, 403, 404)) +_lock = threading.Lock() + + +class _ThreadSafeSession: + """Wraps CachedSession with a lock to prevent concurrent SQLite access across threads. This allows a single object + to be used freely across the application.""" + + def get(self, *args, **kwargs): + with _lock: + return _session.get(*args, **kwargs) + +# Global object +URL_DATA_CACHE = _ThreadSafeSession() diff --git a/requirements.txt b/requirements.txt index 163a941..9dadff3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -17,4 +17,5 @@ websocket-client~=1.8.0 tornado~=6.4.2 tornado_eventsource~=3.0.0 geopandas~=0.13.2 -simplejson~=4.1.1 \ No newline at end of file +simplejson~=4.1.1 +cachetools~=7.1.6 \ No newline at end of file diff --git a/server/handlers/api/addspot.py b/server/handlers/api/addspot.py index 20c64c1..fdc7003 100644 --- a/server/handlers/api/addspot.py +++ b/server/handlers/api/addspot.py @@ -119,7 +119,7 @@ class APISpotHandler(tornado.web.RequestHandler): # infer missing data, and add it to our database. spot.source = "API" spot.infer_missing() - self._spots.add(spot.id, spot, expire=MAX_SPOT_AGE) + self._spots.set(spot.id, spot) self.write(safe_json_dumps("OK")) self.set_status(201) diff --git a/server/webserver.py b/server/webserver.py index f55bf21..52d646f 100644 --- a/server/webserver.py +++ b/server/webserver.py @@ -25,15 +25,12 @@ _HERE = os.path.dirname(__file__ or "") class WebServer: """Provides the public-facing web server.""" - def __init__(self, spots, alerts, solar_conditions, status_data): + def __init__(self, data_store): """Constructor""" - self._spots = spots - self._alerts = alerts - self._solar_conditions = solar_conditions + self._data_store = data_store self._sse_spot_queues = [] self._sse_alert_queues = [] - self._status_data = status_data self._port = WEB_SERVER_PORT self._api_only_mode = API_ONLY_MODE self._shutdown_event = asyncio.Event() @@ -64,20 +61,20 @@ class WebServer: # API endpoints are always enabled api_routes = [ - (r"/api/v1/spots", APISpotsHandler, {"spots": self._spots, **handler_opts}), - (r"/api/v1/alerts", APIAlertsHandler, {"alerts": self._alerts, **handler_opts}), + (r"/api/v1/spots", APISpotsHandler, {"spots": self._data_store.spots, **handler_opts}), + (r"/api/v1/alerts", APIAlertsHandler, {"alerts": self._data_store.alerts, **handler_opts}), (r"/api/v1/spots/stream", APISpotsStreamHandler, {"sse_spot_queues": self._sse_spot_queues, **handler_opts}), (r"/api/v1/alerts/stream", APIAlertsStreamHandler, {"sse_alert_queues": self._sse_alert_queues, **handler_opts}), - (r"/api/v1/solar", APISolarConditionsHandler, {"solar_conditions": self._solar_conditions, **handler_opts}), - (r"/api/v1/dxstats", APIDxStatsHandler, {"spots": self._spots, **handler_opts}), - (r"/api/v1/options", APIOptionsHandler, {"status_data": self._status_data, **handler_opts}), - (r"/api/v1/status", APIStatusHandler, {"status_data": self._status_data, **handler_opts}), + (r"/api/v1/solar", APISolarConditionsHandler, {"solar_conditions": self._data_store.solar, **handler_opts}), + (r"/api/v1/dxstats", APIDxStatsHandler, {"spots": self._data_store.spots, **handler_opts}), + (r"/api/v1/options", APIOptionsHandler, {"status_data": self._data_store.status, **handler_opts}), + (r"/api/v1/status", APIStatusHandler, {"status_data": self._data_store.status, **handler_opts}), (r"/api/v1/lookup/call", APILookupCallHandler, {**handler_opts}), (r"/api/v1/lookup/sigref", APILookupSIGRefHandler, {**handler_opts}), (r"/api/v1/lookup/grid", APILookupGridHandler, {**handler_opts}), - (r"/api/v1/spot", APISpotHandler, {"spots": self._spots, **handler_opts}), + (r"/api/v1/spot", APISpotHandler, {"spots": self._data_store.spots, **handler_opts}), ] # If in API-only mode, serve a basic homepage; in normal mode, serve the usual UI routes diff --git a/solarconditionsproviders/giroionosonde.py b/solarconditionsproviders/giroionosonde.py index a0c94a8..b9d5a19 100644 --- a/solarconditionsproviders/giroionosonde.py +++ b/solarconditionsproviders/giroionosonde.py @@ -44,12 +44,12 @@ class GIROIonosonde(SolarConditionsProvider): stations.append({"ursi": row[0].strip(), "name": row[1].strip()}) return stations - def setup(self, solar_conditions, solar_conditions_cache): + def setup(self, solar_conditions): """Pre-populate ionosonde_data with known station names for stations not already present, so the station dropdown is available before the first poll. Does not overwrite existing entries so KC2G cache data is preserved.""" - super().setup(solar_conditions, solar_conditions_cache) + super().setup(solar_conditions) existing = solar_conditions.ionosonde_data or {} new_entries = { s["ursi"]: {"ursi": s["ursi"], "name": s["name"], "fof2": None, "muf": None, diff --git a/solarconditionsproviders/solar_conditions_provider.py b/solarconditionsproviders/solar_conditions_provider.py index bc939f7..65e5da1 100644 --- a/solarconditionsproviders/solar_conditions_provider.py +++ b/solarconditionsproviders/solar_conditions_provider.py @@ -10,18 +10,16 @@ class SolarConditionsProvider: def __init__(self, name, provider_config): """Constructor""" - self._solar_conditions_cache = None self.name = name self.enabled = provider_config["enabled"] self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC) self.status = "Not Started" if self.enabled else "Disabled" self._solar_conditions = None - def setup(self, solar_conditions, solar_conditions_cache): - """Set up the provider, giving it the solar conditions object and its backing cache""" + def setup(self, solar_conditions): + """Set up the provider, giving it the solar conditions object""" self._solar_conditions = solar_conditions - self._solar_conditions_cache = solar_conditions_cache def start(self): """Start the provider. This should return immediately after spawning threads to access the remote resources""" @@ -41,4 +39,3 @@ class SolarConditionsProvider: if hasattr(self._solar_conditions, key): setattr(self._solar_conditions, key, value) self._solar_conditions.infer_descriptions() - self._solar_conditions_cache['solar_conditions'] = self._solar_conditions diff --git a/spothole.py b/spothole.py index d2ed8be..b361952 100644 --- a/spothole.py +++ b/spothole.py @@ -5,23 +5,16 @@ import os import signal import sys -from diskcache import Cache - -from core.cleanup import CleanupTimer from core.config import config, SERVER_OWNER_CALLSIGN, LOG_LEVEL from core.constants import SOFTWARE_VERSION +from core.data_store import DATA_STORE from core.lookup_helper import lookup_helper from core.status_reporter import StatusReporter -from data.solar_conditions import SolarConditions from server.webserver import WebServer # Globals -spots = Cache('cache/spots_cache') -alerts = Cache('cache/alerts_cache') -solar_conditions_cache = Cache('cache/solar_conditions_cache') -solar_conditions = solar_conditions_cache.get('solar_conditions', SolarConditions()) +data_store = DATA_STORE web_server = None -status_data = {} spot_providers = [] alert_providers = [] solar_condition_providers = [] @@ -46,13 +39,7 @@ def shutdown(_signum=None, _frame=None): for scp in solar_condition_providers: if scp.enabled: scp.stop() - if cleanup_timer: - cleanup_timer.stop() - if lookup_helper: - lookup_helper.stop() - spots.close() - alerts.close() - solar_conditions_cache.close() + data_store.close() os._exit(0) @@ -103,13 +90,13 @@ if __name__ == '__main__': lookup_helper.start() # Set up web server - web_server = WebServer(spots=spots, alerts=alerts, solar_conditions=solar_conditions, status_data=status_data) + web_server = WebServer(data_store=data_store) # Fetch, set up and start spot providers for entry in config["spot-providers"]: spot_providers.append(get_spot_provider_from_config(entry)) for p in spot_providers: - p.setup(spots=spots, web_server=web_server) + p.setup(spots=data_store.spots, web_server=web_server) if p.enabled: p.start() @@ -117,7 +104,7 @@ if __name__ == '__main__': for entry in config["alert-providers"]: alert_providers.append(get_alert_provider_from_config(entry)) for p in alert_providers: - p.setup(alerts=alerts, web_server=web_server) + p.setup(alerts=data_store.alerts, web_server=web_server) if p.enabled: p.start() @@ -125,17 +112,12 @@ if __name__ == '__main__': for entry in config.get("solar-condition-providers", []): solar_condition_providers.append(get_solar_conditions_provider_from_config(entry)) for p in solar_condition_providers: - p.setup(solar_conditions=solar_conditions, solar_conditions_cache=solar_conditions_cache) + p.setup(solar_conditions=data_store.solar_conditions) if p.enabled: p.start() - # Set up timer to clear spot list of old data - cleanup_timer = CleanupTimer(spots=spots, alerts=alerts, web_server=web_server, cleanup_interval=60) - cleanup_timer.start() - # Set up status reporter - status_reporter = StatusReporter(status_data=status_data, spots=spots, alerts=alerts, web_server=web_server, - cleanup_timer=cleanup_timer, spot_providers=spot_providers, + status_reporter = StatusReporter(data_store=data_store, web_server=web_server, spot_providers=spot_providers, alert_providers=alert_providers, solar_condition_providers=solar_condition_providers, run_interval=5) status_reporter.start() diff --git a/spotproviders/gma.py b/spotproviders/gma.py index 8a7d2f0..0962964 100644 --- a/spotproviders/gma.py +++ b/spotproviders/gma.py @@ -3,7 +3,7 @@ from datetime import datetime import pytz -from core.cache_utils import SEMI_STATIC_URL_DATA_CACHE +from core.url_data_cache import URL_DATA_CACHE from core.constants import HTTP_HEADERS from data.sig_ref import SIGRef from data.spot import Spot @@ -56,8 +56,8 @@ class GMA(HTTPSpotProvider): # GMA doesn't give what programme (SIG) the reference is for until we separately look it up. if "REF" in source_spot: try: - ref_response = SEMI_STATIC_URL_DATA_CACHE.get(self.REF_INFO_URL_ROOT + source_spot["REF"], - headers=HTTP_HEADERS) + ref_response = URL_DATA_CACHE.get(self.REF_INFO_URL_ROOT + source_spot["REF"], + headers=HTTP_HEADERS) # Sometimes this is blank even if it's a 200 response, so handle that if ref_response.ok and ref_response.text is not None and ref_response.text != "": ref_info = ref_response.json() diff --git a/spotproviders/spot_provider.py b/spotproviders/spot_provider.py index 6b19a2c..48aa5a4 100644 --- a/spotproviders/spot_provider.py +++ b/spotproviders/spot_provider.py @@ -59,7 +59,7 @@ class SpotProvider: def _add_spot(self, spot): if not spot.expired(): - self._spots.add(spot.id, spot, expire=MAX_SPOT_AGE) + self._spots.set(spot.id, spot) # Ping the web server in case we have any SSE connections that need to see this immediately if self._web_server: self._web_server.notify_new_spot(spot)