`ncore.data.stores` Package#

Package exposing NCore’s storage APIs: indexed tar archives, the zarr hierarchies stored in them (or in other zarr-python stores), and typed access to their groups and arrays

class ncore.data.stores.Array(array: Array)#

Bases: _Node

An array of an NCore data store

property compression: BloscCompression | None#: The blosc compression of the array (None if uncompressed, raises ValueError for other compressors)

property dtype: dtype#

read( selection: Selection = Ellipsis, *, out: npt.NDArray[Any] | None = None, ) → npt.NDArray[Any]#

Reads (a selection of) the data of the array.

Returns a read-only array of the array’s dtype with the shape of the selection (0-dimensional for selections of single elements). Complete reads of arrays stored as a single uncompressed chunk (e.g., encoded images) are views of the chunk record (no copies), all other reads are decoded by the backend (see ncore.impl.data.backend.Backend.read()).

With out, a writeable C-contiguous array of the array’s dtype and the selection’s shape, the data is read into out, which is returned itself (it stays writeable, and keeps its type, e.g., numpy.memmap). Raises ValueError for other output arrays.

async read_async( selection: Selection = Ellipsis, *, out: npt.NDArray[Any] | None = None, ) → npt.NDArray[Any]#

Reads (a selection of) the data of the array asynchronously, see read().

Concurrent reads (e.g., via asyncio.gather) overlap their I/O (e.g., range requests of remote archives) and decompression, the running event loop isn’t blocked (blocking work runs on worker threads).

read_bytes() → bytes#

Reads a binary blob stored by Group.create_bytes_array().

Blobs stored as a single uncompressed chunk (all blobs written by ncore, e.g. of indexed tar archives) are their chunk record (no copies), other blobs (e.g., of multiple chunks) are copied once from the read array. Raises TypeError for arrays that don’t store a binary blob (see ncore.impl.data.backend.Backend.blob()).

async read_bytes_async() → bytes#: Reads a binary blob asynchronously, see read_bytes() and read_async()

property shape: Tuple[int, ...]#

write(selection: Selection, data: npt.ArrayLike) → None#

Writes data to a basic selection of the array (integers, slices with unit step, Ellipsis), e.g., of arrays created by Group.create_empty_array().

The chunks the selection touches are stored (also chunks equal to the fill value), partially selected chunks are completed with the array’s current data (the fill value if not written yet). Records of indexed tar archives are write-once: each chunk can be written once, so callers write whole chunks (or chunk-aligned blocks). Writes of multiple chunks that fail (e.g., on a chunk written already) can leave other chunks of the selection stored (zarr-python 3 writes them concurrently).

Raises ncore.impl.data.itar.RecordExistsError (a ValueError) for chunks written already (indexed tar archives), and ValueError for other selections and for arrays of stores opened for reading (checked before writing, with either zarr-python version).

property zarr: Array#: The wrapped zarr-python array

class ncore.data.stores.BloscCompression( cname: Literal['blosclz', 'lz4', 'lz4hc', 'zlib', 'zstd'] = 'lz4', clevel: int = 5, shuffle: Literal['noshuffle', 'shuffle', 'bitshuffle'] = 'bitshuffle', blocksize: int = 0, )#

Bases: object

Blosc compression of array chunks

blocksize: int = 0#

automatic)

Type:: Block size in bytes (0

clevel: int = 5#

maximum compression)

Type:: Compression level (0
Type:: no compression, 9

cname: Literal['blosclz', 'lz4', 'lz4hc', 'zlib', 'zstd'] = 'lz4'#: Compressor used by blosc

static from_config_v3( config: BloscConfigV3, ) → BloscCompression#: From the configuration of a zarr format v3 blosc codec

static from_numcodecs( compressor: Blosc, ) → BloscCompression#

shuffle: Literal['noshuffle', 'shuffle', 'bitshuffle'] = 'bitshuffle'#: Shuffle filter applied before compression

to_numcodecs() → Blosc#

Bases: object

An array encoded into chunk records, produced by EncodedArray.encode() (e.g., on worker threads) and stored by passing it to Group.create_encoded_array()

chunk_records: Dict[Tuple[int, ...], bytes]#: Records of the stored chunks, by chunk coordinates

chunks: Tuple[int, ...]#: Chunk shape (resolved, also for “single” and “auto”)

compression: BloscCompression | None#

uncompressed)

