from typing import Dict
from typing import List
from ewokscore import execute_graph
from ewoksorange.gui.workflows.owscheme import ows_to_ewoks
try:
from importlib.resources import files as resource_files
except ImportError:
from importlib_resources import files as resource_files
[docs]
def test_mosaic_scan_single_detector_without_qt(tmp_path):
from orangecontrib.ewoksfluo.categories.demo import tutorials
filename = resource_files(tutorials).joinpath("mosaic_mesh_single_detector.ows")
assert_mosaic_scan_single_detector_without_qt(filename, tmp_path)
[docs]
def test_mosaic_scan_single_detector_with_qt(ewoks_orange_canvas, tmp_path):
from orangecontrib.ewoksfluo.categories.demo import tutorials
filename = resource_files(tutorials).joinpath("mosaic_mesh_single_detector.ows")
assert_mosaic_scan_single_detector_with_qt(ewoks_orange_canvas, filename, tmp_path)
[docs]
def assert_mosaic_scan_single_detector_without_qt(filename, tmp_path):
"""Execute workflow after converting it to an ewoks workflow"""
graph = ows_to_ewoks(filename)
outputs = execute_graph(
graph, inputs=get_inputs(tmp_path), outputs=[{"all": True}], merge_outputs=False
)
expected = get_expected_outputs(tmp_path)
label_to_id = {
attrs["label"]: node_id for node_id, attrs in graph.graph.nodes.items()
}
expected = {label_to_id[k]: v for k, v in expected.items()}
assert outputs == expected
[docs]
def assert_mosaic_scan_single_detector_with_qt(ewoks_orange_canvas, filename, tmp_path):
"""Execute workflow using the Qt widgets and signals"""
ewoks_orange_canvas.load_graph(str(filename), inputs=get_inputs(tmp_path))
ewoks_orange_canvas.start_workflow()
ewoks_orange_canvas.wait_widgets(timeout=30)
outputs = dict(ewoks_orange_canvas.iter_output_values())
assert outputs == get_expected_outputs(tmp_path)
[docs]
def get_expected_outputs(tmp_path) -> Dict[str, dict]:
bliss_scan_uri = str(tmp_path / "concat.h5::/1.1")
output_root_uri = str(tmp_path / "output.h5::/1.1")
return {
"Mosaic Mesh (single detector)": {
"config": str(tmp_path / "input.h5::/1.1/theory/configuration/data"),
"detector_name": "mca0",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
"expo_time": 0.1,
"filenames": [str(tmp_path / "input.h5")],
"monitor_name": "I0",
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"scan_ranges": [(1, 6)],
},
"Pick scans": {
"bliss_scan_uris": [
str(tmp_path / f"input.h5::/{i+1}.1") for i in range(6)
],
},
"Concat BLISS scans": {
"bliss_scan_uri": bliss_scan_uri,
},
"Fit scan (single detector)": {
"bliss_scan_uri": bliss_scan_uri,
"detector_name": "mca0",
"output_root_uri": output_root_uri,
"xrf_results_uri": str(tmp_path / "output.h5::/1.1/fit/mca0/results"),
},
"Normalize": {
"bliss_scan_uri": bliss_scan_uri,
"output_root_uri": output_root_uri,
"xrf_results_uri": str(tmp_path / "output.h5::/1.1/norm/results"),
},
"Raw Counters": {
"bliss_scan_uri": bliss_scan_uri,
"output_root_uri": output_root_uri,
"xrf_results_uri": str(tmp_path / "output.h5::/1.1/merge/results"),
},
"Regrid": {
"bliss_scan_uri": bliss_scan_uri,
"output_root_uri": output_root_uri,
"xrf_results_uri": str(tmp_path / "output.h5::/1.1/regrid/results"),
},
}