Source code for ewoksid16a.tasks.common.masterfile

import os

import h5py
from ewokscore import Task
from silx.io import h5py_utils
from silx.io.url import DataUrl

DEFAULTS = {"resolve_links": True}




class AddEntryToMasterfile(
    Task,
    input_names=["masterfile_uri", "output_root_uri"],
    optional_input_names=["resolve_path"],
    output_names=["masterfile_uri", "output_root_uri"],
):
    """
    This task add or replace an entry in the masterfile
    """

    def _resolve_h5_link(self, output_root_uri, already_visited=None):
        if already_visited is None:
            already_visited = []

        output_root_uri_txt = (
            output_root_uri.file_path() + "::" + output_root_uri.data_path()
        )

        if output_root_uri_txt in already_visited:
            raise RuntimeError(
                f"{output_root_uri} already visited, must have a circular linking issue!"
            )

        already_visited += [
            output_root_uri_txt,
        ]

        with h5py_utils.open_item(output_root_uri.file_path(), "/") as fd:
            obj = fd.get(output_root_uri.data_path(), getlink=True)

            if isinstance(obj, h5py.SoftLink):
                linkval = obj.link

                if not os.path.isabs(linkval):
                    linkval = obj.path.normpath(
                        os.path.join(output_root_uri.data_path(), linkval)
                    )

                return self._resolve_h5_link(
                    DataUrl(output_root_uri.file_path() + "::" + linkval),
                    already_visited,
                )

            elif isinstance(obj, h5py.ExternalLink):
                return self._resolve_h5_link(
                    DataUrl(obj.path + "::" + obj.link), already_visited
                )

        return output_root_uri



    def run(self):

        pars = {**DEFAULTS, **self.get_input_values()}

        muri = pars["masterfile_uri"].split("::")
        ouri = pars["output_root_uri"].split("::")

        masterfile_uri = DataUrl(file_path=muri[0], data_path=muri[1])
        output_root_uri = DataUrl(file_path=ouri[0], data_path=ouri[1])

        if pars["resolve_links"]:
            output_root_uri = self._resolve_h5_link(output_root_uri)

        relpath = os.path.relpath(
            output_root_uri.file_path(), os.path.dirname(masterfile_uri.file_path())
        )

        with h5py.File(masterfile_uri.file_path(), "a") as fd:
            if masterfile_uri.data_path() in fd:
                del fd[masterfile_uri.data_path()]

            fd[masterfile_uri.data_path()] = h5py.ExternalLink(
                relpath, output_root_uri.data_path()
            )

        self.outputs["masterfile_uri"] = self.inputs["masterfile_uri"]
        self.outputs["output_root_uri"] = self.inputs["output_root_uri"]