Source code for ewoksid02.tests.conftest

import h5py
import pytest
from silx.resources import ExternalResources

from ..utils.blissdata import LIMA_URL_TEMPLATE_ID02, get_lima_url_template_args_id02

WINDOW_ROI_SIZE = 100
INDEX_RANGE = [0, 2]
EIGER2_SHAPE = (2162, 2068)
NBPT_AZIM = 360
NBPT_RAD = 1633
SCAN_NB = 9
SCAN_NB_SUBSCAN2 = 8

URL_BASE = "http://ftp.edna-site.org/ewoks/id02test"
RESOURCES = ExternalResources("ewoksid02", timeout=60, url_base=URL_BASE)

RAW_MASTER_FILE = "RAW_DATA/ewoks/ewoks_20250212-112132/ewoks_20250212-112132.h5"
RAW_LIMA_FILE = "RAW_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00.h5"
RAW_SCALERS_FILE = "RAW_DATA/ewoks/ewoks_20250212-112132/ewoks_scalers_00009_00.h5"

RAW_LIMA_FILE_SUBSCAN2 = "RAW_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00008_00.h5"
RAW_SCALERS_FILE_SUBSCAN2 = (
    "RAW_DATA/ewoks/ewoks_20250212-112132/ewoks_scalers_00008_00.h5"
)

FILENAME_PROCESSED_NORM_REFERENCE = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00_norm.h5"
)
FILENAME_PROCESSED_2SCAT_FULL = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00_full_2scat.h5"
)
FILENAME_PROCESSED_CAVE_FULL = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00_full_cave.h5"
)
FILENAME_PROCESSED_AZIM_FULL = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00_full_azim.h5"
)
FILENAME_PROCESSED_AVE_FULL = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00009_00_full_ave.h5"
)
FILENAME_PROCESSED_NORM_DAHU_SUBSCAN2 = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00008_00_norm.h5"
)
FILENAME_PROCESSED_AZIM_DAHU_SUBSCAN2 = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00008_00_azim.h5"
)
FILENAME_PROCESSED_AVE_DAHU_SUBSCAN2 = (
    "PROCESSED_DATA/ewoks/ewoks_20250212-112132/ewoks_eiger2_00008_00_ave.h5"
)

FILENAME_MASK_GAPS = "PROCESSED_DATA/mask_eiger2_gaps.edf"
FILENAME_MASK_BEAMSTOP = "PROCESSED_DATA/mask_eiger4m_beamstop.edf"
FILENAME_FLATFIELD = "PROCESSED_DATA/flat_eiger2_1b1.edf"
FILENAME_WINDOW_PATTERN = "PROCESSED_DATA/window_norm_cave.h5"
DETECTOR_EIGER2 = "eiger2"



[docs]
@pytest.fixture
def cupy_available():
    try:
        import cupy  # noqa: F401

        return True
    except ImportError:
        return False




[docs]
@pytest.fixture
def index_read():
    return INDEX_RANGE




[docs]
@pytest.fixture
def eiger2_shape():
    return EIGER2_SHAPE




[docs]
@pytest.fixture
def nbpt_rad():
    return NBPT_RAD




[docs]
@pytest.fixture
def nbpt_azim():
    return NBPT_AZIM




[docs]
@pytest.fixture
def filename_raw_lima():
    return RESOURCES.getfile(RAW_LIMA_FILE)




[docs]
@pytest.fixture
def filename_raw_master():
    return RESOURCES.getfile(RAW_MASTER_FILE)




[docs]
@pytest.fixture
def filename_raw_scalers():
    return RESOURCES.getfile(RAW_SCALERS_FILE)




[docs]
@pytest.fixture
def filename_raw_lima_subscan2():
    return RESOURCES.getfile(RAW_LIMA_FILE_SUBSCAN2)




[docs]
@pytest.fixture
def filename_raw_scalers_subscan2():
    return RESOURCES.getfile(RAW_SCALERS_FILE_SUBSCAN2)




[docs]
@pytest.fixture
def filename_processed_norm_reference():
    return RESOURCES.getfile(FILENAME_PROCESSED_NORM_REFERENCE)




[docs]
@pytest.fixture
def filename_processed_2scat_full():
    return RESOURCES.getfile(FILENAME_PROCESSED_2SCAT_FULL)




[docs]
@pytest.fixture
def filename_processed_cave_full():
    return RESOURCES.getfile(FILENAME_PROCESSED_CAVE_FULL)




[docs]
@pytest.fixture
def filename_processed_azim_full():
    return RESOURCES.getfile(FILENAME_PROCESSED_AZIM_FULL)




[docs]
@pytest.fixture
def filename_processed_ave_full():
    return RESOURCES.getfile(FILENAME_PROCESSED_AVE_FULL)




[docs]
@pytest.fixture
def filename_processed_norm_dahu_subscan2():
    return RESOURCES.getfile(FILENAME_PROCESSED_NORM_DAHU_SUBSCAN2)




