Source code for ewoksxrdct.tasks.integrate

from pathlib import Path

from ewokscore import Task
from ewokscore.model import BaseInputModel, BaseOutputModel
from integrator.app.integrate_mp import integrate_mp_cli
from integrator.app.utils import get_distributed_integrator_params
from pydantic import Field




class IntegrateInputModel(BaseInputModel, validate_assignment=True):
    config_path: str = Field(
        ...,
        description="Full path to the integrator configuration file.",
    )





class IntegrateOutputModel(BaseOutputModel):
    processed_datasets: list[str] = Field(
        ...,
        description="List of processed datasets.",
    )
    integration_output_path: list[str] = Field(
        ...,
        description="Full path to the azimuthally integrated data files.",
    )





class Integrate(
    Task,
    input_model=IntegrateInputModel,
    output_model=IntegrateOutputModel,
):


    def run(self) -> None:
        """Execute the task to run the azimuthal integration from the config file."""
        try:
            _, _, _, datasets, output_files = get_distributed_integrator_params(
                self.inputs.config_path,
                ignore_already_processed=False,
                parsing_error_handling="raise",
            )
        except Exception as e:
            raise RuntimeError(
                f"Failed to parse config file: {self.inputs.config_path}. {e}"
            ) from e

        if not datasets or not output_files:
            raise RuntimeError(
                f"No datasets found in config file: {self.inputs.config_path}. "
                "Check that the dataset location is correct."
            )

        # check if datasets were already processed
        not_processed = [f for f in output_files if not Path(f).exists()]
        if not_processed:
            integrate_mp_cli(run_with_arg=False, config_path=self.inputs.config_path)

        self.outputs.processed_datasets = [dataset.fname for dataset in datasets]
        self.outputs.integration_output_path = [
            str(output_file) for output_file in output_files
        ]