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

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


[docs]def init_control_transformations( self, all_transforms, controlvect, backup_comps, mapper): """Initialize transforms on the control vector side. Also loops on all components/tracers of the ``datavect`` and for those for which the argument ``unit_conversion`` is specified, applies the :doc:`unit_conversion </documentation/plugins/transforms/basic/unit_conversion>` transform. """ # Initializes the overall transform pipe if not hasattr(controlvect, "transform_pipe"): controlvect.transform_pipe = Transform.from_dict({}) transforms = controlvect.transform_pipe transfs_ids = getattr(transforms, "attributes", []) components = controlvect.datavect.components comps = components.attributes # Loops backwards on available transformations and update inputs/outputs # According to model inputs itransf = len(transfs_ids) ntransf_before = len(all_transforms.attributes) while itransf > 0: itransf -= 1 ntransf_loc = 0 transform = "{}_before".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(0, transform) # Updating the general mapper, and creates a local one transf_mapper = transf.ini_mapper( general_mapper=mapper, backup_comps=backup_comps, ) transf.mapper = transf_mapper mapper[transform] = transf_mapper else: transf_mapper = mapper[transform] # Add default transformations prior to other transforms input_keys = set(itertools.chain( *[list(mapper[tr]["inputs"].keys()) for tr in mapper])) for trid in input_keys: prm = trid[1] cmp = trid[0] cmp_in = cmp if cmp in comps else backup_comps.get(cmp, None) if cmp_in is None or not hasattr(components, cmp_in): continue cmp_plg = getattr(components, cmp_in) # Fetch parameters # If no parameters, handle the component as a whole if not hasattr(cmp_plg, "parameters"): params = cmp_plg parameters = [""] else: params = cmp_plg.parameters parameters = params.attributes[:] param = getattr(params, prm, cmp_plg) new_transf, new_id = None, "" # Rescaling if any if hasattr(param, "unit_conversion"): unit_conv = getattr(param, "unit_conversion") yml_dict = { "plugin": { "name": "unit_conversion", "version": "std", "type": "transform", }, "component": [cmp], "parameter": [prm], "orig_parameter_plg": param, "orig_component_plg": cmp_plg, **{attr: getattr(unit_conv, attr) for attr in getattr(unit_conv, "attributes", []) if attr != "plugin"} } new_transf, new_id = add_default( all_transforms, yml_dict, position="first", mapper=mapper, init=True, backup_comps=backup_comps, precursor=new_id ) # Keep in memory transformations on the controlvect side all_transforms.controlpipe = \ [tr for tr in all_transforms.attributes if tr not in all_transforms.mainpipe]