[docs]
@pytest.fixture
def filename_processed_azim_dahu_subscan2():
    return RESOURCES.getfile(FILENAME_PROCESSED_AZIM_DAHU_SUBSCAN2)




[docs]
@pytest.fixture
def filename_processed_ave_dahu_subscan2():
    return RESOURCES.getfile(FILENAME_PROCESSED_AVE_DAHU_SUBSCAN2)




[docs]
@pytest.fixture
def filename_flatfield():
    return RESOURCES.getfile(FILENAME_FLATFIELD)




[docs]
@pytest.fixture
def filename_mask_gaps():
    return RESOURCES.getfile(FILENAME_MASK_GAPS)




[docs]
@pytest.fixture
def filename_mask_beamstop():
    return RESOURCES.getfile(FILENAME_MASK_BEAMSTOP)




[docs]
@pytest.fixture
def filename_window_pattern():
    return RESOURCES.getfile(FILENAME_WINDOW_PATTERN)




[docs]
@pytest.fixture
def window_roi_size():
    return WINDOW_ROI_SIZE




[docs]
@pytest.fixture
def inputs_task_generic(filename_raw_lima, filename_raw_master):
    return {
        "detector_name": DETECTOR_EIGER2,
        "filename_data": filename_raw_master,
        "filename_lima": filename_raw_lima,
        "Dummy": -10,
        "DDummy": 0.1,
        "scan_nb": SCAN_NB,
        "max_slice_size": 50,
        "lima_url_template": LIMA_URL_TEMPLATE_ID02,
        "lima_url_template_args": get_lima_url_template_args_id02(
            scan_number=SCAN_NB,
            detector_name=DETECTOR_EIGER2,
            collection_name="ewoks",
        ),
        "log_level": "info",
        "datatype": "float32",
        "use_cupy": False,
        "save_variance": True,
        "index_range": INDEX_RANGE,
    }




[docs]
@pytest.fixture
def inputs_task_generic_subscan2(filename_raw_lima_subscan2, filename_raw_master):
    return {
        "detector_name": DETECTOR_EIGER2,
        "filename_data": filename_raw_master,
        "filename_lima": filename_raw_lima_subscan2,
        "Dummy": -10,
        "DDummy": 0.1,
        "scan_nb": SCAN_NB_SUBSCAN2,
        "max_slice_size": 50,
        "lima_url_template": LIMA_URL_TEMPLATE_ID02,
        "lima_url_template_args": get_lima_url_template_args_id02(
            scan_number=SCAN_NB,
            detector_name=DETECTOR_EIGER2,
            collection_name="ewoks",
        ),
        "log_level": "info",
        "datatype": "float32",
        "use_cupy": False,
        "save_variance": True,
        "index_range": INDEX_RANGE,
    }




[docs]
@pytest.fixture
def inputs_task_norm(filename_mask_gaps, filename_flatfield):
    return {
        "filename_mask_normalization": filename_mask_gaps,
        "filename_dark_normalization": None,
        "filename_flat_normalization": filename_flatfield,
        "algorithm_normalization": "cython",
    }




[docs]
@pytest.fixture
def inputs_task_2scat(
    filename_window_pattern, filename_mask_gaps, filename_mask_beamstop
):
    return {
        "filename_window_wagon": filename_window_pattern,
        "WindowRoiSize": WINDOW_ROI_SIZE,
        "filename_mask_static": filename_mask_gaps,
        "filename_mask_reference": filename_mask_beamstop,
        "algorithm_2scat": "numpy",
        "save_secondary_scattering": True,
        "pre_caving": True,
        "flip_caving": False,
    }




[docs]
@pytest.fixture
def inputs_task_cave(filename_mask_beamstop):
    return {
        "filename_mask_static": filename_mask_beamstop,
        "algorithm_cave": "numpy",
        "save_variance": False,
    }




[docs]
@pytest.fixture
def inputs_task_azim(filename_mask_beamstop):
    return {
        "filename_mask_azimuthal": filename_mask_beamstop,
        "method": ("bbox", "csr", "cython"),
    }




[docs]
@pytest.fixture
def inputs_task_ave():
    return {}




