chore: merge main into feat/sentinel3-export

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-  ".": "0.10.1"
+  ".": "0.10.2"
 }

CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## 0.10.2 (2026-07-17)
+
+## What's Changed
+* ci(deps): bump the actions group with 3 updates by @dependabot[bot] in https://github.com/EOPF-Explorer/data-model/pull/185
+* bump zarr conventions; migrate type checking to pyright by @d-v-b in https://github.com/EOPF-Explorer/data-model/pull/199
+* fix(conversion): concurrent-safe source opening (native-chunk-aligned reads + atomic CacheStore) for data-pipeline#339 by @lhoupert in https://github.com/EOPF-Explorer/data-model/pull/220
+
+
+**Full Changelog**: https://github.com/EOPF-Explorer/data-model/compare/v0.10.1...v0.10.2
+
 ## 0.10.1 (2026-06-09)
 
 ## What's Changed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "eopf-geozarr"
-version = "0.10.1"
+version = "0.10.2"
 description = "GeoZarr compliant data model for EOPF datasets"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -29,7 +29,9 @@ requires-python = ">=3.12"
 dependencies = [
     "pydantic-zarr>=0.8.0",
     "pydantic>=2.12",
-    "zarr[cast-value-rs]>=3.2.0",
+    # <3.3.0: open_source_datatree uses zarr's experimental CacheStore, whose
+    # location/API may change between minor releases; bump deliberately.
+    "zarr[cast-value-rs]>=3.2.0,<3.3.0",
     "xarray>=2025.7.1",
     "dask[array,distributed]>=2026.1.0",
     "numpy>=2.3.1",

src/eopf_geozarr/cli.py

@@ -28,6 +28,7 @@
     validate_s3_access,
 )
 from .conversion.geozarr import get_zarr_group
+from .conversion.open_source import open_source_datatree
 
 if TYPE_CHECKING:
     from dask.distributed import Client
@@ -1331,10 +1332,7 @@ def convert_s2_optimized_command(args: argparse.Namespace) -> None:
     try:
         # Load input dataset
         log.info("Loading Sentinel-2 dataset from", input_path=args.input_path)
-        storage_options = get_storage_options(str(args.input_path))
-        dt_input = xr.open_datatree(
-            str(args.input_path), engine="zarr", chunks="auto", storage_options=storage_options
-        )
+        dt_input = open_source_datatree(str(args.input_path))
 
         # Convert
         convert_s2_optimized(

src/eopf_geozarr/conversion/__init__.py

@@ -17,6 +17,7 @@
     iterative_copy,
     setup_datatree_metadata_geozarr_spec_compliant,
 )
