Source code for simvx.core.assets.cache

"""Byte-budget LRU cache used by typed loaders.

Keys are ``(uri, version)`` so the same URI with a new version (etag,
mtime) becomes a fresh entry; consumers don't need explicit invalidation
to pick up updated content.
"""

from __future__ import annotations

import sys
from collections import OrderedDict
from typing import Any




class LRUCache:
    """Simple byte-budget LRU.

    Estimates per-entry size via ``sys.getsizeof`` for bytes-like values
    and falls back to a constant for arbitrary objects. Loaders that
    cache structured assets (textures, sounds) should pass ``size_hint``
    so eviction tracks GPU/audio memory rather than the Python wrapper.
    """

    def __init__(self, max_bytes: int) -> None:
        if max_bytes <= 0:
            raise ValueError(f"max_bytes must be > 0, got {max_bytes}")
        self.max_bytes = max_bytes
        self._items: OrderedDict[tuple[str, str | None], tuple[Any, int]] = OrderedDict()
        self._used = 0



    def __len__(self) -> int:
        return len(self._items)




    @property
    def used_bytes(self) -> int:
        return self._used




    def get(self, uri: str, version: str | None) -> Any | None:
        key = (uri, version)
        if key not in self._items:
            return None
        value, _ = self._items[key]
        self._items.move_to_end(key)
        return value




    def put(self, uri: str, version: str | None, value: Any, size_hint: int | None = None) -> None:
        key = (uri, version)
        if key in self._items:
            old_value, old_size = self._items.pop(key)
            self._used -= old_size
        size = size_hint if size_hint is not None else _estimate_size(value)
        if size > self.max_bytes:
            return  # value too big to cache; skip silently rather than evict everything
        self._items[key] = (value, size)
        self._used += size
        while self._used > self.max_bytes and self._items:
            _, (_, evicted_size) = self._items.popitem(last=False)
            self._used -= evicted_size




    def invalidate(self, uri: str, version: str | None = None) -> None:
        """Drop one entry. If ``version`` is None, drop every version of ``uri``."""
        if version is not None:
            self._items.pop((uri, version), None)
            return
        for key in [k for k in self._items if k[0] == uri]:
            _, size = self._items.pop(key)
            self._used -= size




    def clear(self) -> None:
        self._items.clear()
        self._used = 0




def _estimate_size(value: Any) -> int:
    if isinstance(value, (bytes, bytearray, memoryview)):
        return len(value)
    if hasattr(value, "nbytes"):  # numpy / similar
        try:
            return int(value.nbytes)
        except (AttributeError, TypeError):
            pass
    return sys.getsizeof(value)