from pathlib import Path
from typing import Any
from typing import Tuple
import numpy
import pytest
from silx.io import dictdump
from silx.io import h5py_utils
from ..io.hdf5 import split_h5uri
from ..tasks import hdf5_utils
[docs]
@pytest.fixture()
def dummy_bliss_scan(tmp_path: Path) -> Tuple[str, dict]:
filename = tmp_path / "raw_data.py"
samy = numpy.arange(0, 2 * numpy.pi, 360)
diode1 = numpy.sin(numpy.arange(0, 2 * numpy.pi, 360))
data = {
"@NX_class": "NXroot",
"1.1": {
"@NX_class": "NXentry",
"instrument": {
"@NX_class": "NXinstrument",
"diode1": {
"@NX_class": "NXdetector",
"data": diode1,
},
"samy": {
"@NX_class": "NXpositioner",
"value": samy,
},
"positioners": {"@NX_class": "NXCollection", ">samy": "../samy/value"},
},
"sample": {
"@NX_class": "NXsample",
},
"measurement": {
"@NX_class": "NXcollection",
">diode1": "../instrument/diode1/data",
">samy": "../instrument/samy/value",
},
},
}
dictdump.dicttonx(data, filename, "/")
expected = {
"@NX_class": "NXroot",
"1.1": {
"@NX_class": "NXentry",
"instrument": {
"@NX_class": "NXinstrument",
"diode1": {"@NX_class": "NXdetector", "data": diode1},
"samy": {"@NX_class": "NXpositioner", "value": samy},
"positioners": {"@NX_class": "NXCollection", "samy": samy},
},
"sample": {
"@NX_class": "NXsample",
},
"measurement": {
"@NX_class": "NXcollection",
"diode1": diode1,
"samy": samy,
},
},
}
return f"{filename}::/1.1", expected
[docs]
def test_link_bliss_scan(dummy_bliss_scan: Tuple[str, dict], tmp_path: Path):
bliss_scan_url, expected = dummy_bliss_scan
_, data_h5path = split_h5uri(bliss_scan_url)
entry_name = [s for s in data_h5path.split("/") if s][0]
filename = tmp_path / "processed_data.py"
with h5py_utils.File(filename, mode="a") as root:
data = {
"@NX_class": "NXroot",
entry_name: {
"@NX_class": "NXentry",
"fit": {"@NX_class": "NXprocess"},
},
}
expected[entry_name]["fit"] = data[entry_name]["fit"]
dictdump.dicttonx(data, root)
hdf5_utils.link_bliss_scan(root[entry_name], bliss_scan_url, retry_timeout=0)
_assert_content(filename, expected)
[docs]
def test_link_bliss_scan_same_file(dummy_bliss_scan: Tuple[str, dict]):
bliss_scan_url, expected = dummy_bliss_scan
filename, data_h5path = split_h5uri(bliss_scan_url)
entry_name = [s for s in data_h5path.split("/") if s][0]
with h5py_utils.File(filename, mode="a") as root:
data = {
entry_name: {
"fit": {"@NX_class": "NXprocess"},
},
}
expected[entry_name]["fit"] = data[entry_name]["fit"]
dictdump.dicttonx(data, root)
hdf5_utils.link_bliss_scan(root[entry_name], bliss_scan_url, retry_timeout=0)
_assert_content(filename, expected)
def _assert_content(filename: Path, expected: dict) -> None:
actual = dictdump.nxtodict(
filename, include_attributes=True, dereference_links=True
)
actual = _convert_arrays(actual)
expected = _convert_arrays(expected)
assert actual == expected
def _convert_arrays(obj: Any) -> Any:
if isinstance(obj, numpy.ndarray):
return obj.tolist()
elif isinstance(obj, dict):
return {k: _convert_arrays(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_convert_arrays(v) for v in obj]
else:
return obj