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

import itertools
import numpy as np
import pandas as pd
from logging import debug, info

from .....utils.classes.transforms import Transform
from .utils import \
    check_datavect, dump_transform_description, init_entry
from .init_mainpipe import init_mainpipe
from .init_control_transformations import init_control_transformations
# from .init_default_transformations import init_default_transformations
from .init_obsvect_transformations import init_obsvect_transformations
from .period_pipe import period_pipe
from .utils.connect_pipes import connect_pipes
from .dump_read_inout import dump_read_inout
from .batch_computation import batch_computation




def init_transform(self):
    """Initialize the complete transform pipeline for the observation operator.

    Assembles the ordered
    :class:`~pycif.utils.classes.transforms.Transform` pipeline that the
    operator will execute at run time.  The pipeline is built from four
    sub-pipelines applied in sequence:

    1. **Observation-vector side** — transforms from ``obsvect.transform_pipe``
       plus a mandatory ``toobsvect`` step for each observed species, and a
       ``satellites`` step for satellite components (via
       :func:`~.init_obsvect_transformations`).
    2. **Main pipe** — transforms from ``obsoperator.transform_pipe``
       (defaults to a single ``run_model`` step when none are specified), via
       :func:`~.init_mainpipe`.
    3. **Control-vector side** — transforms from ``controlvect.transform_pipe``
       (via :func:`~.init_control_transformations`).
    4. **Dump / load wrappers** — ``dump2inputs`` and ``loadfromoutputs``
       transforms inserted automatically where ``force_dump`` or
       ``force_loadout`` flags are set (via :func:`~.dump_read_inout`).

    After assembly the pipeline is ordered so that precursor transforms
    always run before their successors via :func:`~.period_pipe`.  Data
    availability is verified by
    :func:`~.utils.check_datavect.check_datavect`, and a human-readable
    description is written to disk by
    :func:`~.utils.dump_transform_description.dump_transform_description`.

    If ``self.batch_computation`` is configured, the pipeline is further
    modified for Monte-Carlo batch execution via
    :func:`~.batch_computation.batch_computation`.

    Args:
        self (ObsOperator): the obs-operator plugin instance.  On return,
            ``self.transform_pipe``, ``self.period_order_fwd``, and
            ``self.period_order_adj`` are populated.
    """

    # Initializes the overall transform pipe
    all_transforms = Transform.from_dict({})
    all_transforms.default_index = 0

    mapper = {}
    backup_comps = {}

    info('Initializing observation operator pipe')

    # Initializing transformation after the observation operator pipe
    # i.e., on the observation side
    self.mainpipe = []
    init_obsvect_transformations(
        self, all_transforms, self.obsvect,
        backup_comps, mapper)

    # Initializing the main pipe as defined by the observation operator
    init_mainpipe(self, all_transforms,
                  backup_comps, mapper)

    # Initializing transformation on the control vector side
    init_control_transformations(
        self, all_transforms, self.controlvect,
        backup_comps, mapper)

    # Add transforms to dump/read inputs/outputs of individual transforms
    dump_read_inout(self, all_transforms,
                    backup_comps, mapper)

    # Now initialize the pipe entry
    init_entry.init_entry(
        self, all_transforms,
        backup_comps, mapper
    )

    # Save final transform pipe to the observation observator
    self.transform_pipe = all_transforms
    self.transform_pipe.mapper = mapper

    # Distribute the order of transforms and sub-periods
    # Computing longer periods in the end
    period_order_fwd, period_order_adj = period_pipe(
        self, all_transforms, mapper)
    self.period_order_fwd = period_order_fwd
    self.period_order_adj = period_order_adj

    # Check whether all data is available
    check_datavect(self, all_transforms, backup_comps, mapper)

    # Dump a full description of the transforms and of the transform pipe
    dump_transform_description(self, all_transforms, mapper)

    # Perturb transform pipeline for batch computation of Monte-Carlo samples
    if hasattr(self, "batch_computation"):
        batch_computation(self, all_transforms, mapper)