"""Tests for the LinearCombinationAnalysis task.
The task wraps ewoksxas.fit.lca_fit.LCAModel: it reads sample and component
Orange Tables, fits each sample as a non-negative linear combination of the
components, and writes three tables (Transformed_Data, Data, Components).
Following fit/tests/test_lca_fit.py, the samples here are exact linear
combinations of known components, so the recovered weights, the fit quality
(rmse ~ 0, r2 ~ 1) and the reconstructed spectrum are all known in advance.
test_lca_outputs_match_direct_fit additionally checks the task reproduces a
direct LCAModel.fit() on the same data, i.e. it is a faithful wrapper.
"""
import numpy as np
import pytest
from ewoksorange.tests.utils import execute_task
from ewoksxas.converters.orange import Converter, VarType
from ewoksxas.fit.lca_fit import LCAModel
from ewoksxas.tasks.lca import LinearCombinationAnalysis
# Mixing fractions used to build the synthetic samples (sum to 1).
WEIGHTS = np.array([0.1, 0.2, 0.3, 0.35, 0.05])
def _component_table(x, components, names):
"""Build a Component_Data table: x as features, names as the Scan Name meta."""
return (
Converter()
.add_features(x, components)
.add_meta("Scan Name", np.asarray(names), VarType.TEXT)
.to_table()
)
def _sample_table(x, spectra, names):
"""Build a Sample_Data table with one row per spectrum."""
return (
Converter()
.add_features(x, np.atleast_2d(spectra))
.add_meta("Scan Name", np.asarray(names), VarType.TEXT)
.to_table()
)
def _run(sample_table, component_table, **parameters):
return execute_task(
LinearCombinationAnalysis,
inputs={
"Sample_Data": sample_table,
"Component_Data": component_table,
"parameters": parameters,
},
)
def _meta(converter, name):
"""Return a meta column as a float array (metas are stored as objects)."""
return np.asarray(converter.get_meta_values(name), dtype=float)
[docs]
@pytest.fixture
def x():
return np.linspace(0, np.pi, 100, dtype=np.float64)
[docs]
@pytest.fixture
def components(x):
"""Five linearly independent components, mirroring fit/tests/conftest.py."""
c1 = np.sin(x)
c2 = np.cos(x)
c3 = (c1 * c2) * 2
c4 = (c3 * c1) * (4 / 3)
c5 = (c3 * c2) * (4 / 3)
return np.array([c1, c2, c3, c4, c5])
[docs]
@pytest.fixture
def component_names():
return [f"Component {i + 1}" for i in range(5)]
[docs]
@pytest.fixture
def component_table(x, components, component_names):
return _component_table(x, components, component_names)
[docs]
@pytest.fixture
def sample_table(x, components):
"""A single sample that is an exact linear combination of the components."""
return _sample_table(x, WEIGHTS @ components, ["Sample 1"])
[docs]
def test_lca_recovers_known_weights(sample_table, component_table, component_names):
"""The fitted weights, fit quality and model spectrum are all recovered."""
outputs = _run(sample_table, component_table)
transformed = Converter.from_table(outputs["Transformed_Data"])
names, solutions = transformed.features
assert list(names) == [f"C{i + 1}" for i in range(5)]
np.testing.assert_allclose(solutions[0], WEIGHTS, atol=1e-6)
np.testing.assert_allclose(_meta(transformed, "rmse"), 0.0, atol=1e-6)
np.testing.assert_allclose(_meta(transformed, "r2"), 1.0, atol=1e-6)
# The reconstructed model matches the original sample spectrum.
_, model_y = Converter.from_table(outputs["Data"]).features
_, sample_y = Converter.from_table(sample_table).features
np.testing.assert_allclose(model_y[0], sample_y[0], atol=1e-6)
[docs]
def test_lca_outputs_match_direct_fit(
x, components, component_names, sample_table, component_table
):
"""Task outputs reproduce a direct LCAModel.fit() on the same data."""
outputs = _run(sample_table, component_table)
lca = LCAModel.from_dicts(
{"x": x, "y": WEIGHTS @ components},
{
name: {"x": x, "y": y}
for name, y in zip(component_names, components, strict=True)
},
)
fit = lca.fit()
transformed = Converter.from_table(outputs["Transformed_Data"])
_, solutions = transformed.features
np.testing.assert_allclose(solutions[0], fit.solution, atol=1e-9)
np.testing.assert_allclose(_meta(transformed, "rmse")[0], fit.rmse, atol=1e-9)
np.testing.assert_allclose(_meta(transformed, "r2")[0], fit.r2, atol=1e-9)
for i in range(len(component_names)):
np.testing.assert_allclose(
_meta(transformed, f"C{i + 1} σ")[0], fit.sigma[i], atol=1e-12
)
# The Data table carries the same model x/y as the direct fit.
model_x, model_y = Converter.from_table(outputs["Data"]).features
np.testing.assert_allclose(model_x, fit.model.x, atol=1e-12)
np.testing.assert_allclose(model_y[0], fit.model.y, atol=1e-9)
[docs]
def test_lca_output_tables_structure(sample_table, component_table, component_names):
"""The three output tables have the expected orientation and metas."""
outputs = _run(sample_table, component_table)
assert set(outputs) == {"Transformed_Data", "Data", "Components"}
keys = [f"C{i + 1}" for i in range(5)]
# Transformed_Data: one row per sample, one column per component (C-keys).
transformed = outputs["Transformed_Data"]
assert transformed.X.shape == (1, 5)
t_conv = Converter.from_table(transformed)
assert list(t_conv.features[0]) == keys
t_metas = [meta.name for meta in t_conv.metas]
assert "rmse" in t_metas
assert "r2" in t_metas
assert all(f"{key} σ" in t_metas for key in keys)
# Data: one row per sample, features on the (interpolated) model x-grid.
data = outputs["Data"]
assert data.X.shape == (1, 100)
d_metas = [meta.name for meta in Converter.from_table(data).metas]
assert all(key in d_metas for key in keys) # per-component weights
# Components: one row per component, one column per sample (the transpose).
comps = outputs["Components"]
assert comps.X.shape == (5, 1)
c_conv = Converter.from_table(comps)
assert list(c_conv.features[0]) == ["Sample 1"]
assert list(c_conv.get_meta_values("ID")) == keys
assert list(c_conv.get_meta_values("Name")) == component_names
[docs]
def test_lca_roi_bounds(sample_table, component_table):
"""xmin/xmax restrict the fit to a region of interest."""
outputs = _run(sample_table, component_table, xmin=1.0, xmax=2.0)
model_x, _ = Converter.from_table(outputs["Data"]).features
assert model_x.size == 32 # same ROI size as test_lca_fit_bounded
assert model_x.min() >= 1.0
assert model_x.max() <= 2.0
transformed = Converter.from_table(outputs["Transformed_Data"])
_, solutions = transformed.features
np.testing.assert_allclose(solutions[0], WEIGHTS, atol=1e-3)
np.testing.assert_allclose(_meta(transformed, "r2"), 1.0, atol=1e-6)
[docs]
def test_lca_normalize(x, components, component_table):
"""normalize=True scales each solution to sum to one."""
# Weights that sum to ten, so normalization is observable.
weights = WEIGHTS * 10
sample_table = _sample_table(x, weights @ components, ["Sample 1"])
raw = _run(sample_table, component_table)
norm = _run(sample_table, component_table, normalize=True)
_, raw_solution = Converter.from_table(raw["Transformed_Data"]).features
_, norm_solution = Converter.from_table(norm["Transformed_Data"]).features
np.testing.assert_allclose(raw_solution[0], weights, atol=1e-4)
np.testing.assert_allclose(norm_solution[0].sum(), 1.0, atol=1e-6)
np.testing.assert_allclose(
norm_solution[0], raw_solution[0] / raw_solution[0].sum(), atol=1e-9
)
# The normalized solution recovers the mixing fractions.
np.testing.assert_allclose(norm_solution[0], WEIGHTS, atol=1e-5)
[docs]
def test_lca_exclude_components(sample_table, component_table, component_names):
"""exclude_components drops the keyed (C-key) components from every table."""
outputs = _run(sample_table, component_table, exclude_components=["C4"])
keys = ["C1", "C2", "C3", "C5"]
names, _ = Converter.from_table(outputs["Transformed_Data"]).features
assert list(names) == keys
assert outputs["Transformed_Data"].X.shape == (1, 4)
assert outputs["Components"].X.shape == (4, 1)
comps = Converter.from_table(outputs["Components"])
assert list(comps.get_meta_values("ID")) == keys
expected_names = [name for name in component_names if name != "Component 4"]
assert list(comps.get_meta_values("Name")) == expected_names
# @pytest.mark.skip(reason="not implemented")
[docs]
def test_lca_component_identifier(x, components, sample_table):
"""component_identifier selects the human-readable name for each component."""
labels = [f"Phase {i}" for i in range(5)]
component_table = (
Converter()
.add_features(x, components)
.add_meta("Label", np.asarray(labels), VarType.TEXT)
.to_table()
)
outputs = _run(sample_table, component_table, component_identifier="Label")
transformed = outputs["Transformed_Data"]
names, _ = Converter.from_table(transformed).features
# Features stay the stable C-keys; the selected identifier supplies the
# display name attached to each C-key variable's .attributes (read the raw
# domain, since .attributes are not recovered by Converter.from_table).
assert list(names) == [f"C{i + 1}" for i in range(5)]
assert [var.attributes["name"] for var in transformed.domain.attributes] == labels
[docs]
def test_lca_multiple_samples(x, components, component_table):
"""Each sample is fitted independently; the tables grow accordingly."""
sample_weights = np.array(
[
[0.10, 0.20, 0.30, 0.35, 0.05],
[0.50, 0.10, 0.10, 0.20, 0.10],
[0.20, 0.20, 0.20, 0.20, 0.20],
]
)
sample_names = [f"Sample {i + 1}" for i in range(3)]
sample_table = _sample_table(x, sample_weights @ components, sample_names)
outputs = _run(sample_table, component_table)
transformed = Converter.from_table(outputs["Transformed_Data"])
_, solutions = transformed.features
assert solutions.shape == (3, 5)
np.testing.assert_allclose(solutions, sample_weights, atol=1e-6)
np.testing.assert_allclose(_meta(transformed, "rmse"), 0.0, atol=1e-6)
# Components is the transpose: one row per component, one column per sample.
components_out = Converter.from_table(outputs["Components"])
names, weights_t = components_out.features
assert outputs["Components"].X.shape == (5, 3)
assert list(names) == sample_names
np.testing.assert_allclose(weights_t, sample_weights.T, atol=1e-6)
[docs]
@pytest.mark.filterwarnings("ignore::RuntimeWarning")
def test_lca_excess_components(x, components, component_names, sample_table):
"""Irrelevant extra components are fitted with ~zero weight."""
extra = np.array([x, np.pi - x, np.sqrt(x), np.power(x, 2)])
all_components = np.vstack([components, extra])
all_names = [*component_names, *(f"Extra {i + 1}" for i in range(len(extra)))]
component_table = _component_table(x, all_components, all_names)
outputs = _run(sample_table, component_table)
transformed = Converter.from_table(outputs["Transformed_Data"])
names, solutions = transformed.features
assert list(names) == [f"C{i + 1}" for i in range(len(all_names))]
expected = np.concatenate([WEIGHTS, np.zeros(len(extra))])
np.testing.assert_allclose(solutions[0], expected, atol=1e-2)
np.testing.assert_allclose(_meta(transformed, "r2"), 1.0, atol=1e-3)
[docs]
def test_lca_component_keys_follow_row_order(x, components, sample_table):
"""C-keys are numbered by component row order (positional, collision-free)."""
names = [f"row{i}" for i in range(5)]
component_table = _component_table(x, components, names)
outputs = _run(sample_table, component_table)
comps = Converter.from_table(outputs["Components"])
assert list(comps.get_meta_values("ID")) == ["C1", "C2", "C3", "C4", "C5"]
assert list(comps.get_meta_values("Name")) == names
[docs]
def test_lca_duplicate_component_scan_name(x, components, sample_table):
"""Issue #40: components sharing a Scan Name still get distinct C-keys.
All output tables build (no duplicate variable names), and exclusion by
C-key drops exactly one component even though the human names repeat.
"""
names = ["1.1", "1.1", "1.1", "2.1", "2.1"]
component_table = _component_table(x, components, names)
outputs = _run(sample_table, component_table)
transformed = Converter.from_table(outputs["Transformed_Data"])
assert list(transformed.features[0]) == ["C1", "C2", "C3", "C4", "C5"]
outputs = _run(sample_table, component_table, exclude_components=["C2"])
assert outputs["Transformed_Data"].X.shape == (1, 4)
assert list(Converter.from_table(outputs["Components"]).get_meta_values("ID")) == [
"C1",
"C3",
"C4",
"C5",
]
[docs]
def test_lca_sample_scan_name_collision(x, components, component_table):
"""Two samples sharing a Scan Name no longer crash the Components table.
Scan Name repeats, so the per-sample column headers fall back to the
converter's content-hash identity (unique per Filename+Scan Name), keeping
the Orange Domain free of duplicate feature names.
"""
sample_weights = np.array(
[
[0.10, 0.20, 0.30, 0.35, 0.05],
[0.50, 0.10, 0.10, 0.20, 0.10],
]
)
sample_table = (
Converter()
.add_features(x, sample_weights @ components)
.add_meta("Filename", np.asarray(["a.h5", "b.h5"]), VarType.TEXT)
.add_meta("Scan Name", np.asarray(["1.1", "1.1"]), VarType.TEXT)
.to_table()
)
outputs = _run(sample_table, component_table)
headers = [var.name for var in outputs["Components"].domain.attributes]
assert len(set(headers)) == 2 # unique headers despite shared Scan Name
assert outputs["Components"].X.shape == (5, 2)