<s> from itertools import izip <EOL> from unittest import TestCase <EOL> import theano <EOL> import numpy as np <EOL> from theano import tensor as T <EOL> import quagga <EOL> from quagga . matrix import Matrix <EOL> from quagga . context import Context <EOL> from quagga . blocks import DotBlock <EOL> from quagga . blocks import LstmBlock <EOL> from quagga . utils import List <EOL> from quagga . connector import Connector <EOL> from quagga . blocks import SoftmaxCeBlock <EOL> from quagga . blocks import SigmoidCeBlock <EOL> from quagga . blocks import SequencerBlock <EOL> class TestSequentialLstmBlock ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . rng = np . random . RandomState ( seed = <NUM_LIT> ) <EOL> cls . N = <NUM_LIT:10> <EOL> @ classmethod <EOL> def get_orthogonal_matrix ( cls , nrows , ncols ) : <EOL> shape = ( nrows , ncols ) <EOL> a = cls . rng . normal ( <NUM_LIT:0.0> , <NUM_LIT:1.0> , shape ) <EOL> u , _ , v = np . linalg . svd ( a , full_matrices = False ) <EOL> q = u if u . shape == shape else v <EOL> q = q . reshape ( shape ) . astype ( np . float32 ) <EOL> return q <EOL> def test_fprop ( self ) : <EOL> """<STR_LIT>""" <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> max_input_sequence_len = self . rng . random_integers ( <NUM_LIT> ) <EOL> sequence_len = max_input_sequence_len if i == <NUM_LIT:0> else self . rng . random_integers ( max_input_sequence_len ) <EOL> batch_size = self . rng . random_integers ( <NUM_LIT> ) <EOL> input_dim , hidden_dim = self . rng . random_integers ( <NUM_LIT> , size = <NUM_LIT:2> ) <EOL> x = [ self . rng . randn ( batch_size , input_dim ) . astype ( np . float32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> mask = ( self . rng . rand ( batch_size , sequence_len ) < <NUM_LIT> ) . astype ( np . float32 ) <EOL> h_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> c_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> W_z = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_i = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_f = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_o = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W = np . hstack ( ( W_z , W_i , W_f , W_o ) ) <EOL> R_z = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_i = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_f = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_o = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R = np . hstack ( ( R_z , R_i , R_f , R_o ) ) <EOL> qh = { } <EOL> for reverse in [ False , True ] : <EOL> for with_mask in [ False , True ] : <EOL> for processor_type in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> quagga . processor_type = processor_type <EOL> context = Context ( ) <EOL> qx = List ( [ Connector ( Matrix . from_npa ( e ) ) for e in x ] ) <EOL> qmask = Matrix . empty ( batch_size , len ( qx ) , '<STR_LIT:float>' ) <EOL> qh_0 = Connector ( Matrix . from_npa ( h_0 ) ) <EOL> qc_0 = Connector ( Matrix . from_npa ( c_0 ) ) <EOL> qW = Connector ( Matrix . from_npa ( W ) ) <EOL> qR = Connector ( Matrix . from_npa ( R ) ) <EOL> sequences = [ qx ] <EOL> if with_mask : <EOL> sequences . append ( List ( [ Connector ( qmask [ : , i ] ) for i in xrange ( len ( qx ) ) ] , len ( qx ) ) ) <EOL> qmask . assign_npa ( context , mask ) <EOL> qmask = sequences [ - <NUM_LIT:1> ] <EOL> else : <EOL> sequences . append ( [ None ] * len ( qx ) ) <EOL> lstm = SequencerBlock ( block_class = LstmBlock , <EOL> params = [ qW , qR ] , <EOL> sequences = sequences , <EOL> output_names = [ '<STR_LIT:h>' ] , <EOL> prev_names = [ '<STR_LIT:c>' , '<STR_LIT:h>' ] , <EOL> paddings = [ qc_0 , qh_0 ] , <EOL> reverse = reverse ) <EOL> qx . length = sequence_len <EOL> if with_mask : <EOL> qmask . fprop ( ) <EOL> qx . fprop ( ) <EOL> qh_0 . fprop ( ) <EOL> qc_0 . fprop ( ) <EOL> qW . fprop ( ) <EOL> qR . fprop ( ) <EOL> lstm . fprop ( ) <EOL> qh [ processor_type ] = lstm . h . to_host ( ) <EOL> for h_gpu , h_cpu in izip ( qh [ '<STR_LIT>' ] , qh [ '<STR_LIT>' ] ) : <EOL> if not np . allclose ( h_gpu , h_cpu , rtol = <NUM_LIT> , atol = <NUM_LIT> ) : <EOL> r . append ( False ) <EOL> break <EOL> else : <EOL> r . append ( True ) <EOL> self . assertEqual ( sum ( r ) , len ( r ) ) <EOL> def test_bprop ( self ) : <EOL> """<STR_LIT>""" <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> max_input_sequence_len = self . rng . random_integers ( <NUM_LIT> ) <EOL> sequence_len = max_input_sequence_len if i == <NUM_LIT:0> else self . rng . random_integers ( max_input_sequence_len ) <EOL> batch_size = self . rng . random_integers ( <NUM_LIT> ) <EOL> input_dim , hidden_dim = self . rng . random_integers ( <NUM_LIT> , size = <NUM_LIT:2> ) <EOL> x = [ self . rng . randn ( batch_size , input_dim ) . astype ( np . float32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> true_labels = [ self . rng . randint ( <NUM_LIT:2> , size = ( batch_size , <NUM_LIT:1> ) ) . astype ( np . float32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> mask = ( self . rng . rand ( batch_size , sequence_len ) < <NUM_LIT> ) . astype ( np . float32 ) <EOL> h_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> c_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> W_z = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_i = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_f = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_o = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W = np . hstack ( ( W_z , W_i , W_f , W_o ) ) <EOL> R_z = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_i = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_f = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_o = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R = np . hstack ( ( R_z , R_i , R_f , R_o ) ) <EOL> lr_W = self . get_orthogonal_matrix ( hidden_dim , <NUM_LIT:1> ) <EOL> lr_b = self . rng . rand ( <NUM_LIT:1> , <NUM_LIT:1> ) . astype ( dtype = np . float32 ) <EOL> device_id = <NUM_LIT:0> <EOL> quagga_grads = { } <EOL> for reverse in [ False , True ] : <EOL> for with_mask in [ False , True ] : <EOL> for learn_inital_states in [ False , True ] : <EOL> for processor_type in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> quagga . processor_type = processor_type <EOL> context = Context ( ) <EOL> qx = List ( [ Connector ( Matrix . from_npa ( e ) , device_id ) for e in x ] ) <EOL> qtrue_labels = List ( [ Connector ( Matrix . from_npa ( e ) ) for e in true_labels ] , len ( qx ) ) <EOL> qmask = Matrix . empty ( batch_size , len ( qx ) ) <EOL> qh_0 = Connector ( Matrix . from_npa ( h_0 ) , device_id if learn_inital_states else None ) <EOL> qc_0 = Connector ( Matrix . from_npa ( c_0 ) , device_id if learn_inital_states else None ) <EOL> qW = Connector ( Matrix . from_npa ( W ) , device_id ) <EOL> qR = Connector ( Matrix . from_npa ( R ) , device_id ) <EOL> qlr_W = Connector ( Matrix . from_npa ( lr_W ) , device_id ) <EOL> qlr_b = Connector ( Matrix . from_npa ( lr_b ) , device_id ) <EOL> sequences = [ qx ] <EOL> if with_mask : <EOL> sequences . append ( List ( [ Connector ( qmask [ : , i ] ) for i in xrange ( len ( qx ) ) ] , len ( qx ) ) ) <EOL> qmask . assign_npa ( context , mask ) <EOL> qmask = sequences [ - <NUM_LIT:1> ] <EOL> else : <EOL> sequences . append ( [ None ] * len ( qx ) ) <EOL> lstm = SequencerBlock ( block_class = LstmBlock , <EOL> params = [ qW , qR ] , <EOL> sequences = sequences , <EOL> output_names = [ '<STR_LIT:h>' ] , <EOL> prev_names = [ '<STR_LIT:c>' , '<STR_LIT:h>' ] , <EOL> paddings = [ qc_0 , qh_0 ] , <EOL> reverse = reverse ) <EOL> seq_dot_block = SequencerBlock ( block_class = DotBlock , <EOL> params = [ qlr_W , qlr_b ] , <EOL> sequences = [ lstm . h ] , <EOL> output_names = [ '<STR_LIT>' ] ) <EOL> seq_sce_block = SequencerBlock ( block_class = SigmoidCeBlock , <EOL> params = [ ] , <EOL> sequences = [ seq_dot_block . output , qtrue_labels ] + ( [ qmask ] if with_mask else [ ] ) ) <EOL> qx . length = sequence_len <EOL> qx . fprop ( ) <EOL> qtrue_labels . fprop ( ) <EOL> if with_mask : <EOL> qmask . fprop ( ) <EOL> qlr_W . fprop ( ) <EOL> qlr_b . fprop ( ) <EOL> qh_0 . fprop ( ) <EOL> qc_0 . fprop ( ) <EOL> qW . fprop ( ) <EOL> qR . fprop ( ) <EOL> lstm . fprop ( ) <EOL> seq_dot_block . fprop ( ) <EOL> seq_sce_block . fprop ( ) <EOL> seq_sce_block . bprop ( ) <EOL> seq_dot_block . bprop ( ) <EOL> lstm . bprop ( ) <EOL> quagga_grads [ processor_type ] = [ qlr_b . backward_matrix . to_host ( ) , <EOL> qlr_W . backward_matrix . to_host ( ) , <EOL> qW . backward_matrix . to_host ( ) , <EOL> qR . backward_matrix . to_host ( ) ] <EOL> if learn_inital_states : <EOL> quagga_grads [ processor_type ] . append ( qc_0 . backward_matrix . to_host ( ) ) <EOL> quagga_grads [ processor_type ] . append ( qh_0 . backward_matrix . to_host ( ) ) <EOL> quagga_grads [ processor_type ] . extend ( e . backward_matrix . to_host ( ) for e in qx ) <EOL> for grad_gpu , grad_cpu in izip ( quagga_grads [ '<STR_LIT>' ] , quagga_grads [ '<STR_LIT>' ] ) : <EOL> r . append ( np . allclose ( grad_gpu , grad_cpu , atol = <NUM_LIT> ) ) <EOL> self . assertEqual ( sum ( r ) , len ( r ) ) <EOL> def test_theano_fprop ( self ) : <EOL> quagga . processor_type = '<STR_LIT>' <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> max_input_sequence_len = self . rng . random_integers ( <NUM_LIT> ) <EOL> sequence_len = max_input_sequence_len if i == <NUM_LIT:0> else self . rng . random_integers ( max_input_sequence_len ) <EOL> batch_size = self . rng . random_integers ( <NUM_LIT> ) <EOL> input_dim , hidden_dim = self . rng . random_integers ( <NUM_LIT> , size = <NUM_LIT:2> ) <EOL> x = [ self . rng . randn ( batch_size , input_dim ) . astype ( np . float32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> mask = ( self . rng . rand ( batch_size , sequence_len ) < <NUM_LIT> ) . astype ( np . float32 ) <EOL> h_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> c_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> W_z = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_i = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_f = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_o = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W = np . hstack ( ( W_z , W_i , W_f , W_o ) ) <EOL> R_z = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_i = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_f = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_o = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R = np . hstack ( ( R_z , R_i , R_f , R_o ) ) <EOL> for reverse in [ False , True ] : <EOL> for with_mask in [ False , True ] : <EOL> context = Context ( ) <EOL> qx = List ( [ Connector ( Matrix . from_npa ( e ) ) for e in x ] ) <EOL> qmask = Connector ( Matrix . empty ( batch_size , len ( qx ) , '<STR_LIT:float>' ) ) <EOL> qh_0 = Connector ( Matrix . from_npa ( h_0 ) ) <EOL> qc_0 = Connector ( Matrix . from_npa ( c_0 ) ) <EOL> qW = Connector ( Matrix . from_npa ( W ) ) <EOL> qR = Connector ( Matrix . from_npa ( R ) ) <EOL> lstm = SequencerBlock ( block_class = LstmBlock , <EOL> params = [ qW , qR ] , <EOL> sequences = [ qx ] + ( [ qmask ] if with_mask else [ ] ) , <EOL> output_names = [ '<STR_LIT:h>' ] , <EOL> prev_names = [ '<STR_LIT:c>' , '<STR_LIT:h>' ] , <EOL> paddings = [ qc_0 , qh_0 ] , <EOL> reverse = reverse ) <EOL> qx . length = sequence_len <EOL> for e in qx : <EOL> e . fprop ( ) <EOL> qmask . assign_npa ( context , mask ) <EOL> qmask . fprop ( ) <EOL> qh_0 . fprop ( ) <EOL> qc_0 . fprop ( ) <EOL> qW . fprop ( ) <EOL> qR . fprop ( ) <EOL> lstm . fprop ( ) <EOL> q_h = lstm . h . to_host ( ) <EOL> th_x = T . ftensor3 ( ) <EOL> lstm_layer = LstmLayer ( W , R , c_0 , h_0 , reverse ) <EOL> if with_mask : <EOL> th_mask = T . fmatrix ( ) <EOL> get_th_h = theano . function ( [ th_x , th_mask ] , lstm_layer . get_output_expr ( th_x , th_mask ) ) <EOL> th_h = get_th_h ( np . dstack ( x [ : sequence_len ] ) , mask [ : , : sequence_len ] ) <EOL> else : <EOL> get_th_h = theano . function ( [ th_x ] , lstm_layer . get_output_expr ( th_x ) ) <EOL> th_h = get_th_h ( np . dstack ( x [ : sequence_len ] ) ) <EOL> for i in xrange ( th_h . shape [ <NUM_LIT:0> ] ) : <EOL> if not np . allclose ( q_h [ i ] , th_h [ i ] ) : <EOL> r . append ( False ) <EOL> break <EOL> else : <EOL> r . append ( True ) <EOL> self . assertEqual ( sum ( r ) , len ( r ) ) <EOL> def test_theano_grad ( self ) : <EOL> quagga . processor_type = '<STR_LIT>' <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> max_input_sequence_len = self . rng . random_integers ( <NUM_LIT> ) <EOL> sequence_len = max_input_sequence_len if i == <NUM_LIT:0> else self . rng . random_integers ( max_input_sequence_len ) <EOL> batch_size = self . rng . random_integers ( <NUM_LIT> ) <EOL> input_dim , hidden_dim , class_num = self . rng . random_integers ( <NUM_LIT> , size = <NUM_LIT:3> ) <EOL> x = [ self . rng . randn ( batch_size , input_dim ) . astype ( np . float32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> true_labels = [ self . rng . randint ( class_num , size = ( batch_size , <NUM_LIT:1> ) ) . astype ( np . int32 ) for _ in xrange ( max_input_sequence_len ) ] <EOL> mask = ( self . rng . rand ( batch_size , sequence_len ) < <NUM_LIT> ) . astype ( np . float32 ) <EOL> h_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> c_0 = self . rng . randn ( batch_size , hidden_dim ) . astype ( np . float32 ) <EOL> W_z = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_i = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_f = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W_o = self . get_orthogonal_matrix ( input_dim , hidden_dim ) <EOL> W = np . hstack ( ( W_z , W_i , W_f , W_o ) ) <EOL> R_z = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_i = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_f = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R_o = self . get_orthogonal_matrix ( hidden_dim , hidden_dim ) <EOL> R = np . hstack ( ( R_z , R_i , R_f , R_o ) ) <EOL> lr_W = self . get_orthogonal_matrix ( hidden_dim , class_num ) <EOL> lr_b = self . rng . rand ( <NUM_LIT:1> , class_num ) . astype ( dtype = np . float32 ) <EOL> device_id = <NUM_LIT:0> <EOL> for reverse in [ False , True ] : <EOL> for with_mask in [ False , True ] : <EOL> for learn_inital_states in [ False , True ] : <EOL> context = Context ( ) <EOL> qx = List ( [ Connector ( Matrix . from_npa ( e ) , device_id ) for e in x ] ) <EOL> qtrue_labels = List ( [ Connector ( Matrix . from_npa ( e ) ) for e in true_labels ] , qx . length ) <EOL> qmask = Matrix . empty ( batch_size , qx . length , '<STR_LIT:float>' ) <EOL> qmask_list = [ Connector ( qmask [ : , i ] ) for i in xrange ( qmask . ncols ) ] <EOL> qmask = Connector ( qmask ) <EOL> qh_0 = Connector ( Matrix . from_npa ( h_0 ) , device_id if learn_inital_states else None ) <EOL> qc_0 = Connector ( Matrix . from_npa ( c_0 ) , device_id if learn_inital_states else None ) <EOL> qW = Connector ( Matrix . from_npa ( W ) , device_id ) <EOL> qR = Connector ( Matrix . from_npa ( R ) , device_id ) <EOL> qlr_W = Connector ( Matrix . from_npa ( lr_W ) , device_id ) <EOL> qlr_b = Connector ( Matrix . from_npa ( lr_b ) , device_id ) <EOL> lstm = SequencerBlock ( block_class = LstmBlock , <EOL> params = [ qW , qR ] , <EOL> sequences = [ qx , qmask_list if with_mask else [ None ] * len ( qx ) ] , <EOL> output_names = [ '<STR_LIT:h>' ] , <EOL> prev_names = [ '<STR_LIT:c>' , '<STR_LIT:h>' ] , <EOL> paddings = [ qc_0 , qh_0 ] , <EOL> reverse = reverse ) <EOL> seq_dot_block = SequencerBlock ( block_class = DotBlock , <EOL> params = [ qlr_W , qlr_b ] , <EOL> sequences = [ lstm . h ] , <EOL> output_names = [ '<STR_LIT>' ] ) <EOL> seq_sce_block = SequencerBlock ( block_class = SoftmaxCeBlock , <EOL> params = [ ] , <EOL> sequences = [ seq_dot_block . output , qtrue_labels , qmask_list if with_mask else [ None ] * len ( qx ) ] ) <EOL> qx . length = sequence_len <EOL> for e in qx : <EOL> e . fprop ( ) <EOL> for e in qtrue_labels : <EOL> e . fprop ( ) <EOL> qmask . assign_npa ( context , mask ) <EOL> qmask . fprop ( ) <EOL> qlr_W . fprop ( ) <EOL> qlr_b . fprop ( ) <EOL> qh_0 . fprop ( ) <EOL> qc_0 . fprop ( ) <EOL> qW . fprop ( ) <EOL> qR . fprop ( ) <EOL> lstm . fprop ( ) <EOL> seq_dot_block . fprop ( ) <EOL> seq_sce_block . fprop ( ) <EOL> seq_sce_block . bprop ( ) <EOL> seq_dot_block . bprop ( ) <EOL> lstm . bprop ( ) <EOL> quagga_grads = [ qlr_b . backward_matrix . to_host ( ) , <EOL> qlr_W . backward_matrix . to_host ( ) , <EOL> qW . backward_matrix . to_host ( ) , <EOL> qR . backward_matrix . to_host ( ) ] <EOL> if learn_inital_states : <EOL> quagga_grads . append ( qc_0 . backward_matrix . to_host ( ) ) <EOL> quagga_grads . append ( qh_0 . backward_matrix . to_host ( ) ) <EOL> quagga_grads . append ( [ e . backward_matrix . to_host ( ) for e in qx ] ) <EOL> del qx <EOL> del qlr_b <EOL> del qlr_W <EOL> del qW <EOL> del qR <EOL> del qmask <EOL> del lstm <EOL> del seq_dot_block <EOL> del seq_sce_block <EOL> th_x = T . ftensor3 ( ) <EOL> th_true_labels = T . imatrix ( ) <EOL> th_mask = T . fmatrix ( ) <EOL> lstm_layer = LstmLayer ( W , R , c_0 , h_0 , reverse = reverse ) <EOL> th_h = lstm_layer . get_output_expr ( th_x , th_mask if with_mask else None ) <EOL> seq_softmax_layer = SequentialSoftmaxLayer ( lr_W , lr_b , reverse ) <EOL> loss = seq_softmax_layer . get_loss ( th_h , th_true_labels , th_mask if with_mask else None ) <EOL> wrt = [ seq_softmax_layer . b , seq_softmax_layer . W , lstm_layer . W , lstm_layer . R ] <EOL> if learn_inital_states : <EOL> wrt . append ( lstm_layer . c0 ) <EOL> wrt . append ( lstm_layer . h0 ) <EOL> wrt . append ( th_x ) <EOL> grads = T . grad ( loss , wrt ) <EOL> if with_mask : <EOL> get_theano_grads = theano . function ( [ th_x , th_true_labels , th_mask ] , grads ) <EOL> theano_grads = get_theano_grads ( np . dstack ( x [ : sequence_len ] ) , np . hstack ( true_labels [ : sequence_len ] ) , mask [ : , : sequence_len ] ) <EOL> else : <EOL> get_theano_grads = theano . function ( [ th_x , th_true_labels ] , grads ) <EOL> theano_grads = get_theano_grads ( np . dstack ( x [ : sequence_len ] ) , np . hstack ( true_labels [ : sequence_len ] ) ) <EOL> for quagga_grad , theano_grad in izip ( quagga_grads [ : - <NUM_LIT:1> ] , theano_grads [ : - <NUM_LIT:1> ] ) : <EOL> r . append ( np . allclose ( quagga_grad , theano_grad , atol = <NUM_LIT> ) ) <EOL> for i in xrange ( theano_grads [ - <NUM_LIT:1> ] . shape [ - <NUM_LIT:1> ] ) : <EOL> if not np . allclose ( quagga_grads [ - <NUM_LIT:1> ] [ i ] , theano_grads [ - <NUM_LIT:1> ] [ ... , i ] , atol = <NUM_LIT> ) : <EOL> r . append ( False ) <EOL> break <EOL> else : <EOL> r . append ( True ) <EOL> self . assertEqual ( sum ( r ) , len ( r ) ) <EOL> class LstmLayer ( object ) : <EOL> def __init__ ( self , W , R , c0 , h0 , reverse ) : <EOL> self . W = theano . shared ( W , name = '<STR_LIT>' ) <EOL> self . R = theano . shared ( R , name = '<STR_LIT>' ) <EOL> self . c0 = theano . shared ( c0 , name = '<STR_LIT>' ) <EOL> self . h0 = theano . shared ( h0 , name = '<STR_LIT>' ) <EOL> self . n = W . shape [ <NUM_LIT:1> ] / <NUM_LIT:4> <EOL> self . reverse = reverse <EOL> def get_output_expr ( self , input_sequence , mask = None ) : <EOL> input_sequence = input_sequence . transpose ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> mask = mask . T if mask else T . ones ( input_sequence . shape [ : <NUM_LIT:2> ] , dtype = np . float32 ) <EOL> if self . reverse : <EOL> input_sequence = input_sequence [ : : - <NUM_LIT:1> ] <EOL> mask = mask [ : : - <NUM_LIT:1> ] <EOL> [ _ , h ] , _ = theano . scan ( fn = self . __get_lstm_mask_step , <EOL> sequences = [ input_sequence , mask ] , <EOL> outputs_info = [ self . c0 , self . h0 ] ) <EOL> return h [ : : - <NUM_LIT:1> ] if self . reverse else h <EOL> def __get_lstm_mask_step ( self , x_t , mask_t , c_tm1 , h_tm1 ) : <EOL> sigm = T . nnet . sigmoid <EOL> tanh = T . tanh <EOL> dot = theano . dot <EOL> zifo_t = dot ( x_t , self . W ) + dot ( h_tm1 , self . R ) <EOL> z_t = tanh ( zifo_t [ : , <NUM_LIT:0> * self . n : <NUM_LIT:1> * self . n ] ) <EOL> i_t = sigm ( zifo_t [ : , <NUM_LIT:1> * self . n : <NUM_LIT:2> * self . n ] ) <EOL> f_t = sigm ( zifo_t [ : , <NUM_LIT:2> * self . n : <NUM_LIT:3> * self . n ] ) <EOL> o_t = sigm ( zifo_t [ : , <NUM_LIT:3> * self . n : <NUM_LIT:4> * self . n ] ) <EOL> mask_t = mask_t . dimshuffle ( <NUM_LIT:0> , '<STR_LIT:x>' ) <EOL> c_t = ( i_t * z_t + f_t * c_tm1 ) * mask_t <EOL> h_t = o_t * tanh ( c_t ) * mask_t <EOL> return mask_t * c_t + ( <NUM_LIT:1.0> - mask_t ) * c_tm1 , mask_t * h_t + ( <NUM_LIT:1.0> - mask_t ) * h_tm1 <EOL> class SequentialSoftmaxLayer ( object ) : <EOL> def __init__ ( self , W , b , reverse ) : <EOL> self . W = theano . shared ( W ) <EOL> self . b = theano . shared ( b , broadcastable = ( True , False ) ) <EOL> self . reverse = reverse <EOL> def get_loss ( self , input_sequence , true_labels , mask = None ) : <EOL> input_sequence = T . dot ( input_sequence , self . W ) + self . b <EOL> true_labels = true_labels . T <EOL> mask = mask . T if mask else T . ones ( input_sequence . shape [ : <NUM_LIT:2> ] , np . float32 ) <EOL> losses , _ = theano . scan ( fn = self . __step_loss , <EOL> sequences = [ input_sequence , mask , true_labels ] ) <EOL> return T . sum ( losses ) <EOL> def __step_loss ( self , x_t , mask_t , true_labels_t ) : <EOL> probs = T . nnet . softmax ( x_t ) <EOL> return T . mean ( mask_t * T . nnet . categorical_crossentropy ( probs , true_labels_t ) ) </s>