Type:: Blosc compression of the chunks (None

dtype: dtype#: dtype of the array

classmethod encode( data: npt.ArrayLike, *, dtype: npt.DTypeLike | None = None, chunks: Tuple[int, ...] | Literal['single', 'auto'] = 'single', compression: BloscCompression | None = BloscCompression(cname='lz4', clevel=5, shuffle='bitshuffle', blocksize=0), zarr_format: ZarrFormat = 2, ) → EncodedArray#

Encodes the chunks of an array as Group.create_array() writes them, to be stored by Group.create_encoded_array().

Uses no store and no shared state, so arrays can be encoded concurrently on worker threads (e.g., to compress the arrays of a group in parallel, and store them in order).

Parameters:

data – Data of the array (defines the shape of the array, and its dtype if dtype isn’t given).
dtype – Optional dtype of the array (the data is converted to it).
chunks – Chunk shape (zero-sized dimensions are clamped to 1, as zarr requires positive chunk sizes): “single” stores the array as a single chunk (default), “auto” uses the chunk shape heuristic of zarr-python (both are resolved in EncodedArray.chunks).
compression – Blosc compression of the chunks (default: lz4, level 5, bit-shuffle), or None for uncompressed chunks.
zarr_format – zarr format of the records (the zarr format of the group the array is stored in, raises ValueError for zarr formats unsupported by the installed zarr-python version).

fill_value: generic | bool | int | float | complex | bytes | str#: Fill value of the array’s node metadata (a numpy scalar of the array’s dtype)

shape: Tuple[int, ...]#: Shape of the array

zarr_format: Literal[2, 3]#: zarr format of the records

class ncore.data.stores.Group( group: ZarrGroup, cache: NodeCache | None = None, store: Store | None = None, )#

Bases: _Node

A group of an NCore data store.

Members of groups using a NodeCache are opened via the cache (also members of member groups).

array(path: str) → Array#: Returns a (nested) member array (raises KeyError if it doesn’t exist or isn’t an array)

property cache: NodeCache | None#: The node cache used to open members (None if not cached)

create_array( name: str, data: npt.ArrayLike, *, dtype: npt.DTypeLike | None = None, chunks: Tuple[int, ...] | Literal['single', 'auto'] = 'single', compression: BloscCompression | None = BloscCompression(cname='lz4', clevel=5, shuffle='bitshuffle', blocksize=0), attributes: Attributes | None = None, ) → Array#

Creates a member array initialized with data.

Parameters:

name – Name of the array within the group.
data – Data of the array (defines the shape of the array, and its dtype if dtype isn’t given).
dtype – Optional dtype of the array (the data is converted to it).
chunks – Chunk shape (zero-sized dimensions are clamped to 1, as zarr requires positive chunk sizes): “single” stores the array as a single chunk (default), “auto” uses the chunk shape heuristic of zarr-python.
compression – Blosc compression of the chunks (default: lz4, level 5, bit-shuffle), or None for uncompressed chunks.
attributes – Optional attributes of the array.

create_bytes_array( name: str, data: bytes, *, attributes: Attributes | None = None, ) → Array#

Stores a binary blob (e.g., an encoded image) as uncompressed member array.

zarr format v2: 0-dimensional fixed-length bytes array (|S<N>, the layout of ncore <= 19.8). zarr format v3: 1-dimensional uint8 array (fixed-length bytes are not part of the zarr format v3 specification).

create_empty_array( name: str, *, shape: Tuple[int, ...], dtype: npt.DTypeLike, chunks: Tuple[int, ...], compression: BloscCompression | None = BloscCompression(cname='lz4', clevel=5, shuffle='bitshuffle', blocksize=0), fill_value: FillValue | None = None, attributes: Attributes | None = None, ) → Array#

Creates a member array without data, whose chunks are written by Array.write() (chunks not written are not stored, and read as fill_value).

The node metadata is the one create_array() writes for the same dtype, chunks and compression (and fill value), and so are the records of written chunks.

Parameters:

name – Name of the array within the group.
shape – Shape of the array.
dtype – dtype of the array.
chunks – Chunk shape (zero-sized dimensions are clamped to 1, as zarr requires positive chunk sizes).
compression – Blosc compression of the chunks (default: lz4, level 5, bit-shuffle), or None for uncompressed chunks.
fill_value – Value of the elements of chunks not written (default: the fill value create_array() writes for the dtype, i.e. zero, False, or empty bytes / strings).
attributes – Optional attributes of the array.

Raises:

ValueError – For groups of stores opened for reading, and for chunks that don’t match the shape.

create_encoded_array( name: str, encoded: EncodedArray, *, attributes: Attributes | None = None, ) → Array#

Creates a member array from chunks encoded by EncodedArray.encode() (e.g., on worker threads).

The records are the ones create_array() writes for the same data, chunks and compression. The encoded array is validated before creating the array.

Parameters:

name – Name of the array within the group.
encoded – The encoded array (of the group’s zarr format).
attributes – Optional attributes of the array.

Raises:

ValueError – For groups of stores opened for reading, and for encoded arrays of another zarr format than the group’s, whose chunk shape doesn’t match their shape, or with records outside of their chunk grid.
ncore.impl.data.itar.RecordExistsError – For chunks written already (indexed tar archives; a ValueError).

create_group( name: str, *, attributes: Attributes | None = None, ) → Group#: Creates a member group (raises if a member with the name exists already)

get( path: str, ) → Group | Array | None#: Returns a (nested) member group or array, or None if it doesn’t exist

group(path: str) → Group#: Returns a (nested) member group (raises KeyError if it doesn’t exist or isn’t a group)

groups() → Iterator[Tuple[str, Group]]#: Iterates over the direct member groups (ordered by name)

items() → Iterator[Tuple[str, Group | Array]]#: Iterates over the direct members (opened on iteration)

keys() → List[str]#: Names of all direct members, see members()

members() → List[str]#: Names of all direct members

node( path: str, ) → Group | Array#: Returns a (nested) member group or array (raises KeyError if it doesn’t exist)

require_group(name: str) → Group#: Returns a member group, creating it if it doesn’t exist yet

property store: Store | None#: The store of the group (None for groups not opened / created via a Store)

with_cache( cache: NodeCache | None, ) → Group#: Returns the group, using the given node cache to open its members

property zarr: Group#: The wrapped zarr-python group

class ncore.data.stores.IndexedTarArchive( itar_path: str | Path | UPath, mode: Literal['r', 'w'] = 'r', index_tail_read_size: int | None = 1048576, remote_range_reads: bool = False, )#

Bases: object

The records of an indexed tar archive, opened for reading or (truncating) writing.

Parameters:

itar_path (string, Path or UPath) – Location of the archive (local or remote).
mode (string, optional, default 'r') – ‘r’ to read an existing archive, or ‘w’ to create (truncate) and write an archive.
index_tail_read_size (int or None, optional, default 1 << 20 (1 MiB)) – Maximum bytes read from the end of the archive in a single I/O call to load the record index. Covers both the small index header and (ideally) the compressed index. Values smaller than the tar block size (512 bytes), or None, result in separate reads of the index header and index, which is more robust for large indices.
remote_range_reads (bool, optional, default False) – Read records of remote archives (see is_remote) synchronously via exact byte-range requests, like read_async(), instead of via the archive’s buffered file object (blocks of the filesystem’s block size, cached by its cache type, one request at a time). Synchronous reads of multiple threads then fetch their records concurrently, which suits random access to high-latency storage (e.g., thread pools reading frames of S3 archives); the buffered file object suits sequential reading of nearby records.

Notes

Records are accessed via a mapping interface (archive[key], archive[key] = value, key in archive, keys(), len(archive)). Records of the index tail read are served from memory, so records stored near the end of the archive don’t repeat its high-latency I/O. Local archives are read via positional reads (os.pread), which don’t serialize reads of multiple threads.

Archives opened for writing need to be closed (close(), or by using the archive as context manager), which appends the node metadata records and the record index. Until then, node metadata records can be rewritten (see the module documentation), and are served from memory. Closed archives release their file resources: records not held in memory can’t be read anymore.

All methods are thread-safe.

close() → None#: Closes the archive: when writing, appends the node metadata records and the record index (which is required to read the archive), and releases the archive’s file resources (records held in memory remain readable)

property is_remote: bool#: True for archives opened for reading on asynchronous fsspec filesystems (e.g., s3fs, gcsfs, http), whose records are read via asynchronous range requests by read_async()

property itar_upath: UPath#: Absolute location of the archive

keys() → List[str]#: Returns the keys of all records (in archive order, followed by node metadata records that are written when closing the archive)

property mode: Literal['r', 'w']#: Access mode of the archive (‘r’ for reading, ‘w’ for writing)

async classmethod open_async( itar_path: str | Path | UPath, index_tail_read_size: int | None = 1048576, remote_range_reads: bool = False, prefetch_keys: Tuple[str, ...] = (), ) → IndexedTarArchive#

Opens an archive for reading asynchronously, without blocking the running event loop (see the class documentation for the parameters).

The index of remote archives (see is_remote) is loaded via asynchronous range requests (one if it lies within the tail read), so opening many archives concurrently (e.g., via asyncio.gather) overlaps their requests. Other archives (e.g., local) are opened synchronously on a worker thread.

Records of prefetch_keys (if they exist) are read asynchronously while opening and held in memory until they are read (once): e.g., the consolidated metadata record of a zarr hierarchy, which is read synchronously when opening the hierarchy, then not needed anymore.

read(key: str) → bytes#

Reads the payload of a record synchronously (raises KeyError if it doesn’t exist, ValueError if it has to be read from a closed archive).

Records held in memory (node metadata records of archives opened for writing, records of the index tail read, empty records) are returned directly, records of local archives are read via positional reads, records of remote archives via the archive’s buffered file object or via exact range requests (see remote_range_reads), all others via the archive’s file object.

async read_async(key: str) → bytes#

Reads the payload of a record asynchronously, without blocking the running event loop (raises KeyError if it doesn’t exist, ValueError if it has to be read from a closed archive).

Records held in memory are returned directly (see read()). Records of remote archives (see is_remote) are read via asynchronous range requests (concurrent reads overlap), records of all other archives (e.g., local) on a worker thread.

property records: Mapping[str, Record]#: The records of the archive by key, with the offset and size of their payload in the archive file (a read-only view: for archives opened for writing, of the records appended so far, which excludes the node metadata records appended when closing the archive)

release_index_tail( retention: Literal['release_if_consolidated', 'release_after_opening', 'keep_complete_archives', 'keep'], *, consolidated: bool, ) → None#: Releases the memory of the index tail read as per retention, after opening the zarr hierarchy of the archive (from its consolidated metadata if consolidated): records within it are read from the archive’s file after (see IndexTailRetention, archives opened for reading, no-op otherwise)

reload_resources() → None#: Re-opens the file resources of the archive’s record source (file object, file descriptor of local archives), and lets the writer of archives opened for writing append to the re-opened file object (required in processes forked after opening the archive)

property remote_range_reads: bool#: True if synchronous reads of remote archives use exact range requests (see the class documentation)

class ncore.data.stores.IndexedTarZarrStore( itar_path: str | Path | UPath, mode: Literal['r', 'w'] = 'r', index_tail_read_size: int | None = 1048576, remote_range_reads: bool = False, )#

Bases: Store

A zarr hierarchy in an indexed tar archive (local or remote, see IndexedTarArchive for the parameters).

The constructor opens the archive synchronously (blocking), see open_async() for opening stores for reading asynchronously.

property archive: IndexedTarArchive#: The records of the archive

close() → None#: Closes the store (finishes writing, no-op by default)

async classmethod open_async( itar_path: str | Path | UPath, index_tail_read_size: int | None = 1048576, remote_range_reads: bool = False, ) → IndexedTarZarrStore#: Opens a store for reading asynchronously, see IndexedTarArchive.open_async() (the consolidated metadata record, read when opening the hierarchy, is fetched asynchronously as well)

release_index_tail( retention: Literal['release_if_consolidated', 'release_after_opening', 'keep_complete_archives', 'keep'], *, consolidated: bool, ) → None#: Releases the memory of the archive’s index tail read as per retention, after opening the hierarchy of the store (from its consolidated metadata if consolidated), see IndexedTarArchive.release_index_tail()

reload_resources() → None#: Re-initializes the resources of the store (e.g., file objects in forked processes), no-op by default

class ncore.data.stores.Record(offset: int, size: int)#

Bases: NamedTuple

The payload of a record within the archive file

offset: int#: File offset of the first byte of the payload

size: int#: Size of the payload in bytes

exception ncore.data.stores.RecordExistsError(key: str, message: str | None = None)#

Bases: ValueError

A record written to an archive exists already (records are write-once)

key#: the key of the record

class ncore.data.stores.Store(zarr_store: object)#

Bases: ABC

A store of a zarr hierarchy.

Stores wrap a zarr-python store of the installed zarr-python version (zarr), and are context managers (closing them on exit). Other zarr-python stores (e.g., in-memory stores) can be used via ZarrStoreAdapter.

close() → None#: Closes the store (finishes writing, no-op by default)

consolidate() → None#: Stores the consolidated metadata of the hierarchy of the store (after writing all nodes): a single compressed record with the node metadata of all members (zarr format v2: .zmetadata.cbor.xz, v3: zarr.json.cbor.zlib), from which open_root_group() opens the hierarchy without reading node metadata records. Node metadata records are complete as well (e.g., for zarr-python, which opens such hierarchies as unconsolidated).

create_root_group( attributes: Attributes, *, zarr_format: ZarrFormat = 2, ) → Group#: Creates the root group of the hierarchy of the store, in zarr format v2 (default, readable with all zarr-python versions) or v3 (requires zarr-python 3 for writing and reading)

static is_consolidated_metadata_key(key: str) → bool#: True for keys of consolidated metadata records (ncore’s compressed consolidated metadata of both zarr formats, and zarr-python 2’s .zmetadata), which Store.consolidate() regenerates (e.g., to skip when copying the records of stores)

open_root_group( consolidated: bool | Literal['auto'] = True, ) → Group#: Opens the root group of the hierarchy of the store for reading: from its consolidated metadata record if consolidated (raises KeyError if the store has none, see consolidate()), otherwise from its node metadata records. "auto" opens it from its consolidated metadata record if the store has one.

reload_resources() → None#: Re-initializes the resources of the store (e.g., file objects in forked processes), no-op by default

property zarr: object#: The wrapped zarr-python store (of the installed zarr-python version)

class ncore.data.stores.ZarrStoreAdapter(zarr_store: object)#

Bases: Store

A zarr hierarchy in any other zarr-python store of the installed zarr-python version (e.g., an in-memory store, or a local directory, see directory())

classmethod directory( path: str | Path | UPath, ) → ZarrStoreAdapter#: A zarr hierarchy in a local directory (a directory per group / array, a file per record, e.g., for debugging). Existing files of the directory are kept when writing.

ncore.data.stores Package#

`ncore.data.stores` Package#