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 transform pipeline according to user choices. This includes the
    explicit definition of sub-pipelines (main, control vector side and observation
    vector side), definition based on aliases in the ``datavect``, and
    transforms automatically added depending on compatibility of successive
    input/output formats (including domain definition, dates and units)."""

    # 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
    )

    # for transform in all_transforms.attributes:
    #     print(transform)
    #     print("\tPrecursors:")
    #     for trid in mapper[transform]["precursors"]:
    #         print(f"\t\t{trid}, {mapper[transform]['precursors'][trid]}")
    #     print("\tSuccessors:")
    #     for trid in mapper[transform]["successors"]:
    #         print(f"\t\t{trid}, {mapper[transform]['successors'][trid]}")
    #     print()

    # mapper["dump2inputs_std_00046"]

    # print(__file__)
    # import code
    # code.interact(local=dict(locals(), **globals()))

    # 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

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

    # 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)