Update py-hamt to 3.4.0

DATASET_CATALOG_USAGE.md

@@ -139,6 +139,7 @@ async def load_dataset(
     organization: Optional[str] = None,  # Organization/agency that owns the collection
     cid: Optional[str] = None,       # Direct CID override (bypasses STAC)
     return_xarray: bool = False,     # Return raw xarray.Dataset instead of GeotemporalData
+    zarr_group: Optional[str] = None, # Explicit Zarr group for grouped/pyramid stores
 ) -> Union[
     Tuple[GeotemporalData, DatasetMetadata],
     Tuple[xr.Dataset, DatasetMetadata]
@@ -155,6 +156,7 @@ async def load_dataset(
   inferred from the root catalog metadata.
 - `cid`: Optional direct CID to bypass STAC catalog resolution
 - `return_xarray`: If True, return raw `xarray.Dataset`, otherwise return `GeotemporalData` wrapper
+- `zarr_group`: Optional Zarr group to open. If omitted, multi-group py-hamt v2 stores default to group `"0"` when available.
 
 **Returns:**
 - Tuple of (dataset, metadata)
@@ -169,6 +171,7 @@ async def load_dataset(
     - `url`: Always None for STAC-based loading
     - `timestamp`: Always None for STAC-based loading
     - `organization`: The resolved organization id when available
+    - `zarr_group`: Present when an explicit or inferred Zarr group was opened
 
 **Raises:**
 - `RuntimeError`: If client is not used as async context manager
@@ -203,7 +206,8 @@ async with dClimateClient() as client:
         dataset="temperature",  # Used for metadata only
         collection="ecmwf_ifs",
         organization="ecmwf",
-        cid="bafybeiabc123..."
+        cid="bafybeiabc123...",
+        zarr_group="0"
     )
     # metadata['source'] will be 'direct_cid'
 ```

README.md

@@ -120,6 +120,7 @@ async def main_xarray():
             collection="era5",
             organization="ecmwf",
             variant="finalized",
+            zarr_group="0",  # Optional for grouped/pyramid sharded stores
             return_xarray=True  # Returns xarray.Dataset
         )
         print(xr_dataset)

dclimate_client_py/datasets.py

@@ -96,6 +96,7 @@ class DatasetMetadata(TypedDict, total=False):
         "catalog", "stac", "direct_cid"
     ]  # How the dataset was loaded
     organization: Optional[str]
+    zarr_group: Optional[str]
 
 
 # --- Helper Functions ---

dclimate_client_py/dclimate_client.py

@@ -121,6 +121,12 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
             await self._kubo_cas.__aexit__(exc_type, exc_val, exc_tb)
             self._kubo_cas = None
 
+    @staticmethod
+    def _apply_zarr_group_metadata(ds: xr.Dataset, metadata: DatasetMetadata) -> None:
+        loaded_zarr_group = ds.attrs.get("_ipfs_zarr_group")
+        if isinstance(loaded_zarr_group, str):
+            metadata["zarr_group"] = loaded_zarr_group
+
     async def load_dataset(
         self,
         dataset: str,
@@ -129,6 +135,7 @@ async def load_dataset(
         organization: typing.Optional[str] = None,
         cid: typing.Optional[str] = None,
         return_xarray: bool = False,
+        zarr_group: typing.Optional[str] = None,
     ) -> typing.Union[
         typing.Tuple[GeotemporalData, DatasetMetadata],
         typing.Tuple[xr.Dataset, DatasetMetadata],
@@ -160,6 +167,10 @@ async def load_dataset(
         return_xarray : bool, optional
             If True, return raw xarray.Dataset. If False (default), return
             GeotemporalData wrapper.
+        zarr_group : str, optional
+            Explicit Zarr group to open for grouped/pyramid sharded stores. If
+            omitted, py-hamt v2 stores with multiple top-level groups default
+            to group "0" when available.
 
         Returns
         -------
@@ -223,6 +234,7 @@ async def load_dataset(
             ds = await _load_dataset_from_ipfs_cid(
                 ipfs_cid=cid,
                 kubo_cas=self._kubo_cas,
+                zarr_group=zarr_group,
             )
 
             # Build metadata for direct CID case
@@ -242,6 +254,7 @@ async def load_dataset(
                     else None
                 ),
             }
+            self._apply_zarr_group_metadata(ds, metadata)
 
             if return_xarray:
                 return ds, metadata
@@ -299,6 +312,7 @@ async def load_dataset(
         ds = await _load_dataset_from_ipfs_cid(
             ipfs_cid=final_cid,
             kubo_cas=self._kubo_cas,
+            zarr_group=zarr_group,
         )
 
         # Build metadata for STAC case
@@ -322,6 +336,7 @@ async def load_dataset(
                 else None
             ),
         }
+        self._apply_zarr_group_metadata(ds, metadata)
 
         if return_xarray:
             return ds, metadata

dclimate_client_py/ipfs_retrieval.py

@@ -6,13 +6,20 @@
 
 import logging
 import time
+import warnings
 from typing import Any
 
 import xarray as xr
 from multiformats import CID
 from opentelemetry import metrics, trace
 from opentelemetry.trace import Span, Status, StatusCode
-from py_hamt import HAMT, KuboCAS, ShardedZarrStore, ZarrHAMTStore
+from py_hamt import (
+    HAMT,
+    KuboCAS,
+    ShardedZarrStore,
+    ShardedZarrV1DeprecationWarning,
+    ZarrHAMTStore,
+)
 
 try:
     import py_hamt.instrumentation as ipfs_instrumentation
@@ -146,12 +153,95 @@ def _is_connection_error(exc: Exception) -> bool:
     )
 
 
+def _normalize_zarr_group(zarr_group: str | None) -> str | None:
+    """Normalize optional Zarr group names for xarray."""
+    if zarr_group is None:
+        return None
+    normalized = zarr_group.strip("/")
+    return normalized or None
+
+
+def _zarr_group_candidates(store: Any) -> list[str]:
+    """Return safe default Zarr groups from py-hamt v2 stores."""
+    groups_getter = getattr(store, "_v2_top_level_groups", None)
+    if not callable(groups_getter):
+        return []
+
+    try:
+        groups = groups_getter()
+    except Exception:
+        return []
+
+    normalized_groups = sorted(
+        group.strip("/") for group in groups if isinstance(group, str) and group
+    )
+    if "0" not in normalized_groups:
+        return []
+    return ["0"]
+
+
+def _store_requires_explicit_zarr_group(store: Any) -> bool:
+    """Return whether py-hamt says a root read needs an explicit group."""
+    requires_getter = getattr(store, "_v2_requires_explicit_group_for_root_read", None)
+    if not callable(requires_getter):
+        return False
+
+    try:
+        return bool(requires_getter())
+    except Exception:
+        return False
+
+
+def _is_explicit_zarr_group_error(exc: Exception) -> bool:
+    """Classify py-hamt's multi-group root-read error."""
+    text = str(exc)
+    return "explicit Zarr group" in text or "group='0'" in text
+
+
+def _open_zarr_from_store(
+    store: Any,
+    *,
+    zarr_group: str | None = None,
+) -> tuple[xr.Dataset, str | None]:
+    """Open a Zarr store, choosing a default group for py-hamt v2 pyramids."""
+    normalized_group = _normalize_zarr_group(zarr_group)
+    if normalized_group is not None:
+        return xr.open_zarr(store=store, group=normalized_group), normalized_group
+
+    if _store_requires_explicit_zarr_group(store):
+        for candidate_group in _zarr_group_candidates(store):
+            return xr.open_zarr(store=store, group=candidate_group), candidate_group
+
+    try:
+        return xr.open_zarr(store=store), None
+    except ValueError as exc:
+        if not _is_explicit_zarr_group_error(exc):
+            raise
+        for candidate_group in _zarr_group_candidates(store):
+            return xr.open_zarr(store=store, group=candidate_group), candidate_group
+        raise
+
+
+async def _open_sharded_zarr_store(
+    *,
+    ipfs_cid: str,
+    kubo_cas: KuboCAS,
+) -> ShardedZarrStore:
+    """Open a py-hamt sharded store without surfacing legacy v1 read warnings."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", category=ShardedZarrV1DeprecationWarning)
+        return await ShardedZarrStore.open(
+            root_cid=ipfs_cid, cas=kubo_cas, read_only=True
+        )
+
+
 # --- Zarr Dataset Loading ---
 
 
 async def _load_dataset_from_ipfs_cid(
     ipfs_cid: str,
     kubo_cas: KuboCAS,
+    zarr_group: str | None = None,
 ) -> xr.Dataset:
     """
     Internal function to load a Zarr dataset from IPFS using a provided KuboCAS instance.
@@ -163,6 +253,9 @@ async def _load_dataset_from_ipfs_cid(
     Args:
         ipfs_cid (str): The IPFS CID of the Zarr dataset's root node.
         kubo_cas (KuboCAS): An active KuboCAS instance to use for loading.
+        zarr_group (str, optional): Explicit Zarr group to open. If omitted and
+            py-hamt reports a multi-group sharded v2 store, group "0" is used
+            when available.
 
     Returns:
         xr.Dataset: The loaded dataset.
@@ -206,6 +299,7 @@ async def _load_dataset_from_ipfs_cid(
                     f"Attempting to load as ShardedZarrStore from CID: {ipfs_cid}"
                 )
                 sharded_start = time.perf_counter()
+                sharded_store_opened = False
                 with _TRACER.start_as_current_span(
                     "dclimate_client.ipfs.open_sharded_store",
                     attributes=_attributes(
@@ -218,10 +312,15 @@ async def _load_dataset_from_ipfs_cid(
                     set_status_on_exception=False,
                 ) as sharded_span:
                     try:
-                        sharded_store = await ShardedZarrStore.open(
-                            root_cid=ipfs_cid, cas=kubo_cas, read_only=True
+                        sharded_store = await _open_sharded_zarr_store(
+                            ipfs_cid=ipfs_cid,
+                            kubo_cas=kubo_cas,
+                        )
+                        sharded_store_opened = True
+                        ds, opened_zarr_group = _open_zarr_from_store(
+                            sharded_store,
+                            zarr_group=zarr_group,
                         )
-                        ds = xr.open_zarr(store=sharded_store)
                     except Exception as sharded_err:
                         sharded_seconds = time.perf_counter() - sharded_start
                         _record_store_open(
@@ -240,6 +339,8 @@ async def _load_dataset_from_ipfs_cid(
                         seconds=sharded_seconds,
                     )
                 ds.attrs["_ipfs_store_type"] = "ShardedZarrStore"
+                if opened_zarr_group is not None:
+                    ds.attrs["_ipfs_zarr_group"] = opened_zarr_group
                 ipfs_instrumentation.observe(
                     "dclimate_client.open_sharded_store_seconds",
                     sharded_seconds,
@@ -257,6 +358,8 @@ async def _load_dataset_from_ipfs_cid(
                         f"IPFS connection failed while loading dataset from CID {ipfs_cid}. Details: {sharded_err}"
                     )
                     raise connection_error from sharded_err
+                if sharded_store_opened:
+                    raise
 
                 # Fall back to HAMT store if sharded loading fails
                 if dataset_span.is_recording():
@@ -290,7 +393,10 @@ async def _load_dataset_from_ipfs_cid(
                         # Wrap with ZarrHAMTStore adapter
                         zarr_hamt_store = ZarrHAMTStore(hamt_store, read_only=True)
 
-                        ds = xr.open_zarr(store=zarr_hamt_store)
+                        ds, opened_zarr_group = _open_zarr_from_store(
+                            zarr_hamt_store,
+                            zarr_group=zarr_group,
+                        )
                     except Exception as hamt_err:
                         hamt_seconds = time.perf_counter() - hamt_start
                         _record_store_open(
@@ -309,6 +415,8 @@ async def _load_dataset_from_ipfs_cid(
                         seconds=hamt_seconds,
                     )
                 ds.attrs["_ipfs_store_type"] = "ZarrHAMTStore"
+                if opened_zarr_group is not None:
+                    ds.attrs["_ipfs_zarr_group"] = opened_zarr_group
                 ipfs_instrumentation.observe(
                     "dclimate_client.open_hamt_store_seconds",
                     hamt_seconds,
@@ -365,6 +473,7 @@ async def _get_dataset_by_ipfs_cid(
     ipfs_cid: str,
     gateway_uri_stem: str | None = None,
     rpc_uri_stem: str | None = None,
+    zarr_group: str | None = None,
 ) -> xr.Dataset:
     """
     Gets an xarray dataset directly from its Zarr root IPFS CID.
@@ -379,6 +488,7 @@ async def _get_dataset_by_ipfs_cid(
         ipfs_cid (str): The IPFS CID of the Zarr dataset's root node.
         gateway_uri_stem (str, optional): Custom IPFS HTTP Gateway URI stem.
         rpc_uri_stem (str, optional): Custom IPFS RPC API URI stem.
+        zarr_group (str, optional): Explicit Zarr group to open.
 
     Returns:
         xr.Dataset: The loaded dataset.
@@ -390,4 +500,8 @@ async def _get_dataset_by_ipfs_cid(
     async with KuboCAS(
         rpc_base_url=rpc_uri_stem, gateway_base_url=gateway_uri_stem
     ) as kubo_cas:
-        return await _load_dataset_from_ipfs_cid(ipfs_cid, kubo_cas)
+        return await _load_dataset_from_ipfs_cid(
+            ipfs_cid,
+            kubo_cas,
+            zarr_group=zarr_group,
+        )

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "pdm.backend"
 
 [project]
 name = "dclimate-client-py"
-version = "0.5.9" # Set a static version or handle it in versioning strategy
+version = "0.5.10" # Set a static version or handle it in versioning strategy
 description = "Python client library for accessing dClimate weather and climate data"
 readme = "README.md"
 license = {text = "MIT"}
@@ -34,7 +34,7 @@ dependencies = [
     "rioxarray",
     "zarr>=3.0.8",
     "numpy>=2.1.3",
-    "py_hamt>=3.3.1",
+    "py_hamt>=3.4.0",
     "requests",
     "pyarrow",
     "geopandas",