grib2io.icechunk

Icechunk Virtual Store Writer

Provides IcechunkWriter, which writes grib2io reference manifests (Kerchunk v1 format) into an Icechunk virtual store, creating versioned Zarr v3 datasets backed by the original GRIB2 file bytes.

The writer translates each entry in the manifest's "refs" dict into either a virtual chunk reference (for data chunks pointing at byte ranges in GRIB2 files) or native Zarr metadata (for .zarray, .zattrs, .zgroup entries and inline base64-encoded coordinate arrays).

Icechunk uses Zarr v3 format internally, so the writer converts Kerchunk v1 (Zarr v2) metadata and chunk keys to Zarr v3 equivalents:

  • .zgroupzarr.json with node_type: "group"
  • .zarray + .zattrs<var>/zarr.json with node_type: "array"
  • Chunk keys VAR/0.0.0VAR/c/0/0/0
Example
>>> from grib2io.kerchunk import ReferenceGenerator
>>> from grib2io.icechunk import IcechunkWriter
>>> gen = ReferenceGenerator("gfs.grib2")
>>> manifest = gen.generate()
>>> writer = IcechunkWriter("/tmp/gfs_icechunk")
>>> writer.write(manifest)
>>> snapshot_id = writer.commit("Initial ingest of GFS data")
  1"""
  2Icechunk Virtual Store Writer
  3=============================
  4
  5Provides :class:`IcechunkWriter`, which writes grib2io reference manifests
  6(Kerchunk v1 format) into an `Icechunk <https://icechunk.io/>`_ virtual
  7store, creating versioned Zarr v3 datasets backed by the original GRIB2 file
  8bytes.
  9
 10The writer translates each entry in the manifest's ``"refs"`` dict into
 11either a virtual chunk reference (for data chunks pointing at byte ranges
 12in GRIB2 files) or native Zarr metadata (for ``.zarray``, ``.zattrs``,
 13``.zgroup`` entries and inline base64-encoded coordinate arrays).
 14
 15Icechunk uses Zarr v3 format internally, so the writer converts Kerchunk v1
 16(Zarr v2) metadata and chunk keys to Zarr v3 equivalents:
 17
 18- ``.zgroup`` → ``zarr.json`` with ``node_type: "group"``
 19- ``.zarray`` + ``.zattrs`` → ``<var>/zarr.json`` with ``node_type: "array"``
 20- Chunk keys ``VAR/0.0.0`` → ``VAR/c/0/0/0``
 21
 22Example
 23-------
 24>>> from grib2io.kerchunk import ReferenceGenerator
 25>>> from grib2io.icechunk import IcechunkWriter
 26>>> gen = ReferenceGenerator("gfs.grib2")
 27>>> manifest = gen.generate()
 28>>> writer = IcechunkWriter("/tmp/gfs_icechunk")
 29>>> writer.write(manifest)
 30>>> snapshot_id = writer.commit("Initial ingest of GFS data")
 31"""
 32
 33from __future__ import annotations
 34
 35import base64
 36import json
 37import logging
 38from typing import Any, Dict, Optional, Set
 39
 40_logger = logging.getLogger(__name__)
 41
 42
 43# ---------------------------------------------------------------------------
 44# Lazy import guard
 45# ---------------------------------------------------------------------------
 46
 47
 48def _ensure_icechunk():
 49    """Raise ``ImportError`` if *icechunk* is not available."""
 50    try:
 51        import icechunk  # noqa: F401
 52    except ImportError:
 53        raise ImportError("icechunk is required for virtual store support. Install with: pip install grib2io[icechunk]")
 54
 55
 56# ---------------------------------------------------------------------------
 57# Helpers
 58# ---------------------------------------------------------------------------
 59
 60
 61def _is_metadata_key(key: str) -> bool:
 62    """Return ``True`` if *key* is a Zarr metadata key."""
 63    basename = key.rsplit("/", 1)[-1] if "/" in key else key
 64    return basename.startswith(".z")
 65
 66
 67def _is_inline_data(value) -> bool:
 68    """Return ``True`` if *value* is inline data (a JSON string or
 69    base64-encoded bytes), as opposed to a ``[uri, offset, length]``
 70    reference list."""
 71    return isinstance(value, str)
 72
 73
 74def _is_virtual_ref(value) -> bool:
 75    """Return ``True`` if *value* is a virtual chunk reference
 76    ``[uri, offset, length]``."""
 77    return isinstance(value, list) and len(value) == 3
 78
 79
 80def _decode_inline_value(value: str) -> bytes:
 81    """Decode an inline manifest value to raw bytes.
 82
 83    Inline values are either:
 84    - ``"base64:<data>"`` — base64-encoded binary data (coordinate arrays)
 85    - A plain JSON string (metadata like ``.zarray``, ``.zattrs``)
 86
 87    Parameters
 88    ----------
 89    value : str
 90        The inline value from the manifest.
 91
 92    Returns
 93    -------
 94    bytes
 95        The decoded bytes.
 96    """
 97    if value.startswith("base64:"):
 98        return base64.b64decode(value[7:])
 99    # Plain JSON string — encode to UTF-8 bytes for storage
