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 obsvervation 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, } 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] # Temporal re-indexing if any # tinterp = param.time_interpolation # yml_dict = { # "plugin": { # "name": "time_interpolation", # "version": "std", # "type": "transform", # }, # "method": getattr(tinterp, "method", "bilinear"), # "component": [cmp], # "parameter": [prm], # } # add_default( # all_transforms, # yml_dict, # position="index", # index=itransf, # mapper=mapper, # init=True, # inputs=dict_all_inputs, # outputs=dict_all_outputs, # backup_comps=backup_comps, # ) # # # Vertical re-gridding if any # vinterp = param.vertical_interpolation # yml_dict = { # "plugin": { # "name": "vertical_interpolation", # "version": "std", # "type": "transform", # }, # "method": getattr(vinterp, "method", "linear"), # "psurf": getattr(vinterp, "psurf", 1013.25), # "fill_nans": getattr(vinterp, "fill_nans", False), # "fill_nans_value": getattr(vinterp, "fill_nans_value", # None), # "coord_out": getattr(vinterp, "coord_out", "pres"), # "component": [cmp], # "parameter": [prm], # } # add_default( # all_transforms, # yml_dict, # position="index", # index=itransf, # mapper=mapper, # init=True, # inputs=dict_all_inputs, # outputs=dict_all_outputs, # backup_comps=backup_comps, # ) # # # Regridding if any # regrid = getattr(param, "regrid", None) # yml_dict = { # "plugin": { # "name": "regrid", # "version": "std", # "type": "transform", # }, # "dir_wgt": getattr(regrid, "dir_wgt", ""), # "min_weight": getattr(regrid, "min_weight", 1e-10), # "method": getattr(regrid, "method", "gridcell"), # "component": [cmp], # "parameter": [prm], # } # add_default( # all_transforms, # yml_dict, # position="index", # index=itransf, # mapper=mapper, # init=True, # inputs=all_inputs, # outputs=all_outputs, # backup_comps=backup_comps, # ) # 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]