Source code for ewoksxrpd.tests.test_multi_integrate

from os import PathLike
import numpy
import json
import h5py

from ewoksxrpd.tasks.integrate import MultiConfigIntegrate1D
from ewoksxrpd.tasks.nexus import SaveNexusMultiPattern1D
from .xrpd_theory import Measurement, Setup, xPattern


[docs]def test_multiintegrate1d( imageSetup1SampleA: Measurement, setup1: Setup, xSampleA: xPattern ): configs = [ {"nbpt_rad": 50, "unit": "2th_rad"}, {"nbpt_rad": 50, "unit": "q_A^-1"}, {"nbpt_rad": 100, "unit": "q_A^-1"}, ] inputs = { "image": imageSetup1SampleA.image, "detector": setup1.detector, "geometry": setup1.geometry, "energy": setup1.energy, "integration_options": dict(xSampleA.integration_options), "configs": configs, } task = MultiConfigIntegrate1D(inputs=inputs) task.execute() output_values = task.get_output_values() assert len(output_values["xunits"]) == 3 for unit, config in zip(output_values["xunits"], configs): assert unit == config["unit"] assert len(output_values["x"]) == 3 for x, config in zip(output_values["x"], configs): assert len(x) == config["nbpt_rad"] assert len(output_values["y"]) == 3 for y, config in zip(output_values["y"], configs): assert len(y) == config["nbpt_rad"] assert len(output_values["info"]) == 3 for info, config in zip(output_values["info"], configs): assert info["nbpt_rad"] == config["nbpt_rad"] assert info["unit"] == config["unit"]
[docs]def test_multisave(tmpdir: PathLike, setup1: Setup): npt_list = (50, 100, 200) inputs = { "url": str(tmpdir / "result.h5"), "x_list": [numpy.arange(npt) for npt in npt_list], "y_list": [numpy.random.random(npt) for npt in npt_list], "xunits": ["2th_deg", "q_A^-1", "q_A^-1"], "header_list": [ { "energy": 10.2, "detector": setup1.detector, "geometry": setup1.geometry, "npt_radial": npt, } for npt in npt_list ], } task = SaveNexusMultiPattern1D(inputs=inputs) task.execute() with h5py.File(str(tmpdir / "result.h5")) as root: assert set(root["results"].keys()) == { "measurement", "integrate_0", "integrate_1", "integrate_2", # "instrument", # "dummy", } for i, (npt, unit) in enumerate(zip(npt_list, inputs["xunits"])): nxprocess = root[f"results/integrate_{i}"] numpy.testing.assert_array_equal( nxprocess[f"integrated/{unit.split('_')[0]}"], inputs["x_list"][i], ) numpy.testing.assert_array_equal( nxprocess["integrated/intensity"], inputs["y_list"][i] ) numpy.testing.assert_array_equal( root[f"results/measurement/integrated_{i}"], inputs["y_list"][i] ) config = json.loads(nxprocess["configuration"]["data"][()]) assert set(config) == { "ewoks_version", "energy", "detector", "geometry", "npt_radial", } numpy.testing.assert_array_equal(config["npt_radial"], npt)