import pytest
from ewoksorange.tests.utils import execute_task
from orangecontrib.ewoksfluo.categories.demo.mesh_single_scan_multi_det import (
OWMeshSingleScanMultiDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mesh_single_scan_single_det import (
OWMeshSingleScanSingleDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mesh_stack_multi_det import (
OWMeshStackMultiDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mesh_stack_single_det import (
OWMeshStackSingleDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mosaic_mesh_multi_det import (
OWMosaicMeshMultiDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mosaic_mesh_single_det import (
OWMosaicMeshSingleDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mosaic_mesh_stack_multi_det import (
OWMosaicMeshStackMultiDetector,
)
from orangecontrib.ewoksfluo.categories.demo.mosaic_mesh_stack_single_det import (
OWMosaicMeshStackSingleDetector,
)
from ..utils import hdf5
_TASK_CLASSES = {
OWMeshSingleScanSingleDetector: {"nscans": 1, "ndetectors": 1, "mosaic": None},
OWMeshSingleScanMultiDetector: {"nscans": 1, "ndetectors": 2, "mosaic": None},
OWMeshStackSingleDetector: {"nscans": 2, "ndetectors": 1, "mosaic": None},
OWMeshStackMultiDetector: {"nscans": 2, "ndetectors": 2, "mosaic": None},
OWMosaicMeshSingleDetector: {"nscans": 1, "ndetectors": 1, "mosaic": (2, 3)},
OWMosaicMeshMultiDetector: {"nscans": 1, "ndetectors": 2, "mosaic": (2, 3)},
OWMosaicMeshStackSingleDetector: {"nscans": 2, "ndetectors": 1, "mosaic": (3, 4)},
OWMosaicMeshStackMultiDetector: {"nscans": 2, "ndetectors": 2, "mosaic": (3, 4)},
}
[docs]
@pytest.mark.parametrize("task_cls", _TASK_CLASSES)
def test_mesh_tasks(tmp_path, task_cls):
_test_example_xrf_scan(tmp_path, task_cls)
def _test_example_xrf_scan(tmp_path, task_cls, widget: bool = False):
params = _TASK_CLASSES[task_cls]
nscans = params["nscans"]
ndetectors = params["ndetectors"]
mosaic = params["mosaic"]
if not widget:
task_cls = task_cls.ewokstaskclass
filename = str(tmp_path / "test.h5")
inputs = {
"output_filename": filename,
"rois": [(100, 200), (300, 600)],
"emission_line_groups": ["Si-K", "Ca-K", "Ce-L", "Fe-K"],
}
if ndetectors > 1:
inputs["ndetectors"] = ndetectors
if nscans > 1:
inputs["nscans"] = nscans
if mosaic:
inputs["mosaic"] = mosaic
npoints = 50 // mosaic[0] * 60 // mosaic[1]
nscans_total = nscans * mosaic[0] * mosaic[1]
else:
npoints = 50 * 60
nscans_total = nscans
for _ in range(2): # Repeat twice to test overwrite
outputs = execute_task(task_cls, inputs)
assert outputs == expected_task_outputs(nscans, ndetectors, mosaic, tmp_path)
expected = expected_h5content(nscans_total, npoints, ndetectors)
hdf5.assert_hdf5_content(filename, expected)
[docs]
def expected_task_outputs(nscans, ndetectors, mosaic, tmp_path):
filename = str(tmp_path / "test.h5")
outputs = {
"config": f"{filename}::/1.1/theory/configuration/data",
"expo_time": 0.1,
"monitor_name": "I0",
}
if mosaic and nscans == 1 and ndetectors == 1:
nmosaic = mosaic[0] * mosaic[1]
return {
"detector_name": "mca0",
"filenames": [filename],
"scan_ranges": [(1, nmosaic)],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if mosaic and nscans == 1 and ndetectors > 1:
nmosaic = mosaic[0] * mosaic[1]
return {
"configs": [f"{filename}::/1.1/theory/configuration/data"] * ndetectors,
"detector_names": ["mca0", "mca1"],
"filenames": [filename],
"scan_ranges": [(1, nmosaic)],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if mosaic and nscans > 1 and ndetectors == 1:
nmosaic = mosaic[0] * mosaic[1]
return {
"detector_name": "mca0",
"filenames": [[filename]] * nscans,
"scan_ranges": [
[(i * nmosaic + 1, (i + 1) * nmosaic)] for i in range(nscans)
],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if mosaic and nscans > 1 and ndetectors > 1:
nmosaic = mosaic[0] * mosaic[1]
return {
"configs": [f"{filename}::/1.1/theory/configuration/data"] * ndetectors,
"detector_names": ["mca0", "mca1"],
"filenames": [[filename]] * nscans,
"scan_ranges": [
[(i * nmosaic + 1, (i + 1) * nmosaic)] for i in range(nscans)
],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if not mosaic and nscans == 1 and ndetectors == 1:
return {
"detector_name": "mca0",
"filename": filename,
"scan_number": 1,
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if not mosaic and nscans > 1 and ndetectors == 1:
return {
"detector_name": "mca0",
"filenames": [filename],
"scan_ranges": [(1, nscans)],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if not mosaic and nscans == 1 and ndetectors > 1:
return {
"configs": [f"{filename}::/1.1/theory/configuration/data"] * ndetectors,
"detector_names": ["mca0", "mca1"],
"filename": filename,
"scan_number": 1,
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
if not mosaic and nscans > 1 and ndetectors > 1:
return {
"configs": [f"{filename}::/1.1/theory/configuration/data"] * ndetectors,
"detector_names": ["mca0", "mca1"],
"filenames": [filename],
"scan_ranges": [(1, nscans)],
"monitor_normalization_template": "1000000/<instrument/{}/data>",
"detector_normalization_template": "0.1/<instrument/{}/live_time>",
**outputs,
}
raise NotImplementedError(f"{nscans=}, {ndetectors=}, {mosaic=}")
[docs]
def expected_h5content(nscans, npoints, ndetectors):
content = {"@NX_class": "NXroot", "@creator": "ewoksfluo"}
mca_detector = {
"@NX_class": "NXdetector",
"data@shape": (npoints, 1024),
"elapsed_time@shape": (npoints,),
"event_count_rate@shape": (npoints,),
"events@shape": (npoints,),
"fractional_dead_time@shape": (npoints,),
"live_time@shape": (npoints,),
"roi1@shape": (npoints,),
"roi2@shape": (npoints,),
"spectrum@shape": (npoints, 1024),
"trigger_count_rate@shape": (npoints,),
"trigger_live_time@shape": (npoints,),
"triggers@shape": (npoints,),
}
mca_meas = {
"mca{detector}@shape": (npoints, 1024),
"mca{detector}_elapsed_time@shape": (npoints,),
"mca{detector}_event_count_rate@shape": (npoints,),
"mca{detector}_events@shape": (npoints,),
"mca{detector}_fractional_dead_time@shape": (npoints,),
"mca{detector}_live_time@shape": (npoints,),
"mca{detector}_roi1@shape": (npoints,),
"mca{detector}_roi2@shape": (npoints,),
"mca{detector}_trigger_count_rate@shape": (npoints,),
"mca{detector}_trigger_live_time@shape": (npoints,),
"mca{detector}_triggers@shape": (npoints,),
}
for scan in range(1, nscans + 1):
scan_content = {
"@NX_class": "NXentry",
"instrument": {
"@NX_class": "NXinstrument",
"I0": {"@NX_class": "NXdetector", "data@shape": (npoints,)},
"positioners": {
"@NX_class": "NXcollection",
"energy@units": "keV",
"energy@shape": (),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
"positioners_start": {
"@NX_class": "NXcollection",
"energy@units": "keV",
"energy@shape": (),
"sampy@units": "um",
"sampy@shape": (),
"sampz@units": "um",
"sampz@shape": (),
},
"positioners_end": {
"@NX_class": "NXcollection",
"energy@units": "keV",
"energy@shape": (),
"sampy@units": "um",
"sampy@shape": (),
"sampz@units": "um",
"sampz@shape": (),
},
"sampy": {
"@NX_class": "NXpositioner",
"value@units": "um",
"value@shape": (npoints,),
},
"sampz": {
"@NX_class": "NXpositioner",
"value@units": "um",
"value@shape": (npoints,),
},
},
"measurement": {
"@NX_class": "NXcollection",
"I0@shape": (npoints,),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
"theory": {
"@NX_class": "NXprocess",
"configuration": {
"@NX_class": "NXnote",
"data@shape": (),
"type@shape": (),
},
"description": {
"@NX_class": "NXnote",
"data@shape": (),
"type@shape": (),
},
"I0_reference@shape": (),
"parameters": {
"@NX_class": "NXdata",
"@axes": ["sampz", "sampy"],
"@signals": {
"Si-K",
"Ca-K",
"Ce-L",
"Fe-K",
"Compton000",
"Peak000",
},
"Ca-K@shape": (npoints,),
"Ce-L@shape": (npoints,),
"Compton000@shape": (npoints,),
"Fe-K@shape": (npoints,),
"Peak000@shape": (npoints,),
"Si-K@shape": (npoints,),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
"parameters_norm": {
"@NX_class": "NXdata",
"@axes": ["sampz", "sampy"],
"@signals": {
"Si-K",
"Ca-K",
"Ce-L",
"Fe-K",
"Compton000",
"Peak000",
},
"Ca-K@shape": (npoints,),
"Ce-L@shape": (npoints,),
"Compton000@shape": (npoints,),
"Fe-K@shape": (npoints,),
"Peak000@shape": (npoints,),
"Si-K@shape": (npoints,),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
"rois": {
"@NX_class": "NXdata",
"@axes": ["sampz", "sampy"],
"@signals": {"roi1", "roi2"},
"roi1@shape": (npoints,),
"roi2@shape": (npoints,),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
"rois_norm": {
"@NX_class": "NXdata",
"@axes": ["sampz", "sampy"],
"@signals": {"roi1", "roi2"},
"roi1@shape": (npoints,),
"roi2@shape": (npoints,),
"sampy@units": "um",
"sampy@shape": (npoints,),
"sampz@units": "um",
"sampz@shape": (npoints,),
},
},
"title@shape": (),
"end_time@shape": (),
}
content[f"{scan}.1"] = scan_content
for detector in range(ndetectors):
scan_content["instrument"][f"mca{detector}"] = mca_detector
add = {k.format(detector=detector): v for k, v in mca_meas.items()}
scan_content["measurement"].update(add)
return content