[docs]
@pytest.fixture
def dataset_signal_norm(filename_processed_norm_reference):
    with h5py.File(filename_processed_norm_reference, "r") as f:
        dataset = f["entry_0000/PyFAI/result_norm/data"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sigma_norm(filename_processed_norm_reference):
    with h5py.File(filename_processed_norm_reference, "r") as f:
        dataset = f["entry_0000/PyFAI/result_norm/data_errors"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_signal_2scat(filename_processed_2scat_full):
    with h5py.File(filename_processed_2scat_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_2scat/data"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sigma_2scat(filename_processed_2scat_full):
    with h5py.File(filename_processed_2scat_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_2scat/data_errors"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_signal_cave_new(filename_processed_cave_full):
    with h5py.File(filename_processed_cave_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_cave/data"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sigma_cave_new(filename_processed_cave_full):
    with h5py.File(filename_processed_cave_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_cave/data_errors"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_signal_azim_new(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/data"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sigma_azim_new(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/data_errors"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sumsignal_azim_new(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/sum_signal"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sumnorm_azim_new(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/sum_normalization"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_sumvariance_azim_new(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/sum_variance"][
            INDEX_RANGE[0] : INDEX_RANGE[1]
        ]
    return dataset




[docs]
@pytest.fixture
def dataset_radial_array(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/q"][:]
    return dataset




[docs]
@pytest.fixture
def dataset_azimuthal_array(filename_processed_azim_full):
    with h5py.File(filename_processed_azim_full, "r") as f:
        dataset = f["entry_0000/PyFAI/result_azim/chi"][:]
    return dataset




[docs]
@pytest.fixture
def workflow_norm():
    return {
        "graph": {"id": "workflow_norm"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
            },
        ],
    }




[docs]
@pytest.fixture
def workflow_norm_2scat():
    return {
        "graph": {"id": "workflow_norm_2scat"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
            },
            {
                "id": "2scat",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.secondaryscatteringtask.SecondaryScatteringTask",
            },
        ],
        "links": [
            {
                "source": "norm",
                "target": "2scat",
                "map_all_data": True,
            },
        ],
    }




[docs]
@pytest.fixture
def workflow_norm_2scat_cave():
    return {
        "graph": {"id": "workflow_norm_2scat_cave"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
            },
            {
                "id": "2scat",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.secondaryscatteringtask.SecondaryScatteringTask",
            },
            {
                "id": "cave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.cavingtask.CavingBeamstopTask",
            },
        ],
        "links": [
            {
                "source": "norm",
                "target": "2scat",
                "map_all_data": True,
            },
            {
                "source": "2scat",
                "target": "cave",
                "map_all_data": True,
            },
        ],
    }




[docs]
@pytest.fixture
def workflow_norm_2scat_cave_azim():
    return {
        "graph": {"id": "workflow_norm_2scat_cave_azim"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
            },
            {
                "id": "2scat",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.secondaryscatteringtask.SecondaryScatteringTask",
            },
            {
                "id": "cave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.cavingtask.CavingBeamstopTask",
            },
            {
                "id": "azim",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.azimuthaltask.AzimuthalTask",
            },
        ],
        "links": [
            {
                "source": "norm",
                "target": "2scat",
                "map_all_data": True,
            },
            {
                "source": "2scat",
                "target": "cave",
                "map_all_data": True,
            },
            {
                "source": "cave",
                "target": "azim",
                "map_all_data": True,
            },
        ],
    }




[docs]
@pytest.fixture
def workflow_norm_2scat_cave_azim_ave():
    return {
        "graph": {"id": "workflow_norm_2scat_cave_azim_ave"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
            },
            {
                "id": "2scat",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.secondaryscatteringtask.SecondaryScatteringTask",
            },
            {
                "id": "cave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.cavingtask.CavingBeamstopTask",
            },
            {
                "id": "azim",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.azimuthaltask.AzimuthalTask",
            },
            {
                "id": "ave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.averagetask.AverageTask",
            },
        ],
        "links": [
            {
                "source": "norm",
                "target": "2scat",
                "map_all_data": True,
            },
            {
                "source": "2scat",
                "target": "cave",
                "map_all_data": True,
            },
            {
                "source": "cave",
                "target": "azim",
                "map_all_data": True,
            },
            {
                "source": "azim",
                "target": "ave",
                "map_all_data": True,
            },
        ],
    }




[docs]
@pytest.fixture
def workflow_saxs_loop():
    return {
        "graph": {"id": "workflow_saxs_loop"},
        "nodes": [
            {
                "id": "norm",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.normalizationtask.NormalizationTask",
                "force_start_node": True,
                "default_inputs": [{"name": "reading_node", "value": True}],
            },
            {
                "id": "2scat",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.secondaryscatteringtask.SecondaryScatteringTask",
            },
            {
                "id": "cave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.cavingtask.CavingBeamstopTask",
            },
            {
                "id": "azim",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.azimuthaltask.AzimuthalTask",
            },
            {
                "id": "ave",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.averagetask.AverageTask",
            },
            {
                "id": "scalers",
                "task_type": "class",
                "task_identifier": "ewoksid02.tasks.scalerstask.ScalersTask",
            },
        ],
        "links": [
            {
                "source": "norm",
                "target": "2scat",
                "map_all_data": True,
                "conditions": [{"source_output": "continue_pipeline", "value": True}],
            },
            {
                "source": "2scat",
                "target": "cave",
                "map_all_data": True,
            },
            {
                "source": "cave",
                "target": "azim",
                "map_all_data": True,
            },
            {
                "source": "azim",
                "target": "ave",
                "map_all_data": True,
            },
            {
                "source": "ave",
                "target": "scalers",
                "map_all_data": True,
            },
            {
                "source": "scalers",
                "target": "norm",
                "map_all_data": True,
            },
        ],
    }