<s> '''<STR_LIT>''' <EOL> import logging <EOL> import george <EOL> import numpy as np <EOL> from scipy import optimize <EOL> from copy import deepcopy <EOL> from robo . models . base_model import BaseModel <EOL> logger = logging . getLogger ( __name__ ) <EOL> class GaussianProcess ( BaseModel ) : <EOL> def __init__ ( self , kernel , prior = None , <EOL> yerr = <NUM_LIT> , use_gradients = True , <EOL> basis_func = None , dim = None , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . kernel = kernel <EOL> self . model = None <EOL> self . prior = prior <EOL> self . yerr = yerr <EOL> self . use_gradients = use_gradients <EOL> self . basis_func = basis_func <EOL> self . dim = dim <EOL> def scale ( self , x , new_min , new_max , old_min , old_max ) : <EOL> return ( ( new_max - new_min ) * <EOL> ( x - old_min ) / ( old_max - old_min ) ) + new_min <EOL> def train ( self , X , Y , do_optimize = True ) : <EOL> """<STR_LIT>""" <EOL> self . X = X <EOL> if self . basis_func is not None : <EOL> self . X = deepcopy ( X ) <EOL> self . X [ : , self . dim ] = self . basis_func ( self . X [ : , self . dim ] ) <EOL> self . Y = Y <EOL> self . mean = np . mean ( Y , axis = <NUM_LIT:0> ) <EOL> self . model = george . GP ( self . kernel , mean = self . mean ) <EOL> while ( True ) : <EOL> try : <EOL> self . model . compute ( self . X , yerr = self . yerr ) <EOL> break <EOL> except np . linalg . LinAlgError : <EOL> if self . yerr == <NUM_LIT:0.0> : <EOL> self . yerr = <NUM_LIT> <EOL> else : <EOL> self . yerr *= <NUM_LIT:10> <EOL> logger . error ( <EOL> "<STR_LIT>" % <EOL> self . yerr ) <EOL> if do_optimize : <EOL> self . hypers = self . optimize ( ) <EOL> logger . info ( "<STR_LIT>" + str ( self . hypers ) ) <EOL> self . model . kernel [ : ] = self . hypers <EOL> else : <EOL> self . hypers = self . model . kernel [ : ] <EOL> def get_noise ( self ) : <EOL> return self . yerr <EOL> assert self . kernel . k2 . k2 . kernel_type == <NUM_LIT:1> <EOL> return self . kernel . k2 . k2 . pars [ <NUM_LIT:0> ] <EOL> def nll ( self , theta ) : <EOL> """<STR_LIT>""" <EOL> if np . any ( ( - <NUM_LIT:10> > theta ) + ( theta > <NUM_LIT:10> ) ) : <EOL> return <NUM_LIT> <EOL> self . model . kernel [ : ] = theta <EOL> ll = self . model . lnlikelihood ( self . Y [ : , <NUM_LIT:0> ] , quiet = True ) <EOL> if self . prior is not None : <EOL> ll += self . prior . lnprob ( theta ) <EOL> return - ll if np . isfinite ( ll ) else <NUM_LIT> <EOL> def grad_nll ( self , theta ) : <EOL> self . model . kernel [ : ] = theta <EOL> gll = self . model . grad_lnlikelihood ( self . Y [ : , <NUM_LIT:0> ] , quiet = True ) <EOL> if self . prior is not None : <EOL> gll += self . prior . gradient ( theta ) <EOL> return - gll <EOL> def optimize ( self ) : <EOL> p0 = self . model . kernel . vector <EOL> if self . use_gradients : <EOL> results = optimize . minimize ( self . nll , p0 , jac = self . grad_nll ) <EOL> else : <EOL> results = optimize . minimize ( self . nll , p0 ) <EOL> return results . x <EOL> def predict_variance ( self , X1 , X2 ) : <EOL> r"""<STR_LIT>""" <EOL> if self . basis_func is not None : <EOL> X_test_1 = deepcopy ( X1 ) <EOL> X_test_1 [ : , self . dim ] = self . basis_func ( X_test_1 [ : , self . dim ] ) <EOL> X_test_2 = deepcopy ( X2 ) <EOL> X_test_2 [ : , self . dim ] = self . basis_func ( X_test_2 [ : , self . dim ] ) <EOL> else : <EOL> X_test_1 = X1 <EOL> X_test_2 = X2 <EOL> x_ = np . concatenate ( ( X_test_1 , X_test_2 ) ) <EOL> _ , var = self . predict ( x_ ) <EOL> var = var [ : - <NUM_LIT:1> , - <NUM_LIT:1> , np . newaxis ] <EOL> return var <EOL> def predict ( self , X , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> if self . basis_func is not None : <EOL> X_test = deepcopy ( X ) <EOL> X_test [ : , self . dim ] = self . basis_func ( X_test [ : , self . dim ] ) <EOL> else : <EOL> X_test = X <EOL> if self . model is None : <EOL> logger . error ( "<STR_LIT>" ) <EOL> raise ValueError <EOL> mu , var = self . model . predict ( self . Y [ : , <NUM_LIT:0> ] , X_test ) <EOL> if var . shape [ <NUM_LIT:0> ] == <NUM_LIT:1> : <EOL> var = np . clip ( var , np . finfo ( var . dtype ) . eps , np . inf ) <EOL> else : <EOL> var [ np . diag_indices ( var . shape [ <NUM_LIT:0> ] ) ] = np . clip ( <EOL> var [ np . diag_indices ( var . shape [ <NUM_LIT:0> ] ) ] , <EOL> np . finfo ( var . dtype ) . eps , np . inf ) <EOL> var [ np . where ( ( var < np . finfo ( var . dtype ) . eps ) & ( var > - np . finfo ( var . dtype ) . eps ) ) ] = <NUM_LIT:0> <EOL> return mu [ : , np . newaxis ] , var <EOL> def sample_functions ( self , X_test , n_funcs = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> return self . model . sample_conditional ( self . Y [ : , <NUM_LIT:0> ] , X_test , n_funcs ) <EOL> def predictive_gradients ( self , X_test ) : <EOL> dmdx , dvdx = self . m . predictive_gradients ( Xnew ) <EOL> return dmdx [ : , <NUM_LIT:0> , : ] , dvdx </s>
