Source code for ewoksmx.tasks.characterisation.xds_background_image

from esrf_pathlib import ESRFPath
from ewokscore.model import BaseInputModel
from ewokscore.model import BaseOutputModel
from ewokscore.task import Task
from pydantic import Field

from ewoksmx.models.characterisation_models import MetadataModel
from ewoksmx.shell_utils.execute_local import execute_bash_commands
from ewoksmx.utils.xds_utils import generate_xdsinp




class InputModel(BaseInputModel):
    aggregated_hdf5_path: ESRFPath = Field(
        ...,
        description="Paths to aggregated HDF5 master file",
    )
    metadata: MetadataModel





class OutputModel(BaseOutputModel):
    xds_background_path: ESRFPath = Field(
        ..., description="Path to XDS background image file"
    )
    metadata: MetadataModel





class XDSBackgroundImage(Task, input_model=InputModel, output_model=OutputModel):
    """Run XDS INIT job to generate background image."""



    def run(self):
        metadata_dict = self.inputs.metadata.model_dump()
        list_xdsinp = generate_xdsinp(
            job_list=["INIT"],
            master_path=self.inputs.aggregated_hdf5_path,
            metadata=metadata_dict,
        )
        # Write XDS.INP file in working directory
        working_directory = self.inputs.metadata.workflow_parameters.working_directory
        xds_inp_path = working_directory / "XDS.INP"
        if xds_inp_path.exists():
            xds_inp_path.unlink()
        with open(xds_inp_path, "w") as f:
            for line in list_xdsinp:
                f.write(f"{line}\n")
        # Execute XDS
        list_command = ["module load xds", "xds_par"]
        result = execute_bash_commands(list_command, working_directory)
        result.raise_on_error()
        # Path to background image
        xds_backhround_path = working_directory / "BKGINIT.cbf"
        if not xds_backhround_path.exists():
            raise FileNotFoundError(
                f"XDS background image not found: {xds_backhround_path}"
            )
        self.outputs.xds_background_path = ESRFPath(xds_backhround_path)
        self.outputs.metadata = self.inputs.metadata