Source code for kerchunk.hdf

import base64
import io
import logging
from typing import Union, BinaryIO

import fsspec.core
from fsspec.implementations.reference import LazyReferenceMapper
import numpy as np
import zarr
from zarr.meta import encode_fill_value
import numcodecs

from .codecs import FillStringsCodec
from .utils import _encode_for_JSON

    import h5py
except ModuleNotFoundError:  # pragma: no cover
    raise ImportError(
        "h5py is required for kerchunking HDF5/NetCDF4 files. Please install with "
        "`pip/conda install h5py`. See "
        "for more details."

lggr = logging.getLogger("h5-to-zarr")
_HIDDEN_ATTRS = {  # from h5netcdf.attrs

[docs] class SingleHdf5ToZarr: """Translate the content of one HDF5 file into Zarr metadata. HDF5 groups become Zarr groups. HDF5 datasets become Zarr arrays. Zarr array chunks remain in the HDF5 file. Parameters ---------- h5f : file-like or str Input HDF5 file. Can be a binary Python file-like object (duck-typed, adhering to BinaryIO is optional), in which case must also provide url. If a str, file will be opened using fsspec and storage_options. url : string URI of the HDF5 file, if passing a file-like object or h5py File/Group spec : int The version of output to produce (see README of this repo) inline_threshold : int Include chunks smaller than this value directly in the output. Zero or negative to disable storage_options: dict passed to fsspec if h5f is a str error: "warn" (default) | "pdb" | "ignore" | "raise" vlen_encode: ["embed", "null", "leave", "encode"] What to do with VLEN string variables or columns of tabular variables leave: pass through the 16byte garbage IDs unaffected, but requires no codec null: set all the strings to None or empty; required that this library is available at read time embed: include all the values in the output JSON (should not be used for large tables) encode: save the ID-to-value mapping in a codec, to produce the real values at read time; requires this library to be available. Can be efficient storage where there are few unique values. out: dict-like or None This allows you to supply an fsspec.implementations.reference.LazyReferenceMapper to write out parquet as the references get filled, or some other dictionary-like class to customise how references get stored """ def __init__( self, h5f: "BinaryIO | str | h5py.File | h5py.Group", url: str = None, spec=1, inline_threshold=500, storage_options=None, error="warn", vlen_encode="embed", out=None, ): # Open HDF5 file in read mode... lggr.debug(f"HDF5 file: {h5f}") if isinstance(h5f, str): fs, path = fsspec.core.url_to_fs(h5f, **(storage_options or {})) self.input_file =, "rb") url = h5f self._h5f = h5py.File(self.input_file, mode="r") elif isinstance(h5f, io.IOBase): self.input_file = h5f self._h5f = h5py.File(self.input_file, mode="r") elif isinstance(h5f, (h5py.File, h5py.Group)): # assume h5py object (File or group/dataset) self._h5f = h5f fs, path = fsspec.core.url_to_fs(url, **(storage_options or {})) self.input_file =, "rb") else: raise ValueError("type of input `h5f` not recognised") self.spec = spec self.inline = inline_threshold if vlen_encode not in ["embed", "null", "leave", "encode"]: raise NotImplementedError self.vlen = vlen_encode = out or {} self._zroot =, overwrite=True) self._uri = url self.error = error lggr.debug(f"HDF5 file URI: {self._uri}")
[docs] def translate(self): """Translate content of one HDF5 file into Zarr storage format. This method is the main entry point to execute the workflow, and returns a "reference" structure to be used with zarr/kerchunk No data is copied out of the HDF5 file. Returns ------- dict Dictionary containing reference structure. """ lggr.debug("Translation begins") self._transfer_attrs(self._h5f, self._zroot) self._h5f.visititems(self._translator) if self.spec < 1: return elif isinstance(, LazyReferenceMapper): return else: store = _encode_for_JSON( return {"version": 1, "refs": store}
def _unref(self, ref): name = h5py.h5r.get_name(ref, return self._h5f[name] def _transfer_attrs( self, h5obj: Union[h5py.Dataset, h5py.Group], zobj: Union[zarr.Array, zarr.Group], ): """Transfer attributes from an HDF5 object to its equivalent Zarr object. Parameters ---------- h5obj : h5py.Group or h5py.Dataset An HDF5 group or dataset. zobj : zarr.hierarchy.Group or zarr.core.Array An equivalent Zarr group or array to the HDF5 group or dataset with attributes. """ for n, v in h5obj.attrs.items(): if n in _HIDDEN_ATTRS: continue # Fix some attribute values to avoid JSON encoding exceptions... if isinstance(v, bytes): v = v.decode("utf-8") or " " elif isinstance(v, (np.ndarray, np.number, np.bool_)): if v.dtype.kind == "S": v = v.astype(str) if n == "_FillValue": continue # strip it out! elif v.size == 1: v = v.flatten()[0] if isinstance(v, (np.ndarray, np.number, np.bool_)): v = v.tolist() else: v = v.tolist() elif isinstance(v, h5py._hl.base.Empty): v = "" if v == "DIMENSION_SCALE": continue try: zobj.attrs[n] = v except TypeError: lggr.debug( f"TypeError transferring attr, skipping:\n {n}@{} = {v} ({type(v)})" ) def _decode_filters(self, h5obj: Union[h5py.Dataset, h5py.Group]): if h5obj.scaleoffset: raise RuntimeError( f"{} uses HDF5 scaleoffset filter - not supported by kerchunk" ) if h5obj.compression in ("szip", "lzf"): raise RuntimeError( f"{} uses szip or lzf compression - not supported by kerchunk" ) filters = [] if h5obj.shuffle and h5obj.dtype.kind != "O": # cannot use shuffle if we materialised objects filters.append(numcodecs.Shuffle(elementsize=h5obj.dtype.itemsize)) for filter_id, properties in h5obj._filters.items(): if str(filter_id) == "32001": blosc_compressors = ( "blosclz", "lz4", "lz4hc", "snappy", "zlib", "zstd", ) ( _1, _2, bytes_per_num, total_bytes, clevel, shuffle, compressor, ) = properties pars = dict( blocksize=total_bytes, clevel=clevel, shuffle=shuffle, cname=blosc_compressors[compressor], ) filters.append(numcodecs.Blosc(**pars)) elif str(filter_id) == "32015": filters.append(numcodecs.Zstd(level=properties[0])) elif str(filter_id) == "gzip": filters.append(numcodecs.Zlib(level=properties)) elif str(filter_id) == "32004": raise RuntimeError( f"{} uses lz4 compression - not supported by kerchunk" ) elif str(filter_id) == "32008": raise RuntimeError( f"{} uses bitshuffle compression - not supported by kerchunk" ) elif str(filter_id) == "shuffle": # already handled before this loop pass else: raise RuntimeError( f"{} uses filter id {filter_id} with properties {properties}," f" not supported by kerchunk." ) return filters def _translator(self, name: str, h5obj: Union[h5py.Dataset, h5py.Group]): """Produce Zarr metadata for all groups and datasets in the HDF5 file.""" try: # method must not raise exception kwargs = {} if isinstance(h5obj, h5py.Dataset): lggr.debug(f"HDF5 dataset: {}") lggr.debug(f"HDF5 compression: {h5obj.compression}") if == h5py.h5d.COMPACT: # Only do if h5obj.nbytes < self.inline?? kwargs["data"] = h5obj[:] filters = [] else: filters = self._decode_filters(h5obj) dt = None # Get storage info of this HDF5 dataset... cinfo = self._storage_info(h5obj) if "data" in kwargs: fill = None else: # encodings if h5obj.dtype.kind in "US": fill = h5obj.fillvalue or " " # cannot be None elif h5obj.dtype.kind == "O": if self.vlen == "embed": if np.isscalar(h5obj): out = str(h5obj) elif h5obj.ndim == 0: out = np.array(h5obj).tolist().decode() else: out = h5obj[:] out2 = out.ravel() for i, val in enumerate(out2): if isinstance(val, bytes): out2[i] = val.decode() elif isinstance(val, str): out2[i] = val elif isinstance(val, h5py.h5r.Reference): # TODO: recursively recreate references out2[i] = None else: out2[i] = [ v.decode() if isinstance(v, bytes) else v for v in val ] kwargs["data"] = out kwargs["object_codec"] = numcodecs.JSON() fill = None elif self.vlen == "null": dt = "O" kwargs["object_codec"] = FillStringsCodec(dtype="S16") fill = " " elif self.vlen == "leave": dt = "S16" fill = " " elif self.vlen == "encode": assert len(cinfo) == 1 v = list(cinfo.values())[0] data = _read_block(self.input_file, v["offset"], v["size"]) indexes = np.frombuffer(data, dtype="S16") labels = h5obj[:] mapping = { index.decode(): label.decode() for index, label in zip(indexes, labels) } kwargs["object_codec"] = FillStringsCodec( dtype="S16", id_map=mapping ) fill = " " else: raise NotImplementedError elif _is_netcdf_datetime(h5obj) or _is_netcdf_variable(h5obj): fill = None else: fill = h5obj.fillvalue if h5obj.dtype.kind == "V": fill = None if self.vlen == "encode": assert len(cinfo) == 1 v = list(cinfo.values())[0] dt = [ ( v, ( "S16" if h5obj.dtype[v].kind == "O" else str(h5obj.dtype[v]) ), ) for v in h5obj.dtype.names ] data = _read_block(self.input_file, v["offset"], v["size"]) labels = h5obj[:] arr = np.frombuffer(data, dtype=dt) mapping = {} for field in labels.dtype.names: if labels[field].dtype == "O": mapping.update( { index.decode(): label.decode() for index, label in zip( arr[field], labels[field] ) } ) kwargs["object_codec"] = FillStringsCodec( dtype=str(dt), id_map=mapping ) dt = [ ( v, ( "O" if h5obj.dtype[v].kind == "O" else str(h5obj.dtype[v]) ), ) for v in h5obj.dtype.names ] elif self.vlen == "null": dt = [ ( v, ( "S16" if h5obj.dtype[v].kind == "O" else str(h5obj.dtype[v]) ), ) for v in h5obj.dtype.names ] kwargs["object_codec"] = FillStringsCodec(dtype=str(dt)) dt = [ ( v, ( "O" if h5obj.dtype[v].kind == "O" else str(h5obj.dtype[v]) ), ) for v in h5obj.dtype.names ] elif self.vlen == "leave": dt = [ ( v, ( "S16" if h5obj.dtype[v].kind == "O" else h5obj.dtype[v] ), ) for v in h5obj.dtype.names ] elif self.vlen == "embed": # embed fails due to data = h5obj[:].tolist() data2 = [] for d in data: data2.append( [ ( _.decode(errors="ignore") if isinstance(_, bytes) else _ ) for _ in d ] ) dt = "O" kwargs["data"] = data2 kwargs["object_codec"] = numcodecs.JSON() fill = None else: raise NotImplementedError if h5py.h5ds.is_scale( and not cinfo: return if h5obj.attrs.get("_FillValue") is not None: fill = encode_fill_value( h5obj.attrs.get("_FillValue"), dt or h5obj.dtype ) # Create a Zarr array equivalent to this HDF5 dataset... za = self._zroot.create_dataset(, shape=h5obj.shape, dtype=dt or h5obj.dtype, chunks=h5obj.chunks or False, fill_value=fill, compression=None, filters=filters, overwrite=True, **kwargs, ) lggr.debug(f"Created Zarr array: {za}") self._transfer_attrs(h5obj, za) adims = self._get_array_dims(h5obj) za.attrs["_ARRAY_DIMENSIONS"] = adims lggr.debug(f"_ARRAY_DIMENSIONS = {adims}") if "data" in kwargs: return # embedded bytes, no chunks to copy # Store chunk location metadata... if cinfo: for k, v in cinfo.items(): if h5obj.fletcher32:"Discarding fletcher32 checksum") v["size"] -= 4 if ( self.inline and isinstance(v, dict) and v["size"] < self.inline ):["offset"]) data =["size"]) try: # easiest way to test if data is ascii data.decode("ascii") except UnicodeDecodeError: data = b"base64:" + base64.b64encode(data)[za._chunk_key(k)] = data else:[za._chunk_key(k)] = [ self._uri, v["offset"], v["size"], ] elif isinstance(h5obj, h5py.Group): lggr.debug(f"HDF5 group: {}") zgrp = self._zroot.create_group( self._transfer_attrs(h5obj, zgrp) except Exception as e: import traceback msg = "\n".join( [ "The following excepion was caught and quashed while traversing HDF5", str(e), traceback.format_exc(limit=5), ] ) if self.error == "ignore": return elif self.error == "pdb": print(msg) import pdb pdb.post_mortem() elif self.error == "raise": raise else: # "warn" or anything else, the default import warnings warnings.warn(msg) del e # garbage collect def _get_array_dims(self, dset): """Get a list of dimension scale names attached to input HDF5 dataset. This is required by the xarray package to work with Zarr arrays. Only one dimension scale per dataset dimension is allowed. If dataset is dimension scale, it will be considered as the dimension to itself. Parameters ---------- dset : h5py.Dataset HDF5 dataset. Returns ------- list List with HDF5 path names of dimension scales attached to input dataset. """ dims = list() rank = len(dset.shape) if rank: for n in range(rank): num_scales = len(dset.dims[n]) if num_scales == 1: dims.append(dset.dims[n][0].name[1:]) elif h5py.h5ds.is_scale( dims.append([1:]) elif num_scales > 1: raise RuntimeError( f"{}: {len(dset.dims[n])} " f"dimension scales attached to dimension #{n}" ) elif num_scales == 0: # Some HDF5 files do not have dimension scales. # If this is the case, `num_scales` will be 0. # In this case, we mimic netCDF4 and assign phony dimension names. # See dims.append(f"phony_dim_{n}") return dims def _storage_info(self, dset: h5py.Dataset) -> dict: """Get storage information of an HDF5 dataset in the HDF5 file. Storage information consists of file offset and size (length) for every chunk of the HDF5 dataset. Parameters ---------- dset : h5py.Dataset HDF5 dataset for which to collect storage information. Returns ------- dict HDF5 dataset storage information. Dict keys are chunk array offsets as tuples. Dict values are pairs with chunk file offset and size integers. """ # Empty (null) dataset... if dset.shape is None: return dict() dsid = if dset.chunks is None: # Contiguous dataset... if dsid.get_offset() is None: # No data ever written... return dict() else: key = (0,) * (len(dset.shape) or 1) return { key: {"offset": dsid.get_offset(), "size": dsid.get_storage_size()} } else: # Chunked dataset... num_chunks = dsid.get_num_chunks() if num_chunks == 0: # No data ever written... return dict() # Go over all the dataset chunks... stinfo = dict() chunk_size = dset.chunks def get_key(blob): return tuple([a // b for a, b in zip(blob.chunk_offset, chunk_size)]) def store_chunk_info(blob): stinfo[get_key(blob)] = {"offset": blob.byte_offset, "size": blob.size} has_chunk_iter = callable(getattr(dsid, "chunk_iter", None)) if has_chunk_iter: dsid.chunk_iter(store_chunk_info) else: for index in range(num_chunks): store_chunk_info(dsid.get_chunk_info(index)) return stinfo
def _simple_type(x): if isinstance(x, bytes): return x.decode() if isinstance(x, np.number): if x.dtype.kind == "i": return int(x) return float(x) return x def _read_block(open_file, offset, size): place = open_file.tell() data = return data def _is_netcdf_datetime(dataset: h5py.Dataset): units = dataset.attrs.get("units") if isinstance(units, bytes): units = units.decode("utf-8") # This is the same heuristic used by xarray # return units and "since" in units def _is_netcdf_variable(dataset: h5py.Dataset): return any("_Netcdf4" in _ for _ in dataset.attrs)