Source code for ewoksfluo.tests.utils.hdf5

from pathlib import Path
from typing import Any
from typing import Dict
from typing import Union

import h5py
import numpy

from ...io import hdf5




def assert_hdf5_content(file_path: Union[str, Path], expected: Dict[str, Any]) -> None:
    with h5py.File(file_path, "r") as f:
        actual = _read_content(f)

    assert actual == expected



def _read_content(group: hdf5.GroupType) -> Dict[str, Any]:
    info = dict()
    if group.attrs:
        info.update(_read_attrs("", dict(group.attrs)))

    for name, item in group.items():
        if hdf5.is_group(item):
            info[name] = _read_content(item)
        else:
            info[f"{name}@shape"] = item.shape
            if item.attrs:
                info.update(_read_attrs(name, dict(item.attrs)))

    return info


def _read_attrs(name, attrs: Dict[str, Any]) -> Dict[str, Any]:
    info = {}

    for attr_name in (
        "long_name",
        "units",
        "axes",
        "interpretation",
        "NX_class",
        "creator",
    ):
        if attr_name in attrs:
            value = attrs.pop(attr_name)
            if isinstance(value, numpy.ndarray):
                value = value.tolist()
            info[f"{name}@{attr_name}"] = value

    signals = set()
    if "signal" in attrs:
        signals.add(attrs.pop("signal"))

    if "auxiliary_signals" in attrs:
        signals.update(attrs.pop("auxiliary_signals"))

    if signals:
        info[f"{name}@signals"] = signals

    if attrs:
        info[f"{name}@extra_attrs"] = set(attrs)

    return info