import os
import json
import numpy as np
import pytest
import h5py
from ewoks import execute_graph
from importlib.resources import files
WORKFLOW_JSON = str(files("ewoksbm32").joinpath("workflows/xrr_workflow.json"))
[docs]
@pytest.fixture
def abs_test_paths():
base = os.path.dirname(__file__)
h5 = os.path.abspath(
os.path.join(base, "data", "run3_CaSO4_H2O_4nm_pos2_measurement.h5")
)
return {"h5": h5}
[docs]
def test_xrr_end_to_end(caplog, abs_test_paths, tmp_path):
caplog.set_level("INFO")
assert os.path.exists(WORKFLOW_JSON), f"Workflow JSON not found: {WORKFLOW_JSON}"
with open(WORKFLOW_JSON, "r") as f:
graph = json.load(f)
attnFactors27keV = np.array(
[
1.00000000e00,
2.66000000e00,
7.02000000e00,
1.86732000e01,
4.92000000e01,
1.30872000e02,
3.45384000e02,
9.18721440e02,
2.42000000e03,
6.43720000e03,
1.69884000e04,
4.51891440e04,
1.19064000e05,
3.16710240e05,
8.35829280e05,
2.22330588e06,
],
dtype=float,
)
out_file = tmp_path / "CaSO4_H2O_4nm_pos2.nx.h5"
overrides = [
# ---------------- LoadRawData ----------------
{"label": "load_raw", "name": "filename", "value": abs_test_paths["h5"]},
{"label": "load_raw", "name": "tth_path", "value": "1.1/measurement/psi"},
{"label": "load_raw", "name": "det_path", "value": "1.1/measurement/det"},
{"label": "load_raw", "name": "mon_path", "value": "1.1/measurement/mon4"},
{"label": "load_raw", "name": "itime", "value": 1},
# optional attenuation correction
{"label": "load_raw", "name": "attn_path", "value": "1.1/measurement/attn"},
{"label": "load_raw", "name": "attn_factors", "value": attnFactors27keV},
# Note: no background passed; LoadRawData will use zeros/ones defaults.
# ---------------- Preprocess ----------------
{"label": "preprocess", "name": "sample_length", "value": 4},
{"label": "preprocess", "name": "energy", "value": 27000}, # eV
{"label": "preprocess", "name": "dark_current", "value": 1203},
{"label": "preprocess", "name": "beam_size", "value": 0.05},
{"label": "preprocess", "name": "tthmin", "value": 0.05},
{"label": "preprocess", "name": "direct_beam_amp", "value": None},
# ---------------- CleanSignal ----------------
{"label": "clean", "name": "qmax", "value": 0.8},
{"label": "clean", "name": "qvaluesRangesToRemove", "value": [[0.36, 0.41]]},
# ---------------- InverseFT ----------------
{"label": "inverse_ft", "name": "zrange", "value": 100},
{"label": "inverse_ft", "name": "peak_distance", "value": None},
{"label": "inverse_ft", "name": "baseline_type", "value": "pchip"},
{"label": "inverse_ft", "name": "peak_prominence", "value": 1e-10},
# ---------------- ExportHDF5 ----------------
{"label": "export", "name": "sample_name", "value": "CaSO4_H2O_4nm_pos2"},
{"label": "export", "name": "filename", "value": str(out_file)},
]
execute_graph(graph, inputs=overrides)
assert out_file.exists(), f"Exporter did not write file: {out_file}"
with h5py.File(out_file, "r") as f:
assert "entry" in f
entry = f["entry"]
assert entry.attrs.get("NX_class") == "NXentry"
assert entry.attrs.get("default") == "data"
assert entry["data"].attrs.get("default") == "delta_rho_bl"
rtth = entry["data"]["R_of_tth"]
assert rtth.attrs.get("NX_class") == "NXdata"
assert "R" in rtth and "tth" in rtth
assert rtth["R"].shape == rtth["tth"].shape
assert rtth["R"].shape[0] > 0
rq = entry["data"]["R_of_q"]
assert rq.attrs.get("NX_class") == "NXdata"
assert "R" in rq and "q" in rq
assert rq["R"].shape == rq["q"].shape
assert rq["R"].shape[0] > 0
dr_bl = entry["data"]["delta_rho_bl"]
assert dr_bl.attrs.get("NX_class") == "NXdata"
assert "delta_rho_bl" in dr_bl and "z" in dr_bl
assert dr_bl["delta_rho_bl"].shape == dr_bl["z"].shape
assert dr_bl["delta_rho_bl"].shape[0] > 0
proc = entry["process"]
for name in ["z", "delta_rho"]:
assert name in proc and proc[name].shape[0] > 0
detgrp = entry["instrument"]["detector"]
assert "det" in detgrp