100    return value.encode("utf-8")
101
102
103def _collect_virtual_chunk_prefixes(refs: Dict[str, Any]) -> Set[str]:
104    """Scan all virtual refs and collect the unique URI prefixes needed
105    for Icechunk virtual chunk containers.
106
107    For ``file://`` URIs, the prefix is the directory portion.
108    For ``s3://``, ``gcs://``, ``https://`` URIs, the prefix is the
109    bucket/host + path up to the last ``/``.
110
111    Parameters
112    ----------
113    refs : dict
114        The ``"refs"`` dict from a Kerchunk v1 manifest.
115
116    Returns
117    -------
118    set of str
119        Unique URI prefixes (each ending with ``/``).
120    """
121    prefixes: Set[str] = set()
122    for key, value in refs.items():
123        if _is_virtual_ref(value):
124            uri = value[0]
125            # Extract the directory prefix from the URI
126            last_slash = uri.rfind("/")
127            if last_slash > 0:
128                prefix = uri[: last_slash + 1]
129                prefixes.add(prefix)
130    return prefixes
131
132
133def _chunk_key_v2_to_v3(key: str) -> str:
134    """Convert a Kerchunk v1 (Zarr v2) chunk key to Zarr v3 format.
135
136    Zarr v2 chunk keys use dot-separated indices under the variable name:
137    ``"TMP/0.0.0.0"`` or ``"level/0"``.
138
139    Zarr v3 chunk keys use ``c/`` prefix with slash-separated indices:
140    ``"TMP/c/0/0/0/0"`` or ``"level/c/0"``.
141
142    Parameters
143    ----------
144    key : str
145        A Zarr v2 chunk key like ``"TMP/0.0.0"`` or ``"level/0"``.
146
147    Returns
148    -------
149    str
150        The Zarr v3 equivalent like ``"TMP/c/0/0/0"`` or ``"level/c/0"``.
151    """
152    if "/" not in key:
153        # Root-level key, no conversion needed
154        return key
155
156    parts = key.split("/")
157    var_name = parts[0]
158    index_str = parts[1]
159
160    # Split dot-separated indices and rejoin with slashes under c/
161    indices = index_str.split(".")
162    return var_name + "/c/" + "/".join(indices)
163
164
165def _make_zarr_v3_group_metadata(v2_zgroup_str: str) -> str:
166    """Convert a Zarr v2 ``.zgroup`` JSON string to Zarr v3 ``zarr.json``
167    group metadata.
168
169    Parameters
170    ----------
171    v2_zgroup_str : str
172        JSON string like ``'{"zarr_format": 2}'``.
173
174    Returns
175    -------
176    str
177        Zarr v3 group metadata JSON string.
178    """
179    return json.dumps(
180        {
181            "zarr_format": 3,
182            "node_type": "group",
183            "attributes": {},
184        }
185    )
186
187
188def _make_zarr_v3_array_metadata(
189    v2_zarray_str: str,
190    v2_zattrs_str: Optional[str] = None,
191) -> str:
192    """Convert Zarr v2 ``.zarray`` and ``.zattrs`` JSON strings to a
193    single Zarr v3 ``zarr.json`` array metadata string.
194
195    Parameters
196    ----------
197    v2_zarray_str : str
198        JSON string with Zarr v2 array metadata (shape, chunks, dtype, etc.).
199    v2_zattrs_str : str, optional
200        JSON string with Zarr v2 attributes (_ARRAY_DIMENSIONS, etc.).
201
202    Returns
203    -------
204    str
205        Zarr v3 array metadata JSON string.
206    """
207    v2_zarray = json.loads(v2_zarray_str)
208    v2_zattrs = json.loads(v2_zattrs_str) if v2_zattrs_str else {}
209
210    shape = v2_zarray.get("shape", [])
211    chunks = v2_zarray.get("chunks", shape)
212    dtype = v2_zarray.get("dtype", "<f4")
213    fill_value = v2_zarray.get("fill_value", None)
214
215    # Convert numpy dtype string to Zarr v3 data_type
216    dtype_map = {
217        "<f4": "float32",
218        "<f8": "float64",
219        "<i4": "int32",
220        "<i8": "int64",
221        "<i2": "int16",
222        "<u1": "uint8",
223        "<u2": "uint16",
224        "<u4": "uint32",
225        ">f4": "float32",
226        ">f8": "float64",
227        ">i4": "int32",
228        ">i8": "int64",
229    }
230    data_type = dtype_map.get(dtype, "float32")
231
232    # Determine endianness for the bytes codec
233    endian = "big" if dtype.startswith(">") else "little"
234
235    # Extract dimension names from attributes
236    dim_names = v2_zattrs.get("_ARRAY_DIMENSIONS", [])
237
238    # Build codecs — for Zarr v3 / Icechunk, we use a simple bytes codec.
239    # The grib2io compressor is a Zarr v2 concept used by the Kerchunk
240    # reference filesystem; Icechunk accesses data via virtual refs so
241    # the codec pipeline just needs to handle raw bytes.
242    codecs = [{"name": "bytes", "configuration": {"endian": endian}}]
243
244    # Preserve the grib2io compressor config in attributes so it can be
245    # recovered if needed, but don't put it in the codecs list where
246    # Zarr v3 would try to instantiate it.
247    compressor = v2_zarray.get("compressor", None)
248    if compressor and compressor.get("id") == "grib2io":
249        v2_zattrs["_grib2io_compressor"] = compressor
250
251    # Handle fill_value — "NaN" string is valid in Zarr v3
252    if fill_value == "NaN" or fill_value is None:
253        fill_value_v3 = "NaN"
254    else:
255        fill_value_v3 = fill_value
256
257    v3_meta = {
258        "zarr_format": 3,
259        "node_type": "array",
260        "shape": shape,
261        "data_type": data_type,
262        "chunk_grid": {
263            "name": "regular",
264            "configuration": {"chunk_shape": chunks},
265        },
266        "chunk_key_encoding": {
267            "name": "default",
268            "configuration": {"separator": "/"},
269        },
270        "fill_value": fill_value_v3,
271        "codecs": codecs,
272        "dimension_names": dim_names if dim_names else None,
273        "attributes": v2_zattrs,
274    }
275
276    return json.dumps(v3_meta)
277
278
279def _store_set_sync(store, key: str, data: bytes) -> None:
280    """Synchronously write bytes to an Icechunk store.
281
282    Icechunk v2's ``store.set()`` is an async coroutine. This helper
283    wraps it using the store's ``_sync`` method and converts raw bytes
284    to a Zarr ``Buffer`` object.
285
286    Parameters
287    ----------
288    store : IcechunkStore
289        The Icechunk store.
290    key : str
291        The Zarr key to write.
292    data : bytes
293        The raw bytes to write.
294    """
295    from zarr.core.buffer import default_buffer_prototype
296
297    buf = default_buffer_prototype().buffer.from_bytes(data)
298    store._sync(store.set(key, buf))
299
300
301def _store_get_sync(store, key: str):
302    """Synchronously read bytes from an Icechunk store.
303
304    Parameters
305    ----------
306    store : IcechunkStore
307        The Icechunk store.
308    key : str
309        The Zarr key to read.
310
311    Returns
312    -------
313    bytes or None
314        The raw bytes, or ``None`` if the key does not exist.
315    """
316    from zarr.core.buffer import default_buffer_prototype
317
318    buf = store._sync(store.get(key, default_buffer_prototype()))
319    if buf is not None:
320        return buf.to_bytes()
321    return None
322
323
324# ---------------------------------------------------------------------------
325# Public API
326# ---------------------------------------------------------------------------
327
328
329class IcechunkWriter:
330    """Write grib2io reference manifests into an Icechunk virtual store.
331
332    Parameters
333    ----------
334    store_path : str
335        Path or URI for the Icechunk store.  For local filesystem stores
336        this is a directory path.  For cloud stores, provide the
337        appropriate URI (e.g. ``s3://bucket/prefix``).
338    storage_config : optional
339        An Icechunk ``Storage`` object.  If ``None`` (the default), a
340        local filesystem storage is created at *store_path*.
341    """
342
343    def __init__(
344        self,
345        store_path: str,
346        storage_config: Optional[Any] = None,
347    ):
348        _ensure_icechunk()
349
350        self._store_path = store_path
351        self._storage_config = storage_config
352        self._repo = None
353        self._session = None
354
355    # ------------------------------------------------------------------
356    # Internal helpers
357    # ------------------------------------------------------------------
358
359    def _get_storage(self):
360        """Return an Icechunk ``Storage`` object."""
361        if self._storage_config is not None:
362            return self._storage_config
363
364        import icechunk
365
366        return icechunk.local_filesystem_storage(path=self._store_path)
367
368    def _create_or_open_repo(self, mode: str, virtual_prefixes: Set[str]):
369        """Create or open an Icechunk repository.
370
371        Parameters
372        ----------
373        mode : str
374            ``'w'`` to create a new repo (or overwrite), ``'a'`` to open
375            an existing repo for appending.
376        virtual_prefixes : set of str
377            URI prefixes that need to be registered as virtual chunk
378            containers.
379        """
380        import icechunk
381
382        storage = self._get_storage()
383
384        # Build repository config with virtual chunk containers
385        config = icechunk.config.RepositoryConfig.default()
386        for prefix in virtual_prefixes:
387            container_store = self._make_container_store(prefix)
388            config.set_virtual_chunk_container(icechunk.virtual.VirtualChunkContainer(prefix, container_store))
389
390        # Build authorize_virtual_chunk_access mapping
391        # Use None for credentials (will use environment or anonymous)
392        authorize = {prefix: None for prefix in virtual_prefixes}
393
394        if mode == "w":
395            self._repo = icechunk.Repository.open_or_create(
396                storage,
397                config=config,
398                authorize_virtual_chunk_access=authorize if authorize else None,
399            )
400        else:
401            # Append mode — open existing
402            self._repo = icechunk.Repository.open(
403                storage,
404                config=config,
405                authorize_virtual_chunk_access=authorize if authorize else None,
406            )
407
408        self._session = self._repo.writable_session("main")
409
410    def _make_container_store(self, prefix: str):
411        """Create an appropriate Icechunk storage backend for a virtual
412        chunk container based on the URI prefix scheme.
413
414        Parameters
415        ----------
416        prefix : str
417            URI prefix like ``file:///path/to/dir/`` or
418            ``s3://bucket/prefix/``.
419
420        Returns
421        -------
422        An Icechunk storage store object.
423        """
424        import icechunk
425
426        if prefix.startswith("file://"):
427            # Extract local path from file:// URI
428            local_path = prefix[7:]  # Remove "file://"
429            # Remove trailing slash for the store path
430            if local_path.endswith("/"):
431                local_path = local_path[:-1]
432            return icechunk.storage.local_filesystem_store(local_path)
433        elif prefix.startswith("s3://"):
434            return icechunk.storage.s3_store(region="us-east-1")
435        elif prefix.startswith("gcs://"):
436            return icechunk.storage.gcs_store(opts={})
437        elif prefix.startswith("http://") or prefix.startswith("https://"):
438            return icechunk.storage.http_store(opts={})
439        else:
440            # Default to local filesystem
441            return icechunk.storage.local_filesystem_store(prefix)
442
443    # ------------------------------------------------------------------
444    # Public methods
445    # ------------------------------------------------------------------
446
447    def write(
448        self,
449        manifest: dict,
450        mode: str = "w",
451        append_dim: Optional[str] = None,
452    ) -> None:
453        """Write a reference manifest into the Icechunk store.
454
455        Parameters
456        ----------
457        manifest : dict
458            Kerchunk v1 reference manifest dict with ``"version"`` and
459            ``"refs"`` keys.
460        mode : str
461            ``'w'`` for create/overwrite, ``'a'`` for append.
462        append_dim : str, optional
463            Dimension along which to append when ``mode='a'``.  Required
464            when appending multi-file data along a specific dimension
465            (e.g. ``"refDate"`` or ``"leadTime"``).
466        """
467        refs = manifest.get("refs", {})
468
469        # Collect virtual chunk URI prefixes for container registration
470        virtual_prefixes = _collect_virtual_chunk_prefixes(refs)
471
472        # Create or open the repository
473        self._create_or_open_repo(mode, virtual_prefixes)
474
475        store = self._session.store
476
477        if mode == "a" and append_dim is not None:
478            self._write_append(refs, store, append_dim)
479        else:
480            self._write_create(refs, store)
481
482    def _write_create(self, refs: Dict[str, Any], store) -> None:
483        """Write all refs to a fresh store using zarr's Python API.
484
485        Creates the root group and all arrays via ``zarr.open_group`` /
486        ``group.create_array`` so that Zarr v3 metadata is stored through
487        the proper path that icechunk and zarr recognise.  Coordinate
488        (inline) data is written via the zarr Array write API, and virtual
489        chunk refs for data variables are registered with
490        ``store.set_virtual_ref``.
491
492        Parameters
493        ----------
494        refs : dict
495            The ``"refs"`` dict from the manifest.
496        store : IcechunkStore
497            The Icechunk store to write to.
498        """
499        import numpy as np
500        import zarr
501
502        # Collect all metadata by variable
503        zarray_refs: Dict[str, str] = {}  # var_name -> .zarray JSON str
504        zattrs_refs: Dict[str, str] = {}  # var_name -> .zattrs JSON str
505        data_refs: Dict[str, Any] = {}  # chunk key -> [uri, offset, len]
506        inline_data_refs: Dict[str, str] = {}  # chunk key -> inline value
507
508        for key, value in refs.items():
509            if key == ".zgroup":
510                pass  # zarr.open_group handles the root group metadata
511            elif key.endswith("/.zarray"):
512                var_name = key.rsplit("/.zarray", 1)[0]
513                zarray_refs[var_name] = value
514            elif key.endswith("/.zattrs"):
515                var_name = key.rsplit("/.zattrs", 1)[0]
516                zattrs_refs[var_name] = value
517            elif _is_virtual_ref(value):
518                data_refs[key] = value
519            elif _is_inline_data(value) and not _is_metadata_key(key):
520                inline_data_refs[key] = value
521            elif _is_metadata_key(key):
522                pass  # skip other metadata keys
523            else:
524                _logger.warning(
525                    "Skipping unrecognized manifest entry: %s = %r",
526                    key,
527                    value,
528                )
529
530        # Create root group and all arrays via zarr's Python API so that
531        # metadata is stored in the location that zarr/icechunk actually
532        # reads, rather than as raw chunk bytes at the zarr.json key.
533        root = zarr.open_group(store, mode="w", zarr_format=3)
534
535        # Identify which variables have virtual (data) refs — those need
536        # Grib2SerializerCodec as their zarr v3 serializer.
537        data_var_names: set = {key.split("/")[0] for key in data_refs}
538
539        # Import the zarr v3 serializer codec (registered at import time)
540        from grib2io.codecs import Grib2SerializerCodec as _Grib2Ser
541
542        for var_name, zarray_str in zarray_refs.items():
543            v2_zarray = json.loads(zarray_str)
544            zattrs_str = zattrs_refs.get(var_name)
545            v2_zattrs = json.loads(zattrs_str) if zattrs_str else {}
546
547            shape = v2_zarray["shape"]
548            chunks = v2_zarray.get("chunks", shape)
549            dtype_str = v2_zarray.get("dtype", "<f4")
550            fill_value = v2_zarray.get("fill_value", None)
551            dim_names = v2_zattrs.get("_ARRAY_DIMENSIONS", [])
552
553            dtype = np.dtype(dtype_str)
554            if fill_value == "NaN" or fill_value is None:
555                fv: Any = float("nan") if np.issubdtype(dtype, np.floating) else 0
556            else:
557                fv = fill_value
558
559            if var_name in data_var_names:
560                # Data variable — use Grib2SerializerCodec so zarr v3 can
561                # decode the raw GRIB2 section 7 bytes from virtual chunks.
562                compressor = v2_zarray.get("compressor", {}) or {}
563                codec_cfg = {k: v for k, v in compressor.items() if k != "id"}
564                serializer = _Grib2Ser(**codec_cfg)
565                root.create_array(
566                    var_name,
567                    shape=shape,
568                    chunks=chunks,
569                    dtype=dtype,
570                    fill_value=fv,
571                    serializer=serializer,
572                    compressors=[],
573                    filters=[],
574                    dimension_names=dim_names if dim_names else None,
575                    attributes=v2_zattrs if v2_zattrs else None,
576                )
577            else:
578                # Coordinate array — plain bytes serializer (default)
579                root.create_array(
580                    var_name,
581                    shape=shape,
582                    chunks=chunks,
583                    dtype=dtype,
584                    fill_value=fv,
585                    dimension_names=dim_names if dim_names else None,
586                    attributes=v2_zattrs if v2_zattrs else None,
587                )
588
589        # Write coordinate (inline) data via zarr's Array write API so
590        # that encoding goes through the proper codec pipeline.
591        for key, value in inline_data_refs.items():
592            parts = key.split("/")
593            var_name = parts[0]
594            if var_name not in zarray_refs:
595                _logger.warning("Inline data for unknown variable %s, skipping", var_name)
596                continue
597
598            raw_bytes = _decode_inline_value(value)
599            v2_zarray = json.loads(zarray_refs[var_name])
600            dtype = np.dtype(v2_zarray["dtype"])
601            shape = v2_zarray["shape"]
602            coord_values = np.frombuffer(raw_bytes, dtype=dtype).reshape(shape)
603            root[var_name][...] = coord_values
604
605        # Write virtual chunk references for data arrays (unchanged path)
606        for key, value in data_refs.items():
607            uri, offset, length = value
608            v3_key = _chunk_key_v2_to_v3(key)
609            store.set_virtual_ref(
610                v3_key,
611                uri,
612                offset=int(offset),
613                length=int(length),
614                validate_container=True,
615            )
616
617    def _write_append(
618        self,
619        refs: Dict[str, Any],
620        store,
621        append_dim: str,
622    ) -> None:
623        """Append new data to an existing store along *append_dim*.
624
625        Opens the existing zarr group and uses zarr's Python API to resize
626        arrays and write new coordinate data, ensuring metadata is stored
627        correctly.
628
629        Parameters
630        ----------
631        refs : dict
632            The ``"refs"`` dict from the new manifest.
633        store : IcechunkStore
634            The Icechunk store to append to.
635        append_dim : str
636            Dimension name along which to append.
637        """
638        import numpy as np
639        import zarr
640
641        # Separate refs by type
642        zarray_refs: Dict[str, str] = {}
643        zattrs_refs: Dict[str, str] = {}
644        data_refs: Dict[str, Any] = {}
645        inline_refs: Dict[str, str] = {}
646
647        for key, value in refs.items():
648            if key in {".zgroup"}:
649                pass
650            elif key.endswith("/.zarray"):
651                var_name = key.rsplit("/.zarray", 1)[0]
652                zarray_refs[var_name] = value
653            elif key.endswith("/.zattrs"):
654                var_name = key.rsplit("/.zattrs", 1)[0]
655                zattrs_refs[var_name] = value
656            elif _is_virtual_ref(value):
657                data_refs[key] = value
658            elif _is_inline_data(value) and not _is_metadata_key(key):
659                inline_refs[key] = value
660
661        # Open the existing group via zarr API
662        root = zarr.open_group(store, mode="r+", zarr_format=3)
663
664        # Group data refs by variable
665        var_data_refs: Dict[str, Dict[str, list]] = {}
666        for key, value in data_refs.items():
667            var_name = key.split("/")[0]
668            var_data_refs.setdefault(var_name, {})[key] = value
669
670        # Extend each data variable and register new virtual refs
671        for var_name, chunk_refs in var_data_refs.items():
672            new_zarray_str = zarray_refs.get(var_name)
673            if new_zarray_str is None:
674                continue
675
676            new_zarray = json.loads(new_zarray_str)
677            new_shape_per_append = new_zarray["shape"]
678
679            # Determine append axis from dimension labels
680            new_zattrs_str = zattrs_refs.get(var_name)
681            dim_labels: list = []
682            if new_zattrs_str:
683                dim_labels = json.loads(new_zattrs_str).get("_ARRAY_DIMENSIONS", [])
684
685            append_axis = dim_labels.index(append_dim) if append_dim in dim_labels else None
686
687            if var_name in root and append_axis is not None:
688                zarr_arr = root[var_name]
689                existing_shape = list(zarr_arr.shape)
690                offset = existing_shape[append_axis]
691
692                # Resize array along append axis
693                new_size = existing_shape[append_axis] + new_shape_per_append[append_axis]
694                new_shape_full = list(existing_shape)
695                new_shape_full[append_axis] = new_size
696                zarr_arr.resize(new_shape_full)
697            else:
698                offset = 0
699
700            # Write virtual refs with adjusted chunk indices
701            for key, value in chunk_refs.items():
702                uri, chunk_offset, length = value
703                if append_axis is not None and offset > 0:
704                    adjusted_key = self._adjust_chunk_key(key, var_name, append_axis, offset)
705                else:
706                    adjusted_key = key
707                v3_key = _chunk_key_v2_to_v3(adjusted_key)
708                store.set_virtual_ref(
709                    v3_key,
710                    uri,
711                    offset=int(chunk_offset),
712                    length=int(length),
713                    validate_container=True,
714                )
715
716        # Handle coordinate arrays
717        for key, inline_value in inline_refs.items():
718            coord_name = key.split("/")[0] if "/" in key else None
719            if coord_name is None:
720                continue
721
722            raw_bytes = _decode_inline_value(inline_value)
723            new_zarray_str = zarray_refs.get(coord_name)
724            if new_zarray_str is None:
725                continue
726            dtype = np.dtype(json.loads(new_zarray_str)["dtype"])
727            new_values = np.frombuffer(raw_bytes, dtype=dtype)
728
729            if coord_name == append_dim and coord_name in root:
730                # Concatenate with existing coordinate values
731                existing_arr = root[coord_name][...]
732                combined = np.concatenate([existing_arr, new_values])
733                root[coord_name].resize(len(combined))
734                root[coord_name][...] = combined
735            elif coord_name not in root:
736                # Create new coordinate array via zarr API
737                v2_zattrs = json.loads(zattrs_refs[coord_name]) if coord_name in zattrs_refs else {}
738                dim_names = v2_zattrs.get("_ARRAY_DIMENSIONS", [])
739                shape = [len(new_values)]
740                root.create_array(
741                    coord_name,
742                    shape=shape,
743                    chunks=shape,
744                    dtype=dtype,
745                    fill_value=0,
746                    dimension_names=dim_names if dim_names else None,
747                    attributes=v2_zattrs if v2_zattrs else None,
748                )
749                root[coord_name][...] = new_values
750            # else: coordinate already exists and isn't the append dim — leave as-is
751
752    def _adjust_chunk_key(
753        self,
754        key: str,
755        var_name: str,
756        append_axis: int,
757        offset: int,
758    ) -> str:
759        """Adjust a chunk key's index along the append axis.
760
761        Parameters
762        ----------
763        key : str
764            Original chunk key like ``"TMP/0.0.0.0"``.
765        var_name : str
766            Variable name prefix.
767        append_axis : int
768            Axis index along which to offset.
769        offset : int
770            Number of existing chunks along the append axis.
771
772        Returns
773        -------
774        str
775            Adjusted chunk key (still in Zarr v2 format — caller
776            converts to v3).
777        """
778        # Extract the index portion after the variable name
779        prefix = var_name + "/"
780        if not key.startswith(prefix):
781            return key
782
783        index_str = key[len(prefix) :]
784        indices = index_str.split(".")
785        if append_axis < len(indices):
786            indices[append_axis] = str(int(indices[append_axis]) + offset)
787        return prefix + ".".join(indices)
788
789    def _extend_coord(
790        self,
791        store,
792        coord_name: str,
793        data_key: str,
794        new_value: str,
795        zarray_refs: Dict[str, str],
796        zattrs_refs: Dict[str, str],
797    ) -> None:
798        """Extend a coordinate array with new values for append mode.
799
800        Parameters
801        ----------
802        store : IcechunkStore
803            The Icechunk store.
804        coord_name : str
805            Coordinate name (e.g. ``"refDate"``).
806        data_key : str
807            The data chunk key in v2 format (e.g. ``"refDate/0"``).
808        new_value : str
809            The new inline value (base64-encoded).
810        zarray_refs : dict
811            Variable name -> .zarray JSON string from the new manifest.
812        zattrs_refs : dict
813            Variable name -> .zattrs JSON string from the new manifest.
814        """
815        import numpy as np
816
817        new_bytes = _decode_inline_value(new_value)
818        v3_data_key = _chunk_key_v2_to_v3(data_key)
819
820        # Try to read existing coordinate data
821        existing_bytes = None
822        try:
823            existing_bytes = _store_get_sync(store, v3_data_key)
824        except Exception:
825            pass
826
827        if existing_bytes is not None:
828            # Read existing zarr.json to get dtype
829            existing_v3_bytes = None
830            try:
831                existing_v3_bytes = _store_get_sync(store, f"{coord_name}/zarr.json")
832            except Exception:
833                pass
834
835            dtype = np.float64
836            if existing_v3_bytes is not None:
837                existing_v3_meta = json.loads(existing_v3_bytes.decode("utf-8"))
838                dt_str = existing_v3_meta.get("data_type", "float64")
839                dtype_map = {
840                    "float32": np.float32,
841                    "float64": np.float64,
842                    "int32": np.int32,
843                    "int64": np.int64,
844                }
845                dtype = dtype_map.get(dt_str, np.float64)
846
847            # Concatenate existing and new coordinate values
848            existing_arr = np.frombuffer(existing_bytes, dtype=dtype)
849            new_arr = np.frombuffer(new_bytes, dtype=dtype)
850            combined = np.concatenate([existing_arr, new_arr])
851
852            # Write combined data
853            _store_set_sync(store, v3_data_key, combined.tobytes())
854
855            # Update zarr.json with new shape
856            new_zarray_str = zarray_refs.get(coord_name)
857            if new_zarray_str:
858                coord_zarray = json.loads(new_zarray_str)
859            else:
860                # Reconstruct from existing v3 metadata
861                coord_zarray = {
862                    "shape": [len(combined)],
863                    "chunks": [len(combined)],
864                    "dtype": "<f8",
865                    "fill_value": "NaN",
866                    "order": "C",
867                    "zarr_format": 2,
868                    "compressor": None,
869                }
870
871            coord_zarray["shape"] = [len(combined)]
872            coord_zarray["chunks"] = [len(combined)]
873
874            zattrs_str = zattrs_refs.get(coord_name)
875            v3_array = _make_zarr_v3_array_metadata(json.dumps(coord_zarray), zattrs_str)
876            _store_set_sync(
877                store,
878                f"{coord_name}/zarr.json",
879                v3_array.encode("utf-8"),
880            )
881        else:
882            # No existing data — write as-is
883            _store_set_sync(store, v3_data_key, new_bytes)
884            zarray_str = zarray_refs.get(coord_name)
885            zattrs_str = zattrs_refs.get(coord_name)
886            if zarray_str:
887                v3_array = _make_zarr_v3_array_metadata(zarray_str, zattrs_str)
888                _store_set_sync(
889                    store,
890                    f"{coord_name}/zarr.json",
891                    v3_array.encode("utf-8"),
892                )
893
894    def commit(self, message: str = "") -> str:
895        """Commit the current transaction.
896
897        Parameters
898        ----------
899        message : str
900            Commit message describing the changes.
901
902        Returns
903        -------
904        str
905            The snapshot ID of the committed transaction.
906
907        Raises
908        ------
909        RuntimeError
910            If no session is active (i.e. :meth:`write` has not been
911            called).
912        """
913        if self._session is None:
914            raise RuntimeError("No active session. Call write() before commit().")
915        snapshot_id = self._session.commit(message)
916        return str(snapshot_id)
class IcechunkWriter:
330class IcechunkWriter:
331    """Write grib2io reference manifests into an Icechunk virtual store.
332
333    Parameters
334    ----------
335    store_path : str
336        Path or URI for the Icechunk store.  For local filesystem stores
337        this is a directory path.  For cloud stores, provide the
338        appropriate URI (e.g. ``s3://bucket/prefix``).
339    storage_config : optional
340        An Icechunk ``Storage`` object.  If ``None`` (the default), a
341        local filesystem storage is created at *store_path*.
342    """
343
344    def __init__(
345        self,
346        store_path: str,
347        storage_config: Optional[Any] = None,
348    ):
349        _ensure_icechunk()
350
351        self._store_path = store_path
352        self._storage_config = storage_config
353        self._repo = None
354        self._session = None
355
356    # ------------------------------------------------------------------
357    # Internal helpers
358    # ------------------------------------------------------------------
359
360    def _get_storage(self):
361        """Return an Icechunk ``Storage`` object."""
362        if self._storage_config is not None:
363            return self._storage_config
364
365        import icechunk
366
367        return icechunk.local_filesystem_storage(path=self._store_path)
368
369    def _create_or_open_repo(self, mode: str, virtual_prefixes: Set[str]):
370        """Create or open an Icechunk repository.
371
372        Parameters
373        ----------
374        mode : str
375            ``'w'`` to create a new repo (or overwrite), ``'a'`` to open
376            an existing repo for appending.
377        virtual_prefixes : set of str
378            URI prefixes that need to be registered as virtual chunk
379            containers.
380        """
381        import icechunk
382
383        storage = self._get_storage()
384
385        # Build repository config with virtual chunk containers
386        config = icechunk.config.RepositoryConfig.default()
387        for prefix in virtual_prefixes:
388            container_store = self._make_container_store(prefix)
389            config.set_virtual_chunk_container(icechunk.virtual.VirtualChunkContainer(prefix, container_store))
390
391        # Build authorize_virtual_chunk_access mapping
392        # Use None for credentials (will use environment or anonymous)
393        authorize = {prefix: None for prefix in virtual_prefixes}
394
395        if mode == "w":
396            self._repo = icechunk.Repository.open_or_create(
397                storage,
398                config=config,
399                authorize_virtual_chunk_access=authorize if authorize else None,
400            )
401        else:
402            # Append mode — open existing
403            self._repo = icechunk.Repository.open(
404                storage,
405                config=config,
406                authorize_virtual_chunk_access=authorize if authorize else None,
407            )
408
409        self._session = self._repo.writable_session("main")
410
411    def _make_container_store(self, prefix: str):
412        """Create an appropriate Icechunk storage backend for a virtual
413        chunk container based on the URI prefix scheme.
414
415        Parameters
416        ----------
417        prefix : str
418            URI prefix like ``file:///path/to/dir/`` or
419            ``s3://bucket/prefix/``.
420
421        Returns
422        -------
423        An Icechunk storage store object.
424        """
425        import icechunk
426
427        if prefix.startswith("file://"):
428            # Extract local path from file:// URI
429            local_path = prefix[7:]  # Remove "file://"
430            # Remove trailing slash for the store path
431            if local_path.endswith("/"):
432                local_path = local_path[:-1]
433            return icechunk.storage.local_filesystem_store(local_path)
434        elif prefix.startswith("s3://"):
435            return icechunk.storage.s3_store(region="us-east-1")
436        elif prefix.startswith("gcs://"):
437            return icechunk.storage.gcs_store(opts={})
438        elif prefix.startswith("http://") or prefix.startswith("https://"):
439            return icechunk.storage.http_store(opts={})
440        else:
441            # Default to local filesystem
442            return icechunk.storage.local_filesystem_store(prefix)
443
444    # ------------------------------------------------------------------
445    # Public methods
446    # ------------------------------------------------------------------
447
448    def write(
449        self,
450        manifest: dict,
451        mode: str = "w",
452        append_dim: Optional[str] = None,
453    ) -> None:
454        """Write a reference manifest into the Icechunk store.
455
456        Parameters
457        ----------
458        manifest : dict
459            Kerchunk v1 reference manifest dict with ``"version"`` and
460            ``"refs"`` keys.
461        mode : str
462            ``'w'`` for create/overwrite, ``'a'`` for append.
463        append_dim : str, optional
464            Dimension along which to append when ``mode='a'``.  Required
465            when appending multi-file data along a specific dimension
466            (e.g. ``"refDate"`` or ``"leadTime"``).
467        """
468        refs = manifest.get("refs", {})
469
470        # Collect virtual chunk URI prefixes for container registration
471        virtual_prefixes = _collect_virtual_chunk_prefixes(refs)
472
473        # Create or open the repository
474        self._create_or_open_repo(mode, virtual_prefixes)
475
476        store = self._session.store
477
478        if mode == "a" and append_dim is not None:
479            self._write_append(refs, store, append_dim)
480        else:
481            self._write_create(refs, store)
482
483    def _write_create(self, refs: Dict[str, Any], store) -> None:
484        """Write all refs to a fresh store using zarr's Python API.
485
486        Creates the root group and all arrays via ``zarr.open_group`` /
487        ``group.create_array`` so that Zarr v3 metadata is stored through
488        the proper path that icechunk and zarr recognise.  Coordinate
489        (inline) data is written via the zarr Array write API, and virtual
490        chunk refs for data variables are registered with
491        ``store.set_virtual_ref``.
492
493        Parameters
494        ----------
495        refs : dict
496            The ``"refs"`` dict from the manifest.
497        store : IcechunkStore
498            The Icechunk store to write to.
499        """
500        import numpy as np
501        import zarr
502
503        # Collect all metadata by variable
504        zarray_refs: Dict[str, str] = {}  # var_name -> .zarray JSON str
505        zattrs_refs: Dict[str, str] = {}  # var_name -> .zattrs JSON str
506        data_refs: Dict[str, Any] = {}  # chunk key -> [uri, offset, len]
507        inline_data_refs: Dict[str, str] = {}  # chunk key -> inline value
508
509        for key, value in refs.items():
510            if key == ".zgroup":
511                pass  # zarr.open_group handles the root group metadata
512            elif key.endswith("/.zarray"):
513                var_name = key.rsplit("/.zarray", 1)[0]
514                zarray_refs[var_name] = value
515            elif key.endswith("/.zattrs"):
516                var_name = key.rsplit("/.zattrs", 1)[0]
517                zattrs_refs[var_name] = value
518            elif _is_virtual_ref(value):
519                data_refs[key] = value
520            elif _is_inline_data(value) and not _is_metadata_key(key):
521                inline_data_refs[key] = value
522            elif _is_metadata_key(key):
523                pass  # skip other metadata keys
524            else:
525                _logger.warning(
526                    "Skipping unrecognized manifest entry: %s = %r",
527                    key,
528                    value,
529                )
530
531        # Create root group and all arrays via zarr's Python API so that
532        # metadata is stored in the location that zarr/icechunk actually
533        # reads, rather than as raw chunk bytes at the zarr.json key.
534        root = zarr.open_group(store, mode="w", zarr_format=3)
535
536        # Identify which variables have virtual (data) refs — those need
537        # Grib2SerializerCodec as their zarr v3 serializer.
538        data_var_names: set = {key.split("/")[0] for key in data_refs}
539
540        # Import the zarr v3 serializer codec (registered at import time)
541        from grib2io.codecs import Grib2SerializerCodec as _Grib2Ser
542
543        for var_name, zarray_str in zarray_refs.items():
544            v2_zarray = json.loads(zarray_str)
545            zattrs_str = zattrs_refs.get(var_name)
546            v2_zattrs = json.loads(zattrs_str) if zattrs_str else {}
547
548            shape = v2_zarray["shape"]
549            chunks = v2_zarray.get("chunks", shape)
550            dtype_str = v2_zarray.get("dtype", "<f4")
551            fill_value = v2_zarray.get("fill_value", None)
552            dim_names = v2_zattrs.get("_ARRAY_DIMENSIONS", [])
553
554            dtype = np.dtype(dtype_str)
555            if fill_value == "NaN" or fill_value is None:
556                fv: Any = float("nan") if np.issubdtype(dtype, np.floating) else 0
557            else:
558                fv = fill_value
559
560            if var_name in data_var_names:
561                # Data variable — use Grib2SerializerCodec so zarr v3 can
562                # decode the raw GRIB2 section 7 bytes from virtual chunks.
563                compressor = v2_zarray.get("compressor", {}) or {}
564                codec_cfg = {k: v for k, v in compressor.items() if k != "id"}
565                serializer = _Grib2Ser(**codec_cfg)
566                root.create_array(
567                    var_name,
568                    shape=shape,
569                    chunks=chunks,
570                    dtype=dtype,
571                    fill_value=fv,
572                    serializer=serializer,
573                    compressors=[],
574                    filters=[],
575                    dimension_names=dim_names if dim_names else None,
576                    attributes=v2_zattrs if v2_zattrs else None,
577                )
578            else:
579                # Coordinate array — plain bytes serializer (default)
580                root.create_array(
581                    var_name,
582                    shape=shape,
583                    chunks=chunks,
584                    dtype=dtype,
585                    fill_value=fv,
586                    dimension_names=dim_names if dim_names else None,
587                    attributes=v2_zattrs if v2_zattrs else None,
588                )
589
590        # Write coordinate (inline) data via zarr's Array write API so
591        # that encoding goes through the proper codec pipeline.
592        for key, value in inline_data_refs.items():
593            parts = key.split("/")
594            var_name = parts[0]
595            if var_name not in zarray_refs:
596                _logger.warning("Inline data for unknown variable %s, skipping", var_name)
597                continue
598
599            raw_bytes = _decode_inline_value(value)
600            v2_zarray = json.loads(zarray_refs[var_name])
601            dtype = np.dtype(v2_zarray["dtype"])
602            shape = v2_zarray["shape"]
603            coord_values = np.frombuffer(raw_bytes, dtype=dtype).reshape(shape)
604            root[var_name][...] = coord_values
605
606        # Write virtual chunk references for data arrays (unchanged path)
607        for key, value in data_refs.items():
608            uri, offset, length = value
609            v3_key = _chunk_key_v2_to_v3(key)
610            store.set_virtual_ref(
611                v3_key,
612                uri,
613                offset=int(offset),
614                length=int(length),
615                validate_container=True,
616            )
617
618    def _write_append(
619        self,
620        refs: Dict[str, Any],
621        store,
622        append_dim: str,
623    ) -> None:
624        """Append new data to an existing store along *append_dim*.
625
626        Opens the existing zarr group and uses zarr's Python API to resize
627        arrays and write new coordinate data, ensuring metadata is stored
628        correctly.
629
630        Parameters
631        ----------
632        refs : dict
633            The ``"refs"`` dict from the new manifest.
634        store : IcechunkStore
635            The Icechunk store to append to.
636        append_dim : str
637            Dimension name along which to append.
638        """
639        import numpy as np
640        import zarr
641
642        # Separate refs by type
643        zarray_refs: Dict[str, str] = {}
644        zattrs_refs: Dict[str, str] = {}
645        data_refs: Dict[str, Any] = {}
646        inline_refs: Dict[str, str] = {}
647
648        for key, value in refs.items():
649            if key in {".zgroup"}:
650                pass
651            elif key.endswith("/.zarray"):
652                var_name = key.rsplit("/.zarray", 1)[0]
653                zarray_refs[var_name] = value
654            elif key.endswith("/.zattrs"):
655                var_name = key.rsplit("/.zattrs", 1)[0]
656                zattrs_refs[var_name] = value
657            elif _is_virtual_ref(value):
658                data_refs[key] = value
659            elif _is_inline_data(value) and not _is_metadata_key(key):
660                inline_refs[key] = value
661
662        # Open the existing group via zarr API
663        root = zarr.open_group(store, mode="r+", zarr_format=3)
664
665        # Group data refs by variable
666        var_data_refs: Dict[str, Dict[str, list]] = {}
667        for key, value in data_refs.items():
668            var_name = key.split("/")[0]
669            var_data_refs.setdefault(var_name, {})[key] = value
670
671        # Extend each data variable and register new virtual refs
672        for var_name, chunk_refs in var_data_refs.items():
673            new_zarray_str = zarray_refs.get(var_name)
674            if new_zarray_str is None:
675                continue
676
677            new_zarray = json.loads(new_zarray_str)
678            new_shape_per_append = new_zarray["shape"]
679
680            # Determine append axis from dimension labels
681            new_zattrs_str = zattrs_refs.get(var_name)
682            dim_labels: list = []
683            if new_zattrs_str:
684                dim_labels = json.loads(new_zattrs_str).get("_ARRAY_DIMENSIONS", [])
685
686            append_axis = dim_labels.index(append_dim) if append_dim in dim_labels else None
687
688            if var_name in root and append_axis is not None:
689                zarr_arr = root[var_name]
690                existing_shape = list(zarr_arr.shape)
691                offset = existing_shape[append_axis]
692
693                # Resize array along append axis
694                new_size = existing_shape[append_axis] + new_shape_per_append[append_axis]
695                new_shape_full = list(existing_shape)
696                new_shape_full[append_axis] = new_size
697                zarr_arr.resize(new_shape_full)
698            else:
699                offset = 0
700
701            # Write virtual refs with adjusted chunk indices
702            for key, value in chunk_refs.items():
703                uri, chunk_offset, length = value
704                if append_axis is not None and offset > 0:
705                    adjusted_key = self._adjust_chunk_key(key, var_name, append_axis, offset)
706                else:
707                    adjusted_key = key
708                v3_key = _chunk_key_v2_to_v3(adjusted_key)
709                store.set_virtual_ref(
710                    v3_key,
711                    uri,
712                    offset=int(chunk_offset),
713                    length=int(length),
714                    validate_container=True,
715                )
716
717        # Handle coordinate arrays
718        for key, inline_value in inline_refs.items():
719            coord_name = key.split("/")[0] if "/" in key else None
720            if coord_name is None:
721                continue
722
723            raw_bytes = _decode_inline_value(inline_value)
724            new_zarray_str = zarray_refs.get(coord_name)
725            if new_zarray_str is None:
726                continue
727            dtype = np.dtype(json.loads(new_zarray_str)["dtype"])
728            new_values = np.frombuffer(raw_bytes, dtype=dtype)
729
730            if coord_name == append_dim and coord_name in root:
731                # Concatenate with existing coordinate values
732                existing_arr = root[coord_name][...]
733                combined = np.concatenate([existing_arr, new_values])
734                root[coord_name].resize(len(combined))
735                root[coord_name][...] = combined
736            elif coord_name not in root:
737                # Create new coordinate array via zarr API
738                v2_zattrs = json.loads(zattrs_refs[coord_name]) if coord_name in zattrs_refs else {}
739                dim_names = v2_zattrs.get("_ARRAY_DIMENSIONS", [])
740                shape = [len(new_values)]
741                root.create_array(
742                    coord_name,
743                    shape=shape,
744                    chunks=shape,
745                    dtype=dtype,
746                    fill_value=0,
747                    dimension_names=dim_names if dim_names else None,
748                    attributes=v2_zattrs if v2_zattrs else None,
749                )
750                root[coord_name][...] = new_values
751            # else: coordinate already exists and isn't the append dim — leave as-is
752
753    def _adjust_chunk_key(
754        self,
755        key: str,
756        var_name: str,
757        append_axis: int,
758        offset: int,
759    ) -> str:
760        """Adjust a chunk key's index along the append axis.
761
762        Parameters
763        ----------
764        key : str
765            Original chunk key like ``"TMP/0.0.0.0"``.
766        var_name : str
767            Variable name prefix.
768        append_axis : int
769            Axis index along which to offset.
770        offset : int
771            Number of existing chunks along the append axis.
772
773        Returns
774        -------
775        str
776            Adjusted chunk key (still in Zarr v2 format — caller
777            converts to v3).
778        """
779        # Extract the index portion after the variable name
780        prefix = var_name + "/"
781        if not key.startswith(prefix):
782            return key
783
784        index_str = key[len(prefix) :]
785        indices = index_str.split(".")
786        if append_axis < len(indices):
787            indices[append_axis] = str(int(indices[append_axis]) + offset)
788        return prefix + ".".join(indices)
789
790    def _extend_coord(
791        self,
792        store,
793        coord_name: str,
794        data_key: str,
795        new_value: str,
796        zarray_refs: Dict[str, str],
797        zattrs_refs: Dict[str, str],
798    ) -> None:
799        """Extend a coordinate array with new values for append mode.
800
801        Parameters
802        ----------
803        store : IcechunkStore
804            The Icechunk store.
805        coord_name : str
806            Coordinate name (e.g. ``"refDate"``).
807        data_key : str
808            The data chunk key in v2 format (e.g. ``"refDate/0"``).
809        new_value : str
810            The new inline value (base64-encoded).
811        zarray_refs : dict
812            Variable name -> .zarray JSON string from the new manifest.
813        zattrs_refs : dict
814            Variable name -> .zattrs JSON string from the new manifest.
815        """
816        import numpy as np
817
818        new_bytes = _decode_inline_value(new_value)
819        v3_data_key = _chunk_key_v2_to_v3(data_key)
820
821        # Try to read existing coordinate data
822        existing_bytes = None
823        try:
824            existing_bytes = _store_get_sync(store, v3_data_key)
825        except Exception:
826            pass
827
828        if existing_bytes is not None:
829            # Read existing zarr.json to get dtype
830            existing_v3_bytes = None
831            try:
832                existing_v3_bytes = _store_get_sync(store, f"{coord_name}/zarr.json")
833            except Exception:
834                pass
835
836            dtype = np.float64
837            if existing_v3_bytes is not None:
838                existing_v3_meta = json.loads(existing_v3_bytes.decode("utf-8"))
839                dt_str = existing_v3_meta.get("data_type", "float64")
840                dtype_map = {
841                    "float32": np.float32,
842                    "float64": np.float64,
843                    "int32": np.int32,
844                    "int64": np.int64,
845                }
846                dtype = dtype_map.get(dt_str, np.float64)
847
848            # Concatenate existing and new coordinate values
849            existing_arr = np.frombuffer(existing_bytes, dtype=dtype)
850            new_arr = np.frombuffer(new_bytes, dtype=dtype)
851            combined = np.concatenate([existing_arr, new_arr])
852
853            # Write combined data
854            _store_set_sync(store, v3_data_key, combined.tobytes())
855
856            # Update zarr.json with new shape
857            new_zarray_str = zarray_refs.get(coord_name)
858            if new_zarray_str:
859                coord_zarray = json.loads(new_zarray_str)
860            else:
861                # Reconstruct from existing v3 metadata
862                coord_zarray = {
863                    "shape": [len(combined)],
864                    "chunks": [len(combined)],
865                    "dtype": "<f8",
866                    "fill_value": "NaN",
867                    "order": "C",
868                    "zarr_format": 2,
869                    "compressor": None,
870                }
871
872            coord_zarray["shape"] = [len(combined)]
873            coord_zarray["chunks"] = [len(combined)]
874
875            zattrs_str = zattrs_refs.get(coord_name)
876            v3_array = _make_zarr_v3_array_metadata(json.dumps(coord_zarray), zattrs_str)
877            _store_set_sync(
878                store,
879                f"{coord_name}/zarr.json",
880                v3_array.encode("utf-8"),
881            )
882        else:
883            # No existing data — write as-is
884            _store_set_sync(store, v3_data_key, new_bytes)
885            zarray_str = zarray_refs.get(coord_name)
886            zattrs_str = zattrs_refs.get(coord_name)
887            if zarray_str:
888                v3_array = _make_zarr_v3_array_metadata(zarray_str, zattrs_str)
889                _store_set_sync(
890                    store,
891                    f"{coord_name}/zarr.json",
892                    v3_array.encode("utf-8"),
893                )
894
895    def commit(self, message: str = "") -> str:
896        """Commit the current transaction.
897
898        Parameters
899        ----------
900        message : str
901            Commit message describing the changes.
902
903        Returns
904        -------
905        str
906            The snapshot ID of the committed transaction.
907
908        Raises
909        ------
910        RuntimeError
911            If no session is active (i.e. :meth:`write` has not been
912            called).
913        """
914        if self._session is None:
915            raise RuntimeError("No active session. Call write() before commit().")
916        snapshot_id = self._session.commit(message)
917        return str(snapshot_id)

