From 818fd2d5044476a53bb4b68eda458b553ad76fa1 Mon Sep 17 00:00:00 2001 From: Ian Renton Date: Fri, 31 Jul 2026 15:18:23 +0100 Subject: [PATCH] Refactor of caching & data storage part 2 #118 --- alertproviders/alert_provider.py | 14 +--- core/live_data_cache.py | 32 ++++++++- core/lookup_helper.py | 2 +- core/status_reporter.py | 28 ++++---- core/utils.py | 12 +--- server/handlers/api/addspot.py | 2 +- server/handlers/api/alerts.py | 69 +++++------------- server/handlers/api/dxstats.py | 2 +- server/handlers/api/spots.py | 70 +++++------------- server/sse_broadcaster.py | 36 ++++++++++ server/webserver.py | 72 ++++--------------- sigrefdataproviders/sig_ref_data_provider.py | 28 ++++++++ solarconditionsproviders/giroionosonde.py | 27 ++++--- .../solar_conditions_provider.py | 9 +-- spothole.py | 10 +-- spotproviders/gma.py | 2 +- spotproviders/spot_provider.py | 14 +--- 17 files changed, 186 insertions(+), 243 deletions(-) create mode 100644 server/sse_broadcaster.py create mode 100644 sigrefdataproviders/sig_ref_data_provider.py diff --git a/alertproviders/alert_provider.py b/alertproviders/alert_provider.py index 4498679..27d019b 100644 --- a/alertproviders/alert_provider.py +++ b/alertproviders/alert_provider.py @@ -2,7 +2,7 @@ from datetime import datetime import pytz -from core.config import MAX_ALERT_AGE +from core.data_store import DATA_STORE class AlertProvider: @@ -15,14 +15,7 @@ class AlertProvider: 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._alerts = None - self._web_server = None - - def setup(self, alerts, web_server): - """Set up the provider, e.g. giving it the alert list to work from""" - - self._alerts = alerts - self._web_server = web_server + self._alerts = DATA_STORE.alerts def start(self): """Start the provider. This should return immediately after spawning threads to access the remote resources""" @@ -45,9 +38,6 @@ class AlertProvider: def _add_alert(self, alert): if not alert.expired(): 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) def stop(self): """Stop any threads and prepare for application shutdown""" diff --git a/core/live_data_cache.py b/core/live_data_cache.py index 9528ae9..6afbcec 100644 --- a/core/live_data_cache.py +++ b/core/live_data_cache.py @@ -7,14 +7,17 @@ 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 + """Cache for spots and alerts. Uses the fast 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.""" + Also adds thread safety so spots and alerts can come from any thread, and a listener mechanism so the web server + can get a callback when new spots/alerts are added, and send them to any SSE clients.""" 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._listeners = [] + self._listeners_lock = threading.Lock() self._snapshot_dir = snapshot_dir self._disk_cache = diskcache.Cache(str(snapshot_dir)) self._load_snapshot() @@ -24,6 +27,16 @@ class LiveDataCache: with self._lock: self._cache[key] = value + # Notify listeners + with self._listeners_lock: + listeners = list(self._listeners) + for callback in listeners: + try: + callback(value) + except Exception: + logging.error("Listener raised an exception for key %s", key, exc_info=True) + + def get(self, key, default=None): with self._lock: return self._cache.get(key, default) @@ -32,10 +45,25 @@ class LiveDataCache: with self._lock: self._cache.pop(key, None) + def keys(self): + with self._lock: + return list(self._cache.keys()) + def values(self): with self._lock: return list(self._cache.values()) + def add_listener(self, callback): + """Register callback(value) which will be called whenever a new spot/alert item is added via set(). Used by the + web server (via SSEBroadcaster) to send SSE clients an update on every new spot.""" + + with self._listeners_lock: + self._listeners.append(callback) + + def remove_listener(self, callback): + with self._listeners_lock: + self._listeners.remove(callback) + 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 diff --git a/core/lookup_helper.py b/core/lookup_helper.py index c0fdbaf..09d84d8 100644 --- a/core/lookup_helper.py +++ b/core/lookup_helper.py @@ -14,10 +14,10 @@ from pyhamtools.locator import latlong_to_locator from requests.exceptions import ConnectionError, ReadTimeout, ConnectTimeout from requests_cache import CachedSession -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 +from core.url_data_cache import URL_DATA_CACHE # QRZ XML field names differ from pyhamtools' normalised names; map them here. _QRZ_FIELD_MAP = { diff --git a/core/status_reporter.py b/core/status_reporter.py index 5b42de1..5da9421 100644 --- a/core/status_reporter.py +++ b/core/status_reporter.py @@ -7,16 +7,16 @@ import pytz from core.config import SERVER_OWNER_CALLSIGN from core.constants import SOFTWARE_VERSION +from core.data_store import DATA_STORE from core.prometheus_metrics_handler import memory_use_gauge, spots_gauge, alerts_gauge class StatusReporter: """Provides a timed update of the application's status data.""" - def __init__(self, data_store, run_interval, web_server,spot_providers, alert_providers, solar_condition_providers): + def __init__(self, run_interval, web_server,spot_providers, alert_providers, solar_condition_providers): """Constructor""" - self._data_store = data_store self._run_interval = run_interval self._web_server = web_server self._spot_providers = spot_providers @@ -26,8 +26,8 @@ class StatusReporter: self._stop_event = Event() self._startup_time = datetime.now(pytz.UTC) - self._data_store.status_data["software-version"] = SOFTWARE_VERSION - self._data_store.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN + DATA_STORE.status_data["software-version"] = SOFTWARE_VERSION + DATA_STORE.status_data["server-owner-callsign"] = SERVER_OWNER_CALLSIGN def start(self): """Start the reporter thread""" @@ -51,28 +51,28 @@ class StatusReporter: def _report(self): """Write status information""" - 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( + DATA_STORE.status_data["uptime"] = (datetime.now(pytz.UTC) - self._startup_time).total_seconds() + DATA_STORE.status_data["mem_use_mb"] = round(psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024), 3) + DATA_STORE.status_data["num_spots"] = len(DATA_STORE.spots.values()) + DATA_STORE.status_data["num_alerts"] = len(DATA_STORE.alerts.values()) + 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( + 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( + 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"], + 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[ @@ -87,5 +87,5 @@ class StatusReporter: # Update Prometheus metrics memory_use_gauge.set(psutil.Process(os.getpid()).memory_info().rss) - spots_gauge.set(len(self._data_store.spots.values())) - alerts_gauge.set(len(self._data_store.alerts.values())) + spots_gauge.set(len(DATA_STORE.spots.values())) + alerts_gauge.set(len(DATA_STORE.alerts.values())) diff --git a/core/utils.py b/core/utils.py index 9c99210..54a48ed 100644 --- a/core/utils.py +++ b/core/utils.py @@ -5,14 +5,4 @@ def safe_json_dumps(obj): """Safe version of json.dumps that also converts objects to dicts so they can be output, and ignores NaN floats which are invalid in JSON.""" - return simplejson.dumps(obj, ensure_ascii=False, ignore_nan=True, default=lambda o: o.__dict__) - - -def empty_queue(q): - """Empty a queue""" - - while not q.empty(): - try: - q.get_nowait() - except: - break + return simplejson.dumps(obj, ensure_ascii=False, ignore_nan=True, default=lambda o: o.__dict__) \ No newline at end of file diff --git a/server/handlers/api/addspot.py b/server/handlers/api/addspot.py index fdc7003..e93e090 100644 --- a/server/handlers/api/addspot.py +++ b/server/handlers/api/addspot.py @@ -8,7 +8,7 @@ import tornado from tornado import httputil from tornado.web import Application -from core.config import ALLOW_SPOTTING, MAX_SPOT_AGE +from core.config import ALLOW_SPOTTING from core.constants import UNKNOWN_BAND from core.lookup_helper import infer_band_from_freq from core.prometheus_metrics_handler import api_requests_counter diff --git a/server/handlers/api/alerts.py b/server/handlers/api/alerts.py index b556fbc..54c7aa9 100644 --- a/server/handlers/api/alerts.py +++ b/server/handlers/api/alerts.py @@ -1,7 +1,6 @@ import copy import logging from datetime import datetime -from queue import Queue from typing import Any import pytz @@ -11,12 +10,9 @@ from tornado import httputil from tornado.web import Application from core.prometheus_metrics_handler import api_requests_counter -from core.utils import safe_json_dumps, empty_queue +from core.utils import safe_json_dumps from data.lookup_credentials import extract_credentials -SSE_HANDLER_MAX_QUEUE_SIZE = 100 -SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000 - class APIAlertsHandler(tornado.web.RequestHandler): """API request handler for /api/v1/alerts""" @@ -73,16 +69,14 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler): """API request handler for /api/v1/alerts/stream""" def __init__(self, application, request, **kwargs: Any): - self._sse_alert_queues = None + self._sse_alert_broadcaster = None self._web_server_metrics = None self._query_params = None self._credentials = None - self._alert_queue = None - self._heartbeat = None super().__init__(application, request, **kwargs) - def initialize(self, sse_alert_queues, web_server_metrics): - self._sse_alert_queues = sse_alert_queues + def initialize(self, _sse_alert_broadcaster, web_server_metrics): + self._sse_alert_broadcaster = _sse_alert_broadcaster self._web_server_metrics = web_server_metrics def custom_headers(self): @@ -104,59 +98,32 @@ class APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler): self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()} self._credentials = extract_credentials(self._query_params) - # Create a alert queue and add it to the web server's list. The web server will fill this when alerts arrive - self._alert_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE) - self._sse_alert_queues.append(self._alert_queue) - - # Set up a timed callback to check if anything is in the queue - self._heartbeat = tornado.ioloop.PeriodicCallback(self._callback, SSE_HANDLER_QUEUE_CHECK_INTERVAL) - self._heartbeat.start() - # Flush headers immediately so nginx doesn't time out waiting for a response self.write_message("keepalive", "") + # Register to handle new alerts arriving. The callback() method will get called with the new alert as an + # argument. + self._sse_alert_broadcaster.register(self) + except Exception as e: logging.warning("Exception when serving SSE socket: %s", e, exc_info=True) self.close() def close(self): - """When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it""" + """When the user closes the socket, deregister ourselves from the alert broadcaster""" - try: - if self._alert_queue in self._sse_alert_queues: - self._sse_alert_queues.remove(self._alert_queue) - empty_queue(self._alert_queue) - except: - pass - try: - self._heartbeat.stop() - except: - pass - self._alert_queue = None + self._sse_alert_broadcaster.unregister(self) super().close() - def _callback(self): - """Callback to check if anything has arrived in the queue, and if so send it to the client""" + def callback(self, alert): + """Callback when a new alert arrives""" try: - if self._alert_queue: - if not self._alert_queue.empty(): - while not self._alert_queue.empty(): - alert = self._alert_queue.get() - # If the new alert matches our param filters, send it to the client. If not, ignore it. - if alert_allowed_by_query(alert, self._query_params): - if self._credentials: - alert = copy.deepcopy(alert) - alert.infer_missing(self._credentials) - self.write_message(msg=safe_json_dumps(alert)) - - else: - # Send a keepalive comment if the queue was empty - self.write_message("keepalive", "") - - if self._alert_queue not in self._sse_alert_queues: - logging.error("Web server cleared up a queue of an active connection!") - self.close() + if alert_allowed_by_query(alert, self._query_params): + if self._credentials: + alert = copy.deepcopy(alert) + alert.infer_missing(self._credentials) + self.write_message(msg=safe_json_dumps(alert)) except Exception as e: logging.warning("Exception in SSE callback, connection will be closed: %s", e, exc_info=True) self.close() @@ -169,7 +136,7 @@ def get_alert_list_with_filters(all_alerts, query): # Create a shallow copy of the alert list ordered by start time, then filter the list to reduce it only to alerts # that match the filter parameters in the query string. Finally, apply a limit to the number of alerts returned. # The list of query string filters is defined in the API docs. - alert_ids = list(all_alerts.iterkeys()) + alert_ids = all_alerts.keys() alerts = [] for k in alert_ids: a = all_alerts.get(k) diff --git a/server/handlers/api/dxstats.py b/server/handlers/api/dxstats.py index 7a683a5..22dc7c9 100644 --- a/server/handlers/api/dxstats.py +++ b/server/handlers/api/dxstats.py @@ -40,7 +40,7 @@ class APIDxStatsHandler(tornado.web.RequestHandler): one_hour_ago = (datetime.now(pytz.UTC) - timedelta(hours=1)).timestamp() counts = Counter() - for key in self._spots.iterkeys(): + for key in self._spots.keys(): spot = self._spots.get(key) if spot is None: continue diff --git a/server/handlers/api/spots.py b/server/handlers/api/spots.py index 675076f..b86b42f 100644 --- a/server/handlers/api/spots.py +++ b/server/handlers/api/spots.py @@ -1,7 +1,6 @@ import copy import logging from datetime import datetime, timedelta -from queue import Queue from typing import Any import pytz @@ -11,12 +10,9 @@ from tornado import httputil from tornado.web import Application from core.prometheus_metrics_handler import api_requests_counter -from core.utils import safe_json_dumps, empty_queue +from core.utils import safe_json_dumps from data.lookup_credentials import extract_credentials -SSE_HANDLER_MAX_QUEUE_SIZE = 1000 -SSE_HANDLER_QUEUE_CHECK_INTERVAL = 5000 - class APISpotsHandler(tornado.web.RequestHandler): """API request handler for /api/v1/spots""" @@ -73,16 +69,14 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler): """API request handler for /api/v1/spots/stream""" def __init__(self, application, request, **kwargs: Any): - self._sse_spot_queues = None + self._sse_spot_broadcaster = None self._web_server_metrics = None self._query_params = None self._credentials = None - self._spot_queue = None - self._heartbeat = None super().__init__(application, request, **kwargs) - def initialize(self, sse_spot_queues, web_server_metrics): - self._sse_spot_queues = sse_spot_queues + def initialize(self, sse_spot_broadcaster, web_server_metrics): + self._sse_spot_broadcaster = sse_spot_broadcaster self._web_server_metrics = web_server_metrics def custom_headers(self): @@ -106,59 +100,33 @@ class APISpotsStreamHandler(tornado_eventsource.handler.EventSourceHandler): self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()} self._credentials = extract_credentials(self._query_params) - # Create a spot queue and add it to the web server's list. The web server will fill this when spots arrive - self._spot_queue = Queue(maxsize=SSE_HANDLER_MAX_QUEUE_SIZE) - self._sse_spot_queues.append(self._spot_queue) - - # Set up a timed callback to check if anything is in the queue - self._heartbeat = tornado.ioloop.PeriodicCallback(self._callback, SSE_HANDLER_QUEUE_CHECK_INTERVAL) - self._heartbeat.start() - # Flush headers immediately so nginx doesn't time out waiting for a response self.write_message("keepalive", "") + # Register to handle new spots arriving. The callback() method will get called with the new spot as an + # argument. + self._sse_spot_broadcaster.register(self) + except Exception as e: logging.warning("Exception when serving SSE socket: %s", e, exc_info=True) self.close() def close(self): - """When the user closes the socket, empty our queue and remove it from the list so the server no longer fills it""" + """When the user closes the socket, deregister ourselves from the spot broadcaster""" - try: - if self._spot_queue in self._sse_spot_queues: - self._sse_spot_queues.remove(self._spot_queue) - empty_queue(self._spot_queue) - except: - pass - try: - self._heartbeat.stop() - except: - pass - self._spot_queue = None + self._sse_spot_broadcaster.unregister(self) super().close() - def _callback(self): - """Callback to check if anything has arrived in the queue, and if so send it to the client""" + def callback(self, spot): + """Callback when a new spot arrives""" try: - if self._spot_queue: - if not self._spot_queue.empty(): - while not self._spot_queue.empty(): - spot = self._spot_queue.get() - # If the new spot matches our param filters, send it to the client. If not, ignore it. - if spot_allowed_by_query(spot, self._query_params): - if self._credentials: - spot = copy.deepcopy(spot) - spot.infer_missing(self._credentials) - self.write_message(msg=safe_json_dumps(spot)) - - else: - # Send a keepalive comment if the queue was empty - self.write_message("keepalive", "") - - if self._spot_queue not in self._sse_spot_queues: - logging.error("Web server cleared up a queue of an active connection!") - self.close() + # If the new spot matches our param filters, send it to the client. If not, ignore it. + if spot_allowed_by_query(spot, self._query_params): + if self._credentials: + spot = copy.deepcopy(spot) + spot.infer_missing(self._credentials) + self.write_message(msg=safe_json_dumps(spot)) except Exception as e: logging.warning("Exception in SSE callback, connection will be closed: %s", e, exc_info=True) self.close() @@ -171,7 +139,7 @@ def get_spot_list_with_filters(all_spots, query): # Create a shallow copy of the spot list, ordered by spot time, then filter the list to reduce it only to spots # that match the filter parameters in the query string. Finally, apply a limit to the number of spots returned. # The list of query string filters is defined in the API docs. - spot_ids = list(all_spots.iterkeys()) + spot_ids = all_spots.keys() spots = [] for k in spot_ids: s = all_spots.get(k) diff --git a/server/sse_broadcaster.py b/server/sse_broadcaster.py new file mode 100644 index 0000000..a538117 --- /dev/null +++ b/server/sse_broadcaster.py @@ -0,0 +1,36 @@ +import logging +import threading + +from tornado.ioloop import IOLoop + + +class SSEBroadcaster: + """Bridge between DataStore listener callbacks (which fire on provider threads) to Tornado's async SSE handlers + (which live on the IOLoop thread) to avoid any interdependency between them.""" + + def __init__(self): + self._handlers = set() + self._lock = threading.Lock() + self._loop = IOLoop.current() + + def register(self, handler): + with self._lock: + self._handlers.add(handler) + + def unregister(self, handler): + with self._lock: + self._handlers.discard(handler) + + def publish(self, value): + self._loop.add_callback(self._fan_out, value) + + def _fan_out(self, value): + with self._lock: + handlers = list(self._handlers) + for handler in handlers: + try: + handler.callback(value) + except Exception: + # Connection probably dropped, ignore and de-register the handler to stop getting future items. + logging.debug("Failed to push to an SSE client; dropping it") + self.unregister(handler) \ No newline at end of file diff --git a/server/webserver.py b/server/webserver.py index 52d646f..8ba1c93 100644 --- a/server/webserver.py +++ b/server/webserver.py @@ -6,7 +6,7 @@ import tornado from tornado.web import StaticFileHandler from core.config import ALLOW_SPOTTING, WEB_SERVER_PORT, API_ONLY_MODE, LOG_WEB_REQUESTS, BASE_URL -from core.utils import empty_queue +from core.data_store import DATA_STORE from server.handlers.api.addspot import APISpotHandler from server.handlers.api.alerts import APIAlertsHandler, APIAlertsStreamHandler from server.handlers.api.dxstats import APIDxStatsHandler @@ -18,6 +18,7 @@ from server.handlers.api.status import APIStatusHandler from server.handlers.manifesthandler import ManifestHandler from server.handlers.metrics import PrometheusMetricsHandler from server.handlers.pagetemplate import PageTemplateHandler +from server.sse_broadcaster import SSEBroadcaster _HERE = os.path.dirname(__file__ or "") @@ -25,12 +26,12 @@ _HERE = os.path.dirname(__file__ or "") class WebServer: """Provides the public-facing web server.""" - def __init__(self, data_store): + def __init__(self): """Constructor""" - self._data_store = data_store - self._sse_spot_queues = [] - self._sse_alert_queues = [] + self._data_store = DATA_STORE + self._spot_broadcaster = SSEBroadcaster() + self._alert_broadcaster = SSEBroadcaster() self._port = WEB_SERVER_PORT self._api_only_mode = API_ONLY_MODE self._shutdown_event = asyncio.Event() @@ -42,6 +43,10 @@ class WebServer: "status": "Starting" } + # Listen for new spots and alerts being added to the cache, so we can notify SSE clients immediately + DATA_STORE.spots.add_listener(self._spot_broadcaster.publish) + DATA_STORE.alerts.add_listener(self._alert_broadcaster.publish) + def start(self): """Start the web server""" @@ -64,13 +69,13 @@ class WebServer: (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}), + {"sse_spot_broadcaster": self._spot_broadcaster, **handler_opts}), (r"/api/v1/alerts/stream", APIAlertsStreamHandler, - {"sse_alert_queues": self._sse_alert_queues, **handler_opts}), + {"sse_alert_broadcaster": self._alert_broadcaster, **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/options", APIOptionsHandler, {"status_data": self._data_store.status_data, **handler_opts}), + (r"/api/v1/status", APIStatusHandler, {"status_data": self._data_store.status_data, **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}), @@ -118,55 +123,6 @@ class WebServer: logging.info("You can access your copy of Spothole at " + BASE_URL) await self._shutdown_event.wait() - def notify_new_spot(self, spot): - """Internal method called when a new spot is added to the system. This is used to ping any SSE clients that are - awaiting a server-sent message with new spots.""" - - for queue in self._sse_spot_queues: - try: - queue.put(spot) - except: - # Cleanup thread was probably deleting the queue, that's fine - pass - pass - - def notify_new_alert(self, alert): - """Internal method called when a new alert is added to the system. This is used to ping any SSE clients that are - awaiting a server-sent message with new spots.""" - - for queue in self._sse_alert_queues: - try: - queue.put(alert) - except: - # Cleanup thread was probably deleting the queue, that's fine - pass - pass - - def clean_up_sse_queues(self): - """Clean up any SSE queues that are growing too large; probably their client disconnected and we didn't catch it - properly for some reason.""" - - for q in self._sse_spot_queues: - try: - if q.full(): - logging.warning( - "A full SSE spot queue was found, presumably because the client disconnected strangely. It has been removed.") - self._sse_spot_queues.remove(q) - empty_queue(q) - except: - # Probably got deleted already on another thread - pass - for q in self._sse_alert_queues: - try: - if q.full(): - logging.warning( - "A full SSE alert queue was found, presumably because the client disconnected strangely. It has been removed.") - self._sse_alert_queues.remove(q) - empty_queue(q) - except: - # Probably got deleted already on another thread - pass - pass def request_log(handler): """Custom log function to provide more data about requests when enabled, and to provide the ability to turn off diff --git a/sigrefdataproviders/sig_ref_data_provider.py b/sigrefdataproviders/sig_ref_data_provider.py new file mode 100644 index 0000000..cf2e7f8 --- /dev/null +++ b/sigrefdataproviders/sig_ref_data_provider.py @@ -0,0 +1,28 @@ +from datetime import datetime + +import pytz + + +class SIGRefDataProvider: + """Generic SIG reference data provider class. Subclasses of this query the individual URLs or files for data.""" + + def __init__(self, name, provider_config): + """Constructor""" + + self.name = name + self.enabled = provider_config["enabled"] + self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC) + self.last_spot_time = datetime.min.replace(tzinfo=pytz.UTC) + self.status = "Not Started" if self.enabled else "Disabled" + + + def start(self): + """Start the provider. This should return immediately after spawning threads to access the remote resources""" + + raise NotImplementedError("Subclasses must implement this method") + + + def stop(self): + """Stop any threads and prepare for application shutdown""" + + raise NotImplementedError("Subclasses must implement this method") diff --git a/solarconditionsproviders/giroionosonde.py b/solarconditionsproviders/giroionosonde.py index b9d5a19..10ffaae 100644 --- a/solarconditionsproviders/giroionosonde.py +++ b/solarconditionsproviders/giroionosonde.py @@ -35,6 +35,18 @@ class GIROIonosonde(SolarConditionsProvider): self._thread = None self._stop_event = Event() + # 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. + existing = self._solar_conditions.ionosonde_data or {} + new_entries = { + s["ursi"]: {"ursi": s["ursi"], "name": s["name"], "fof2": None, "muf": None, + "luf": None, "band_states": None} + for s in self._stations if s["ursi"] not in existing + } + if new_entries: + self.update_data({"ionosonde_data": {**existing, **new_entries}}) + @staticmethod def _load_stations(): stations = [] @@ -44,21 +56,6 @@ class GIROIonosonde(SolarConditionsProvider): stations.append({"ursi": row[0].strip(), "name": row[1].strip()}) return stations - 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) - existing = solar_conditions.ionosonde_data or {} - new_entries = { - s["ursi"]: {"ursi": s["ursi"], "name": s["name"], "fof2": None, "muf": None, - "luf": None, "band_states": None} - for s in self._stations if s["ursi"] not in existing - } - if new_entries: - self.update_data({"ionosonde_data": {**existing, **new_entries}}) - def start(self): logging.info(f"Set up query of GIRO ionosonde data API every {POLL_INTERVAL} seconds.") self._thread = Thread(target=self._run, daemon=True) diff --git a/solarconditionsproviders/solar_conditions_provider.py b/solarconditionsproviders/solar_conditions_provider.py index 65e5da1..f36ecd5 100644 --- a/solarconditionsproviders/solar_conditions_provider.py +++ b/solarconditionsproviders/solar_conditions_provider.py @@ -2,6 +2,8 @@ from datetime import datetime import pytz +from core.data_store import DATA_STORE + class SolarConditionsProvider: """Generic solar conditions provider class. Subclasses of this query individual APIs for space weather and @@ -14,12 +16,7 @@ class SolarConditionsProvider: 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): - """Set up the provider, giving it the solar conditions object""" - - self._solar_conditions = solar_conditions + self._solar_conditions = DATA_STORE.solar_conditions def start(self): """Start the provider. This should return immediately after spawning threads to access the remote resources""" diff --git a/spothole.py b/spothole.py index b361952..ad9016a 100644 --- a/spothole.py +++ b/spothole.py @@ -13,7 +13,6 @@ from core.status_reporter import StatusReporter from server.webserver import WebServer # Globals -data_store = DATA_STORE web_server = None spot_providers = [] alert_providers = [] @@ -39,7 +38,7 @@ def shutdown(_signum=None, _frame=None): for scp in solar_condition_providers: if scp.enabled: scp.stop() - data_store.close() + DATA_STORE.close() os._exit(0) @@ -90,13 +89,12 @@ if __name__ == '__main__': lookup_helper.start() # Set up web server - web_server = WebServer(data_store=data_store) + web_server = WebServer() # 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=data_store.spots, web_server=web_server) if p.enabled: p.start() @@ -104,7 +102,6 @@ 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=data_store.alerts, web_server=web_server) if p.enabled: p.start() @@ -112,12 +109,11 @@ 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=data_store.solar_conditions) if p.enabled: p.start() # Set up status reporter - status_reporter = StatusReporter(data_store=data_store, web_server=web_server, spot_providers=spot_providers, + status_reporter = StatusReporter(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 0962964..35e2a2a 100644 --- a/spotproviders/gma.py +++ b/spotproviders/gma.py @@ -3,8 +3,8 @@ from datetime import datetime import pytz -from core.url_data_cache import URL_DATA_CACHE from core.constants import HTTP_HEADERS +from core.url_data_cache import URL_DATA_CACHE from data.sig_ref import SIGRef from data.spot import Spot from spotproviders.http_spot_provider import HTTPSpotProvider diff --git a/spotproviders/spot_provider.py b/spotproviders/spot_provider.py index 48aa5a4..a6ff2c6 100644 --- a/spotproviders/spot_provider.py +++ b/spotproviders/spot_provider.py @@ -2,7 +2,7 @@ from datetime import datetime import pytz -from core.config import MAX_SPOT_AGE +from core.data_store import DATA_STORE class SpotProvider: @@ -16,14 +16,7 @@ class SpotProvider: self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC) self.last_spot_time = datetime.min.replace(tzinfo=pytz.UTC) self.status = "Not Started" if self.enabled else "Disabled" - self._spots = None - self._web_server = None - - def setup(self, spots, web_server): - """Set up the provider, e.g. giving it the spot list to work from""" - - self._spots = spots - self._web_server = web_server + self._spots = DATA_STORE.spots def start(self): """Start the provider. This should return immediately after spawning threads to access the remote resources""" @@ -60,9 +53,6 @@ class SpotProvider: def _add_spot(self, spot): if not spot.expired(): 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) def stop(self): """Stop any threads and prepare for application shutdown"""