Source code for pycif.plugins.obsoperators.standard.transforms.utils.dump_transform_description

import copy
from ......utils.yml import ordered_dump




def dump_transform_description(self, all_transforms, mapper):
    towrite = []
    for transform in all_transforms.attributes:
        transf = getattr(all_transforms, transform)
        transf_name = transf.plugin.name
        transf_version = transf.plugin.version
        transf_mapper = mapper[transform]
        precursors = transf_mapper["precursors"]
        successors = transf_mapper["successors"]
        inputs = transf_mapper["inputs"]
        outputs = transf_mapper["outputs"]

        towrite.extend([
            "######################################",
            f"{transform} ({transf_name}/{transf_version}):",
            "",
        ])

        if inputs != {}:
            towrite.append("--------")
            towrite.append("Inputs:")
            for key in inputs:
                towrite.append(f"    - {key}")
            towrite.append("")

        if outputs != {}:
            towrite.append("--------")
            towrite.append("Outputs:")
            for key in outputs:
                towrite.append(f"    - {key}")

        if precursors != {}:
            towrite.append("--------")
            towrite.append("Precursors:")
            for key in precursors:
                towrite.append(f"    - {key}:")
                for tr in precursors[key]:
                    towrite.append(f"        - {tr}")

        if successors != {}:
            towrite.append("--------")
            towrite.append("Successors:")
            for key in successors:
                towrite.append(f"    - {key}:")
                for tr in successors[key]:
                    towrite.append(f"        - {tr}")

    # Dump required inputs for debugging purposes
    with open(f"{self.workdir}/obsoperator/transform_description.txt", "w") as f:
        f.write("\n".join(towrite))

    # Now dump the transform pipe
    if self.use_dask:
        return

    for mode, order_pipe in zip(
            ["forward", "adjoint"],
            [self.period_order_fwd, self.period_order_adj]
    ):
        towrite = []
        for ddi, transform, direction in order_pipe:
            transf_mapper = mapper[transform]
            inputs = transf_mapper["inputs"]
            name = getattr(all_transforms, transform).plugin.name
            version = getattr(all_transforms, transform).plugin.version
            towrite.append(f"{transform:15} ({name:25} / {version:15}) "
                           f"{ddi} / {direction:8} / {list(inputs.keys())}")

        with open(f"{self.workdir}/obsoperator/"
                  f"transform_pipe_{mode}.txt", "w") as f:
            f.write("\n".join(towrite))