Spaces:
Running
Running
| """Thread-safe bounded LRU cache for TIDAL inference results.""" | |
| from __future__ import annotations | |
| import threading | |
| from collections import OrderedDict | |
| from typing import Any | |
| class InferenceCache: | |
| """Bounded LRU cache keyed by SHA-256 of raw image bytes. | |
| Thread-safe via a single lock. Oldest entries are evicted when | |
| ``max_size`` is reached. | |
| """ | |
| def __init__(self, max_size: int = 128) -> None: | |
| self._max_size = max(1, max_size) | |
| self._store: OrderedDict[str, Any] = OrderedDict() | |
| self._hits = 0 | |
| self._misses = 0 | |
| self._lock = threading.Lock() | |
| # ── Public API ───────────────────────────────────────────────────────── | |
| def get(self, key: str) -> Any | None: | |
| """Return cached value or ``None`` if not present.""" | |
| with self._lock: | |
| if key not in self._store: | |
| self._misses += 1 | |
| return None | |
| # Move to end (most recently used) | |
| self._store.move_to_end(key) | |
| self._hits += 1 | |
| return self._store[key] | |
| def set(self, key: str, value: Any) -> None: | |
| """Store a value; evict the LRU entry when at capacity.""" | |
| with self._lock: | |
| if key in self._store: | |
| self._store.move_to_end(key) | |
| self._store[key] = value | |
| return | |
| self._store[key] = value | |
| if len(self._store) > self._max_size: | |
| self._store.popitem(last=False) # evict oldest | |
| def stats(self) -> dict: | |
| with self._lock: | |
| total = self._hits + self._misses | |
| return { | |
| "size": len(self._store), | |
| "max_size": self._max_size, | |
| "hits": self._hits, | |
| "misses": self._misses, | |
| "hit_rate": round(self._hits / total, 4) if total else 0.0, | |
| } | |
| def clear(self) -> None: | |
| with self._lock: | |
| self._store.clear() | |
| self._hits = 0 | |
| self._misses = 0 | |
| def __contains__(self, key: str) -> bool: | |
| with self._lock: | |
| return key in self._store | |