src/modules/cache_manager.py

"""
Cache Manager for RVC Models
================================
Thread-safe LRU cache for AI models (RVC, MDX, HuBERT).

Stores loaded model objects keyed by their file path. When the cache
is full, the least recently used entry is evicted.
"""

import threading
import time
from collections import OrderedDict
from typing import Any, Optional


class ModelCache:
    """Thread-safe LRU cache for AI models."""

    def __init__(self, max_entries: int = 4, max_age_seconds: int = 1800):
        self._cache: OrderedDict[str, dict] = OrderedDict()
        self._max_entries = max_entries
        self._max_age = max_age_seconds
        self._lock = threading.RLock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> Optional[Any]:
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                self._misses += 1
                return None
            if time.time() - entry["loaded_at"] > self._max_age:
                self._evict_entry(key)
                self._misses += 1
                return None
            self._cache.move_to_end(key)
            self._hits += 1
            return entry["model"]

    def put(self, key: str, model: Any, cleanup_fn: Optional[callable] = None) -> None:
        with self._lock:
            if key in self._cache:
                self._cache.move_to_end(key)
            self._cache[key] = {
                "model": model,
                "loaded_at": time.time(),
                "cleanup_fn": cleanup_fn,
            }
            while len(self._cache) > self._max_entries:
                oldest_key, oldest_entry = self._cache.popitem(last=False)
                self._run_cleanup(oldest_key, oldest_entry)

    def clear(self) -> None:
        with self._lock:
            for key in list(self._cache.keys()):
                self._evict_entry(key)

    def stats(self) -> dict:
        with self._lock:
            total = self._hits + self._misses
            hit_rate = (self._hits / total * 100) if total > 0 else 0.0
            return {
                "entries": len(self._cache),
                "max_entries": self._max_entries,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate": f"{hit_rate:.1f}%",
                "keys": list(self._cache.keys()),
            }

    def _evict_entry(self, key: str) -> None:
        entry = self._cache.pop(key, None)
        if entry is not None:
            self._run_cleanup(key, entry)

    def _run_cleanup(self, key: str, entry: dict) -> None:
        cleanup_fn = entry.get("cleanup_fn")
        if cleanup_fn is not None:
            try:
                cleanup_fn(entry["model"])
            except Exception as e:
                print(f"[cache] cleanup error for {key}: {e}")