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

import copy
from ......utils.yml import ordered_dump
from ......utils.check.errclass import CifError




def check_datavect(self, all_transforms, backup_comps, mapper):
    """Check that the data vector has no missing required inputs.

    Dumps ``self.required_inputs`` to ``pipe_inputs.txt`` for debugging.
    If ``self.missing`` is empty, returns immediately. Otherwise, dumps
    the full list of missing inputs to ``missing_inputs.txt`` and the
    subset needed for initialization (``self.init_missing``) to
    ``missing_inputs_init.txt``, both under
    ``{self.workdir}/obsoperator/``. Raises a :class:`CifError` unless
    ``self`` has an ``init_inputs`` attribute and no inputs are missing
    for initialization.

    Args:
        self: The obs operator, exposing ``workdir``, ``required_inputs``,
            ``missing`` and ``init_missing``.
        all_transforms: Namespace holding all registered transform instances,
            used to resolve each transform's plugin name/version.
        backup_comps: Unused backup of components, kept for interface
            consistency with related functions.
        mapper: Unused transform mapper, kept for interface consistency
            with related functions.

    Raises:
        CifError: If required inputs are missing from the data vector.
    """
    # Dump required inputs for debugging purposes
    with open(f"{self.workdir}/obsoperator/pipe_inputs.txt", "w") as f:
        ordered_dump(f, self.required_inputs)
    
    # Dump a detailed log about missing inputs
    if self.missing == {}:
        return

    todump = []
    for transform in self.missing:
        transf = getattr(all_transforms, transform)
        todump.extend([
            "##################",
            f"{transform} ({transf.plugin.name}/{transf.plugin.version}):",
            ""
        ])
        for cmp in self.missing[transform]:
            todump.extend(
                [f"   - {cmp}:"] + [
                    f"       - {prm}" for prm in self.missing[transform][cmp]
                ])

    todump_missing = []
    for transform in self.init_missing:
        transf = getattr(all_transforms, transform)
        todump_missing.extend([
            "##################",
            f"{transform} ({transf.plugin.name}/{transf.plugin.version}):",
            ""
        ])
        for cmp in self.init_missing[transform]:
            todump_missing.extend(
                [f"   - {cmp}:"] + [
                    f"       - {prm}" for prm in self.init_missing[transform][cmp]
                ])

        todump_missing.append("")

    # Dump to file
    file_missing = f"{self.workdir}/obsoperator/missing_inputs.txt"
    with open(file_missing, "w") as f:
        f.write("\n".join(todump))

    file_missing_init = f"{self.workdir}/obsoperator/missing_inputs_init.txt"
    with open(file_missing_init, "w") as f:
        f.write("\n".join(todump_missing))

    # Raise exception only if missing inputs for initialization if required
    ignore_missing = hasattr(self, "init_inputs")
    if ignore_missing and todump_missing == []:
        if todump_missing == []:
            return

        # Otherwise raise exception
        raise CifError(f"There are missing elements in your data vector. Only missing elements for initialization ('init_inputs' is switched on) are reported.Please see {file_missing_init} for details")

    # Otherwise raise exception
    
    raise CifError(f"There are missing elements in your data vector. Please see {file_missing} for details")