""" Memory Cache Handles in-memory caching with TTL support, size limits, and automatic cleanup. """ import os import time import threading import logging from typing import Dict, Any, Optional # Historical fixed ceiling, kept as the fallback when RAM cannot be read. DEFAULT_MAX_SIZE = 1000 def _total_memory_mb() -> Optional[float]: """Physical RAM in MB, or None where /proc/meminfo is unavailable.""" try: with open('/proc/meminfo', 'r', encoding='utf-8') as fh: for line in fh: if line.startswith('MemTotal:'): return int(line.split()[1]) / 1024 except (OSError, ValueError, IndexError): return None return None def default_max_size() -> int: """Entry ceiling scaled to this machine's RAM. One fixed ceiling cannot serve both a 512 MB Pi Zero 2 W and an 8 GB Pi 5. Entries here are parsed API payloads that routinely run tens of kilobytes each, so a thousand of them is a comfortable cache on a large board and a substantial fraction of total RAM on a small one — where the process competing for that RAM is also driving the panel. Set LEDMATRIX_CACHE_MAX_ENTRIES to override. """ override = os.environ.get('LEDMATRIX_CACHE_MAX_ENTRIES') if override: try: value = int(override) if value > 0: return value except ValueError: pass total_mb = _total_memory_mb() if total_mb is None: return DEFAULT_MAX_SIZE if total_mb < 1536: # 512 MB and 1 GB boards return 150 if total_mb < 3072: # 2 GB return 400 if total_mb < 6144: # 4 GB return 800 return 1500 # 8 GB and up class MemoryCache: """Manages in-memory cache with TTL and size limits.""" def __init__(self, max_size: int = 1000, cleanup_interval: float = 300.0) -> None: """ Initialize memory cache. Args: max_size: Maximum number of entries in cache cleanup_interval: Seconds between automatic cleanups """ self.logger = logging.getLogger(__name__) self._cache: Dict[str, Dict[str, Any]] = {} self._timestamps: Dict[str, float] = {} self._lock = threading.Lock() self._max_size = max_size self._cleanup_interval = cleanup_interval self._last_cleanup = time.time() def get(self, key: str, max_age: Optional[int] = None) -> Optional[Dict[str, Any]]: """ Get value from memory cache. Args: key: Cache key max_age: Maximum age in seconds (None = no expiration) Returns: Cached value or None if not found or expired """ now = time.time() with self._lock: if key not in self._cache: return None timestamp = self._timestamps.get(key) if isinstance(timestamp, str): try: timestamp = float(timestamp) except ValueError: self.logger.error(f"Invalid timestamp format for key {key}: {timestamp}") timestamp = None if timestamp is None: return None # An explicit per-entry ttl wins over the caller's max_age, matching # DiskCache. max_age is inferred from substrings in the key and is # only a fallback for records that did not say what they wanted. record = self._cache[key] if isinstance(record, dict): stored_ttl = record.get('ttl') if isinstance(stored_ttl, (int, float)) and not isinstance(stored_ttl, bool) \ and stored_ttl >= 0: max_age = stored_ttl # Check expiration if max_age is not None and (now - timestamp) > max_age: # Expired - remove it self._cache.pop(key, None) self._timestamps.pop(key, None) return None return self._cache[key] def set(self, key: str, value: Dict[str, Any]) -> None: """ Set value in memory cache. Args: key: Cache key value: Value to cache """ with self._lock: self._cache[key] = value self._timestamps[key] = time.time() # Enforce the ceiling here rather than leaving it to the periodic # cleanup, which only runs every cleanup_interval seconds (300 by # default). A burst of inserts between two sweeps could otherwise # take the cache far past _max_size, which is the memory growth this # limit exists to prevent -- and on a 1GB board that is the # difference between a bounded cache and an unreachable Pi. self._evict_over_limit_locked() def _evict_over_limit_locked(self) -> int: """Drop oldest entries until the cache is within _max_size. Caller must hold self._lock. Returns the number of entries removed. """ excess = len(self._cache) - self._max_size if excess <= 0: return 0 oldest = sorted( self._timestamps.items(), key=lambda item: float(item[1]) if isinstance(item[1], (int, float)) else 0.0 ) removed = 0 for key, _ in oldest[:excess]: self._cache.pop(key, None) self._timestamps.pop(key, None) removed += 1 return removed def clear(self, key: Optional[str] = None) -> None: """ Clear cache entry or all entries. Args: key: Specific key to clear, or None to clear all """ with self._lock: if key: self._cache.pop(key, None) self._timestamps.pop(key, None) else: self._cache.clear() self._timestamps.clear() def cleanup(self, force: bool = False) -> int: """ Clean up expired entries and enforce size limits. Args: force: If True, perform cleanup regardless of time interval Returns: Number of entries removed """ now = time.time() # Check if cleanup is needed if not force and (now - self._last_cleanup) < self._cleanup_interval: return 0 with self._lock: removed_count = 0 current_time = time.time() # Remove expired entries (entries older than 1 hour without access are considered expired) max_age_for_cleanup = 3600 # 1 hour expired_keys = [] for key, timestamp in list(self._timestamps.items()): if isinstance(timestamp, str): try: timestamp = float(timestamp) except ValueError: timestamp = None if timestamp is None or (current_time - timestamp) > max_age_for_cleanup: expired_keys.append(key) # Remove expired entries for key in expired_keys: self._cache.pop(key, None) self._timestamps.pop(key, None) removed_count += 1 # Same ceiling enforcement set() uses, so the two cannot drift. removed_count += self._evict_over_limit_locked() self._last_cleanup = current_time if removed_count > 0: self.logger.debug("Memory cache cleanup: removed %d entries (current size: %d)", removed_count, len(self._cache)) return removed_count def size(self) -> int: """Get current cache size.""" with self._lock: return len(self._cache) def max_size(self) -> int: """Get maximum cache size.""" return self._max_size def get_stats(self) -> Dict[str, Any]: """ Get cache statistics. Returns: Dictionary with cache statistics """ with self._lock: return { 'size': len(self._cache), 'max_size': self._max_size, 'usage_percent': (len(self._cache) / self._max_size * 100) if self._max_size > 0 else 0, 'last_cleanup': self._last_cleanup, 'cleanup_interval': self._cleanup_interval }