Write grib2io reference manifests into an Icechunk virtual store.

Parameters
  • store_path (str): Path or URI for the Icechunk store. For local filesystem stores this is a directory path. For cloud stores, provide the appropriate URI (e.g. s3://bucket/prefix).
  • storage_config (optional): An Icechunk Storage object. If None (the default), a local filesystem storage is created at store_path.
IcechunkWriter(store_path: str, storage_config: Optional[Any] = None)
344    def __init__(
345        self,
346        store_path: str,
347        storage_config: Optional[Any] = None,
348    ):
349        _ensure_icechunk()
350
351        self._store_path = store_path
352        self._storage_config = storage_config
353        self._repo = None
354        self._session = None
def write( self, manifest: dict, mode: str = 'w', append_dim: Optional[str] = None) -> None:
448    def write(
449        self,
450        manifest: dict,
451        mode: str = "w",
452        append_dim: Optional[str] = None,
453    ) -> None:
454        """Write a reference manifest into the Icechunk store.
455
456        Parameters
457        ----------
458        manifest : dict
459            Kerchunk v1 reference manifest dict with ``"version"`` and
460            ``"refs"`` keys.
461        mode : str
462            ``'w'`` for create/overwrite, ``'a'`` for append.
463        append_dim : str, optional
464            Dimension along which to append when ``mode='a'``.  Required
465            when appending multi-file data along a specific dimension
466            (e.g. ``"refDate"`` or ``"leadTime"``).
467        """
468        refs = manifest.get("refs", {})
469
470        # Collect virtual chunk URI prefixes for container registration
471        virtual_prefixes = _collect_virtual_chunk_prefixes(refs)
472
473        # Create or open the repository
474        self._create_or_open_repo(mode, virtual_prefixes)
475
476        store = self._session.store
477
478        if mode == "a" and append_dim is not None:
479            self._write_append(refs, store, append_dim)
480        else:
481            self._write_create(refs, store)

Write a reference manifest into the Icechunk store.

Parameters
  • manifest (dict): Kerchunk v1 reference manifest dict with "version" and "refs" keys.
  • mode (str): 'w' for create/overwrite, 'a' for append.
  • append_dim (str, optional): Dimension along which to append when mode='a'. Required when appending multi-file data along a specific dimension (e.g. "refDate" or "leadTime").
def commit(self, message: str = '') -> str:
895    def commit(self, message: str = "") -> str:
896        """Commit the current transaction.
897
898        Parameters
899        ----------
900        message : str
901            Commit message describing the changes.
902
903        Returns
904        -------
905        str
906            The snapshot ID of the committed transaction.
907
908        Raises
909        ------
910        RuntimeError
911            If no session is active (i.e. :meth:`write` has not been
912            called).
913        """
914        if self._session is None:
915            raise RuntimeError("No active session. Call write() before commit().")
916        snapshot_id = self._session.commit(message)
917        return str(snapshot_id)

Commit the current transaction.

Parameters
  • message (str): Commit message describing the changes.
Returns
  • str: The snapshot ID of the committed transaction.
Raises
  • RuntimeError: If no session is active (i.e. write() has not been called).