Source code for ewoksxas.tasks.xftf

import numpy as np
from ewokscore import Task
from ewokscore.model import BaseInputModel, BaseOutputModel
from larch import xafs
from larch.symboltable import Group
from Orange.data import Table
from pydantic import Field

from ewoksxas.converters import larch, orange
from ewoksxas.converters.resample import Alignment, resample_to_common_grid




class Inputs(BaseInputModel):
    Data: Table
    parameters: dict = Field(default_factory=dict)





class Outputs(BaseOutputModel):
    Data: Table
    Groups: list[Group]





class Xftf(Task, input_model=Inputs, output_model=Outputs):  # type: ignore
    """Task to calculate k-space for X-ray absorption spectra using Larch."""



    def run(self):
        data = self.inputs.Data
        parameters: dict = dict(self.inputs.parameters)

        # Alignment mode is consumed here; never forward it to xafs.xftf.
        alignment = Alignment(parameters.pop("resample", Alignment.INTERPOLATION))

        input_converter = orange.Converter.from_table(data)
        k, chi = input_converter.features
        groups = larch.create_groups(x=k, y=chi, mode="k")

        for group in groups:
            xafs.xftf(group, group=group, **parameters)

        r_list = [group.r for group in groups]
        mag_list = [group.chir_mag for group in groups]

        if len({r.size for r in r_list}) == 1:
            # Every spectrum already shares one R grid: no alignment needed.
            x_out = r_list[0]
            y_out = np.asarray(mag_list)
        else:
            x_out, y_out = resample_to_common_grid(r_list, mag_list, alignment)

        self.outputs.Data = input_converter.with_features(x_out, y_out).to_table()
        self.outputs.Groups = groups