<s> from datetime import timedelta <EOL> from time import time <EOL> import warnings <EOL> from gdbn . dbn import buildDBN <EOL> from gdbn import activationFunctions <EOL> import numpy as np <EOL> from sklearn . base import BaseEstimator <EOL> from sklearn . preprocessing import LabelEncoder <EOL> from sklearn . preprocessing import OneHotEncoder <EOL> warnings . warn ( """<STR_LIT>""" ) <EOL> class DBN ( BaseEstimator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , <EOL> layer_sizes = None , <EOL> scales = <NUM_LIT> , <EOL> fan_outs = None , <EOL> output_act_funct = None , <EOL> real_valued_vis = True , <EOL> use_re_lu = True , <EOL> uniforms = False , <EOL> learn_rates = <NUM_LIT:0.1> , <EOL> learn_rate_decays = <NUM_LIT:1.0> , <EOL> learn_rate_minimums = <NUM_LIT:0.0> , <EOL> momentum = <NUM_LIT> , <EOL> l2_costs = <NUM_LIT> , <EOL> dropouts = <NUM_LIT:0> , <EOL> nesterov = True , <EOL> nest_compare = True , <EOL> rms_lims = None , <EOL> learn_rates_pretrain = None , <EOL> momentum_pretrain = None , <EOL> l2_costs_pretrain = None , <EOL> nest_compare_pretrain = None , <EOL> epochs = <NUM_LIT:10> , <EOL> epochs_pretrain = <NUM_LIT:0> , <EOL> loss_funct = None , <EOL> minibatch_size = <NUM_LIT:64> , <EOL> minibatches_per_epoch = None , <EOL> pretrain_callback = None , <EOL> fine_tune_callback = None , <EOL> random_state = None , <EOL> verbose = <NUM_LIT:0> , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> if layer_sizes is None : <EOL> layer_sizes = [ - <NUM_LIT:1> , - <NUM_LIT:1> ] <EOL> if output_act_funct is None : <EOL> output_act_funct = activationFunctions . Softmax ( ) <EOL> elif isinstance ( output_act_funct , str ) : <EOL> output_act_funct = getattr ( activationFunctions , output_act_funct ) ( ) <EOL> if random_state is not None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . layer_sizes = layer_sizes <EOL> self . scales = scales <EOL> self . fan_outs = fan_outs <EOL> self . output_act_funct = output_act_funct <EOL> self . real_valued_vis = real_valued_vis <EOL> self . use_re_lu = use_re_lu <EOL> self . uniforms = uniforms <EOL> self . learn_rates = learn_rates <EOL> self . learn_rate_decays = learn_rate_decays <EOL> self . learn_rate_minimums = learn_rate_minimums <EOL> self . momentum = momentum <EOL> self . l2_costs = l2_costs <EOL> self . dropouts = dropouts <EOL> self . nesterov = nesterov <EOL> self . nest_compare = nest_compare <EOL> self . rms_lims = rms_lims <EOL> self . learn_rates_pretrain = learn_rates_pretrain <EOL> self . momentum_pretrain = momentum_pretrain <EOL> self . l2_costs_pretrain = l2_costs_pretrain <EOL> self . nest_compare_pretrain = nest_compare_pretrain <EOL> self . epochs = epochs <EOL> self . epochs_pretrain = epochs_pretrain <EOL> self . loss_funct = loss_funct <EOL> self . use_dropout = True if dropouts else False <EOL> self . minibatch_size = minibatch_size <EOL> self . minibatches_per_epoch = minibatches_per_epoch <EOL> self . pretrain_callback = pretrain_callback <EOL> self . fine_tune_callback = fine_tune_callback <EOL> self . random_state = random_state <EOL> self . verbose = verbose <EOL> def _fill_missing_layer_sizes ( self , X , y ) : <EOL> layer_sizes = self . layer_sizes <EOL> if layer_sizes [ <NUM_LIT:0> ] == - <NUM_LIT:1> : <EOL> layer_sizes [ <NUM_LIT:0> ] = X . shape [ <NUM_LIT:1> ] <EOL> if layer_sizes [ - <NUM_LIT:1> ] == - <NUM_LIT:1> and y is not None : <EOL> layer_sizes [ - <NUM_LIT:1> ] = y . shape [ <NUM_LIT:1> ] <EOL> def _vp ( self , value ) : <EOL> num_weights = len ( self . layer_sizes ) - <NUM_LIT:1> <EOL> if not hasattr ( value , '<STR_LIT>' ) : <EOL> value = [ value ] * num_weights <EOL> return list ( value ) <EOL> def _build_net ( self , X , y = None ) : <EOL> v = self . _vp <EOL> self . _fill_missing_layer_sizes ( X , y ) <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" . format ( self . layer_sizes ) ) <EOL> if self . random_state is not None : <EOL> np . random . seed ( self . random_state ) <EOL> net = buildDBN ( <EOL> self . layer_sizes , <EOL> v ( self . scales ) , <EOL> v ( self . fan_outs ) , <EOL> self . output_act_funct , <EOL> self . real_valued_vis , <EOL> self . use_re_lu , <EOL> v ( self . uniforms ) , <EOL> ) <EOL> return net <EOL> def _configure_net_pretrain ( self , net ) : <EOL> v = self . _vp <EOL> self . _configure_net_finetune ( net ) <EOL> learn_rates = self . learn_rates_pretrain <EOL> momentum = self . momentum_pretrain <EOL> l2_costs = self . l2_costs_pretrain <EOL> nest_compare = self . nest_compare_pretrain <EOL> if learn_rates is None : <EOL> learn_rates = self . learn_rates <EOL> if momentum is None : <EOL> momentum = self . momentum <EOL> if l2_costs is None : <EOL> l2_costs = self . l2_costs <EOL> if nest_compare is None : <EOL> nest_compare = self . nest_compare <EOL> net . learnRates = v ( learn_rates ) <EOL> net . momentum = momentum <EOL> net . L2Costs = v ( l2_costs ) <EOL> net . nestCompare = nest_compare <EOL> return net <EOL> def _configure_net_finetune ( self , net ) : <EOL> v = self . _vp <EOL> net . learnRates = v ( self . learn_rates ) <EOL> net . momentum = self . momentum <EOL> net . L2Costs = v ( self . l2_costs ) <EOL> net . dropouts = v ( self . dropouts ) <EOL> net . nesterov = self . nesterov <EOL> net . nestCompare = self . nest_compare <EOL> net . rmsLims = v ( self . rms_lims ) <EOL> return net <EOL> def _minibatches ( self , X , y = None ) : <EOL> while True : <EOL> idx = np . random . randint ( X . shape [ <NUM_LIT:0> ] , size = ( self . minibatch_size , ) ) <EOL> X_batch = X [ idx ] <EOL> if hasattr ( X_batch , '<STR_LIT>' ) : <EOL> X_batch = X_batch . todense ( ) <EOL> if y is not None : <EOL> yield ( X_batch , y [ idx ] ) <EOL> else : <EOL> yield X_batch <EOL> def _onehot ( self , y ) : <EOL> return np . array ( <EOL> OneHotEncoder ( ) . fit_transform ( y . reshape ( - <NUM_LIT:1> , <NUM_LIT:1> ) ) . todense ( ) ) <EOL> def _num_mistakes ( self , targets , outputs ) : <EOL> if hasattr ( targets , '<STR_LIT>' ) : <EOL> targets = targets . as_numpy_array ( ) <EOL> if hasattr ( outputs , '<STR_LIT>' ) : <EOL> outputs = outputs . as_numpy_array ( ) <EOL> return np . sum ( outputs . argmax ( <NUM_LIT:1> ) != targets . argmax ( <NUM_LIT:1> ) ) <EOL> def _learn_rate_adjust ( self ) : <EOL> if self . learn_rate_decays == <NUM_LIT:1.0> : <EOL> return <EOL> learn_rate_decays = self . _vp ( self . learn_rate_decays ) <EOL> learn_rate_minimums = self . _vp ( self . learn_rate_minimums ) <EOL> for index , decay in enumerate ( learn_rate_decays ) : <EOL> new_learn_rate = self . net_ . learnRates [ index ] * decay <EOL> if new_learn_rate >= learn_rate_minimums [ index ] : <EOL> self . net_ . learnRates [ index ] = new_learn_rate <EOL> if self . verbose >= <NUM_LIT:2> : <EOL> print ( "<STR_LIT>" . format ( self . net_ . learnRates ) ) <EOL> def fit ( self , X , y ) : <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" % ( X . shape , ) ) <EOL> self . _enc = LabelEncoder ( ) <EOL> y = self . _enc . fit_transform ( y ) <EOL> y = self . _onehot ( y ) <EOL> self . net_ = self . _build_net ( X , y ) <EOL> minibatches_per_epoch = self . minibatches_per_epoch <EOL> if minibatches_per_epoch is None : <EOL> minibatches_per_epoch = X . shape [ <NUM_LIT:0> ] / self . minibatch_size <EOL> loss_funct = self . loss_funct <EOL> if loss_funct is None : <EOL> loss_funct = self . _num_mistakes <EOL> errors_pretrain = self . errors_pretrain_ = [ ] <EOL> losses_fine_tune = self . losses_fine_tune_ = [ ] <EOL> errors_fine_tune = self . errors_fine_tune_ = [ ] <EOL> if self . epochs_pretrain : <EOL> self . epochs_pretrain = self . _vp ( self . epochs_pretrain ) <EOL> self . _configure_net_pretrain ( self . net_ ) <EOL> for layer_index in range ( len ( self . layer_sizes ) - <NUM_LIT:1> ) : <EOL> errors_pretrain . append ( [ ] ) <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" . format ( layer_index + <NUM_LIT:1> ) ) <EOL> time0 = time ( ) <EOL> for epoch , err in enumerate ( <EOL> self . net_ . preTrainIth ( <EOL> layer_index , <EOL> self . _minibatches ( X ) , <EOL> self . epochs_pretrain [ layer_index ] , <EOL> minibatches_per_epoch , <EOL> ) ) : <EOL> errors_pretrain [ - <NUM_LIT:1> ] . append ( err ) <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" . format ( epoch + <NUM_LIT:1> , err ) ) <EOL> elapsed = str ( timedelta ( seconds = time ( ) - time0 ) ) <EOL> print ( "<STR_LIT>" . format ( elapsed . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] ) ) <EOL> time0 = time ( ) <EOL> if self . pretrain_callback is not None : <EOL> self . pretrain_callback ( <EOL> self , epoch + <NUM_LIT:1> , layer_index ) <EOL> self . _configure_net_finetune ( self . net_ ) <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" ) <EOL> time0 = time ( ) <EOL> for epoch , ( loss , err ) in enumerate ( <EOL> self . net_ . fineTune ( <EOL> self . _minibatches ( X , y ) , <EOL> self . epochs , <EOL> minibatches_per_epoch , <EOL> loss_funct , <EOL> self . verbose , <EOL> self . use_dropout , <EOL> ) ) : <EOL> losses_fine_tune . append ( loss ) <EOL> errors_fine_tune . append ( err ) <EOL> self . _learn_rate_adjust ( ) <EOL> if self . verbose : <EOL> print ( "<STR_LIT>" . format ( epoch + <NUM_LIT:1> ) ) <EOL> print ( "<STR_LIT>" . format ( loss ) ) <EOL> print ( "<STR_LIT>" . format ( err ) ) <EOL> elapsed = str ( timedelta ( seconds = time ( ) - time0 ) ) <EOL> print ( "<STR_LIT>" . format ( elapsed . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] ) ) <EOL> time0 = time ( ) <EOL> if self . fine_tune_callback is not None : <EOL> self . fine_tune_callback ( self , epoch + <NUM_LIT:1> ) <EOL> def predict ( self , X ) : <EOL> y_ind = np . argmax ( self . predict_proba ( X ) , axis = <NUM_LIT:1> ) <EOL> return self . _enc . inverse_transform ( y_ind ) <EOL> def predict_proba ( self , X ) : <EOL> if hasattr ( X , '<STR_LIT>' ) : <EOL> return self . _predict_proba_sparse ( X ) <EOL> res = np . zeros ( ( X . shape [ <NUM_LIT:0> ] , self . layer_sizes [ - <NUM_LIT:1> ] ) ) <EOL> for i , el in enumerate ( self . net_ . predictions ( X , asNumpy = True ) ) : <EOL> res [ i ] = el <EOL> return res <EOL> def _predict_proba_sparse ( self , X ) : <EOL> batch_size = self . minibatch_size <EOL> res = [ ] <EOL> for i in range ( <NUM_LIT:0> , X . shape [ <NUM_LIT:0> ] , batch_size ) : <EOL> X_batch = X [ i : min ( i + batch_size , X . shape [ <NUM_LIT:0> ] ) ] . todense ( ) <EOL> res . extend ( self . net_ . predictions ( X_batch ) ) <EOL> return np . array ( res ) . reshape ( X . shape [ <NUM_LIT:0> ] , - <NUM_LIT:1> ) <EOL> def score ( self , X , y ) : <EOL> loss_funct = self . loss_funct <EOL> if loss_funct is None : <EOL> loss_funct = self . _num_mistakes <EOL> outputs = self . predict_proba ( X ) <EOL> targets = self . _onehot ( self . _enc . transform ( y ) ) <EOL> mistakes = loss_funct ( outputs , targets ) <EOL> return - float ( mistakes ) / len ( y ) + <NUM_LIT:1> <EOL> @ property <EOL> def classes_ ( self ) : <EOL> return self . _enc . classes_ </s>
