"""Apply a mathematical equation to groups of rows selected by metadata.
The typical use-case is combining tabular data where multiple rows represent
different measurements or channels that should be merged. For example, a
dataset might have one row per counter (signal, monitor, …). This task
lets the user write an equation such as ``signal / monitor`` and receive
a new table whose result row is appended (or replaces) the originals.
Row selection
-------------
Two metadata fields govern which rows participate in the calculation:
* **grouping** - identifies which rows belong to *the same group*.
Rows that share the same value of this column are treated as one group.
Example: ``"Group ID"``.
* **equation_terms** - the metadata column whose values name the rows
within each group. All rows in the group are made available in the
equation namespace by their value string (e.g. ``signal``, ``monitor``).
Equation
--------
Inside the equation all rows of a group are available by their metadata
value (e.g. ``signal``, ``monitor``).
Common numpy/math helpers (``sqrt``, ``log``, ``exp``, ``abs``, …) are available.
Values that are not valid Python identifiers (e.g. ``"det 1"``) cannot be
referenced in the equation and are ignored. Duplicate values within a group
raise an error because the equation result would depend on row order.
Output
------
* ``"insert"`` - inserts one result row per group immediately after the
last row of that group; the new row gets the same grouping metadata as
the group and the *equation* as the value for *equation_terms*.
* ``"replace"`` - collapses each group to its single result row, in
first-seen group order.
"""
import numpy as np
from asteval import Interpreter
from ewokscore import Task
from ewokscore.model import BaseInputModel, BaseOutputModel
from Orange.data import Table
from ewoksxas.converters.orange import Converter, VarType
[docs]
def evaluate_equation(equation: str, rows: dict[str, np.ndarray]) -> np.ndarray:
"""Evaluate the equation with rows available as local names.
Args:
equation: A Python equation string such as ``"a / b"`` or ``"log(a)"``.
rows: Mapping from metadata value string to the corresponding array.
Returns:
The computed result array with the same length as the input rows.
Raises:
ValueError: If the equation fails to evaluate or the result does not
have the same length as the input rows.
"""
interpreter = Interpreter(user_symbols=rows, use_numpy=True)
result = interpreter(equation)
if interpreter.error:
error_type, message = interpreter.error[0].get_error()
available = sorted(rows)
raise ValueError(
f"{error_type} in equation: {message}. "
f"Available row references: {available}"
)
if result is None:
raise ValueError(f"Equation '{equation}' did not produce a result.")
result = np.asarray(result, dtype=np.float64)
if rows:
expected = len(next(iter(rows.values())))
if result.shape != (expected,):
raise ValueError(
f"Equation result has shape {result.shape}, expected "
f"({expected},); the equation must return one value per point."
)
return result
[docs]
class Outputs(BaseOutputModel):
Data: Table
[docs]
class CombineRows(Task, input_model=Inputs, output_model=Outputs): # type: ignore
"""Apply a mathematical equation to groups of rows."""
[docs]
def run(self) -> None:
data: Table = self.inputs.Data
grouping_column: str | None = self.inputs.grouping_column
equation_terms_column: str | None = self.inputs.equation_terms_column
equation: str = self.inputs.equation.strip()
mode: str = self.inputs.mode
if mode not in ("insert", "replace"):
raise ValueError(f"Unknown mode {mode!r}; use 'insert' or 'replace'.")
# Pass through unchanged if nothing is configured yet.
if not equation or not equation_terms_column:
self.outputs.Data = data
return
source = Converter.from_table(data)
meta_names = source.get_meta_names()
if grouping_column and grouping_column not in meta_names:
raise ValueError(
f"Grouping column '{grouping_column}' not found in metadata."
)
if equation_terms_column not in meta_names:
raise ValueError(
f"Equation terms column '{equation_terms_column}' "
"not found in metadata."
)
_, spectra = source.features
term_values = source.get_meta_values(equation_terms_column)
if grouping_column:
group_values = source.get_group_id((grouping_column,))
else:
group_values = source.get_group_id(default_by_row=False)
if source.get_column(equation_terms_column).is_numeric:
source.add_meta(
equation_terms_column,
np.asarray(term_values, dtype=object),
VarType.TEXT,
)
# Group row indices by their grouping value, preserving first-seen order.
groups: dict[str, list[int]] = {}
for index in range(len(spectra)):
key = str(group_values[index])
groups.setdefault(key, []).append(index)
result_rows = self._build_result_rows(
source, groups, term_values, equation, equation_terms_column
)
if mode == "insert":
combined = self._insert(source, groups, result_rows)
else:
combined = self._replace(groups, result_rows)
self.outputs.Data = combined.to_table()
@staticmethod
def _build_result_rows(
source: Converter,
groups: dict[str | None, list[int]],
term_values: np.ndarray,
equation: str,
equation_terms_column: str,
) -> dict[str | None, Converter]:
"""Build one result row per group.
Each result row clones the first row in the group (to inherit its
metadata), overwrites the spectrum with the computed result, and
labels the equation-terms column with the equation itself. Within a
group each row is exposed to the equation by its equation-terms value
(e.g. ``signal``, ``monitor``).
"""
x_axis, spectra = source.features
result_rows: dict[str | None, Converter] = {}
for key, indices in groups.items():
rows: dict[str, np.ndarray] = {}
for index in indices:
term = str(term_values[index])
if not term.isidentifier():
continue
if term in rows:
group_part = f" in group '{key}'" if key is not None else ""
raise ValueError(
f"Duplicate equation term '{term}'{group_part}; rows "
"within a group must have unique values in the "
f"'{equation_terms_column}' column."
)
rows[term] = spectra[index]
computed = evaluate_equation(equation, rows)
result_rows[key] = (
source.take_rows([indices[0]])
.add_features(x_axis, computed)
.add_meta(
equation_terms_column,
np.array([equation], dtype=object),
)
)
return result_rows
@staticmethod
def _insert(
source: Converter,
groups: dict[str | None, list[int]],
result_rows: dict[str | None, Converter],
) -> Converter:
"""Insert each group result after the last source row in the group."""
n_rows = sum(len(indices) for indices in groups.values())
group_order = list(groups)
combined = Converter.concatenate(
[source] + [result_rows[key] for key in group_order]
)
# In the concatenated converter the result row of group k sits at
# index n_rows + k; interleave those indices into the source order.
result_position = {key: n_rows + k for k, key in enumerate(group_order)}
last_index_to_key = {max(indices): key for key, indices in groups.items()}
order: list[int] = []
for index in range(n_rows):
order.append(index)
if index in last_index_to_key:
order.append(result_position[last_index_to_key[index]])
return combined.take_rows(order)
@staticmethod
def _replace(
groups: dict[str | None, list[int]],
result_rows: dict[str | None, Converter],
) -> Converter:
"""Collapse each group to its single result row, in first-seen order."""
return Converter.concatenate([result_rows[key] for key in groups])