Repository navigation
Expand file tree
/
Copy pathraster_display_cache.py
More file actions
164 lines (127 loc) · 5.11 KB
/
Copy pathraster_display_cache.py
File metadata and controls
164 lines (127 loc) · 5.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""LRU cache for raster display planes and prebuilt image pyramids."""
from __future__ import annotations
import os
import threading
from collections import OrderedDict
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
import numpy as np
from nicewidgets.raster_viewer.backend.image_model import BackendImage, RasterGridSpec
from nicewidgets.raster_viewer.backend.pyramid import ImagePyramid
# Default LRU cap for cloud deployments with limited memory. Override with
# ``CLOUDSCOPE_RASTER_DISPLAY_CACHE_MAX_ENTRIES`` (minimum 1).
DEFAULT_RASTER_DISPLAY_CACHE_MAX_ENTRIES = 20
# Grid used only when constructing ``BackendImage`` for pyramid build; pyramid
# math ignores calibration and uses raw array shape only.
_PYRAMID_BUILD_GRID = RasterGridSpec(dx=1.0, dy=1.0, x_unit='', y_unit='')
class RasterDisplayPlaneKind(str, Enum):
"""Distinguish primary slice planes from reference-image planes."""
PRIMARY = 'primary'
REFERENCE = 'reference'
@dataclass(frozen=True, slots=True)
class RasterDisplayCacheKey:
"""Cache key for one display plane within one file, channel, and slice."""
file_id: str
channel: int
z: int
t: int
kind: RasterDisplayPlaneKind
@dataclass(slots=True)
class RasterDisplayEntry:
"""Cached plane array and its prebuilt pyramid."""
plane: np.ndarray
pyramid: ImagePyramid
def resolve_raster_display_cache_max_entries() -> int:
"""Resolve the LRU capacity from the environment or factory default.
Returns:
Maximum number of cached ``(file_id, channel, z, t, kind)`` entries, at least 1.
"""
raw = os.environ.get('CLOUDSCOPE_RASTER_DISPLAY_CACHE_MAX_ENTRIES', '').strip()
if not raw:
return DEFAULT_RASTER_DISPLAY_CACHE_MAX_ENTRIES
try:
value = int(raw)
except ValueError:
return DEFAULT_RASTER_DISPLAY_CACHE_MAX_ENTRIES
return max(1, value)
class RasterDisplayCache:
"""Process-local LRU cache of raster planes and display pyramids.
Pyramid levels depend only on pixel array shape and values. Physical
calibration (``RasterGridSpec``) is applied when constructing a fresh
:class:`~nicewidgets.raster_viewer.backend.raster_service.RasterViewService`
for display; it does not require rebuilding the cached pyramid.
"""
def __init__(self, *, max_entries: int | None = None) -> None:
"""Create an empty cache.
Args:
max_entries: LRU capacity. When ``None``, uses
:func:`resolve_raster_display_cache_max_entries`.
"""
resolved = max_entries if max_entries is not None else resolve_raster_display_cache_max_entries()
self._max_entries = max(1, int(resolved))
self._entries: OrderedDict[RasterDisplayCacheKey, RasterDisplayEntry] = OrderedDict()
self._lock = threading.Lock()
@property
def max_entries(self) -> int:
"""Return the configured LRU capacity."""
return self._max_entries
def __len__(self) -> int:
"""Return the number of cached entries."""
with self._lock:
return len(self._entries)
def get_or_build(
self,
key: RasterDisplayCacheKey,
*,
plane_loader: Callable[[], np.ndarray],
) -> RasterDisplayEntry:
"""Return a cached entry or load the plane and build a pyramid on miss.
Args:
key: File, channel, ``z``/``t`` slice indices, and plane kind.
plane_loader: Callable that returns a 2D ``(Y, X)`` array. Invoked
only on cache miss.
Returns:
Cached or newly built display entry.
Raises:
ValueError: If ``plane_loader`` returns a non-2D array.
"""
with self._lock:
existing = self._entries.get(key)
if existing is not None:
self._entries.move_to_end(key)
return existing
plane = np.asarray(plane_loader())
if plane.ndim != 2:
raise ValueError(f'Expected 2D plane (Y, X), got shape={plane.shape}')
source = BackendImage(plane, grid=_PYRAMID_BUILD_GRID)
pyramid = ImagePyramid(source)
entry = RasterDisplayEntry(plane=plane, pyramid=pyramid)
with self._lock:
cached = self._entries.get(key)
if cached is not None:
self._entries.move_to_end(key)
return cached
self._entries[key] = entry
self._entries.move_to_end(key)
while len(self._entries) > self._max_entries:
self._entries.popitem(last=False)
return entry
def invalidate_file(self, file_id: str) -> None:
"""Remove all cached entries for one file id.
Args:
file_id: Acquisition file identifier.
Returns:
None.
"""
with self._lock:
keys = [key for key in self._entries if key.file_id == file_id]
for key in keys:
del self._entries[key]
def clear(self) -> None:
"""Remove all cached entries.
Returns:
None.
"""
with self._lock:
self._entries.clear()