Integration boundaries¶
For the complete path from indexing syntax to chunk coordinates, local selectors, and result positions, start with From a selection to chunk operations.
The core package supplies indexing plans and synchronous readers. It does not
provide a general scheduler, codec pipeline, or async execution engine. A
synchronous cache is included as an example. 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))[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 paired projection objects are materialized. The consumer reads table
columns, constructs selectors, and assembles each chunk contribution,
including axis permutation and expansion where needed.
"""
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 general selection its joint_sets rows, whose
local coordinates already have the chunk origin subtracted. Both tables
compute local once on first access and return the same read-only array
afterward, so accessing a row does not recalculate every point.
One slab read or many part reads¶
A backend with an efficient native subset operation may benefit from receiving one complete dense selection so it can choose its own chunk dispatch. Other backends may benefit from partitioning, including for strided or fancy selections. The tradeoff depends on the backend, chunk layout, latency, memory, and selection; a single read is not universally fastest.
A read's cover is the smallest unit-step slab enclosing its coordinates. With this package's basic readers, partitioning limits each source read to the part's selected cover. This may reduce over-reading, but a partition that spans the source can still require the entire hull. Custom readers choose their own source operations under the reader contract.
The composed view carries enough to make that call at materialization time,
and with_parts() returns a view with a new partitioning. This example uses
unit strides as a sufficient condition for its independently mapped selections:
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[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.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¶
For affine selections, basic_reader uses positive-step slices and applies
reversal or layout changes in memory. Fancy selections can require reading a
cover containing unselected values. The source must accept the emitted steps.
Use unit_step_reader for a source that
accepts only unit steps. Every key it receives is an ascending unit-step slice per axis, with
strides, reversals, and gathers applied to the in-memory block instead:
For strided selections, the ratio of cover cells to selected cells depends on stride, length, and endpoint alignment. Partitioning can reduce that cover.
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)[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 parts; evict after the outermost request succeeds, not on failure."""
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
view = lazy.oindex[key] if orthogonal else lazy[key]
# One prepared tuple is the request plan: queue its chunks and defer
# eviction during the request, 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]]. |
The example defers eviction until a successful outermost request finishes, so it can temporarily exceed capacity during assembly. Capacity counts decoded chunks, not bytes, event records, or temporary arrays. Failed requests skip that eviction step. The example assumes an unchanged source and one source/grid per reader; it does not implement invalidation or synchronization for concurrent requests. The prepared tuple supplies the projections used for each read.
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.
Dask tokenization and source mutation¶
LazyArray.__dask_tokenize__() combines the serialized transform with a source
value token. For plain NumPy arrays without object fields, it hashes all bytes
on every call, including large arrays. Time and temporary memory are linear in
source byte size. Equal supported contents and serialized transforms have equal
tokens; changes to numeric contents are visible on the next tokenization call.
Other sources must define an explicit __dask_tokenize__ hook. This includes
object arrays (also structured object fields), array subclasses, memory-mapped
arrays, and remote arrays. Known numpy.memmap and mmap.mmap backing is
rejected through ndarray base and memoryview object chains. Arbitrary buffer
provenance cannot be inferred from a plain ndarray. The wrapper does not
convert or serialize unsupported sources to discover their values. Unsupported sources raise TypeError, and
exceptions from explicit hooks propagate. Installing Dask does not change this
policy. A hook must describe the source's values or immutable version, and owns
its determinism and any I/O it performs.
For an opaque source such as a Zarr array, use
dask.array.from_array(EagerArrayAdapter(view), chunks=..., name=False) to request a fresh graph
name without content tokenization. This opts out of content-based task sharing.
Import EagerArrayAdapter from zarr_indexing: it materializes each indexed block while the wrapped view keeps lazy indexing. See Dask's from_array documentation
and tokenization contract.
Tokens describe values at tokenization time, not a snapshot. Mutating a source after building a graph does not update existing Dask keys or invalidate cached results. Concurrent mutation during hashing is unsupported. Applications must manage source lifetimes and versions; these tokens are not persistent cache identities. Readers and partitioning are omitted because they must preserve values.