"""Decode object crops lazily and cache the most recent ones for hover
previews.
A hover preview shows the crop under the cursor. A PNG decode takes a few
milliseconds and a mouse sweep produces about a hundred hover events per
second, so decoding on every event makes the plot lag behind the cursor.
Decoding therefore never happens on the paint path, and each crop is decoded
once:
* :meth:`CropThumbnails.peek` answers from memory or not at all. It is what a
hover handler calls, because the answer has to be instant or absent.
* :meth:`CropThumbnails.pixmap` decodes on a miss. A caller runs it behind a
short debounce (or on a click), never once per mouse-move event.
* The cache is keyed on ``(path, mtime, size, px)``, following
:mod:`spacr.crops`, so a crop re-written by a re-run is re-read rather than
served stale from the last session's decode.
Why not ``QPixmap(path)``
-------------------------
Crop PNGs are not ordinary images. Anything spaCR wrote before the BGR fix has
the first stain in the *blue* channel, and a 16-bit single-channel crop opened
with ``convert('RGB')`` is clipped to solid white by PIL. Both are corrected by
:func:`spacr.qt.annotate_engine.load_crop_image`, which is therefore the only
loader used here, so a hover preview shows the same colours as the annotation
grid.
"""
from __future__ import annotations
import logging
import os
import sys
import time
import weakref
from collections import OrderedDict
from typing import Any, Dict, Optional, Tuple
from PySide6.QtGui import QPixmap
LOG = logging.getLogger(__name__)
__all__ = [
"CropThumbnails",
"resolve_crop_path",
"DEFAULT_CAPACITY",
"DEFAULT_SIZE",
]
#: How many decoded thumbnails to keep. Sized for a hover sweep across a
#: cluster rather than for the whole plot: at 192 px RGBA that is ~14 MB, and
#: the working set of "the points I am moving the mouse over" is small even
#: when the plot is not.
DEFAULT_CAPACITY = 96
#: The longest edge of a cached thumbnail, in pixels. Big enough to see a
#: parasite in, small enough that a hundred of them do not cost real memory.
DEFAULT_SIZE = 192
_LIVE_CACHES: "weakref.WeakSet[CropThumbnails]" = weakref.WeakSet()
[docs]
def live_thumbnail_caches():
"""A snapshot of the decoded-thumbnail caches that still have owners."""
return tuple(_LIVE_CACHES)
[docs]
def resolve_crop_path(path: str, db_path: str = "") -> str:
"""The crop's path on *this* machine, re-anchored if the dataset moved.
``png_list`` stores absolute paths built at measure time, so a dataset
copied to another disk resolves to nothing and every preview is blank.
:func:`spacr.qt.screens.annotate._reanchor_png_path` already knows how to
rebuild one from the ``/data/`` segment under the database's own root;
this borrows it rather than growing a third copy of that rule, and falls
back to the stored path when the Annotate screen is not importable (a
headless test, a trimmed install).
:param path: crop path as stored in ``png_list``; returned unchanged when
it is empty or already exists as a file.
"""
text = str(path or "")
if not text or os.path.isfile(text) or not db_path:
return text
try:
from .screens.annotate import _reanchor_png_path
except Exception:
LOG.debug("cannot re-anchor crop paths without the Annotate screen",
exc_info=True)
return text
try:
return _reanchor_png_path(text, str(db_path))
except Exception:
LOG.debug("could not re-anchor %s", text, exc_info=True)
return text
[docs]
class CropThumbnails:
"""A bounded LRU of decoded crop thumbnails.
:param db_path: the ``measurements.db`` the crops belong to. Used to
re-anchor moved paths and to resolve the folder's stored channel
order; optional, because a folder of crops is a legitimate source too.
:param size: longest edge of the thumbnails, in pixels.
:param capacity: how many to keep.
Not a ``QObject``, and holds no widget: a screen can own one, a test can
drive one, and neither keeps the other alive.
"""
def __init__(self, db_path: str = "", *, size: int = DEFAULT_SIZE,
capacity: int = DEFAULT_CAPACITY):
"""Create the thumbnail cache.
:param db_path: measurements database the crops are read from.
:param size: longest edge of a decoded thumbnail, in pixels.
:param capacity: how many thumbnails to keep; the least recently used
are dropped first.
"""
self.db_path = str(db_path or "")
self.size = max(16, int(size))
self.capacity = max(1, int(capacity))
self._cache: "OrderedDict[Tuple[Any, ...], Optional[QPixmap]]" = \
OrderedDict()
self._last_used: Dict[Tuple[Any, ...], float] = {}
self._bytes: Dict[Tuple[Any, ...], int] = {}
#: Counters, for a status line and for the tests that assert a hover
#: sweep decoded each crop once rather than once per mouse event.
self.hits = 0
self.misses = 0
self.decodes = 0
self.failures = 0
_LIVE_CACHES.add(self)
cleanup = sys.modules.get("spacr.qt.resource_cleanup")
install = getattr(cleanup, "install_budget_sweep", None)
if callable(install):
install()
def _key(self, path: str) -> Tuple[Any, ...]:
"""``(abspath, mtime_ns, size, px)`` — the identity of one decode.
Stat is included so a re-run that rewrote the crop invalidates the
entry. A crop that cannot be stat'ed still gets a key (with zeros), so
a missing file is remembered as missing rather than re-attempted on
every hover.
"""
resolved = os.path.abspath(path) if path else ""
try:
stat = os.stat(resolved)
stamp: Tuple[Any, ...] = (stat.st_mtime_ns, stat.st_size)
except OSError:
stamp = (0, 0)
return (resolved, stamp[0], stamp[1], self.size)
[docs]
def peek(self, path: str) -> Optional[QPixmap]:
"""The thumbnail if it is already decoded, else ``None``. Never blocks.
The call a mouse-move handler makes. Returning ``None`` means "not
yet", not "there is no crop" — ask :meth:`pixmap` for that, off the
hover path.
:param path: path of the crop image; an empty value returns ``None``.
"""
if not path:
return None
key = self._key(str(path))
if key not in self._cache:
return None
self._cache.move_to_end(key)
self._last_used[key] = time.time()
self.hits += 1
return self._cache[key]
[docs]
def pixmap(self, path: str) -> Optional[QPixmap]:
"""The thumbnail, decoding it if this is the first time.
:param path: path of the crop image to decode; an empty value returns
``None``.
:returns: the ``QPixmap``, or ``None`` when the crop cannot be read.
A failure is *cached* as ``None`` rather than raised: a missing
file under the cursor must not throw out of a mouse handler, and
it must not be retried sixty times a second either.
"""
if not path:
return None
key = self._key(str(path))
if key in self._cache:
self._cache.move_to_end(key)
self._last_used[key] = time.time()
self.hits += 1
return self._cache[key]
self.misses += 1
pixmap = self._decode(str(path))
self._store(key, pixmap)
return pixmap
def _decode(self, path: str) -> Optional[QPixmap]:
"""Read one crop through the corrected reader, or ``None``."""
self.decodes += 1
resolved = resolve_crop_path(path, self.db_path)
try:
from PIL.ImageQt import ImageQt
from .annotate_engine import load_crop_image
image = load_crop_image(resolved, db_path=self.db_path or None)
image.thumbnail((self.size, self.size))
return QPixmap.fromImage(ImageQt(image).copy())
except Exception:
self.failures += 1
LOG.debug("could not decode the crop %s", resolved, exc_info=True)
return None
def _store(self, key: Tuple[Any, ...],
pixmap: Optional[QPixmap]) -> None:
"""Put one decoded thumbnail in the cache and evict down to capacity.
A ``None`` is cached too: a crop that could not be decoded must not be
retried on every hover.
:param key: the cache key.
:param pixmap: the decoded thumbnail, or ``None`` for a failed decode.
"""
self._cache[key] = pixmap
self._cache.move_to_end(key)
self._last_used[key] = time.time()
self._bytes[key] = self._pixmap_bytes(pixmap)
while len(self._cache) > self.capacity:
old, _ = self._cache.popitem(last=False)
self._last_used.pop(old, None)
self._bytes.pop(old, None)
@staticmethod
def _pixmap_bytes(pixmap: Optional[QPixmap]) -> int:
"""Storage represented by a pixmap, without copying its pixels."""
if pixmap is None or pixmap.isNull():
return 0
try:
return max(0, int(pixmap.width()) * int(pixmap.height())
* int(pixmap.depth()) // 8)
except (AttributeError, TypeError, ValueError):
return 0
[docs]
def cache_budget_entries(self):
"""``(key, bytes, last use, in use)`` rows for the global sweep.
A QLabel/QGraphicsItem that is displaying a pixmap owns its own
implicitly-shared Qt value. Removing this lookup entry cannot blank
that control, so no thumbnail entry needs pinning here.
"""
now = time.time()
return [
(key, int(self._bytes.get(key, self._pixmap_bytes(pixmap))),
float(self._last_used.get(key, now)), False)
for key, pixmap in list(self._cache.items())
]
[docs]
def drop_cache_budget_entry(self, key) -> bool:
"""Evict one decoded thumbnail selected by the memory policy.
:param key: cache key of the entry to evict, an
``(abspath, mtime_ns, size, px)`` tuple; an unknown key is ignored
and ``False`` is returned.
"""
existed = key in self._cache
self._cache.pop(key, None)
self._last_used.pop(key, None)
self._bytes.pop(key, None)
return existed
[docs]
def prime(self, path: str) -> Optional[QPixmap]:
"""Decode ``path`` now so a later :meth:`peek` is instant.
The same as :meth:`pixmap`; named separately because a debounce timer
runs it to fill the cache, not a hover handler.
:param path: path of the crop image to decode and cache, as for
:meth:`pixmap`.
"""
return self.pixmap(path)
[docs]
def __len__(self) -> int:
"""Return how many entries the cache holds."""
return len(self._cache)
[docs]
def __contains__(self, path: object) -> bool:
"""Report whether a crop path is cached.
:param path: the crop path; coerced with :func:`str`.
:returns: ``True`` if it has an entry, including a cached failure.
"""
return self._key(str(path)) in self._cache
[docs]
def clear(self) -> None:
"""Drop everything. For a new source, or a screen closing."""
self._cache.clear()
self._last_used.clear()
self._bytes.clear()
[docs]
def describe(self) -> str:
"""One line of cache health, for a status bar."""
return (f"{len(self._cache)}/{self.capacity} crops cached · "
f"{self.decodes} decode(s), {self.hits} hit(s), "
f"{self.failures} unreadable")
[docs]
def crop_paths_for_keys(db_path: str, keys) -> Dict[str, str]:
"""``{object key: crop path}``, resolved ONCE for a whole plot.
:func:`spacr.active_learning.crops_for_object_keys` scans the crop table
per call — suitable for opening a subset, too slow to run once per hover
event. This is the plot-time call: resolve every plotted point up front and
let hover be a dict lookup.
That function keeps the caller's order but *drops* keys it cannot resolve,
so when everything resolves the answer is a zip and costs one scan. When
some keys miss, the returned list is a subsequence and no longer lines up,
so the range is bisected: each half is asked separately until a half is
either fully resolved (zip it) or a single key (resolve it or record the
miss). That is a handful of extra scans for a handful of missing crops,
rather than one scan per key.
Keys with no crop are absent from the result rather than mapped to ``""``,
which would read as a crop path.
"""
wanted = [str(k) for k in keys]
if not wanted or not db_path:
return {}
from ..active_learning import crops_for_object_keys
out: Dict[str, str] = {}
def resolve(batch) -> None:
"""Resolve one batch of keys, bisecting when some are missing.
Never called with an empty batch: the caller refuses an empty key list
and a bisection of two or more cannot produce an empty half.
"""
rows = crops_for_object_keys(db_path, batch)
if len(rows) == len(batch):
for key, (path, _annotation) in zip(batch, rows):
out[key] = path
return
if len(batch) == 1:
if rows:
out[batch[0]] = rows[0][0]
return
middle = len(batch) // 2
resolve(batch[:middle])
resolve(batch[middle:])
resolve(wanted)
return out