Source code for ewoksfluo.tests.tutorials.test_mosaic_stack_mdet

from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple

import pytest

from . import assert_results
from . import expected_results




[docs]
@pytest.mark.parametrize("interrupt_fraction", [None, 0.5])
def test_mosaic_stack_mdet_without_qt(tmp_path, interrupt_fraction):
    assert_results.assert_without_qt(
        "mosaic_stack_mdet.ows",
        get_inputs(tmp_path, interrupt_fraction=interrupt_fraction),
        get_expected_outputs(tmp_path),
        get_expected_hdf5(tmp_path),
    )






[docs]
def test_mosaic_stack_mdet_with_qt(ewoks_orange_canvas, tmp_path):
    assert_results.assert_with_qt(
        ewoks_orange_canvas,
        "mosaic_stack_mdet.ows",
        get_inputs(tmp_path),
        get_expected_outputs(tmp_path),
        get_expected_hdf5(tmp_path),
    )






[docs]
def get_inputs(
    tmp_path, interrupt_fraction: Optional[float] = None
) -> List[Dict[str, Any]]:
    return [
        {
            "label": "Mosaic mesh stack (multi detector)",
            "name": "output_filename",
            "value": str(tmp_path / "input.h5"),
        },
        {
            "label": "Mosaic mesh stack (multi detector)",
            "name": "interrupt_fraction",
            "value": interrupt_fraction,
        },
        {
            "label": "Concat BLISS scan groups",
            "name": "output_root_uri",
            "value": str(tmp_path / "concat.h5"),
        },
        {
            "label": "Fit stack (multi detector)",
            "name": "output_root_uri",
            "value": str(tmp_path / "output.h5::/1.1"),
        },
    ]






[docs]
def get_expected_outputs(tmp_path) -> Dict[str, Dict[str, Any]]:
    nscans = 4
    nscans_per_mosaic = 2 * 3
    bliss_scan_uris = [
        str(tmp_path / f"concat.h5::/{i}.1") for i in range(1, nscans + 1)
    ]
    output_root_uri = str(tmp_path / "output.h5::/1.1")
    return {
        "Mosaic mesh stack (multi detector)": {
            "config": str(tmp_path / "input.h5::/1.1/theory/configuration/data"),
            "configs": [str(tmp_path / "input.h5::/1.1/theory/configuration/data")] * 2,
            "detector_names": ["mca0", "mca1"],
            "detector_normalization_template": "0.1/<instrument/{}/live_time>",
            "expo_time": 0.1,
            "filenames": [[str(tmp_path / "input.h5")]] * 4,
            "monitor_name": "I0",
            "monitor_normalization_template": "1000000/<instrument/{}/data>",
            "scan_ranges": [[(1, 6)], [(7, 12)], [(13, 18)], [(19, 24)]],
        },
        "Pick scan groups": {
            "bliss_scan_uris": [
                [
                    str(tmp_path / f"input.h5::/{j*nscans_per_mosaic+i+1}.1")
                    for i in range(nscans_per_mosaic)
                ]
                for j in range(nscans)
            ],
        },
        "Concat BLISS scan groups": {
            "bliss_scan_uris": bliss_scan_uris,
        },
        "Fit stack (multi detector)": {
            "bliss_scan_uris": bliss_scan_uris,
            "detector_names": ["mca0", "mca1"],
            "output_root_uri": output_root_uri,
            "output_root_group": None,
            "xrf_results_uris": [
                str(tmp_path / "output.h5::/1.1/fit/mca0/results"),
                str(tmp_path / "output.h5::/1.1/fit/mca1/results"),
            ],
        },
        "Sum Fit Results Stack": {
            "xrf_results_uri": str(tmp_path / "output.h5::/1.1/sum/results"),
            "bliss_scan_uris": bliss_scan_uris,
            "output_root_uri": output_root_uri,
            "output_root_group": None,
        },
        "Normalize Stack": {
            "bliss_scan_uris": bliss_scan_uris,
            "output_root_uri": output_root_uri,
            "output_root_group": None,
            "xrf_results_uri": str(tmp_path / "output.h5::/1.1/norm/results"),
        },
        "Raw Counters Stack": {
            "bliss_scan_uris": bliss_scan_uris,
            "output_root_uri": output_root_uri,
            "output_root_group": None,
            "xrf_results_uri": str(tmp_path / "output.h5::/1.1/merge/results"),
        },
        "Regrid Stack": {
            "bliss_scan_uris": bliss_scan_uris,
            "output_root_uri": output_root_uri,
            "output_root_group": None,
            "xrf_results_uri": str(tmp_path / "output.h5::/1.1/regrid/results"),
        },
    }






[docs]
def get_expected_hdf5(tmp_path) -> Tuple[str, Dict[str, Any]]:
    output_path = tmp_path / "output.h5"
    datatype = expected_results.DataType(
        nmca=2,
        nmcasum=0,
        nscans=4,
        ndim0=54,
        ndim1=64,
        mosaic=True,
        partial_mca_sum=False,
    )
    expected = expected_results.build_expected(datatype)
    return output_path, expected