Source code for ewoksxas.tasks.tests.test_read_scans

import os

import numpy as np
import pytest
import silx.resources
from ewoksorange.tests.utils import execute_task
from Orange.data import ContinuousVariable, DiscreteVariable, StringVariable
from silx.io import h5py_utils

from ewoksxas.converters.orange import Converter, VarType
from ewoksxas.tasks.read_scans import CounterInfo, Inputs, MetadataInfo, ReadScans




def test_inputs_coerce_dict_configs_to_dataclasses():
    """Plain dict counter/metadata configs are coerced into the dataclasses."""
    converter = Converter()
    converter.add_meta("Filename", ["f"], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)

    inputs = Inputs(
        Data=converter.to_table(),
        x="energy",
        counters=[{"name": "mu", "path": "measurement/mu"}],
        metadata=[
            {"name": "T", "path": "pos/T", "type": "categorical"},
            {"name": "Legacy", "path": "pos/legacy"},  # no "type" -> "auto"
        ],
    )

    assert inputs.counters == [CounterInfo(name="mu", path="measurement/mu")]
    assert inputs.metadata == [
        MetadataInfo(name="T", path="pos/T", type="categorical"),
        MetadataInfo(name="Legacy", path="pos/legacy", type="auto"),
    ]





def test_read_scans():
    """Test reading a single scan with one counter from HDF5."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    assert os.path.exists(file_path)

    # Create input table with one scan.
    converter = Converter()
    converter.add_meta("Filename", [file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
        "metadata": [
            {
                "name": "Emono",
                "path": "instrument/positioners/Emono",
            }
        ],
    }

    outputs = execute_task(ReadScans, inputs)
    data = outputs["Data"]

    assert len(data) == 1
    assert len(data.domain.attributes) > 0
    assert "Filename" in [m.name for m in data.domain.metas]
    assert "Scan Name" in [m.name for m in data.domain.metas]
    assert "Counter" in [m.name for m in data.domain.metas]
    assert "Emono" in [m.name for m in data.domain.metas]

    # Check Counter value. Counter is a DiscreteVariable, so read the label
    # via the row instance rather than the raw (index) metas array.
    assert str(data[0]["Counter"]) == "mu_trans"

    # Check Emono value (scalar in HDF5).
    with h5py_utils.File(file_path, "r") as h5:
        expected_emono = h5["1.1/instrument/positioners/Emono"][()]

    emono_idx = [m.name for m in data.domain.metas].index("Emono")
    assert data.metas[0, emono_idx] == expected_emono





def test_read_scans_from_table():
    """Test reading multiple scans with one counter from HDF5."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")

    converter = Converter()
    converter.add_meta("Filename", [file_path, file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1", "2.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
        "metadata": [],
    }

    outputs = execute_task(ReadScans, inputs)
    data = outputs["Data"]

    assert len(data) == 2
    assert "Filename" in [m.name for m in data.domain.metas]
    assert "Scan Name" in [m.name for m in data.domain.metas]
    assert "Counter" in [m.name for m in data.domain.metas]





def test_read_scans_multiple_counters():
    """Test reading scans with multiple counters creates row per scan*counter."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")

    converter = Converter()
    converter.add_meta("Filename", [file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [
            {"name": "mu_trans", "path": "measurement/mu_trans"},
            {"name": "mu_ref", "path": "measurement/mu_ref"},
        ],
        "metadata": [],
    }

    outputs = execute_task(ReadScans, inputs)
    data = outputs["Data"]

    # 1 scan * 2 counters = 2 rows.
    assert len(data) == 2

    # Counter is a DiscreteVariable; read decoded labels via the row instances.
    counter_names = [str(data[i]["Counter"]) for i in range(len(data))]
    assert "mu_trans" in counter_names
    assert "mu_ref" in counter_names





def test_read_scans_multiple_scans_multiple_counters():
    """Test reading multiple scans with multiple counters."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")

    converter = Converter()
    converter.add_meta("Filename", [file_path, file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1", "2.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [
            {"name": "mu_trans", "path": "measurement/mu_trans"},
            {"name": "mu_ref", "path": "measurement/mu_ref"},
        ],
        "metadata": [],
    }

    outputs = execute_task(ReadScans, inputs)
    data = outputs["Data"]

    # 2 scans * 2 counters = 4 rows.
    assert len(data) == 4

    # No visible ID meta column; the per-scan identity is the converter's
    # internal content hash, identical within a (Filename, Scan Name) pair and
    # distinct across pairs. Here both counter rows of scan "1.1" share a hash,
    # both rows of scan "2.1" share another, and the two scans differ.
    assert "ID" not in [m.name for m in data.domain.metas]
    group_id = Converter.from_table(data).get_group_id()
    assert group_id is not None
    assert group_id[0] == group_id[1]
    assert group_id[2] == group_id[3]
    assert group_id[0] != group_id[2]





def test_read_scans_custom_grid():
    """Test reading scans with a custom X grid specification."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")

    converter = Converter()
    converter.add_meta("Filename", [file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    # Custom grid: 7000 to 7100 with 11 points.
    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
        "metadata": [],
        "x_interp_grid": [7000.0, 7100.0, 11],
    }

    outputs = execute_task(ReadScans, inputs)
    data = outputs["Data"]

    # Verify number of points (features) corresponds to npoints.
    assert len(data.domain.attributes) == 11

    # Verify X values (features) are as expected.
    expected_x = np.linspace(7000.0, 7100.0, 11)
    feature_names = [attr.name for attr in data.domain.attributes]
    # Orange features are usually named by index or value, let's check the values.
    # Our Converter.add_features uses the X values as names if possible or serial names.
    # Let's just check the data length for now and if we can extract names.
    for i, x_val in enumerate(expected_x):
        assert float(feature_names[i]) == x_val





def test_read_scans_array_metadata():
    """Test reading scans where a metadata item is an array (not scalar)."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")

    converter = Converter()
    converter.add_meta("Filename", [file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)
    input_table = converter.to_table()

    # Use an array dataset as metadata (e.g. energy array).  Typically metadata
    # are scalars, but if user provides array path, it should not crash.
    inputs = {
        "Data": input_table,
        "x": "instrument/energy_enc/data",
        "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
        "metadata": [
            {
                "name": "EnergyArr",
                "path": "instrument/energy_enc/data",  # This is an array
            }
        ],
    }

    # This should now raise a ValueError (wrapped in RuntimeError) because array
    # metadata is not allowed.
    with pytest.raises(RuntimeError, match="contains non-scalar values"):
        execute_task(ReadScans, inputs)



def _single_scan_input_table(file_path):
    """Build a one-scan input table (Filename + Scan Name metas)."""
    converter = Converter()
    converter.add_meta("Filename", [file_path], var_type=VarType.TEXT)
    converter.add_meta("Scan Name", ["1.1"], var_type=VarType.TEXT)
    return converter.to_table()


def _run_with_metadata(file_path, metadata):
    """Execute ReadScans for a single scan with the given metadata config."""
    inputs = {
        "Data": _single_scan_input_table(file_path),
        "x": "instrument/energy_enc/data",
        "counters": [{"name": "mu_trans", "path": "measurement/mu_trans"}],
        "metadata": metadata,
    }
    return execute_task(ReadScans, inputs)["Data"]




def test_read_scans_counter_is_discrete():
    """The standard Counter meta is a DiscreteVariable (issue #15/#25)."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    data = _run_with_metadata(file_path, [])

    assert isinstance(data.domain["Counter"], DiscreteVariable)
    assert isinstance(data.domain["Filename"], DiscreteVariable)
    assert isinstance(data.domain["Scan Name"], DiscreteVariable)





def test_read_scans_metadata_type_categorical():
    """A 'categorical' type token produces a DiscreteVariable meta."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    data = _run_with_metadata(
        file_path,
        [
            {
                "name": "Emono",
                "path": "instrument/positioners/Emono",
                "type": "categorical",
            }
        ],
    )

    assert isinstance(data.domain["Emono"], DiscreteVariable)





def test_read_scans_metadata_type_text_overrides_autodetect():
    """A 'text' token forces StringVariable even for low-cardinality data."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    data = _run_with_metadata(
        file_path,
        [{"name": "Emono", "path": "instrument/positioners/Emono", "type": "text"}],
    )

    assert isinstance(data.domain["Emono"], StringVariable)





def test_read_scans_metadata_type_numeric():
    """A 'numeric' token produces a ContinuousVariable meta."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    data = _run_with_metadata(
        file_path,
        [{"name": "Emono", "path": "instrument/positioners/Emono", "type": "numeric"}],
    )

    assert isinstance(data.domain["Emono"], ContinuousVariable)





def test_read_scans_categorical_missing_values_no_nan_category():
    """Missing categorical values are padded with '' (no spurious 'nan')."""
    file_path = silx.resources.resource_filename("ewoksxas:data/Fe_foil_0001.h5")
    data = _run_with_metadata(
        file_path,
        [{"name": "Missing", "path": "does/not/exist", "type": "categorical"}],
    )

    var = data.domain["Missing"]
    assert isinstance(var, DiscreteVariable)
    assert "nan" not in var.values