<s> import numpy as np <EOL> import tensorflow as tf <EOL> from tensorflow . python . framework import ops <EOL> from models . rbm_models import rbm <EOL> from utils import utilities <EOL> import model <EOL> class DBN ( model . Model ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , layers , do_pretrain = True , rbm_num_epochs = list ( [ <NUM_LIT:10> ] ) , rbm_batch_size = list ( [ <NUM_LIT:10> ] ) , model_name = '<STR_LIT>' , <EOL> rbm_learning_rate = list ( [ <NUM_LIT> ] ) , rbm_names = list ( [ '<STR_LIT>' ] ) , rbm_gibbs_k = list ( [ <NUM_LIT:1> ] ) , gauss_visible = False , <EOL> stddev = <NUM_LIT:0.1> , learning_rate = <NUM_LIT> , momentum = <NUM_LIT> , num_epochs = <NUM_LIT:10> , batch_size = <NUM_LIT:10> , dropout = <NUM_LIT:1> , <EOL> opt = '<STR_LIT>' , verbose = <NUM_LIT:1> , act_func = '<STR_LIT:relu>' , loss_func = '<STR_LIT>' , main_dir = '<STR_LIT>' , <EOL> dataset = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> model . Model . __init__ ( self , model_name , main_dir ) <EOL> self . layers = layers <EOL> self . n_layers = len ( layers ) <EOL> self . do_pretrain = do_pretrain <EOL> self . rbm_num_epochs = rbm_num_epochs <EOL> self . rbm_batch_size = rbm_batch_size <EOL> self . rbm_learning_rate = rbm_learning_rate <EOL> self . rbm_names = rbm_names <EOL> self . rbm_gibbs_k = rbm_gibbs_k <EOL> self . gauss_visible = gauss_visible <EOL> self . stddev = stddev <EOL> self . act_func = act_func <EOL> self . learning_rate = learning_rate <EOL> self . momentum = momentum <EOL> self . num_epochs = num_epochs <EOL> self . batch_size = batch_size <EOL> self . dropout = dropout <EOL> self . opt = opt <EOL> self . loss_func = loss_func <EOL> self . dataset = dataset <EOL> self . verbose = verbose <EOL> self . W_pretrain = None <EOL> self . bh_pretrain = None <EOL> self . bv_pretrain = None <EOL> self . W_vars = None <EOL> self . bh_vars = None <EOL> self . bv_vars = None <EOL> self . tf_merged_summaries = None <EOL> self . tf_summary_writer = None <EOL> self . tf_session = None <EOL> self . tf_saver = None <EOL> if self . do_pretrain : <EOL> self . rbms = [ ] <EOL> for l in range ( self . n_layers ) : <EOL> if l == <NUM_LIT:0> and self . gauss_visible : <EOL> self . rbms . append ( rbm . RBM ( <EOL> visible_unit_type = '<STR_LIT>' , stddev = self . stddev , <EOL> model_name = self . rbm_names [ l ] + str ( l ) , num_hidden = self . layers [ l ] , <EOL> main_dir = self . main_dir , learning_rate = self . rbm_learning_rate [ l ] , <EOL> verbose = self . verbose , num_epochs = self . rbm_num_epochs [ l ] , batch_size = self . rbm_batch_size [ l ] ) ) <EOL> else : <EOL> self . rbms . append ( rbm . RBM ( <EOL> model_name = self . rbm_names [ l ] + str ( l ) , num_hidden = self . layers [ l ] , <EOL> main_dir = self . main_dir , learning_rate = self . rbm_learning_rate [ l ] , <EOL> verbose = self . verbose , num_epochs = self . rbm_num_epochs [ l ] , batch_size = self . rbm_batch_size [ l ] ) ) <EOL> def pretrain ( self , train_set , validation_set = None ) : <EOL> """<STR_LIT>""" <EOL> self . W_pretrain = [ ] <EOL> self . bh_pretrain = [ ] <EOL> self . bv_pretrain = [ ] <EOL> next_train = train_set <EOL> next_valid = validation_set <EOL> for l , rboltz in enumerate ( self . rbms ) : <EOL> print ( '<STR_LIT>' . format ( l + <NUM_LIT:1> ) ) <EOL> next_train , next_valid = self . _pretrain_rbm_and_gen_feed ( rboltz , next_train , next_valid ) <EOL> ops . reset_default_graph ( ) <EOL> return next_train , next_valid <EOL> def _pretrain_rbm_and_gen_feed ( self , rboltz , train_set , validation_set ) : <EOL> """<STR_LIT>""" <EOL> rboltz . build_model ( train_set . shape [ <NUM_LIT:1> ] ) <EOL> rboltz . fit ( train_set , validation_set ) <EOL> params = rboltz . get_model_parameters ( ) <EOL> self . W_pretrain . append ( params [ '<STR_LIT>' ] ) <EOL> self . bh_pretrain . append ( params [ '<STR_LIT>' ] ) <EOL> self . bv_pretrain . append ( params [ '<STR_LIT>' ] ) <EOL> return rboltz . transform ( train_set ) , rboltz . transform ( validation_set ) <EOL> def fit ( self , train_set , train_labels , validation_set = None , validation_labels = None ) : <EOL> """<STR_LIT>""" <EOL> with tf . Session ( ) as self . tf_session : <EOL> self . _initialize_tf_utilities_and_ops ( ) <EOL> self . _train_model ( train_set , train_labels , validation_set , validation_labels ) <EOL> self . tf_saver . save ( self . tf_session , self . model_path ) <EOL> def _initialize_tf_utilities_and_ops ( self ) : <EOL> """<STR_LIT>""" <EOL> self . tf_merged_summaries = tf . merge_all_summaries ( ) <EOL> init_op = tf . initialize_all_variables ( ) <EOL> self . tf_saver = tf . train . Saver ( ) <EOL> self . tf_session . run ( init_op ) <EOL> self . tf_summary_writer = tf . train . SummaryWriter ( self . tf_summary_dir , self . tf_session . graph ) <EOL> def _train_model ( self , train_set , train_labels , validation_set = None , validation_labels = None ) : <EOL> """<STR_LIT>""" <EOL> shuff = zip ( train_set , train_labels ) <EOL> for i in range ( self . num_epochs ) : <EOL> np . random . shuffle ( shuff ) <EOL> batches = [ _ for _ in utilities . gen_batches ( shuff , self . batch_size ) ] <EOL> for batch in batches : <EOL> x_batch , y_batch = zip ( * batch ) <EOL> feed = self . _create_feed ( x_batch , y_batch , self . dropout ) <EOL> self . tf_session . run ( self . train_step , feed_dict = feed ) <EOL> if validation_set is not None : <EOL> self . _run_validation_error_and_summaries ( i , validation_set , validation_labels ) <EOL> def _run_validation_error_and_summaries ( self , epoch , validation_set , validation_labels ) : <EOL> """<STR_LIT>""" <EOL> feed = self . _create_feed ( validation_set , validation_labels , <NUM_LIT:1> ) <EOL> result = self . tf_session . run ( [ self . tf_merged_summaries , self . cost ] , feed_dict = feed ) <EOL> summary_str = result [ <NUM_LIT:0> ] <EOL> err = result [ <NUM_LIT:1> ] <EOL> self . tf_summary_writer . add_summary ( summary_str , epoch ) <EOL> if self . verbose == <NUM_LIT:1> : <EOL> print ( "<STR_LIT>" % ( epoch , err ) ) <EOL> def build_model ( self , n_features , n_classes ) : <EOL> """<STR_LIT>""" <EOL> self . _create_placeholders ( n_features , n_classes ) <EOL> self . _create_variables ( ) <EOL> next_train = self . _forward_pass ( ) <EOL> self . _create_softmax_layer ( next_train , n_classes ) <EOL> self . cost = self . _create_cost_function_node ( self . loss_func , self . y , self . y_ ) <EOL> self . train_step = self . _create_train_step_node ( self . opt , self . learning_rate , self . cost , self . momentum ) <EOL> self . _create_test_node ( ) <EOL> def _create_placeholders ( self , n_features , n_classes ) : <EOL> """<STR_LIT>""" <EOL> self . keep_prob = tf . placeholder ( '<STR_LIT:float>' ) <EOL> self . hrand = [ tf . placeholder ( '<STR_LIT:float>' , [ None , self . layers [ l + <NUM_LIT:1> ] ] ) for l in range ( self . n_layers - <NUM_LIT:1> ) ] <EOL> self . vrand = [ tf . placeholder ( '<STR_LIT:float>' , [ None , self . layers [ l ] ] ) for l in range ( self . n_layers - <NUM_LIT:1> ) ] <EOL> self . x = tf . placeholder ( '<STR_LIT:float>' , [ None , n_features ] ) <EOL> self . y_ = tf . placeholder ( '<STR_LIT:float>' , [ None , n_classes ] ) <EOL> def _create_variables ( self ) : <EOL> """<STR_LIT>""" <EOL> self . W_vars = [ tf . Variable ( self . W_pretrain [ l ] ) for l in range ( self . n_layers - <NUM_LIT:1> ) ] <EOL> self . bh_vars = [ tf . Variable ( self . bh_pretrain [ l ] ) for l in range ( self . n_layers - <NUM_LIT:1> ) ] <EOL> self . bv_vars = [ tf . Variable ( self . bv_pretrain [ l ] ) for l in range ( self . n_layers - <NUM_LIT:1> ) ] <EOL> def _forward_pass ( self ) : <EOL> """<STR_LIT>""" <EOL> next_train = self . x <EOL> for l in range ( self . n_layers - <NUM_LIT:1> ) : <EOL> activation = tf . matmul ( next_train , self . W_vars [ l ] ) + self . bh_vars [ l ] <EOL> if self . act_func == '<STR_LIT>' : <EOL> hprobs = tf . nn . sigmoid ( activation ) <EOL> elif self . act_func == '<STR_LIT>' : <EOL> hprobs = tf . nn . tanh ( activation ) <EOL> elif self . act_func == '<STR_LIT:relu>' : <EOL> hprobs = tf . nn . relu ( activation ) <EOL> else : <EOL> hprobs = None <EOL> next_train = tf . nn . dropout ( hprobs , self . keep_prob ) <EOL> return next_train <EOL> def _create_softmax_layer ( self , next_train , n_classes ) : <EOL> """<STR_LIT>""" <EOL> self . softmax_W = tf . Variable ( tf . truncated_normal ( [ self . layers [ - <NUM_LIT:1> ] , n_classes ] ) , <EOL> name = '<STR_LIT>' ) <EOL> self . softmax_b = tf . Variable ( tf . constant ( <NUM_LIT:0.1> , shape = [ n_classes ] ) , name = '<STR_LIT>' ) <EOL> with tf . name_scope ( "<STR_LIT>" ) : <EOL> self . y = tf . matmul ( next_train , self . softmax_W ) + self . softmax_b <EOL> def _create_test_node ( self ) : <EOL> """<STR_LIT>""" <EOL> with tf . name_scope ( "<STR_LIT:test>" ) : <EOL> correct_prediction = tf . equal ( tf . argmax ( self . y , <NUM_LIT:1> ) , tf . argmax ( self . y_ , <NUM_LIT:1> ) ) <EOL> self . accuracy = tf . reduce_mean ( tf . cast ( correct_prediction , "<STR_LIT:float>" ) ) <EOL> _ = tf . scalar_summary ( '<STR_LIT>' , self . accuracy ) <EOL> def get_model_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> with tf . Session ( ) as self . tf_session : <EOL> self . tf_saver . restore ( self . tf_session , self . model_path ) <EOL> def eval_tensors ( tens , prefix ) : <EOL> eval_tens = { } <EOL> for i , t in enumerate ( tens ) : <EOL> eval_tens [ prefix + str ( i ) ] = t . eval ( ) <EOL> return { <EOL> '<STR_LIT>' : eval_tensors ( self . W_vars , '<STR_LIT:w>' ) , <EOL> '<STR_LIT>' : eval_tensors ( self . bh_vars , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : eval_tensors ( self . bv_vars , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : self . softmax_W . eval ( ) , <EOL> '<STR_LIT>' : self . softmax_b . eval ( ) <EOL> } <EOL> def predict ( self , test_set , test_labels ) : <EOL> """<STR_LIT>""" <EOL> with tf . Session ( ) as self . tf_session : <EOL> self . tf_saver . restore ( self . tf_session , self . model_path ) <EOL> feed = self . _create_feed ( test_set , test_labels , <NUM_LIT:1> ) <EOL> return self . accuracy . eval ( feed ) <EOL> def _create_feed ( self , datax , datay , keep_prob ) : <EOL> """<STR_LIT>""" <EOL> feed = { self . x : datax , self . y_ : datay , self . keep_prob : keep_prob } <EOL> hrand = [ np . random . rand ( len ( datax ) , l ) for l in self . layers [ <NUM_LIT:1> : ] ] <EOL> for j , h in enumerate ( hrand ) : <EOL> feed [ self . hrand [ j ] ] = h <EOL> vrand = [ np . random . rand ( len ( datax ) , l ) for l in self . layers [ : - <NUM_LIT:1> ] ] <EOL> for j , v in enumerate ( vrand ) : <EOL> feed [ self . vrand [ j ] ] = v <EOL> return feed </s>
