Source code for ewoksxas.tasks.peak_fit
from typing import TYPE_CHECKING
import numpy as np
from ewokscore import Task
from ewokscore.model import BaseInputModel, BaseOutputModel
from Orange.data import Table
from pydantic import Field
from ewoksxas.converters import orange
from ewoksxas.fit.peak_fit import MultiPeakFitResult, PeakModel, peak_descriptors
if TYPE_CHECKING:
from ewoksxas.fit.data import ArrayF64
# Per-fit scalar descriptors (one value per spectrum, not per peak).
_FIT_DESCRIPTORS = ("r2", "rmse")
[docs]
def descriptor_name(base: str, peak_index: int) -> str:
"""Return the per-peak metadata column name for a descriptor.
The single place that encodes the per-peak column-naming scheme, shared by
this task and the Orange widget that reads the output table back.
"""
return f"{base}_{peak_index}"
[docs]
class Outputs(BaseOutputModel):
Data: Table
[docs]
class PeakFitting(Task, input_model=Inputs, output_model=Outputs): # type: ignore
"""Task to fit one or several peaks of a single shape to each spectrum.
Wraps :class:`ewoksxas.fit.peak_fit.PeakModel`: every spectrum in the input
table is fitted with the configured peak shape and baseline (read from
``parameters`` via ``PeakModel.from_dict``), and the fitted model curves are
returned with the per-peak descriptors (center, height, area, FWHM and its
Gaussian/Lorentzian contributions and the mixing fraction) as metadata.
Each per-peak descriptor ``base`` becomes the columns ``base_0``, ``base_1``,
... up to the largest peak count fitted across the run, NaN-padded for
spectra with fewer peaks; the bare ``base`` column aliases the first peak so
single-peak consumers keep working. The per-fit quality (``r2``, ``rmse``)
and the integer ``n_peaks`` are added once per spectrum. The input table's
metadata (e.g. ``Scan Name`` and any acquisition metas) is carried through to
the output, with these fit descriptors taking precedence on name collisions.
"""
[docs]
def run(self):
"""Fit each spectrum and assemble the model curves and descriptors."""
converter = orange.Converter.from_table(self.inputs.Data)
x, all_y = converter.features
parameters = self.inputs.parameters
results: list[MultiPeakFitResult] = []
model_x: ArrayF64 = np.array([], dtype=np.float64)
model_y: list[ArrayF64] = []
for y in all_y:
fit = PeakModel.from_dict({"x": x, "y": y}, parameters).fit()
if not model_x.size:
model_x = fit.model.x
model_y.append(fit.model.y)
results.append(fit)
out = converter.with_features(model_x, np.stack(model_y))
self._add_descriptor_metas(out, results)
self.outputs.Data = out.to_table()
@staticmethod
def _add_descriptor_metas(
out: "orange.Converter", results: list[MultiPeakFitResult]
) -> None:
"""Add per-peak, bare-alias and per-fit descriptor columns to ``out``."""
max_peaks = max(result.n_peaks for result in results)
# Every spectrum is fitted with the same shape, so the per-peak
# descriptor set (which FWHM columns, whether eta) is shared.
for base in peak_descriptors(results[0].shape):
for index in range(max_peaks):
column = [
result.peaks[index][base] if index < result.n_peaks else np.nan
for result in results
]
out.add_meta(descriptor_name(base, index), _as_float_array(column))
# Bare alias to the first (leftmost) peak for single-peak consumers.
out.add_meta(base, _as_float_array([r.peaks[0][base] for r in results]))
for scalar in _FIT_DESCRIPTORS:
out.add_meta(scalar, _as_float_array([getattr(r, scalar) for r in results]))
out.add_meta("n_peaks", _as_float_array([r.n_peaks for r in results]))
def _as_float_array(values: list[float]) -> "ArrayF64":
"""Return ``values`` as a float64 array (NaN-friendly metadata column)."""
return np.asarray(values, dtype=np.float64)
if __name__ == "__main__":
main()