Integration boundaries¶
This package supplies indexing plans. It does not supply scheduling,
caching, codecs, or async orchestration. A consumer decides when projections
run, how decoded chunks are obtained, and where completed values are retained.
An IndexTransform says which source values belong in a result; a Reader
lowers that complete transform for one backend. The reader does not choose
indexing semantics or result ownership.
Zarr chunk dispatch¶
A Zarr-oriented reader can consume each public ChunkProjection and use its
chunk_coords to obtain one decoded chunk from its own storage and codec
layers. This tiny source keeps four in-memory chunks keyed by their global
chunk coordinates and records the exact reads. For each projection, the
consumer enumerates the shared synthetic input cell domain, evaluates
chunk_transform to read zero-origin chunk-local coordinates, and evaluates
cell_transform to place each value at its literal request coordinate.
class RecordingChunkSource:
"""A decoded-chunk source keyed by public chunk coordinates."""
def __init__(self, chunks: dict[tuple[int, ...], np.ndarray[Any, Any]]) -> None:
self.chunks = chunks
self.reads: list[tuple[int, ...]] = []
def read(self, chunk_coords: tuple[int, ...]) -> np.ndarray[Any, Any]:
self.reads.append(chunk_coords)
return self.chunks[chunk_coords]
def _domain_points(domain: IndexDomain) -> np.ndarray[Any, np.dtype[np.intp]]:
"""Enumerate a rectangular domain with a trailing coordinate axis."""
if domain.ndim == 0:
return np.empty((1, 0), dtype=np.intp)
points = np.moveaxis(np.indices(domain.shape, dtype=np.intp), 0, -1).reshape(
-1, domain.ndim
)
points += np.asarray(domain.inclusive_min, dtype=np.intp)
return points
def _gather_and_scatter(
destination: np.ndarray[Any, Any],
source: np.ndarray[Any, Any],
source_points: np.ndarray[Any, np.dtype[np.intp]],
destination_points: np.ndarray[Any, np.dtype[np.intp]],
) -> np.ndarray[Any, Any]:
"""Gather and scatter a flattened point batch, including rank zero."""
values = np.asarray(source[tuple(source_points.T)]).reshape(-1)
if destination_points.shape[-1] == 0:
destination[()] = values.reshape(destination.shape)[()]
else:
destination[tuple(destination_points.T)] = values
return values
zarr_image = np.arange(12).reshape(3, 4)
zarr_chunks = {
(chunk_row, chunk_column): zarr_image[
chunk_row * 2 : (chunk_row + 1) * 2,
chunk_column * 2 : (chunk_column + 1) * 2,
]
for chunk_row in range(2)
for chunk_column in range(2)
}
zarr_source = RecordingChunkSource(zarr_chunks)
zarr_view = LazyArray.from_numpy(zarr_image).with_parts((2, 2)).lazy[1, 0:4]
ZARR_RESULT = np.empty(zarr_view.shape, dtype=zarr_image.dtype)
shared_domains: list[tuple[IndexDomain, IndexDomain]] = []
chunk_local_coords: list[tuple[tuple[int, ...], ...]] = []
request_coords: list[tuple[tuple[int, ...], ...]] = []
read_values: list[tuple[int, ...]] = []
for part in zarr_view.parts():
projection = part.projection
assert projection.chunk_transform.domain == projection.cell_transform.domain
shared_domains.append(
(projection.chunk_transform.domain, projection.cell_transform.domain)
)
domain = projection.chunk_transform.domain
cell_points = _domain_points(domain)
local_points_array = projection.chunk_transform.apply_many(cell_points)
result_points_array = projection.cell_transform.apply_many(cell_points)
chunk = zarr_source.read(projection.chunk_coords)
values_array = _gather_and_scatter(
ZARR_RESULT, chunk, local_points_array, result_points_array
)
local_points = tuple(tuple(point) for point in local_points_array.tolist())
result_points = tuple(tuple(point) for point in result_points_array.tolist())
values = tuple(int(value) for value in values_array)
chunk_local_coords.append(local_points)
request_coords.append(result_points)
read_values.append(values)
ZARR_SOURCE_KEYS = tuple(zarr_source.chunks)
ZARR_SOURCE_READS = tuple(zarr_source.reads)
ZARR_DISPATCHED_CHUNKS = ZARR_SOURCE_READS
ZARR_SHARED_DOMAINS = tuple(shared_domains)
ZARR_CHUNK_LOCAL_COORDS = tuple(chunk_local_coords)
ZARR_REQUEST_COORDS = tuple(request_coords)
ZARR_READ_VALUES = tuple(read_values)
assert ZARR_SOURCE_KEYS == ((0, 0), (0, 1), (1, 0), (1, 1))
assert ZARR_SOURCE_READS == ((0, 0), (0, 1))
assert ZARR_CHUNK_LOCAL_COORDS == (((1, 0), (1, 1)), ((1, 0), (1, 1)))
assert ZARR_REQUEST_COORDS == (((0,), (1,)), ((2,), (3,)))
assert ZARR_READ_VALUES == ((4, 5), (6, 7))
assert ZARR_RESULT.tolist() == [4, 5, 6, 7]
The two reads are exactly (0, 0) and (0, 1); untouched chunks (1, 0) and
(1, 1) are never read. The assembled request is [4, 5, 6, 7]. The
example intentionally begins with already decoded in-memory chunks: storage
keys, codecs, scheduling, caching, and asynchronous orchestration remain the
consumer's policy rather than responsibilities of the plan.
Reading the tables directly¶
A consumer that fetches many chunks per request — a codec pipeline, a
prefetcher — does not need a ChunkProjection object per chunk. The plan's
factored form is a few
NumPy arrays per axis, and everything a chunk copy needs is a row of each:
the chunk index, the chunk-local start and extent, and where the cells land
in the request. chunk_coords() alone answers "which chunks?" for a prefetch,
without materializing anything.
The example assembles a strided box from its StridedSet tables, one slice
per chunk. Three things a real consumer also has to get right are checked at
the end: a descending stride needs the same care with a negative stop that any
NumPy slice does, a chunk slab comes out in storage-axis order while the
result is in request-axis order, and a request axis no map reads (None in
the selection) is a length-one axis of the result.
def read_box_through_tables(
source: np.ndarray[Any, Any], chunks: tuple[int, ...], transform: IndexTransform
) -> tuple[np.ndarray[Any, Any], list[tuple[int, ...]]]:
"""Assemble a box selection from the per-axis tables, one slice per chunk.
No projection is materialized: every column read here is a NumPy array of
the table, and the only per-chunk Python work is building the two selector
tuples the copy needs.
"""
partition = plan_chunks(
transform, dimension_grids_from_chunks(chunks, shape=source.shape)
).partition()
domain = transform.domain
out = np.empty(domain.shape, dtype=source.dtype)
tables = [table for table in partition.sets if isinstance(table, StridedSet)]
assert len(tables) == len(partition.sets) # a box: strided tables only
# A chunk slab comes out in storage-axis order; the result is in request-axis
# order, and request axes no map reads (`None` in the selection) are length 1.
read_axes = [table.input_dimension for table in tables if table.input_dimension is not None]
to_request_order = tuple(np.argsort(read_axes))
unread_axes = [axis for axis in range(domain.ndim) if axis not in read_axes]
reads: list[tuple[int, ...]] = []
for table_rows in np.ndindex(*partition.row_shape):
chunk_key: list[int] = []
chunk_sel: list[int | slice] = []
out_sel: list[slice] = [slice(None)] * domain.ndim
for table, row in zip(tables, table_rows, strict=True):
chunk_key.append(int(table.chunk[row]))
start = int(table.local_start[row])
count = int(table.extent[row])
if table.input_dimension is None:
chunk_sel.append(start)
continue
stop: int | None = start + table.stride * count
if table.stride < 0 and stop < 0:
stop = None # a negative stop would count from the end
chunk_sel.append(slice(start, stop, table.stride))
first = int(table.origin[row])
out_sel[table.input_dimension] = slice(first, first + count)
chunk = source[tuple(slice(k * c, (k + 1) * c) for k, c in zip(chunk_key, chunks, strict=True))]
values = np.transpose(chunk[tuple(chunk_sel)], to_request_order)
for axis in unread_axes:
values = np.expand_dims(values, axis)
out[tuple(out_sel)] = values
reads.append(tuple(chunk_key))
return out, reads
image = np.arange(63).reshape(7, 9)
box = IndexTransform.from_shape((7, 9))[1:6:2, 5:]
values, reads = read_box_through_tables(image, (3, 4), box)
np.testing.assert_array_equal(values, image[1:6:2, 5:])
assert reads == [(0, 1), (0, 2), (1, 1), (1, 2)]
# A reversed axis, an inserted axis, and a transposed transform read the same way.
reversed_box = IndexTransform.from_shape((7, 9))[6:0:-1, None, ::3]
np.testing.assert_array_equal(read_box_through_tables(image, (3, 4), reversed_box)[0], image[6:0:-1, None, ::3])
transposed = IndexTransform(
domain=IndexDomain.from_shape((9, 7)),
output=(DimensionMap(input_dimension=1), DimensionMap(input_dimension=0)),
)
np.testing.assert_array_equal(read_box_through_tables(image, (3, 4), transposed)[0], image.T)
The four reads are the four chunks the two axis tables multiply out to, and no
intermediate transform, domain, or projection was built. A gather reads its
IndexedSet rows the same way — index[pointer[i]:pointer[i + 1]] and the
matching positions — and a vindex selection its JointSet rows, whose
local coordinates already have the chunk origin subtracted.
One slab read or many part reads¶
A backend with its own native subset read — a Rust or C zarr implementation,
a database, an HTTP range endpoint — resolves a dense box (is_box with
every stride 1) best as a single read: hand it the whole selection and let it
dispatch to chunks, decode in parallel, and partial-decode shards on its own
side of the boundary. Splitting that read along this library's partitioning
only adds round-trips. Every other selection — a strided box, an oindex
or vindex gather — is where the partitioning earns its keep. The cover
of a read is the smallest step-1 slab enclosing every coordinate it needs;
partitioned, each part's cover is bounded by that part's box, so a sparse
selection can never force one read of its whole bounding hull (the smallest
rectangle containing every selected coordinate — a thousand rows for the two
of oindex[[0, 999]]).
The composed view carries enough to make that call at materialization time, and re-partitioning is a pure setter, so the policy is three lines:
def materialize(view: LazyArray) -> Any:
"""Read a dense box as one slab; resolve everything else per part."""
strides = view.strides()
if view.is_box and strides is not None and all(s == 1 for s in strides):
view = view.with_parts(view.base_shape)
return view.result()
slab_source = RecordingArray(np.arange(100).reshape(10, 10), chunks=(4, 4))
slab = LazyArray(slab_source)
dense = slab.lazy[2:9, 1:8] # a dense box: every stride 1
assert materialize(dense).shape == (7, 7)
assert len(slab_source.keys) == 1 # one slab read; the source dispatches
slab_source.keys.clear()
gather = slab.lazy.oindex[[0, 9], [0, 9]] # a query: keep the chunk parts
assert materialize(gather).tolist() == [[0, 9], [90, 99]]
assert len(slab_source.keys) == 4 # four covers, each inside one chunk
assert all(
(key[0].stop - key[0].start) * (key[1].stop - key[1].start) == 1
for key in slab_source.keys
)
The corner gather reads four single cells instead of the 10-by-10 hull, and
the dense box becomes exactly one backend call. Both regimes go through
result(); only the partitioning in force differs.
Sources that accept only unit-step slices¶
The default basic_reader pushes strided and descending selections down as
positive-step slices, which reads the minimum but assumes the source accepts
any step. Many backends do not: FFI bindings and range requests often
support nothing but slice(start, stop, 1). Select
unit_step_reader for such a source
and every key it receives is an ascending unit-step slice per axis, with
strides, reversals, and gathers applied to the in-memory block instead:
A strided selection then over-reads its cover by the stride factor, which the partitioning above bounds by one part.
napari-like consumer¶
This is a napari-like consumer, not a napari integration. It models the
boundary a viewport could use without importing or claiming support for
napari. RecordingArray exposes a chunked, basic-indexing source — and its
chunks attribute is why the reads below split along (2, 2) boxes:
LazyArray discovers a partitioning from the wrapped array at construction
(read_chunk_sizes, then chunks), with with_parts as the explicit
override. Composing the
visible slice records no reads. Only result() materializes it, with the exact
source selectors 1:2, 0:2 and 1:2, 2:4; neither selector crosses into an
untouched neighboring chunk.
class RecordingArray:
"""An array-like source that records the basic reads it receives."""
def __init__(self, data: np.ndarray[Any, Any], chunks: tuple[int, ...]) -> None:
self._data = data
self.chunks = chunks
self.keys: list[tuple[slice, ...]] = []
@property
def shape(self) -> tuple[int, ...]:
return self._data.shape
@property
def dtype(self) -> np.dtype[Any]:
return self._data.dtype
def __getitem__(self, key: tuple[slice, ...]) -> np.ndarray[Any, Any]:
self.keys.append(key)
return self._data[key]
viewport_source = RecordingArray(np.arange(12).reshape(3, 4), chunks=(2, 2))
viewport = LazyArray(viewport_source).lazy[1, 0:4]
VIEWPORT_READS_BEFORE_RESULT = tuple(viewport_source.keys)
assert VIEWPORT_READS_BEFORE_RESULT == ()
assert viewport.result().tolist() == [4, 5, 6, 7]
VIEWPORT_SOURCE_KEYS = tuple(viewport_source.keys)
VIEWPORT_SOURCE_CHUNKS = tuple(
(key[0].start // 2, key[1].start // 2) for key in VIEWPORT_SOURCE_KEYS
)
assert VIEWPORT_SOURCE_KEYS == (
(slice(1, 2, 1), slice(0, 2, 1)),
(slice(1, 2, 1), slice(2, 4, 1)),
)
assert VIEWPORT_SOURCE_CHUNKS == ((0, 0), (0, 1))
The viewport owns its interaction loop and any cancellation, caching, or
background execution. LazyArray contributes the composable selection and
the partition plan, then resolves only when the consumer asks for the result.
A system-memory chunk cache¶
Napari accepts NumPy-like array objects and can defer materialization until an image region is displayed. The indexing plan still deliberately owns no cache or scheduler. A viewport adapter can place that policy around the plan, as the executable reference below demonstrates.
This remains a napari-like consumer, not a napari integration. It models only decoded chunks resident in system memory, synchronously.
For setup instructions and the complete executable, see the system-memory chunk cache example.
read succeeds
NEW -> QUEUED -> LOADING -------------> READY -> EVICTED
^ |
| | read fails
| v
+--------- FAILED
retry
EVICTED -> QUEUED
reload
The example keeps the lifecycle records and transitions explicit:
class ChunkState(StrEnum):
NEW = "new"
QUEUED = "queued"
LOADING = "loading"
READY = "ready"
FAILED = "failed"
EVICTED = "evicted"
@dataclass(slots=True)
class ChunkRecord:
state: ChunkState = ChunkState.NEW
buffer: np.ndarray[Any, Any] | None = None
error: Exception | None = None
last_access: int = -1
@dataclass(frozen=True, slots=True)
class ChunkEvent:
chunk_coords: ChunkCoords
previous: ChunkState
current: ChunkState
reason: str
class ChunkLoadError(RuntimeError):
pass
Its source represents already decoded chunks and records each read:
class RecordingChunkSource:
def __init__(self, data: np.ndarray[Any, Any], chunks: tuple[int, ...]) -> None:
self._data = data
self.chunks = chunks
self.reads: list[ChunkCoords] = []
self.failures: set[ChunkCoords] = set()
@property
def shape(self) -> tuple[int, ...]:
return self._data.shape
@property
def dtype(self) -> np.dtype[Any]:
return self._data.dtype
def __getitem__(self, key: Any) -> np.ndarray[Any, Any]:
raise AssertionError("the cache must read complete chunks through read_chunk")
def read_chunk(self, chunk_coords: ChunkCoords) -> np.ndarray[Any, Any]:
self.reads.append(chunk_coords)
if chunk_coords in self.failures:
raise OSError(f"source read failed for chunk {chunk_coords}")
key = tuple(
slice(coord * size, min((coord + 1) * size, extent))
for coord, size, extent in zip(chunk_coords, self.chunks, self.shape, strict=True)
)
return self._data[key].copy()
def _domain_points(domain: IndexDomain) -> np.ndarray[Any, np.dtype[np.intp]]:
"""Enumerate a rectangular domain with a trailing coordinate axis."""
if domain.ndim == 0:
return np.empty((1, 0), dtype=np.intp)
points = np.moveaxis(np.indices(domain.shape, dtype=np.intp), 0, -1).reshape(-1, domain.ndim)
points += np.asarray(domain.inclusive_min, dtype=np.intp)
return points
def _gather_and_scatter(
destination: np.ndarray[Any, Any],
source: np.ndarray[Any, Any],
source_points: np.ndarray[Any, np.dtype[np.intp]],
destination_points: np.ndarray[Any, np.dtype[np.intp]],
) -> np.ndarray[Any, Any]:
"""Gather and scatter a flattened point batch, including rank zero."""
values = np.asarray(source[tuple(source_points.T)]).reshape(-1)
if destination_points.shape[-1] == 0:
destination[()] = values.reshape(destination.shape)[()]
else:
destination[tuple(destination_points.T)] = values
return values
LazyArray converts a cache selection into transforms and partitions, then
allocates and assembles the result. The facade constructs exactly one tuple
from view.parts(): it derives the chunk coordinates to pin from that tuple,
then passes the same owned parts to view.result(parts=parts). Planning is
therefore performed once for the request rather than repeated during
materialization. Neither pinning nor the result call rebuilds the plan; both
reuse those prepared Partition objects.
SystemMemoryChunkReader receives one ReadContext for each materialized
part. Its global context.transform directly addresses the raw source, while
context.projection.chunk_transform addresses the already identified chunk
locally. The reader consumes that supplied projection directly; it never calls
the chunk planner. LazyArray retains responsibility for the projection's
result placement and final assembly. The reader owns only cache state and
source reads, while SystemMemoryChunkCache remains the thin NumPy-style facade
that prepares and pins the one plan:
Its indexing dialects remain explicit: cache[key] accepts basic indexing
(integers, slices, ellipsis, and new axes), while cache.oindex[key] combines
per-axis index arrays as an outer product. Array keys are not silently treated
as orthogonal by plain square brackets; callers choose that behavior through
the named accessor.
LEGAL_TRANSITIONS: dict[ChunkState, frozenset[ChunkState]] = {
ChunkState.NEW: frozenset({ChunkState.QUEUED}),
ChunkState.QUEUED: frozenset({ChunkState.LOADING}),
ChunkState.LOADING: frozenset({ChunkState.READY, ChunkState.FAILED}),
ChunkState.READY: frozenset({ChunkState.EVICTED}),
ChunkState.FAILED: frozenset({ChunkState.QUEUED}),
ChunkState.EVICTED: frozenset({ChunkState.QUEUED}),
}
class _OrthogonalIndexer:
"""Expose outer-product indexing without changing ``cache[key]`` semantics."""
def __init__(self, getitem: Callable[[Any], np.ndarray[Any, Any]]) -> None:
self._getitem = getitem
def __getitem__(self, key: Any) -> np.ndarray[Any, Any]:
return self._getitem(key)
class SystemMemoryChunkReader:
def __init__(self, *, capacity: int) -> None:
self.capacity = capacity
self._records: dict[ChunkCoords, ChunkRecord] = {}
self._queue: list[ChunkCoords] = []
self._clock = 0
self._requests = 0
self.events: list[ChunkEvent] = []
self.projection_uses: list[tuple[str, str]] = []
def state(self, chunk_coords: ChunkCoords) -> ChunkState:
return self._record(chunk_coords).state
def resident(self) -> tuple[ChunkCoords, ...]:
return tuple(
sorted(
coords
for coords, record in self._records.items()
if record.state is ChunkState.READY
)
)
def _record(self, chunk_coords: ChunkCoords) -> ChunkRecord:
return self._records.setdefault(chunk_coords, ChunkRecord())
def _transition(self, chunk_coords: ChunkCoords, current: ChunkState, reason: str) -> None:
record = self._record(chunk_coords)
if current not in LEGAL_TRANSITIONS[record.state]:
raise ValueError(f"illegal chunk transition {record.state} -> {current}")
previous = record.state
record.state = current
self.events.append(ChunkEvent(chunk_coords, previous, current, reason))
def retry(self, chunk_coords: ChunkCoords) -> None:
record = self._record(chunk_coords)
if record.state is not ChunkState.FAILED:
raise ValueError(f"retry requires failed chunk {chunk_coords}, got {record.state}")
record.error = None
self._transition(chunk_coords, ChunkState.QUEUED, "explicit retry")
self._queue.append(chunk_coords)
@contextmanager
def request(self, required: tuple[ChunkCoords, ...]) -> Iterator[None]:
"""Prepare every part and defer eviction until one request completes."""
self._prepare(required)
self._requests += 1
try:
yield
except Exception:
self._requests -= 1
raise
else:
self._requests -= 1
if self._requests == 0:
self._evict(pinned=frozenset())
def _touch(self, record: ChunkRecord) -> None:
self._clock += 1
record.last_access = self._clock
def _queue_once(self, chunk_coords: ChunkCoords) -> None:
record = self._record(chunk_coords)
if record.state in {ChunkState.QUEUED, ChunkState.LOADING, ChunkState.READY}:
return
if record.state is ChunkState.FAILED:
raise ValueError(f"failed chunk {chunk_coords} requires explicit retry")
self._transition(chunk_coords, ChunkState.QUEUED, "requested")
self._queue.append(chunk_coords)
def _prepare(self, required: tuple[ChunkCoords, ...]) -> None:
for chunk_coords in required:
record = self._record(chunk_coords)
if record.state is ChunkState.FAILED:
assert record.error is not None
raise ChunkLoadError(
f"chunk {chunk_coords} is failed; call retry first"
) from record.error
for chunk_coords in required:
record = self._record(chunk_coords)
if record.state is ChunkState.READY:
self._touch(record)
else:
self._queue_once(chunk_coords)
def _ensure_ready(
self,
source: RecordingChunkSource,
required: tuple[ChunkCoords, ...],
) -> None:
if self._requests == 0:
self._prepare(required)
self._drain(source, frozenset(required))
def _drain(self, source: RecordingChunkSource, required: frozenset[ChunkCoords]) -> None:
pending = self._queue
self._queue = []
for index, chunk_coords in enumerate(pending):
if chunk_coords not in required:
self._queue.append(chunk_coords)
continue
record = self._record(chunk_coords)
self._transition(chunk_coords, ChunkState.LOADING, "queue drained")
try:
record.buffer = source.read_chunk(chunk_coords)
except OSError as error:
record.buffer = None
record.error = error
self._transition(chunk_coords, ChunkState.FAILED, "source read failed")
self._queue.extend(pending[index + 1 :])
raise ChunkLoadError(f"could not load chunk {chunk_coords}") from error
record.error = None
self._transition(chunk_coords, ChunkState.READY, "source read completed")
self._touch(record)
def _evict(self, *, pinned: frozenset[ChunkCoords]) -> None:
while len(self.resident()) > self.capacity:
candidates = (
(record.last_access, chunk_coords)
for chunk_coords, record in self._records.items()
if record.state is ChunkState.READY and chunk_coords not in pinned
)
_, chunk_coords = min(candidates)
record = self._record(chunk_coords)
record.buffer = None
self._transition(chunk_coords, ChunkState.EVICTED, "LRU capacity")
def read_into(
self,
source: RecordingChunkSource,
context: ReadContext,
out: np.ndarray[Any, Any],
/,
) -> None:
projection = context.projection
if projection is None:
raise ValueError("SystemMemoryChunkReader requires context.projection")
required = (projection.chunk_coords,)
self._ensure_ready(source, required)
record = self._record(projection.chunk_coords)
assert record.buffer is not None
cell_points = _domain_points(projection.chunk_transform.domain)
chunk_points = projection.chunk_transform.apply_many(cell_points)
destination_points = _domain_points(context.transform.domain)
_gather_and_scatter(out, record.buffer, chunk_points, destination_points)
self.projection_uses.append(("chunk_transform", "context.transform"))
if self._requests == 0:
self._evict(pinned=frozenset())
class SystemMemoryChunkCache:
def __init__(self, source: RecordingChunkSource, *, capacity: int) -> None:
self.source = source
self.reader = SystemMemoryChunkReader(capacity=capacity)
self._lazy = LazyArray(source).with_reader(self.reader)
@property
def shape(self) -> tuple[int, ...]:
return self.source.shape
@property
def dtype(self) -> np.dtype[Any]:
return self.source.dtype
@property
def oindex(self) -> _OrthogonalIndexer:
return _OrthogonalIndexer(lambda key: self._read(key, orthogonal=True))
@property
def events(self) -> list[ChunkEvent]:
return self.reader.events
@property
def projection_uses(self) -> tuple[tuple[str, str], ...]:
return tuple(self.reader.projection_uses)
def state(self, chunk_coords: ChunkCoords) -> ChunkState:
return self.reader.state(chunk_coords)
def resident(self) -> tuple[ChunkCoords, ...]:
return self.reader.resident()
def retry(self, chunk_coords: ChunkCoords) -> None:
self.reader.retry(chunk_coords)
def __getitem__(self, key: Any) -> np.ndarray[Any, Any]:
return self._read(key, orthogonal=False)
def _read(self, key: Any, *, orthogonal: bool) -> np.ndarray[Any, Any]:
self.reader.projection_uses.clear()
lazy = self._lazy.lazy
view = lazy.oindex[key] if orthogonal else lazy[key]
# One prepared tuple is the request plan: pin from it, then hand the
# same owned parts back to LazyArray for assembly without replanning.
parts = tuple(view.parts())
required = tuple(dict.fromkeys(part.base_coords for part in parts))
with self.reader.request(required):
return np.asarray(view.result(parts=parts))
Follow one viewport through the cache¶
image = np.arange(48).reshape(6, 8)
source = RecordingChunkSource(image, chunks=(3, 4))
cache = SystemMemoryChunkCache(source, capacity=2)
READS_BEFORE_SELECTION = tuple(source.reads)
INITIAL_RESULT = cache[1:5, 2]
INITIAL_READS = tuple(source.reads)
before_overlap = len(source.reads)
OVERLAP_RESULT = cache[3:5, 2]
OVERLAP_NEW_READS = tuple(source.reads[before_overlap:])
before_eviction = len(source.reads)
EVICTION_RESULT = cache[0:2, 5]
EVICTION_NEW_READS = tuple(source.reads[before_eviction:])
AFTER_EVICTION_RESIDENT = cache.resident()
before_reload = len(source.reads)
RELOAD_RESULT = cache[1:5, 2]
RELOAD_NEW_READS = tuple(source.reads[before_reload:])
AFTER_RELOAD_RESIDENT = cache.resident()
source.failures.add((1, 1))
failed_once = False
try:
cache[3:5, 4:6]
except ChunkLoadError:
failed_once = True
assert failed_once
FAILED_READ_COUNT = source.reads.count((1, 1))
failed_twice = False
try:
cache[3:5, 4:6]
except ChunkLoadError:
failed_twice = True
assert failed_twice
FAILED_REPEAT_READ_COUNT = source.reads.count((1, 1))
FAILURE_READ_COUNTS = (FAILED_READ_COUNT, FAILED_REPEAT_READ_COUNT)
source.failures.remove((1, 1))
cache.retry((1, 1))
before_retry = len(source.reads)
RETRY_RESULT = cache[3:5, 4:6]
RETRY_NEW_READS = tuple(source.reads[before_retry:])
RETRY_STATE = cache.state((1, 1)).value
WORKED_EVENTS = tuple(cache.events)
FAILED_TRANSITIONS = tuple(
event.current.value for event in WORKED_EVENTS if event.chunk_coords == (1, 1)
)
FAILED_EVENT_ROWS = tuple(
(event.previous.value, event.current.value, event.reason)
for event in WORKED_EVENTS
if event.chunk_coords == (1, 1)
)
The worked example uses a 6-by-8 image, 3-by-4 chunks, and capacity for two decoded chunks. Every read delta follows directly from the viewport request:
| Step | Viewport | New reads | Resident afterward | Why |
|---|---|---|---|---|
| 1 | image[1:5, 2] |
(0, 0), (1, 0) |
(0, 0), (1, 0) |
Both projected chunks are loaded and assembled as [10, 18, 26, 34]. |
| 2 | image[3:5, 2] |
None | (0, 0), (1, 0) |
The ready buffer for (1, 0) is reused and becomes most recently used. |
| 3 | image[0:2, 5] |
(0, 1) |
(0, 1), (1, 0) |
Placement returns [5, 13], then LRU pressure evicts (0, 0). |
| 4 | image[1:5, 2] |
(0, 0) |
(0, 0), (1, 0) |
The evicted chunk is reloaded while the required ready chunk is retained. |
| 5 | image[3:5, 4:6] |
(1, 1) fails; no repeated read; (1, 1) succeeds after retry |
(0, 0), (1, 1) |
Failure is retained until explicit retry; the repaired source then returns [[28, 29], [36, 37]]. |
Chunks required by an active request are pinned through assembly, so a request may temporarily span more chunks than the steady-state capacity. Capacity is counted in decoded chunks—not records or bytes—and eviction occurs only after all requested values have been placed. Because pinning and materialization use the same prepared tuple, those lifecycle decisions cannot drift from the parts that are actually read, and the cache never has to infer or reconstruct a projection.
The event log makes the failure boundary equally explicit:
| Chunk | Transition | Reason |
|---|---|---|
(1, 1) |
NEW -> QUEUED |
requested |
(1, 1) |
QUEUED -> LOADING |
queue drained |
(1, 1) |
LOADING -> FAILED |
source read failed |
(1, 1) |
FAILED -> QUEUED |
explicit retry |
(1, 1) |
QUEUED -> LOADING |
queue drained |
(1, 1) |
LOADING -> READY |
source read completed |
A repeated request while the record is FAILED creates no event and performs
no source read. The retained failure forces the caller to choose when retry is
appropriate. A real viewport adapter could drain the queue in workers and
invalidate its canvas when chunks become ready without changing the selection
or projection semantics shown here.
Napari's image-layer documentation
describes its NumPy-like array boundary. Neuroglancer's
ChunkState
is conceptual prior art for making residency explicit. This example is a
smaller, independently authored, synchronous teaching model; it does not copy
that implementation or reproduce its full worker/GPU lifecycle.