Source code for ewoksbm29.tasks.mesh_rebuild

import os
from typing import Optional

from ..io.read_mesh import read_scan_parameters
from ..models.dahu import MeshRebuildParameters
from .base.dahu_ispyb import DahuJobWithIspybUpload




class DahuMeshRebuild(
    DahuJobWithIspybUpload,
    input_names=[
        "integrated_files",
        "scan_file_path",
        "scan_number",
    ],
    optional_input_names=[
        "mesh_rebuild_parameters",
        "retry_timeout",
        "retry_period",
    ],
):
    """Rebuild the 2D mesh with basic analysis.

    In addition to the inputs from `DahuJobWithIspybUpload`:

    Required inputs:
    - integrated_files (List[str]): Azimuthally integrated SAXS data.
    - scan_file_path (str): Bliss scan file name.
    - scan_number (str): Bliss scan number as a string.
    - retry_timeout (float): Timeout when trying to access the Lima image dataset or wait for the scan to be PREPARED.
    - retry_period (float): Period in retry loops.

    Optional inputs:
    - mesh_rebuild_parameters (dict): Extra mesh parameters (see `MeshRebuildParameters`).
    """

    DAHU_PLUGIN = "bm29.mesh"
    DAHU_INPUTS_MODEL = MeshRebuildParameters
    PROCESS_NAME = "mesh"
    CONFIG_NAME = "mesh"



    def dahu_parameters_initialize(self) -> dict:
        dahu_parameters = super().dahu_parameters_initialize()

        integrated_files = self.get_input_value("integrated_files")
        if integrated_files:
            dahu_parameters["integrated_files"] = integrated_files

        return dahu_parameters




    def dahu_parameters_finalize(self, dahu_parameters: dict) -> None:
        integrated_files = self.get_input_value("integrated_files")
        output_file = dahu_parameters.get("output_file")
        if integrated_files and not output_file:
            _, ext = os.path.splitext(integrated_files[0])
            output_filename = (
                os.path.commonprefix(integrated_files) + ext
            )  # does not exist
            output_file = self.output_filename_from_output_filename(
                output_filename, ".h5"
            )
            dahu_parameters["output_file"] = output_file

        mesh_rebuild_parameters = self.get_input_value("mesh_rebuild_parameters")
        if mesh_rebuild_parameters:
            dahu_parameters.update(mesh_rebuild_parameters)

        scan_file_path = self.get_input_value("scan_file_path")
        scan_number = self.get_input_value("scan_number")
        if scan_file_path and scan_number:
            retry_timeout = self.get_input_value("retry_timeout") or 10
            retry_period = self.get_input_value("retry_period") or 0.2
            dahu_parameters["scan"] = read_scan_parameters(
                scan_file_path,
                scan_number,
                retry_timeout=retry_timeout,
                retry_period=retry_period,
            )

        super().dahu_parameters_finalize(dahu_parameters)




    def dahu_parameters_save_path(
        self, dahu_parameters: MeshRebuildParameters
    ) -> Optional[str]:
        if dahu_parameters.output_file:
            return os.path.splitext(dahu_parameters.output_file)[0] + ".json"
        if dahu_parameters.integrated_files:
            return self.output_filename_from_output_filename(
                dahu_parameters.integrated_files[0], ".json"
            )