Source code for pycif.plugins.obsoperators.standard.transforms.dump_read_inout

import copy
import itertools
from logging import debug

from pycif.utils.check.errclass import CifError
from .utils import add_default

from .utils.propagate_parameters import propagate_parameters




def dump_read_inout(self, all_transforms,
                    backup_comps, mapper):
    """Insert dump and load transforms for explicit input/output file I/O.

    Scans all transforms for inputs flagged with ``force_dump`` and outputs
    flagged with ``force_loadout``, then inserts extra transforms into the
    pipeline accordingly:

    * **force_dump** on an *input*: inserts a ``dump2inputs`` transform
      immediately **before** the flagged transform to write that input to
      disk so it can be inspected or reused.
    * **force_loadout** on an *output*: inserts a ``loadfromoutputs``
      transform immediately **after** the flagged transform to reload that
      output from disk (useful for transforms whose results must survive
      across restarts).

    Before scanning,
    :func:`~.utils.propagate_parameters.propagate_parameters` is called to
    ensure ``force_load`` flags have been propagated through the pipeline.

    Args:
        self (ObsOperator): the obs-operator plugin instance.
        all_transforms: the :class:`~pycif.utils.classes.transforms.Transform`
            object holding all transforms; modified in-place.
        backup_comps (dict): backed-up component definitions forwarded to
            :func:`~.utils.add_default.add_default`.
        mapper (dict): the pipeline mapper dictionary; updated in-place to
            include entries for the newly inserted transforms.
    """

    # First propagate force_load inputs to be sure to load
    # all inputs needed for a given transforms
    propagate_parameters(all_transforms, mapper)

    # Dump to input files if necessary
    ref_transforms = copy.deepcopy(all_transforms.attributes)
    itransf = 0
    dump2inputs_deps = {}
    for transform in ref_transforms:
        transf = getattr(all_transforms, transform)
        transf_mapper = mapper[transform]
        precursors = transf_mapper["precursors"]
        successors = transf_mapper["successors"]

        dump2inputs_ids = {}
        for trid in transf_mapper["inputs"]:
            if not transf_mapper["inputs"][trid].get("force_dump", False):
                continue

            # Adding a dump transformation before the present transform
            yml_dict = {
                "plugin": {
                    "name": "dump2inputs",
                    "version": "std",
                    "type": "transform",
                },
                "component": [trid[0]],
                "parameter": [trid[1]],
                "successor": transform,
                "precursor":
                    precursors[trid][0] if len(precursors[trid]) > 0
                    else None
            }
            ref_successor = {trid: transform}
            ref_precursor = {trid: precursors[trid][0]} \
                if len(precursors[trid]) > 0 else None
            new_transf, dumpin_id = add_default.add_default(
                self,
                all_transforms,
                yml_dict,
                position="index",
                index=all_transforms.attributes.index(transform),
                mapper=mapper,
                init=True,
                backup_comps=backup_comps,
                successor=ref_successor,
                precursor=ref_precursor
            )
            itransf += 1

            # Save ids of dump2inputs transforms to add extra dependencies if needed
            dump2inputs_ids[trid] = dumpin_id

        transf_mapper["dump2inputs_ids"] = dump2inputs_ids

        itransf += 1

    # Read output files if necessary
    ref_transforms = copy.deepcopy(all_transforms.attributes)
    itransf = 0
    for transform in ref_transforms:
        transf = getattr(all_transforms, transform)
        transf_mapper = mapper[transform]
        precursors = transf_mapper["precursors"]
        successors = transf_mapper["successors"]

        itransf += 1
        loadfromoutputs_ids = {}
        for trid in transf_mapper["outputs"]:
            if transf_mapper["outputs"][trid].get("force_loadout", False):
                # Adding a load transformation after the present one
                yml_dict = {
                    "plugin": {
                        "name": "loadfromoutputs",
                        "version": "std",
                        "type": "transform",
                    },
                    "component": [trid[0]],
                    "parameter": [trid[1]],
                    "precursor": transform,
                }

                ref_precursor = {trid: transform}
                ref_successor = {trid: successors[trid]} \
                    if len(successors[trid]) > 0 else None
                new_transf, loadout_id = add_default.add_default(
                    self,
                    all_transforms,
                    yml_dict,
                    position="index",
                    index=all_transforms.attributes.index(transform) + 1,
                    mapper=mapper,
                    init=True,
                    backup_comps=backup_comps,
                    precursor=ref_precursor,
                    successor=ref_successor
                )
                itransf += 1

                # Save ids of loadfromoutputs transforms to add extra dependencies if needed
                loadfromoutputs_ids[trid] = loadout_id

        transf_mapper["loadfromoutputs_ids"] = loadfromoutputs_ids