Source code for ewoksxrpd.tests.test_sum_integrate

import h5py
import numpy
import pytest
from silx.io.dictdump import dicttonx

from ewoksxrpd.tasks.sum_integrate import SumIntegrationResults
from ewoksxrpd.tasks.utils.data_utils import _is_silx_compatible

NXDATA_METADATA = {
    "@NX_class": "NXdata",
    "@interpretation": "spectrum",
    "@signal": "intensity",
}




@pytest.mark.skipif(not _is_silx_compatible(), reason="Silx is too old")
def test_execution(tmp_path):
    q = numpy.linspace(0.01, 10, 100, dtype=numpy.float64)
    input_data = numpy.ones((2, 100), dtype=numpy.float32)

    # Create input file
    input_filename = tmp_path / "input.h5"
    input_nxdata = {
        **NXDATA_METADATA,
        "q": q,
        "@axes": [".", "q"],
        "q@units": "A^-1",
        "intensity": input_data,
        "points": [0, 1],
    }
    dicttonx(input_nxdata, input_filename, "/entry/process/integrated")

    inputs = {
        "nxdata_url": f"{input_filename}::/entry/process/integrated",
        "output_nxprocess_url": f"{input_filename}::/entry/process_summed",
    }
    task = SumIntegrationResults(inputs={**inputs})
    task.execute()

    assert (
        task.get_output_value("nxdata_url")
        == f"{input_filename}::/entry/process_summed/integrated"
    )

    with h5py.File(input_filename, "r") as h5file:
        input_nxdata = h5file["/entry/process/integrated"]
        output_nxdata = h5file["/entry/process_summed/integrated"]
        numpy.testing.assert_allclose(
            output_nxdata["intensity"][()],
            numpy.ones((100), dtype=numpy.float32) * 2,
        )
        numpy.testing.assert_allclose(input_nxdata["q"][()], output_nxdata["q"][()])
        for name, value in inputs.items():
            stored_value = h5file["/entry/process_summed/configuration"][name][()]
            if isinstance(stored_value, bytes):
                stored_value = stored_value.decode()
            assert stored_value == value





@pytest.mark.skipif(not _is_silx_compatible(), reason="Silx is too old")
def test_execution_2d(tmp_path):
    q = numpy.linspace(0.01, 10, 100, dtype=numpy.float64)
    chi = numpy.linspace(0, 360, 360, dtype=numpy.float64)
    input_data = numpy.ones((2, 360, 100), dtype=numpy.float32)

    input_filename = tmp_path / "input.h5"
    input_nxdata = {
        **NXDATA_METADATA,
        "q": q,
        "q@units": "A^-1",
        "chi": chi,
        "@axes": [".", "chi", "q"],
        "intensity": input_data,
        "points": [0, 1],
    }
    dicttonx(input_nxdata, input_filename, "/entry/process/integrated")

    inputs = {
        "nxdata_url": f"{input_filename}::/entry/process/integrated",
        "output_nxprocess_url": f"{input_filename}::/entry/process_summed",
    }
    task = SumIntegrationResults(inputs={**inputs})
    task.execute()

    with h5py.File(input_filename, "r") as h5file:
        input_nxdata = h5file["/entry/process/integrated"]
        output_nxdata = h5file["/entry/process_summed/integrated"]
        numpy.testing.assert_allclose(
            output_nxdata["intensity"][()],
            numpy.ones((100), dtype=numpy.float32) * 2 * 360,
        )