Source code for ewoksmx.tasks.ID29MeshPlot

import pathlib
import tempfile
from typing import Optional

import h5py
import matplotlib.pyplot as plt

from .base_tasks.icat_task import IcatCallbackTask




class ID29MeshPlot(
    IcatCallbackTask,
    input_names=["raw_data_path", "callback", "no_columns", "no_rows"],
    optional_input_names=[],
    output_names=["result_path"],
):


    def run(self):
        self.notify_icat({"status": "RUNNING"})

        # Find raw HDF5 file
        raw_data_path = pathlib.Path(self.inputs.raw_data_path[0])
        if not raw_data_path.exists():
            self.notify_icat(
                {
                    "logs": {
                        "message": f"Raw data directory does not exist: {raw_data_path}"
                    }
                }
            )
            self.notify_icat({"status": "ERROR"})
            raise FileNotFoundError(str(raw_data_path))
        aggregated_path = raw_data_path / "aggregated"
        list_h5_path = list(aggregated_path.glob("*.h5"))
        first_path = list_h5_path[0]
        first_file_name = first_path.name
        prefix = "_".join(first_file_name.split("_")[0:-1])
        data_path = aggregated_path / (prefix + ".h5")

        # Result directory
        prefix_without_master = prefix.replace("master_", "")
        tmp_14_days_path = pathlib.Path("/tmp_14_days")
        opid29_path = tmp_14_days_path / "opid29"
        if not opid29_path.exists():
            opid29_path.mkdir(mode=0o755)
        result_dir = pathlib.Path(
            tempfile.mkdtemp(prefix=prefix_without_master + "_", dir=opid29_path)
        )
        result_dir.chmod(0o755)
        self.notify_icat({"outputFolder": str(result_dir)})

        # Read raw HDF5 file
        no_columns = self.get_input_value("no_columns", None)
        no_rows = self.get_input_value("no_rows", None)
        with h5py.File(data_path, "r") as f:
            isHit = f["/entry_0000/processing/peakfinder/isHit"][()]
            isHit.shape = (no_rows, no_columns)

        # Save HDF5 data as png image
        result_path = result_dir / "mesh_plot.png"
        plt.imshow(isHit, interpolation="none")
        plt.savefig(result_path)

        self.notify_icat({"logs": {"message": f"Result path: {result_path}"}})

        self.outputs.result_path = str(result_path)

        self.notify_icat({"status": "FINISHED"})


    @property
    def _icat_callback_url(self) -> Optional[str]:
        return self.inputs.callback