Source code for ewoksxas.tasks.normalize

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

from ewoksxas.converters import larch, orange




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





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





class Normalize(Task, input_model=Inputs, output_model=Outputs):  # type: ignore
    """Task to normalize X-ray absorption spectra using Larch."""



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

        converter = orange.Converter.from_table(data)

        energy, mu = converter.features
        groups = larch.create_groups(energy, mu)
        for group in groups:
            xafs.pre_edge(group, group=group, **parameters)

        attribute_names = ["flat", "e0", "edge_step", "pre_edge", "post_edge"]
        attributes = larch.get_attribute_values(groups, attribute_names)

        converter = (
            orange.Converter()
            .add_features(energy, attributes["flat"])
            .add_meta("e0", attributes["e0"])
            .add_meta("edge_step", attributes["edge_step"])
        )
        self.outputs.Data = converter.to_table()
        self.outputs.Groups = groups






def main():
    pass



if __name__ == "__main__":
    main()