Source code for ewoksxas.tasks.tests.conftest

import pytest
from ewoksorange.tests.utils import execute_task
from Orange.data import Table
from silx.resources import resource_filename

from ewoksxas.tasks.normalize import Normalize
from ewoksxas.tasks.read_scans import ReadScans
from ewoksxas.tasks.read_sources import ReadSources


[docs] @pytest.fixture(scope="module") def fe_foil_data() -> Table | None: try: data_path = [resource_filename("ewoksxas:data/Fe_foil_0001.h5")] except ValueError as v_err: return pytest.skip(f"Error with silx.resources: {v_err}") source_inputs = {"file_paths": data_path, "filters": []} try: source_outputs = execute_task(ReadSources, inputs=source_inputs) except (RuntimeError, TypeError) as err: return pytest.skip(f"Error with task.read_sources: {err}") scans_inputs = { "Data": source_outputs["Data"], "x": "instrument/energy_enc/data", "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}], "metadata": [], } try: scans_outputs = execute_task(ReadScans, inputs=scans_inputs) except (RuntimeError, TypeError) as err: return pytest.skip(f"Error with task.read_scans: {err}") return scans_outputs["Data"]
[docs] @pytest.fixture(scope="module") def fe_foil_normalized(fe_foil_data) -> Table | None: inputs: dict[str, Table | dict] = {"Data": fe_foil_data} try: outputs = execute_task(Normalize, inputs=inputs) except (RuntimeError, TypeError) as err: return pytest.skip(f"Error with task.normalize: {err}") return outputs["Data"]
[docs] @pytest.fixture(scope="module") def fe_foil_one(fe_foil_normalized) -> Table | None: """A single normalized spectrum. The Fe foil dataset is ten near-identical repeat scans, so one spectrum reproduces the same per-spectrum statistics as the full set while running autobk ten times fewer. Tasks read the table through ``Converter.from_table`` and build fresh Larch groups, so this shared fixture is never mutated. """ return fe_foil_normalized[:1]