from ewoksid02.tasks.normalizationtask import NormalizationTask
from .utils import check_h5groups_common, check_h5groups_equivalent, execute_ewoks
[docs]
def test_eiger2_normalization_cython(
inputs_task_generic,
inputs_task_norm,
tmp_path,
filename_processed_norm_reference,
):
processing_filename_cython = str(tmp_path / "id02test_eiger2_norm_cython.h5")
inputs_cython = {
**inputs_task_generic,
**inputs_task_norm,
"processing_filename": processing_filename_cython,
}
task_norm_cython = NormalizationTask(inputs_cython)
task_norm_cython.run()
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_reference}?path=/entry_0000/PyFAI/result_norm/",
url_test=f"silx://{processing_filename_cython}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_reference,
filename_test=processing_filename_cython,
)
[docs]
def test_eiger2_normalization_cupy(
cupy_available,
inputs_task_generic,
inputs_task_norm,
tmp_path,
filename_processed_norm_reference,
):
if not cupy_available:
return
processing_filename_cupy = str(tmp_path / "id02test_eiger2_norm_cupy.h5")
inputs_cupy = {
**inputs_task_generic,
**inputs_task_norm,
"processing_filename": processing_filename_cupy,
}
inputs_cupy["algorithm"] = "cupy"
task_norm_cupy = NormalizationTask(inputs_cupy)
task_norm_cupy.run()
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_reference}?path=/entry_0000/PyFAI/result_norm/",
url_test=f"silx://{processing_filename_cupy}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_reference,
filename_test=processing_filename_cupy,
)
[docs]
def test_eiger2_normalization_numpy(
cupy_available,
inputs_task_generic,
inputs_task_norm,
tmp_path,
filename_processed_norm_reference,
):
if not cupy_available:
return
processing_filename_numpy = str(tmp_path / "id02test_eiger2_norm_numpy.h5")
inputs_numpy = {
**inputs_task_generic,
**inputs_task_norm,
"processing_filename": processing_filename_numpy,
}
inputs_numpy["algorithm"] = "cupy"
task_norm_numpy = NormalizationTask(inputs_numpy)
task_norm_numpy.run()
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_reference}?path=/entry_0000/PyFAI/result_norm/",
url_test=f"silx://{processing_filename_numpy}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_reference,
filename_test=processing_filename_numpy,
)
[docs]
def test_eiger2_normalization_workflow(
workflow_norm,
inputs_task_generic,
inputs_task_norm,
tmp_path,
filename_processed_norm_reference,
):
inputs = []
for key, value in inputs_task_generic.items():
inputs.append({"name": key, "value": value, "all": True})
for key, value in inputs_task_norm.items():
inputs.append({"name": key, "value": value, "id": "norm"})
filename_processing_norm = str(tmp_path / "id02test_eiger2_norm.h5")
inputs.append(
{"name": "processing_filename", "value": filename_processing_norm, "id": "norm"}
)
_ = execute_ewoks(
graph=workflow_norm,
inputs=inputs,
)
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_reference}?path=/entry_0000/PyFAI/result_norm/",
url_test=f"silx://{filename_processing_norm}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_reference,
filename_test=filename_processing_norm,
)
[docs]
def test_eiger2_normalization_workflow_loop(
workflow_norm,
inputs_task_generic,
inputs_task_norm,
tmp_path,
filename_processed_norm_reference,
):
workflow_norm["links"] = [
{
"source": "norm",
"target": "norm",
"conditions": [{"source_output": "continue_pipeline", "value": True}],
"map_all_data": True,
}
]
inputs = []
for key, value in inputs_task_generic.items():
inputs.append({"name": key, "value": value, "all": True})
for key, value in inputs_task_norm.items():
inputs.append({"name": key, "value": value, "id": "norm"})
filename_processing_norm = str(tmp_path / "id02test_eiger2_norm.h5")
inputs.append(
{"name": "processing_filename", "value": filename_processing_norm, "id": "norm"}
)
inputs.append({"name": "reading_node", "value": True, "id": "norm"})
_ = execute_ewoks(
graph=workflow_norm,
inputs=inputs,
)
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_reference}?path=/entry_0000/PyFAI/result_norm/",
url_test=f"silx://{filename_processing_norm}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_reference,
filename_test=filename_processing_norm,
)
[docs]
def test_eiger2_normalization_subscan2(
inputs_task_generic_subscan2,
inputs_task_norm,
tmp_path,
filename_processed_norm_dahu_subscan2,
):
processing_filename_cython = str(tmp_path / "id02test_eiger2_norm_subscan2.h5")
inputs_cython = {
**inputs_task_generic_subscan2,
**inputs_task_norm,
"processing_filename": processing_filename_cython,
}
task_norm_cython = NormalizationTask(inputs_cython)
task_norm_cython.run()
check_h5groups_equivalent(
url_reference=f"silx://{filename_processed_norm_dahu_subscan2}?path=/entry_0000/PyFAI/result_norm/&slice=0,2",
url_test=f"silx://{processing_filename_cython}?path=/entry_0000/PyFAI/result_norm/",
)
check_h5groups_common(
filename_reference=filename_processed_norm_dahu_subscan2,
filename_test=processing_filename_cython,
)