<s> """<STR_LIT>""" <EOL> from copy import deepcopy <EOL> from openmdao . main . api import Component <EOL> from openmdao . main . datatypes . api import List , Bool , Dict , Float , Slot , Str , VarTree <EOL> from openmdao . main . datatypes . uncertaindist import UncertainDistVar <EOL> from openmdao . main . interfaces import ISurrogate , ICaseRecorder , IUncertainVariable <EOL> from openmdao . main . mp_support import has_interface <EOL> from openmdao . main . vartree import VariableTree <EOL> from openmdao . util . typegroups import int_types , real_types <EOL> class MetaModel ( Component ) : <EOL> """<STR_LIT>""" <EOL> default_surrogate = Slot ( ISurrogate , allow_none = True , <EOL> desc = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> surrogates = Dict ( key_trait = Str , <EOL> value_trait = Slot ( ISurrogate ) , <EOL> desc = '<STR_LIT>' ) <EOL> params = VarTree ( VariableTree ( ) , iotype = '<STR_LIT>' ) <EOL> responses = VarTree ( VariableTree ( ) , iotype = '<STR_LIT>' ) <EOL> recorder = Slot ( ICaseRecorder , <EOL> desc = '<STR_LIT>' ) <EOL> warm_restart = Bool ( False , iotype = '<STR_LIT>' , <EOL> desc = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def __init__ ( self , params = None , responses = None ) : <EOL> super ( MetaModel , self ) . __init__ ( ) <EOL> if params is None or not isinstance ( params , tuple ) : <EOL> msg = "<STR_LIT>" + "<STR_LIT>" <EOL> self . raise_exception ( msg , ValueError ) <EOL> if responses is None or not isinstance ( responses , tuple ) : <EOL> msg = "<STR_LIT>" + "<STR_LIT>" <EOL> self . raise_exception ( msg , ValueError ) <EOL> input_tree = self . get ( '<STR_LIT>' ) <EOL> self . _param_data = [ ] <EOL> for name in params : <EOL> self . add ( name , Float ( <NUM_LIT:0.0> , iotype = '<STR_LIT>' , desc = '<STR_LIT>' ) ) <EOL> input_tree . add ( name , List ( [ ] , desc = '<STR_LIT>' ) ) <EOL> output_tree = self . get ( '<STR_LIT>' ) <EOL> self . _response_data = { } <EOL> for name in responses : <EOL> self . add ( name , Float ( <NUM_LIT:0.0> , iotype = '<STR_LIT>' , desc = '<STR_LIT>' ) ) <EOL> output_tree . add ( name , List ( [ ] , desc = '<STR_LIT>' ) ) <EOL> self . _response_data [ name ] = [ ] <EOL> self . surrogates [ name ] = None <EOL> self . _surrogate_input_names = params <EOL> self . _surrogate_output_names = responses <EOL> self . _train = True <EOL> self . _surrogate_overrides = set ( ) <EOL> self . _default_surrogate_copies = { } <EOL> self . on_trait_change ( self . _surrogate_updated , "<STR_LIT>" ) <EOL> def _input_updated ( self , name , fullpath = None ) : <EOL> '''<STR_LIT>''' <EOL> if fullpath is not None : <EOL> if fullpath . startswith ( '<STR_LIT>' ) or fullpath . startswith ( '<STR_LIT>' ) : <EOL> self . _train = True <EOL> super ( MetaModel , self ) . _input_updated ( name . split ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] ) <EOL> def check_config ( self , strict = False ) : <EOL> '''<STR_LIT>''' <EOL> super ( MetaModel , self ) . check_config ( strict = strict ) <EOL> if self . default_surrogate is None : <EOL> no_sur = [ ] <EOL> for name in self . _surrogate_output_names : <EOL> if name not in self . surrogates or self . surrogates [ name ] is None : <EOL> no_sur . append ( name ) <EOL> if len ( no_sur ) > <NUM_LIT:0> : <EOL> self . raise_exception ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> no_sur , RuntimeError ) <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _train : <EOL> input_data = self . _param_data <EOL> if self . warm_restart is False : <EOL> input_data = [ ] <EOL> base = <NUM_LIT:0> <EOL> else : <EOL> base = len ( input_data ) <EOL> for name in self . _surrogate_input_names : <EOL> train_name = "<STR_LIT>" % name <EOL> val = self . get ( train_name ) <EOL> num_sample = len ( val ) <EOL> for j in xrange ( base , base + num_sample ) : <EOL> if j > len ( input_data ) - <NUM_LIT:1> : <EOL> input_data . append ( [ ] ) <EOL> input_data [ j ] . append ( val [ j - base ] ) <EOL> for name in self . _surrogate_output_names : <EOL> train_name = "<STR_LIT>" % name <EOL> output_data = self . _response_data [ name ] <EOL> if self . warm_restart is False : <EOL> output_data = [ ] <EOL> output_data . extend ( self . get ( train_name ) ) <EOL> surrogate = self . _get_surrogate ( name ) <EOL> if surrogate is not None : <EOL> surrogate . train ( input_data , output_data ) <EOL> self . _train = False <EOL> inputs = [ ] <EOL> for name in self . _surrogate_input_names : <EOL> val = self . get ( name ) <EOL> inputs . append ( val ) <EOL> for name in self . _surrogate_output_names : <EOL> surrogate = self . _get_surrogate ( name ) <EOL> if surrogate is not None : <EOL> setattr ( self , name , surrogate . predict ( inputs ) ) <EOL> def _get_surrogate ( self , name ) : <EOL> """<STR_LIT>""" <EOL> surrogate = self . surrogates . get ( name ) <EOL> if surrogate is None and self . default_surrogate is not None : <EOL> surrogate = self . _default_surrogate_copies . get ( name ) <EOL> return surrogate <EOL> def _default_surrogate_changed ( self , old_obj , new_obj ) : <EOL> """<STR_LIT>""" <EOL> if old_obj : <EOL> old_obj . on_trait_change ( self . _def_surrogate_trait_modified , <EOL> remove = True ) <EOL> if new_obj : <EOL> new_obj . on_trait_change ( self . _def_surrogate_trait_modified ) <EOL> for name in self . _surrogate_output_names : <EOL> if name not in self . _surrogate_overrides : <EOL> surrogate = deepcopy ( self . default_surrogate ) <EOL> self . _default_surrogate_copies [ name ] = surrogate <EOL> self . _update_var_for_surrogate ( surrogate , name ) <EOL> self . config_changed ( ) <EOL> self . _train = True <EOL> def _def_surrogate_trait_modified ( self , surrogate , name , old , new ) : <EOL> for name in self . _default_surrogate_copies : <EOL> surr_copy = deepcopy ( self . default_surrogate ) <EOL> self . _default_surrogate_copies [ name ] = surr_copy <EOL> def _surrogate_updated ( self , obj , name , old , new ) : <EOL> """<STR_LIT>""" <EOL> all_changes = new . changed . keys ( ) + new . added . keys ( ) + new . removed . keys ( ) <EOL> for varname in all_changes : <EOL> surr = self . surrogates . get ( varname ) <EOL> if surr is None : <EOL> if self . default_surrogate : <EOL> def_surr = deepcopy ( self . default_surrogate ) <EOL> self . _default_surrogate_copies [ varname ] = def_surr <EOL> self . _update_var_for_surrogate ( def_surr , varname ) <EOL> if varname in self . _surrogate_overrides : <EOL> self . _surrogate_overrides . remove ( varname ) <EOL> else : <EOL> self . _surrogate_overrides . add ( varname ) <EOL> if name in self . _default_surrogate_copies : <EOL> del self . _default_surrogate_copies [ name ] <EOL> self . _update_var_for_surrogate ( surr , varname ) <EOL> self . config_changed ( ) <EOL> self . _train = True <EOL> def _update_var_for_surrogate ( self , surrogate , varname ) : <EOL> """<STR_LIT>""" <EOL> val = surrogate . get_uncertain_value ( <NUM_LIT:1.0> ) <EOL> if has_interface ( val , IUncertainVariable ) : <EOL> ttype = UncertainDistVar <EOL> elif isinstance ( val , real_types ) : <EOL> ttype = Float <EOL> else : <EOL> self . raise_exception ( "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> val . __class__ . __name__ ) <EOL> self . add ( varname , ttype ( default_value = val , iotype = '<STR_LIT>' , <EOL> desc = self . trait ( varname ) . desc ) ) <EOL> setattr ( self , varname , val ) <EOL> return </s>