+from .open_source import open_source_datatree
 from .utils import (
     calculate_aligned_chunk_size,
     downsample_2d_array,
@@ -36,6 +37,7 @@
     "is_s3_path",
     "iterative_copy",
     "open_s3_zarr_group",
+    "open_source_datatree",
     "parse_s3_path",
     "s3_path_exists",
     "setup_datatree_metadata_geozarr_spec_compliant",

src/eopf_geozarr/conversion/open_source.py

@@ -0,0 +1,90 @@
+"""Open source EOPF zarr stores with dask chunks aligned to native chunks.
+
+With ``chunks="auto"``, dask may split one on-disk zarr chunk into several
+read tasks that each fetch and decompress the same object key — multiplying
+egress and racing on shared cache files (EOPF-Explorer/data-pipeline#339).
+``open_source_datatree`` opens with ``chunks={}`` (the store's native chunk
+grid), so each zarr chunk is read by exactly one dask task; downstream
+conversion code rechunks to its own output encoding anyway. Note that one
+task then materializes one whole native chunk (~241 MB raw for cpm_v270
+aot/wvp) — under tight memory budgets, lower dask task concurrency rather
+than sub-splitting stored chunks.
+"""
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+import xarray as xr
+
+from .fs_utils import S3FsOptions, get_storage_options
+
+
+def open_source_datatree(
+    path: str,
+    *,
+    storage_options: S3FsOptions | dict[str, Any] | None = None,
+    cache_dir: str | None = None,
+    engine: str = "zarr",
+) -> xr.DataTree:
+    """Open a source datatree with dask chunks matching the native zarr chunks.
+
+    Parameters
+    ----------
+    path : str
+        Source store URL (local path, s3://, or https://).
+    storage_options : dict, optional
+        fsspec storage options. Defaults to ``get_storage_options(path)``
+        (S3 credentials/endpoint for s3:// paths, None otherwise).
+    cache_dir : str, optional
+        Local directory for an on-disk read cache of source objects, backed
+        by zarr's ``CacheStore`` over a ``LocalStore``. LocalStore writes are
+        atomic (temp file + rename), so a concurrent read never sees a
+        partially downloaded object. Entries are namespaced per source URL
+        (CacheStore keys entries by relative zarr key, so different sources
+        sharing a directory would otherwise collide) and never expire; use an
+        ephemeral directory.
+    engine : str
+        xarray backend engine, default ``"zarr"``.
+
+    Returns
+    -------
+    xr.DataTree
+        Datatree whose dask arrays are chunked exactly on the store's native
+        chunk grid: each on-disk zarr chunk is read by exactly one dask task.
+    """
+    if storage_options is None:
+        storage_options = get_storage_options(path)
+    if cache_dir is None:
+        return xr.open_datatree(
+            path,
+            engine=engine,
+            chunks={},
+            storage_options=storage_options,
+        )
+    # Imported lazily: CacheStore is zarr's experimental API, so a relocation
+    # in a future zarr release must not break importing this package.
+    try:
+        from zarr.experimental.cache_store import CacheStore
+    except ImportError as exc:
+        raise ImportError(
+            "cache_dir requires zarr.experimental.cache_store.CacheStore "
+            "(present in zarr 3.2.x). Your zarr version no longer provides "
+            "it; pin zarr accordingly or open without cache_dir."
+        ) from exc
+    from zarr.storage import FsspecStore, LocalStore
+
+    source = FsspecStore.from_url(
+        path,
+        storage_options=dict(storage_options) if storage_options else None,
+        read_only=True,
+    )
+    # CacheStore keys entries by relative zarr key ("aot/0.0" is the same key
+    # for every S2 product), so namespace the cache per source URL. Trailing
+    # slashes are stripped so equivalent URLs share a namespace.
+    cache_key = hashlib.sha256(path.rstrip("/").encode()).hexdigest()[:16]
+    source_cache = Path(cache_dir) / cache_key
+    cached = CacheStore(source, cache_store=LocalStore(source_cache))
+    # xarray's open_datatree accepts a zarr store at runtime, but its stub does
+    # not list Store among the accepted input types.
+    return xr.open_datatree(cached, engine=engine, chunks={})  # pyright: ignore[reportArgumentType]

tests/test_open_source.py

@@ -0,0 +1,145 @@
+"""Tests for open_source_datatree: native-chunk-aligned reads and the
+per-source cache (EOPF-Explorer/data-pipeline#339)."""
+
+import os
+import threading
+from collections.abc import Iterator
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+
+import numcodecs
+import numpy as np
+import pytest
+import zarr
+
+from eopf_geozarr.conversion.open_source import open_source_datatree
+
+
+def _build_source_store(store_path: str, seed: int = 42) -> str:
+    """Write a zarr v2 store mimicking a cpm_v270 source group.
+
+    - ``aot``: ONE whole-array chunk (the pathological v270 layout)
+    - ``b02``: regular 32x32 tiles
+
+    ``seed`` varies the values so two stores can share a key layout but differ
+    in content.
+    """
+    group = zarr.open_group(store_path, mode="w", zarr_format=2)
+    rng = np.random.default_rng(seed)
+
+    aot = group.create_array(
+        "aot",
+        shape=(128, 128),
+        chunks=(128, 128),
+        dtype="f8",
+        compressors=numcodecs.Blosc(cname="lz4", clevel=5, shuffle=1),
+    )
+    aot[:] = rng.random((128, 128))
+    aot.attrs["_ARRAY_DIMENSIONS"] = ["y", "x"]
+
+    b02 = group.create_array(
+        "b02",
+        shape=(128, 128),
+        chunks=(32, 32),
+        dtype="f8",
+        compressors=numcodecs.Blosc(cname="lz4", clevel=5, shuffle=1),
+    )
+    b02[:] = rng.random((128, 128))
+    b02.attrs["_ARRAY_DIMENSIONS"] = ["y", "x"]
+
+    zarr.consolidate_metadata(group.store)
+    return store_path
+
+
+@pytest.fixture
+def source_store(tmp_path) -> str:
+    return _build_source_store(str(tmp_path / "source.zarr"))
+
+
+def test_single_chunk_array_is_one_dask_task(source_store) -> None:
+    dt = open_source_datatree(source_store)
+    assert dt["aot"].data.chunks == ((128,), (128,))
+
+
+def test_tiled_array_chunks_match_native_grid(source_store) -> None:
+    dt = open_source_datatree(source_store)
+    assert dt["aot"].data.chunks == ((128,), (128,))
+    assert dt["b02"].data.chunks == ((32,) * 4, (32,) * 4)
+
+
+def test_values_roundtrip(source_store) -> None:
+    dt = open_source_datatree(source_store)
+    aot_expected = np.asarray(zarr.open_array(source_store, mode="r", path="aot")[:])
+    np.testing.assert_array_equal(dt["aot"].values, aot_expected)
+
+
+def test_explicit_storage_options_are_forwarded_to_xarray(monkeypatch) -> None:
+    captured: dict[str, object] = {}
+
+    def fake_open_datatree(path: object, **kwargs: object) -> None:
+        captured.update(kwargs)
+        raise InterruptedError("stop before any I/O")
+
+    monkeypatch.setattr("xarray.open_datatree", fake_open_datatree)
+    sentinel = {"endpoint_url": "https://example.invalid"}
+    with pytest.raises(InterruptedError):
+        open_source_datatree("/some/store.zarr", storage_options=sentinel)
+    assert captured["storage_options"] == sentinel
+    assert captured["chunks"] == {}
+
+
+@pytest.fixture
+def http_source(source_store) -> Iterator[str]:
+    """Serve the source store over local HTTP (mimics the EODC https source)."""
+    root = os.path.dirname(source_store)
+    name = os.path.basename(source_store)
+
+    class Handler(SimpleHTTPRequestHandler):
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            super().__init__(*args, directory=root, **kwargs)  # type: ignore[arg-type]
+
+        def log_message(self, format: str, *args: object) -> None:
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{server.server_address[1]}/{name}"
+    server.shutdown()
+
+
+def test_cache_dir_reads_correct_data_over_http(http_source, source_store, tmp_path) -> None:
+    """cache_dir reads return correct data and populate the cache, with no
+    leftover temp files. Atomicity of the cache writes themselves is zarr
+    LocalStore's contract (temp file + rename)."""
+    cache_dir = str(tmp_path / "source-cache")
+    dt = open_source_datatree(http_source, cache_dir=cache_dir)
+
+    aot_expected = np.asarray(zarr.open_array(source_store, mode="r", path="aot")[:])
+    np.testing.assert_array_equal(dt["aot"].values, aot_expected)
+    assert dt["aot"].data.chunks == ((128,), (128,))  # native alignment preserved
+
+    cached_files = [f for _, _, fs in os.walk(cache_dir) for f in fs]
+    assert cached_files, "cache directory should be populated after reads"
+    assert not [f for f in cached_files if f.endswith(".partial")]
+
+
+def test_two_sources_can_share_one_cache_dir(tmp_path) -> None:
+    """A shared cache_dir must never let one source serve another's bytes.
+
+    CacheStore keys entries by zarr key ("aot/0.0" is identical for every S2
+    product), so open_source_datatree namespaces the cache per source URL.
+    """
+    first = _build_source_store(str(tmp_path / "first.zarr"), seed=1)
+    second = _build_source_store(str(tmp_path / "second.zarr"), seed=2)
+    shared_cache = str(tmp_path / "shared-cache")
+
+    dt_first = open_source_datatree(first, cache_dir=shared_cache)
+    np.testing.assert_array_equal(
+        dt_first["aot"].values,
+        np.asarray(zarr.open_array(first, mode="r", path="aot")[:]),
+    )
+
+    dt_second = open_source_datatree(second, cache_dir=shared_cache)
+    np.testing.assert_array_equal(
+        dt_second["aot"].values,
+        np.asarray(zarr.open_array(second, mode="r", path="aot")[:]),
+    )