Python Examples

Below are code snippets for completing various tasks using the Python API for the Acquire Zarr library. Have an example you'd like to share with the community? Submit a GitHub issue and include "Example:" in your title.

Example: Basic Zarr V2 to filesystem

# Basic Zarr V2 to filesystem
import numpy as np
from acquire_zarr import (
    StreamSettings, ZarrStream, Dimension, DimensionType, ZarrVersion, DataType, Compressor, CompressionCodec,
    CompressionSettings
)


def make_sample_data():
    return np.random.randint(
        0, 65535,
        (32, 48, 64),  # Shape matches chunk size for time dimension
        dtype=np.int32
    )


def main():
    # Configure stream settings
    settings = StreamSettings()

    # Configure compression
    settings.compression = CompressionSettings(
        compressor=Compressor.BLOSC1,
        codec=CompressionCodec.BLOSC_LZ4,
        level=1,
        shuffle=2, # bitshuffle
    )

    # Configure dimensions (t, y, x)
    settings.dimensions.extend([
        Dimension(
            name="t",
            kind=DimensionType.TIME,
            array_size_px=0,  # Unlimited
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
        Dimension(
            name="y",
            kind=DimensionType.SPACE,
            array_size_px=48,
            chunk_size_px=16,
            shard_size_chunks=1,
        ),
        Dimension(
            name="x",
            kind=DimensionType.SPACE,
            array_size_px=64,
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
    ])

    settings.store_path = "output_v2_multiscale.zarr"
    settings.version = ZarrVersion.V2
    settings.data_type = DataType.INT32
    settings.multiscale = True

    # Create stream
    stream = ZarrStream(settings)

    # Create and write sample data
    for i in range(10):
        stream.append(make_sample_data())


if __name__ == "__main__":
    main()

Download this example as a Python script

Example: Zarr V2 with ZSTD compression to S3

# Zarr V2 with ZSTD compression to S3
import numpy as np
from acquire_zarr import (
    StreamSettings, ZarrStream, Dimension, DimensionType, ZarrVersion,
    DataType, Compressor, CompressionCodec, CompressionSettings, S3Settings
)


def make_sample_data():
    return np.random.randint(
        0, 65535,
        (32, 3, 48, 64),  # Shape matches chunk sizes
        dtype=np.int32
    )

def main():
    settings = StreamSettings()

    # Configure S3
    settings.s3 = S3Settings(
        endpoint="http://localhost:9000",
        bucket_name="mybucket",
        access_key_id="myaccesskey",
        secret_access_key="mysecretkey",
        region="us-east-2"
    )

    # Configure compression
    settings.compression = CompressionSettings(
        compressor=Compressor.BLOSC1,
        codec=CompressionCodec.BLOSC_ZSTD,
        level=1,
        shuffle=1,
    )

    # Configure 4D array (t, c, y, x)
    settings.dimensions.extend([
        Dimension(
            name="t",
            kind=DimensionType.TIME,
            array_size_px=0,  # Unlimited
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
        Dimension(
            name="c",
            kind=DimensionType.CHANNEL,
            array_size_px=3,
            chunk_size_px=3,
            shard_size_chunks=1,
        ),
        Dimension(
            name="y",
            kind=DimensionType.SPACE,
            array_size_px=48,
            chunk_size_px=16,
            shard_size_chunks=1,
        ),
        Dimension(
            name="x",
            kind=DimensionType.SPACE,
            array_size_px=64,
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
    ])

    settings.store_path = "output_v2_s3.zarr"
    settings.version = ZarrVersion.V2
    settings.data_type = DataType.INT32

    # Create stream
    stream = ZarrStream(settings)

    # Create and write sample data
    for i in range(10):
        stream.append(make_sample_data())


if __name__ == "__main__":
    main()

Download this example as a Python script

Example: Basic Zarr V2 to filesystem

# Basic Zarr V2 to filesystem
import numpy as np
from acquire_zarr import (
    StreamSettings, ZarrStream, Dimension, DimensionType, ZarrVersion, DataType
)


def make_sample_data():
    return np.random.randint(
        0, 65535,
        (32, 48, 64),  # Shape matches chunk size for time dimension
        dtype=np.int32
    )

def main():
    # Configure stream settings
    settings = StreamSettings()

    # Configure dimensions (t, y, x)
    settings.dimensions.extend([
        Dimension(
            name="t",
            kind=DimensionType.TIME,
            array_size_px=0,  # Unlimited
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
        Dimension(
            name="y",
            kind=DimensionType.SPACE,
            array_size_px=48,
            chunk_size_px=16,
            shard_size_chunks=1,
        ),
        Dimension(
            name="x",
            kind=DimensionType.SPACE,
            array_size_px=64,
            chunk_size_px=32,
            shard_size_chunks=1,
        ),
    ])

    settings.store_path = "output_v2.zarr"
    settings.version = ZarrVersion.V2
    settings.data_type = DataType.INT32

    # Create stream
    stream = ZarrStream(settings)

    # Create and write sample data
    for i in range(10):
        stream.append(make_sample_data())


if __name__ == "__main__":
    main()

Download this example as a Python script

Example: Zarr V3 with LZ4 compression to filesystem

# Zarr V3 with LZ4 compression to filesystem
import numpy as np
from acquire_zarr import (
    StreamSettings, ZarrStream, Dimension, DimensionType, ZarrVersion,
    DataType, Compressor, CompressionCodec, CompressionSettings
)


def make_sample_data():
    return np.random.randint(
        0, 65535,
        (5, 4, 2, 48, 64),  # Shape matches chunk sizes
        dtype=np.uint16
    )

def main():
    settings = StreamSettings()

    # Configure compression
    settings.compression = CompressionSettings(
        compressor=Compressor.BLOSC1,
        codec=CompressionCodec.BLOSC_LZ4,
        level=1,
        shuffle=1,
    )

    # Configure 5D array (t, c, z, y, x)
    settings.dimensions.extend([
        Dimension(
            name="t",
            kind=DimensionType.TIME,
            array_size_px=10,
            chunk_size_px=5,
            shard_size_chunks=2,
        ),
        Dimension(
            name="c",
            kind=DimensionType.CHANNEL,
            array_size_px=8,
            chunk_size_px=4,
            shard_size_chunks=2,
        ),
        Dimension(
            name="z",
            kind=DimensionType.SPACE,
            array_size_px=6,
            chunk_size_px=2,
            shard_size_chunks=1,
        ),
        Dimension(
            name="y",
            kind=DimensionType.SPACE,
            array_size_px=48,
            chunk_size_px=16,
            shard_size_chunks=1,
        ),
        Dimension(
            name="x",
            kind=DimensionType.SPACE,
            array_size_px=64,
            chunk_size_px=16,
            shard_size_chunks=2,
        ),
    ])

    settings.store_path = "output_v3_compressed.zarr"
    settings.version = ZarrVersion.V3
    settings.data_type = DataType.UINT16

    # Create stream
    stream = ZarrStream(settings)

    # Write sample data
    stream.append(make_sample_data())


if __name__ == "__main__":
    main()

Download this example as a Python script

Example: Zarr V3 S3 raw

# zarr_v3_s3_raw.py
import numpy as np
from acquire_zarr import (
    StreamSettings, ZarrStream, Dimension, DimensionType, ZarrVersion,
    DataType, S3Settings
)


def make_sample_data():
    return np.random.randint(
        0, 65535,
        (5, 2, 48, 64),  # Shape matches chunk sizes
        dtype=np.uint16
    )

def main():
    settings = StreamSettings()

    # Configure S3
    settings.s3 = S3Settings(
        endpoint="http://localhost:9000",
        bucket_name="mybucket",
        access_key_id="myaccesskey",
        secret_access_key="mysecretkey",
        region="us-east-2"
    )

    # Configure 4D array (t, z, y, x)
    settings.dimensions.extend([
        Dimension(
            name="t",
            kind=DimensionType.TIME,
            array_size_px=0,  # Unlimited
            chunk_size_px=5,
            shard_size_chunks=2,
        ),
        Dimension(
            name="z",
            kind=DimensionType.SPACE,
            array_size_px=10,
            chunk_size_px=2,
            shard_size_chunks=1,
        ),
        Dimension(
            name="y",
            kind=DimensionType.SPACE,
            array_size_px=48,
            chunk_size_px=16,
            shard_size_chunks=1,
        ),
        Dimension(
            name="x",
            kind=DimensionType.SPACE,
            array_size_px=64,
            chunk_size_px=16,
            shard_size_chunks=2,
        ),
    ])

    settings.store_path = "output_v3_s3.zarr"
    settings.version = ZarrVersion.V3
    settings.data_type = DataType.UINT16

    # Create stream
    stream = ZarrStream(settings)

    # Write sample data
    stream.append(make_sample_data())


if __name__ == "__main__":
    main()

Download this example as a Python script