"""Tests for the CombineRows task.
These tests are self-contained: they build synthetic Orange Tables rather than
relying on external data files, so they run without the test fixtures defined
in conftest.py.
"""
from __future__ import annotations
import numpy as np
import pytest
from ewoksorange.tests.utils import execute_task
from ewoksxas.converters.orange import Converter, VarType
from ewoksxas.tasks.combine_rows import CombineRows, evaluate_equation
def _make_grouped_table(
n_groups: int = 3,
n_points: int = 50,
a_scale: float = 2.0,
b_scale: float = 1.0,
) -> tuple:
"""Build a table with two row types (a, b) across n_groups groups.
Returns:
A tuple of (table, x_axis, a_rows, b_rows) where table is an
Orange.data.Table, x_axis is a 1-D array of length n_points, and
a_rows and b_rows are arrays of shape (n_groups, n_points).
"""
x_axis = np.linspace(0, 100, n_points)
rng = np.random.default_rng(42)
a_rows, b_rows = [], []
group_names_a, group_names_b, type_names_a, type_names_b = [], [], [], []
for group_idx in range(n_groups):
base = rng.random(n_points) + group_idx * 0.1
a_val = base * a_scale
b_val = base * b_scale
a_rows.append(a_val)
b_rows.append(b_val)
group_names_a.append(f"group-{group_idx + 1}")
group_names_b.append(f"group-{group_idx + 1}")
type_names_a.append("a")
type_names_b.append("b")
all_X = np.vstack(a_rows + b_rows)
all_group_names = group_names_a + group_names_b
all_type_names = type_names_a + type_names_b
conv = Converter().add_features(x_axis, all_X)
conv.add_meta(
"group",
np.array(all_group_names, dtype=object),
VarType.TEXT,
)
conv.add_meta(
"type",
np.array(all_type_names, dtype=object),
VarType.TEXT,
)
table = conv.to_table()
return table, x_axis, np.array(a_rows), np.array(b_rows)
def _repro_table(energy: np.ndarray, rows: list) -> object:
"""Build the issue #42 repro table (Filename, Scan Name, Counter, spectrum).
The per-scan identity is the converter's internal content hash of
(Filename, Scan Name); no visible ID meta is added.
"""
spectra = np.array([r[3] for r in rows])
return (
Converter()
.add_features(energy, spectra)
.add_meta(
"Filename",
np.array([r[0] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Scan Name",
np.array([r[1] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Counter",
np.array([r[2] for r in rows], dtype=object),
VarType.TEXT,
)
.to_table()
)
def _decoded_meta(table: object, name: str) -> list:
"""Read a meta column back as decoded labels.
Reading ``table.metas[i, idx]`` directly returns the raw storage value,
which for a DiscreteVariable is the integer category index, not the label.
Concatenation in the task re-infers small text columns (e.g. the equation
label) as categorical, so tests must decode through the Converter to compare
against the original label strings.
"""
return list(Converter.from_table(table).get_meta_values(name))
[docs]
def test_scan_name_duplicate_equation_error():
"""Minimal reproduction of the Combine Rows 'duplicate equation term' error.
Run with the project env, e.g.:
conda run -n conda310-ewoksxas python repro_combine_rows.py
"""
# Two data files, each with a scan called "1.1" measuring the same two counters.
# This mirrors BLISS per-file scan numbering: scan names restart at 1.1 in every
# file, so "Scan Name" alone is NOT unique across files.
energy = np.array([8.97, 8.98, 8.99])
rows = [
# Filename Scan Name Counter spectrum
("xanes_0004.h5", "1.1", "small_roi", [10.0, 20.0, 30.0]),
("xanes_0004.h5", "1.1", "I02", [1.0, 2.0, 3.0]),
("xanes_0002.h5", "1.1", "small_roi", [40.0, 50.0, 60.0]),
("xanes_0002.h5", "1.1", "I02", [4.0, 5.0, 6.0]),
]
spectra = np.array([r[3] for r in rows])
table = (
Converter()
.add_features(energy, spectra)
.add_meta(
"Filename",
np.array([r[0] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Scan Name",
np.array([r[1] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Counter",
np.array([r[2] for r in rows], dtype=object),
VarType.TEXT,
)
.to_table()
)
# Same configuration as the failing .ows workflow.
inputs = {
"Data": table,
"grouping_column": "Scan Name", # <-- groups both files' "1.1" scans together
"equation_terms_column": "Counter",
"equation": "small_roi / I02",
"mode": "replace",
}
with pytest.raises(RuntimeError, match="Duplicate equation term"):
CombineRows(inputs=inputs).execute()
# -> RuntimeError: Execution failed for ewoks task (id: None, task:
# 'ewoksxas.tasks.combine_rows.CombineRows'
# -> # Caused by:
# -> ValueError: Duplicate equation term 'small_roi' in group '1.1'; rows within
# a group must have unique values in the 'Counter' column.
[docs]
def test_filename_duplicate_equation_error():
"""Minimal reproduction of the Combine Rows 'duplicate equation term' error.
Run with the project env, e.g.:
conda run -n conda310-ewoksxas python repro_combine_rows.py
"""
# Two data files, each with a scan called "1.1" measuring the same two counters.
# This mirrors BLISS per-file scan numbering: scan names restart at 1.1 in every
# file, so "Scan Name" alone is NOT unique across files.
energy = np.array([8.97, 8.98, 8.99])
rows = [
# Filename Scan Name Counter spectrum
("xanes_0004.h5", "1.1", "small_roi", [10.0, 20.0, 30.0]),
("xanes_0004.h5", "1.1", "I02", [1.0, 2.0, 3.0]),
("xanes_0004.h5", "2.1", "small_roi", [40.0, 50.0, 60.0]),
("xanes_0004.h5", "2.1", "I02", [4.0, 5.0, 6.0]),
]
spectra = np.array([r[3] for r in rows])
table = (
Converter()
.add_features(energy, spectra)
.add_meta(
"Filename",
np.array([r[0] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Scan Name",
np.array([r[1] for r in rows], dtype=object),
VarType.TEXT,
)
.add_meta(
"Counter",
np.array([r[2] for r in rows], dtype=object),
VarType.TEXT,
)
.to_table()
)
# Same configuration as the failing .ows workflow.
inputs = {
"Data": table,
"grouping_column": "Filename",
"equation_terms_column": "Counter",
"equation": "small_roi / I02",
"mode": "replace",
}
with pytest.raises(RuntimeError, match="Duplicate equation term"):
CombineRows(inputs=inputs).execute()
# -> RuntimeError: Execution failed for ewoks task (id: None, task:
# 'ewoksxas.tasks.combine_rows.CombineRows'
# -> # Caused by:
# -> ValueError: Duplicate equation term 'small_roi' in group 'xanes_0004.h5';
# rows within a group must have unique values in the 'Counter' column.
[docs]
def test_scan_name_collision_no_error_with_default_hash_grouping():
"""Issue #42 regression (scan-name collision across files).
With NO grouping_column given, the task groups by the converter's
content-hash scan identity instead of "Scan Name", so the two files' "1.1"
scans no longer merge and the duplicate-term error does not occur.
"""
energy = np.array([8.97, 8.98, 8.99])
rows = [
("xanes_0004.h5", "1.1", "small_roi", [10.0, 20.0, 30.0]),
("xanes_0004.h5", "1.1", "I02", [1.0, 2.0, 3.0]),
("xanes_0002.h5", "1.1", "small_roi", [40.0, 50.0, 60.0]),
("xanes_0002.h5", "1.1", "I02", [4.0, 5.0, 6.0]),
]
inputs = {
"Data": _repro_table(energy, rows),
# No grouping_column -> the task groups by the content-hash identity.
"equation_terms_column": "Counter",
"equation": "small_roi / I02",
"mode": "replace",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
# One result row per unique scan (two here), each small_roi / I02 == 10.
assert len(output) == 2
_, result = Converter.from_table(output).features
np.testing.assert_allclose(result, [[10.0, 10.0, 10.0], [10.0, 10.0, 10.0]])
[docs]
def test_filename_collision_no_error_with_default_hash_grouping():
"""Issue #42 regression (multiple scans within one file).
A single file with two scans no longer collapses into one group under the
default grouping, because the content-hash identity distinguishes the scans.
"""
energy = np.array([8.97, 8.98, 8.99])
rows = [
("xanes_0004.h5", "1.1", "small_roi", [10.0, 20.0, 30.0]),
("xanes_0004.h5", "1.1", "I02", [1.0, 2.0, 3.0]),
("xanes_0004.h5", "2.1", "small_roi", [40.0, 50.0, 60.0]),
("xanes_0004.h5", "2.1", "I02", [4.0, 5.0, 6.0]),
]
inputs = {
"Data": _repro_table(energy, rows),
"equation_terms_column": "Counter",
"equation": "small_roi / I02",
"mode": "replace",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == 2
_, result = Converter.from_table(output).features
np.testing.assert_allclose(result, [[10.0, 10.0, 10.0], [10.0, 10.0, 10.0]])
[docs]
def test_simple_ratio():
a = np.array([2.0, 4.0, 6.0])
b = np.array([1.0, 2.0, 3.0])
result = evaluate_equation("a / b", {"a": a, "b": b})
np.testing.assert_allclose(result, 2.0)
[docs]
def test_named_references():
num = np.array([10.0, 20.0])
den = np.array([2.0, 4.0])
result = evaluate_equation("num / den", {"num": num, "den": den})
np.testing.assert_allclose(result, 5.0)
[docs]
def test_numpy_function():
a = np.array([1.0, np.e, np.e**2])
result = evaluate_equation("log(a)", {"a": a})
np.testing.assert_allclose(result, [0.0, 1.0, 2.0])
[docs]
def test_np_namespace():
a = np.array([0.0, 1.0, 4.0])
result = evaluate_equation("sqrt(a)", {"a": a})
np.testing.assert_allclose(result, [0.0, 1.0, 2.0])
[docs]
def test_undefined_name_raises():
with pytest.raises(ValueError, match="NameError"):
evaluate_equation("missing_var + 1", {"a": np.array([1.0])})
[docs]
def test_result_is_float64():
a = np.array([1, 2, 3])
result = evaluate_equation("a * 2", {"a": a})
assert result.dtype == np.float64
[docs]
def test_scalar_result_raises():
with pytest.raises(ValueError, match="one value per point"):
evaluate_equation("sum(a)", {"a": np.array([1.0, 2.0])})
[docs]
def test_length_mismatch_raises():
with pytest.raises(ValueError, match="one value per point"):
evaluate_equation("a[:1]", {"a": np.array([1.0, 2.0])})
[docs]
def test_passthrough_when_no_equation():
table, *_ = _make_grouped_table()
inputs = {"Data": table}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == len(table)
[docs]
def test_passthrough_when_no_selection_field():
table, *_ = _make_grouped_table()
inputs = {"Data": table, "equation": "a / b"}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == len(table)
[docs]
def test_insert_mode_row_count():
table, *_ = _make_grouped_table(n_groups=3)
n_original = len(table)
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == n_original + 3
[docs]
def test_insert_mode_values_correct():
table, _, a_rows, b_rows = _make_grouped_table(n_groups=2)
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
meta_names = [var.name for var in output.domain.metas]
type_idx = meta_names.index("type")
result_indices = [
i for i in range(len(output)) if str(output.metas[i, type_idx]) == "a / b"
]
expected = a_rows / b_rows
for group_i, row_i in enumerate(result_indices):
np.testing.assert_allclose(output.X[row_i], expected[group_i], rtol=1e-10)
[docs]
def test_replace_mode_values_correct():
n_groups = 2
table, _, a_rows, b_rows = _make_grouped_table(n_groups=n_groups)
equation = "a / b"
inputs = {
"Data": table,
"equation": equation,
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "replace",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == n_groups
types = _decoded_meta(output, "type")
result_indices = [i for i, t in enumerate(types) if t == equation]
assert len(result_indices) == n_groups
expected = a_rows / b_rows
for group_i, row_i in enumerate(result_indices):
np.testing.assert_allclose(output.X[row_i], expected[group_i], rtol=1e-10)
[docs]
def test_named_row_references():
table, _, a_rows, b_rows = _make_grouped_table(n_groups=1)
equation = "a / b"
inputs = {
"Data": table,
"equation": equation,
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
types = _decoded_meta(output, "type")
result_idx = next(i for i, t in enumerate(types) if t == equation)
expected = (a_rows / b_rows)[0]
np.testing.assert_allclose(output.X[result_idx], expected, rtol=1e-10)
[docs]
def test_missing_selection_field_raises():
table, *_ = _make_grouped_table()
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "NonExistentField",
}
with pytest.raises(RuntimeError, match="NonExistentField"):
execute_task(CombineRows, inputs=inputs)
[docs]
def test_missing_grouping_field_raises():
table, *_ = _make_grouped_table()
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "BadField",
"equation_terms_column": "type",
}
with pytest.raises(RuntimeError, match="BadField"):
execute_task(CombineRows, inputs=inputs)
[docs]
def test_invalid_mode_raises():
table, *_ = _make_grouped_table()
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "invalid_mode",
}
with pytest.raises(RuntimeError, match="mode"):
execute_task(CombineRows, inputs=inputs)
[docs]
def test_x_axis_preserved():
table, x_axis, *_ = _make_grouped_table(n_groups=2)
inputs = {
"Data": table,
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
out_x, _ = Converter.from_table(output).features
np.testing.assert_allclose(out_x, x_axis)
[docs]
def test_no_grouping_field_single_group():
table, _, _, _ = _make_grouped_table(n_groups=1)
n_original = len(table)
equation = "a / b"
inputs = {
"Data": table,
"equation": equation,
"grouping_column": None,
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == n_original + 1
[docs]
def test_result_rows_interleaved_after_groups():
table, *_ = _make_grouped_table(n_groups=3)
equation = "a / b"
inputs = {
"Data": table,
"grouping_column": "group",
"equation_terms_column": "type",
"equation": equation,
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
conv = Converter.from_table(output)
types = list(conv.get_meta_values("type"))
groups = list(conv.get_meta_values("group"))
result_positions = [i for i, t in enumerate(types) if t == equation]
assert result_positions == [4, 6, 8]
for pos, expected_group in zip(
result_positions, ["group-1", "group-2", "group-3"], strict=True
):
assert groups[pos] == expected_group
[docs]
def test_replace_mode_with_no_grouping_column():
table, _, a_rows, b_rows = _make_grouped_table(n_groups=1)
equation = "a / b"
inputs = {
"Data": table,
"equation": equation,
"grouping_column": None,
"equation_terms_column": "type",
"mode": "replace",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == 1
assert _decoded_meta(output, "type")[0] == equation
np.testing.assert_allclose(output.X[0], (a_rows / b_rows)[0], rtol=1e-10)
[docs]
def test_preserves_discrete_target():
"""A DiscreteVariable target survives the from_table -> ... -> to_table run."""
table, *_ = _make_grouped_table(n_groups=2)
converter = Converter.from_table(table)
converter.add_target(
"quality",
np.array(["good", "bad", "good", "bad"]),
VarType.CATEGORICAL,
values=["bad", "good"],
)
inputs = {
"Data": converter.to_table(),
"equation": "a / b",
"grouping_column": "group",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
target_var = output.domain.class_vars[0]
assert target_var.name == "quality"
# Concatenation re-infers the categorical from the combined labels, so the
# category *set* survives even though its order follows first appearance
# rather than the explicit ["bad", "good"] given on input.
assert set(target_var.values) == {"bad", "good"}
# The per-row labels themselves are preserved for the original rows.
decoded = list(Converter.from_table(output).targets[0].get_decoded_data())
assert set(decoded) == {"bad", "good"}
[docs]
def test_duplicate_term_in_group_raises():
x = np.array([1.0, 2.0])
y = np.array([[1.0, 1.0], [2.0, 2.0]])
conv = Converter().add_features(x, y)
conv.add_meta("type", np.array(["a", "a"], dtype=object), VarType.TEXT)
inputs = {
"Data": conv.to_table(),
"equation": "a * 2",
"equation_terms_column": "type",
}
with pytest.raises(RuntimeError, match="Duplicate equation term"):
execute_task(CombineRows, inputs=inputs)
[docs]
def test_non_identifier_terms_ignored():
"""Rows whose term value is not a valid identifier do not break evaluation."""
x = np.array([1.0, 2.0])
y = np.array([[2.0, 4.0], [1.0, 2.0], [9.0, 9.0]])
conv = Converter().add_features(x, y)
conv.add_meta("type", np.array(["a", "b", "det 1"], dtype=object), VarType.TEXT)
inputs = {
"Data": conv.to_table(),
"equation": "a / b",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == 4
np.testing.assert_allclose(output.X[3], [2.0, 2.0])
[docs]
def test_discrete_terms_column():
"""A terms column auto-detected as categorical is coerced to text.
Without an explicit var_type, add_meta stores a small set of repeated
labels as a DiscreteVariable. The equation label written into the result
rows is free-form text, so the task must still work.
"""
x = np.array([1.0, 2.0])
y = np.array([[2.0, 4.0], [1.0, 2.0]])
conv = Converter().add_features(x, y)
conv.add_meta("type", np.array(["a", "b"], dtype=object))
assert conv.metas[0].var_type is VarType.CATEGORICAL
inputs = {
"Data": conv.to_table(),
"equation": "a / b",
"equation_terms_column": "type",
"mode": "insert",
}
output = execute_task(CombineRows, inputs=inputs)["Data"]
assert len(output) == 3
np.testing.assert_allclose(output.X[2], [2.0, 2.0])
assert _decoded_meta(output, "type") == ["a", "b", "a / b"]