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

import itertools
import copy

from .utils import add_default
from .....utils.classes.transforms import Transform


[docs]def init_mainpipe(self, all_transforms, backup_comps, mapper): """Initialize the core of the transform_pipe depending on the list of transformations specified in obsoper.transform_pipe. .. warning:: If no transform_pipe is specified, the CTM model specified in the Yaml is run by default. On the opposite, if ``transform_pipe`` is specified in the observation operator, only transforms explicitly specified will be used. Thus, if custom transforms need to be run on top of the model, one should not forget to include the transform ``run_model`` in the ``transform_pipe``. Another option (recommended for most applications) is to use the ``controlvect`` and ``obsvect`` transform_pipes to define transforms related to the control vector and to the observation vector respectively. """ # Initializes the overall transform pipe # If no pipe is specified, assume that the model should be run if not hasattr(self, "transform_pipe"): if not self.ignore_model: yml_dict = {"transform": { "run_model": { "plugin": { "name": "run_model", "version": "std", "type": "transform", } } }} else: yml_dict = {"transform": {}} self.transform_pipe = Transform.load_from_dict(yml_dict).transform main_transforms = self.transform_pipe transfs_ids = getattr(main_transforms, "attributes", []) # Loops backwards on available transformations and update inputs/out # According to model outputs itransf = len(transfs_ids) while itransf > 0: itransf -= 1 ntransf_loc = 0 transform = transfs_ids[itransf] # Initializes mapper if not already done if transform not in mapper: # Replacing the transform by a transform class # if not already initialized transf = getattr(main_transforms, transform) if transf is None: transf = Transform.from_dict({}, orig_name=transform) setattr(all_transforms, transform, transf) all_transforms.attributes.insert(0, transform) # Updating the general mapper, and creates a local one transf_mapper = transf.ini_mapper( general_mapper=mapper, backup_comps=backup_comps, transform_name=transform ) transf.mapper = transf_mapper mapper[transform] = transf_mapper # dict_all_inputs[transform] = transf_mapper["inputs"] # dict_all_outputs[transform] = transf_mapper["outputs"] else: transf_mapper = mapper[transform] # Keep in memory the transformations in the main pipe all_transforms.mainpipe = copy.deepcopy(all_transforms.attributes)