import numpy as np
from ewokscore import Task
from ewokscore.model import BaseInputModel, BaseOutputModel
from Orange.data import Table
from Orange.util import Enum
from pydantic import Field
from ewoksxas.converters import orange
from ewoksxas.converters.orange import VarType
from ewoksxas.fit.data import ArrayF64, XYData
from ewoksxas.fit.lca_fit import LCAModel
[docs]
class Labels(str, Enum):
COMPONENT_ID = "Component ID"
FILENAME_SCAN = "Filename / Scan Name"
[docs]
def assign_component_keys(converter: orange.Converter) -> list[str]:
"""Return a stable ``"C1".."Cn"`` key per component row, in row order.
Keys are positional (row 0 -> ``"C1"``, row 1 -> ``"C2"``, …), so they are
collision-free even when display names repeat and stay fixed regardless of
which components are excluded. Being a pure function of row position, this
is the single source of truth shared by the task and the widget.
Args:
converter: Converter over the component input table.
"""
return [f"C{i + 1}" for i in range(converter.n_rows)]
[docs]
class Outputs(BaseOutputModel):
Transformed_Data: Table
Data: Table
Components: Table
[docs]
class LinearCombinationAnalysis(Task, input_model=Inputs, output_model=Outputs): # type: ignore
"""Task to do linear combination analysis on X-ray absorption spectra."""
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
# Extract Parameters
parameters = self.inputs.parameters
self._xmin: float = parameters.get("xmin", -np.inf)
self._xmax: float = parameters.get("xmax", np.inf)
self._normalize_results: bool = parameters.get("normalize", False)
self._exclude_components: list[str] = parameters.get("exclude_components", [])
self._components_data: dict[str, XYData] = {}
self._component_keys: list[str] = []
self._component_display_names: dict[str, str] = {}
# Initialize Outputs.
self._solutions: list[ArrayF64] = []
self._sigmas: list[ArrayF64] = []
self._model_x: ArrayF64 = np.array([], dtype=np.float64)
self._model_y: list[ArrayF64] = []
self._scan_names: list[str] = []
self._metas: dict[str, list[float]] = {"rmse": [], "r2": []}
self._sample_converter = orange.Converter()
[docs]
def run(self):
"""Main execution flow for Linear Combination Analysis."""
self.arrange_components()
self._sample_converter = orange.Converter.from_table(self.inputs.Sample_Data)
sample_x, all_sample_y = self._sample_converter.features
self._scan_names = self._sample_converter.get_meta_values(
"Scan Name",
[f"Scan {i}" for i in range(self._sample_converter.n_rows)],
).tolist()
self._process_spectra(sample_x, all_sample_y)
self.outputs.Transformed_Data = self._create_transformed_table()
self.outputs.Data = self._create_data_table()
self.outputs.Components = self._create_components_table()
def _create_transformed_table(self) -> Table:
"""Compiles the LCA transformed data into an Orange Table."""
output_y: ArrayF64 = np.stack(self._solutions)
attributes = [{"name": name} for name in self._component_display_names.values()]
converter = self._sample_converter.with_features(
np.array(self._component_keys), output_y, attributes=attributes
)
for meta, meta_val in self._metas.items():
converter.add_meta(name=meta, data=np.asarray(meta_val))
zipped_components = zip(
self._component_keys, np.asarray(self._sigmas).T, strict=True
)
for key, sigma in zipped_components:
converter.add_meta(f"{key} σ", sigma)
return converter.to_table()
def _create_data_table(self) -> Table:
"""Compiles the optimized model features and metas into an Orange Table.
The features are unchanged (the model spectrum on the model x-grid);
only the per-component weight and error meta names are re-keyed to the
stable C-keys, with each weight meta carrying its display name.
"""
output_y: ArrayF64 = np.stack(self._model_y)
converter = self._sample_converter.with_features(self._model_x, output_y)
for meta, meta_val in self._metas.items():
converter.add_meta(name=meta, data=np.asarray(meta_val))
zipped_components = zip(
self._component_keys,
np.asarray(self._solutions).T,
np.asarray(self._sigmas).T,
self._component_display_names.values(),
strict=True,
)
for key, solution, sigma, name in zipped_components:
(
converter.add_meta(key, solution, attributes={"name": name}).add_meta(
f"{key} σ", sigma, attributes={"name": f"{name} σ"}
)
)
return converter.to_table()
def _create_components_table(self) -> Table:
"""Compiles the component solutions into an Orange Table.
One row per component: the ``ID`` meta holds the stable C-keys and the
``Name`` meta the human-readable names. Feature columns are one per
sample, headed by a unique sample identifier (see
``_resolve_sample_headers``) so two samples sharing a scan name cannot
collide into duplicate column names.
"""
output_y: ArrayF64 = np.stack(self._solutions).T
headers = self._resolve_sample_headers()
names = [self._component_display_names[key] for key in self._component_keys]
converter = (
orange.Converter()
.add_features(np.asarray(headers), output_y)
.add_meta("ID", np.asarray(self._component_keys), VarType.TEXT)
.add_meta("Name", np.asarray(names), VarType.TEXT)
)
return converter.to_table()
def _resolve_sample_headers(self) -> list[str]:
"""Pick unique per-sample column headers for the Components table.
Prefers the user-selected ``sample_identifier`` then ``"Scan Name"``
when either is unique across the samples. Otherwise it falls back to the
converter's content-hash row identity (``group_id``), then Orange's
built-in ``Table.ids``, then synthetic ``"S1".."Sn"`` -- all guaranteed
unique. Uniqueness is required because these become Orange feature
names, which a Domain forbids from repeating.
"""
n_rows = self._sample_converter.n_rows
identifier = self.inputs.parameters.get("sample_identifier", "Scan Name")
if identifier == Labels.FILENAME_SCAN:
group_id = self._sample_converter.get_group_id()
dense: dict = {}
for value in group_id.tolist():
dense.setdefault(value, str(len(dense)))
return [dense[value] for value in group_id.tolist()]
for candidate in (identifier, "Scan Name"):
if not candidate:
continue
try:
values = [
str(value)
for value in self._sample_converter.get_meta_values(candidate)
]
except KeyError:
continue
if len(set(values)) == n_rows:
return values
group_id = self._sample_converter.get_group_id()
if group_id is not None and len(set(group_id.tolist())) == n_rows:
return [str(value) for value in group_id]
ids = self.inputs.Sample_Data.ids
if len(set(ids.tolist())) == n_rows:
return [str(value) for value in ids]
return [f"S{i + 1}" for i in range(n_rows)]
[docs]
def arrange_components(self) -> None:
"""Extracts and labels component data from the input Orange Table.
Components are keyed by the stable C-keys (see ``assign_component_keys``)
so duplicate human-readable names no longer collapse distinct components.
The chosen ``component_identifier`` column supplies each component's
display name, falling back to the C-key when blank or missing.
"""
comp_converter = orange.Converter.from_table(self.inputs.Component_Data)
comp_x, comp_y = comp_converter.features
keys_by_row = assign_component_keys(comp_converter)
identifier = self.inputs.parameters.get("component_identifier", "Scan Name")
names_by_row = comp_converter.get_meta_values(
identifier, default=np.asarray(keys_by_row)
)
self._components_data = {}
self._component_display_names = {}
for row, key in enumerate(keys_by_row):
if key in self._exclude_components:
continue
self._components_data[key] = XYData(x=comp_x, y=comp_y[row])
self._component_display_names[key] = str(names_by_row[row]).strip() or key
self._component_keys = list(self._components_data.keys())
def _process_spectra(self, sample_x: ArrayF64, all_sample_y: ArrayF64) -> None:
"""Iterates over all sample spectra, applying LCA fitting."""
for sample_y in all_sample_y:
sample = XYData(x=sample_x, y=sample_y)
lca = LCAModel(
sample, self._components_data, xmin=self._xmin, xmax=self._xmax
)
fit = lca.fit()
self._model_y.append(fit.model.y)
if not self._model_x.size:
self._model_x = fit.model.x
solution = fit.solution_norm if self._normalize_results else fit.solution
sigma = fit.sigma_norm if self._normalize_results else fit.sigma
self._solutions.append(solution)
self._sigmas.append(sigma)
self._metas["rmse"].append(fit.rmse)
self._metas["r2"].append(fit.r2)