Source code for pycif.plugins.obsoperators.standard.flushrun

from logging import info, warning
from pathlib import Path

from ....utils.check.errclass import PluginError




def flushrun(self, workdir, rundir, mode, transform_pipe, full_flush=True):
    """Remove intermediate files produced by transforms that are no longer needed.

    Iterates over every transform in *transform_pipe* and calls each
    transform's own ``flushrun`` method to clean up its output files in
    *rundir*.  In adjoint mode, when the operator is not running in
    approximate mode, the forward reference directory of each transform is
    also flushed — provided it lies inside *workdir*, to avoid accidentally
    deleting files outside the managed tree.

    Args:
        self (ObsOperator): the obs-operator plugin instance.
        workdir (str): root working directory; used to check that
            ``transf.adj_refdir`` is a safe path to flush.
        rundir (str): the run sub-directory whose files should be cleaned.
        mode (str): execution mode — one of ``'fwd'``, ``'tl'``, or
            ``'adj'``; controls whether ``adj_refdir`` of each transform is
            also flushed.
        transform_pipe: the :class:`~pycif.utils.classes.transforms.Transform`
            object holding all transforms for this run.
        full_flush (bool, optional): forwarded to each transform's own
            ``flushrun``; if ``False`` only a partial cleanup is performed
            (exact behaviour is transform-specific).  Defaults to ``True``.

    Raises:
        PluginError: caught internally and logged as a warning if a
            transform's ``flushrun`` raises it; execution continues with the
            remaining transforms.
    """
    info(f"Flushing unnecessary files in '{rundir}'")

    # Loop over all transforms and flush what's necessary.
    # By default the flushrun function of transforms does nothing
    for transform in transform_pipe.attributes:
        transf = getattr(transform_pipe, transform)

        try:
            transf.flushrun(rundir, mode, transform, full_flush=full_flush)
            if mode == "adj" and not getattr(self, "approx_operator", False):
                if hasattr(transf, "adj_refdir"):
                    # Only flush 'transf.adj_refdir' if it is inside of the workdir
                    if Path(transf.adj_refdir).is_relative_to(workdir):
                        transf.flushrun(
                            transf.adj_refdir, mode, transform, full_flush=full_flush
                        )

        except PluginError as e:
            warning(f"An exception occured in {__file__}: {str(e)}")