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

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

[docs]def init_obsvect_transformations( self, all_transforms, obsvect, backup_comps, mapper): """Initialize transforms on the observation vector side. Also, for the component ``satellite`` of the ``datavect``, includes the transform :doc:`satellites </documentation/plugins/transforms/complex/satellites>` to the pipeline. """ # Initializes the overall transform pipe if not hasattr(obsvect, "transform_pipe"): obsvect.transform_pipe = Transform.from_dict({}) transforms = obsvect.transform_pipe transfs_ids = getattr(transforms, "attributes", []) components = obsvect.datavect.components comps = components.attributes obs_mapper = obsvect.ini_mapper(obsvect) # mainpipe_outputs_keys = list(all_outputs.keys()) # all_outputs.update(obs_mapper["outputs"]) # Add default transformations prior to other transforms ntransf_before = len(all_transforms.attributes) for comp in self.datavect.components.attributes: component = getattr(components, comp) # Fetch parameters # If no parameters, handle the component as a whole if not hasattr(component, "parameters"): continue else: params = component.parameters parameters = params.attributes[:] for trcr in parameters: param = getattr(params, trcr, component) # Do nothing if not in observation vector if not param.isobs: continue # Add toobsvect default transformation yml_dict = { "plugin": { "name": "toobsvect", "version": "std", "type": "transform", "newplg": True, }, "component": comp, "parameter": trcr, "orig_parameter_plg": param, "orig_component_plg": component, **{attr: getattr(param, attr) for attr in getattr(param, "attributes", []) if attr != "plugin"} } new_transf, toobsvect_id = add_default( all_transforms, yml_dict, position="last", mapper=mapper, init=True, backup_comps=backup_comps, ) # Default satellite transformation new_transf, new_id = None, "" if comp == "satellites": yml_dict = { "plugin": { "name": "satellites", "version": "std", "type": "transform", }, "component": [comp], "parameter": [trcr], "orig_parameter_plg": param, "orig_component_plg": component, **{attr: getattr(param, attr) for attr in getattr(param, "attributes", []) if attr != "plugin"} } new_transf, new_id = add_default( all_transforms, yml_dict, position="index", index=ntransf_before, mapper=mapper, init=True, backup_comps=backup_comps, successor=toobsvect_id ) mapper[toobsvect_id]["precursors"][(comp, trcr)] = [new_id] # Loops backwards on available transformations and update inputs/outputs # According to model outputs itransf = len(transfs_ids) while itransf > 0: itransf -= 1 ntransf_loc = 0 transform = "{}_after".format(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(transforms, transfs_ids[itransf]) if transf is None: transf = Transform.from_dict({}, orig_name=transform) setattr(all_transforms, transform, transf) all_transforms.attributes.insert(ntransf_before, transform) # Updating the general mapper, and creates a local one transf_mapper = transf.ini_mapper( general_mapper=mapper, backup_comps=backup_comps, obsvect=obsvect, ) transf.mapper = transf_mapper mapper[transform] = transf_mapper else: transf_mapper = mapper[transform] # Keep in memory transformations on the controlvect side all_transforms.obsvectpipe = \ [tr for tr in all_transforms.attributes if tr not in all_transforms.mainpipe + all_transforms.controlpipe]