import logging import threading import time import diskcache from cachetools import TTLCache class LiveDataCache: """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 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() self._start_periodic_snapshot(snapshot_interval_sec) def set(self, key, value): 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) def delete(self, key): 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 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()