<s> from theano import tensor <EOL> from blocks . bricks import ( <EOL> Brick , Identity , Initializable , NDimensionalSoftmax , application , lazy ) <EOL> from blocks . bricks . recurrent import BaseRecurrent , recurrent <EOL> from blocks . bricks . parallel import Fork , Merge <EOL> from blocks . bricks . sequence_generators import ( <EOL> AbstractEmitter , SequenceGenerator , Readout , SoftmaxEmitter ) <EOL> from blocks . bricks . wrappers import WithExtraDims <EOL> from lvsr . ops import FST , FSTCostsOp , FSTTransitionOp , MAX_STATES , NOT_STATE <EOL> class FSTTransition ( BaseRecurrent , Initializable ) : <EOL> def __init__ ( self , fst , remap_table , no_transition_cost , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> super ( FSTTransition , self ) . __init__ ( ** kwargs ) <EOL> self . fst = fst <EOL> self . transition = FSTTransitionOp ( fst , remap_table ) <EOL> self . probability_computer = FSTCostsOp ( <EOL> fst , remap_table , no_transition_cost ) <EOL> self . out_dim = len ( remap_table ) <EOL> @ recurrent ( sequences = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> states = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , contexts = [ ] ) <EOL> def apply ( self , inputs , states , weights , add , <EOL> mask = None ) : <EOL> new_states , new_weights = self . transition ( states , weights , inputs ) <EOL> if mask : <EOL> new_states = tensor . cast ( mask * new_states + <EOL> ( <NUM_LIT:1.> - mask ) * states , '<STR_LIT>' ) <EOL> new_weights = mask * new_weights + ( <NUM_LIT:1.> - mask ) * weights <EOL> new_add = self . probability_computer ( new_states , new_weights ) <EOL> return new_states , new_weights , new_add <EOL> @ application ( outputs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def initial_states ( self , batch_size , * args , ** kwargs ) : <EOL> states_dict = self . fst . expand ( { self . fst . fst . start : <NUM_LIT:0.0> } ) <EOL> states = tensor . as_tensor_variable ( <EOL> self . transition . pad ( states_dict . keys ( ) , NOT_STATE ) ) <EOL> states = tensor . tile ( states [ None , : ] , ( batch_size , <NUM_LIT:1> ) ) <EOL> weights = tensor . as_tensor_variable ( <EOL> self . transition . pad ( states_dict . values ( ) , <NUM_LIT:0> ) ) <EOL> weights = tensor . tile ( weights [ None , : ] , ( batch_size , <NUM_LIT:1> ) ) <EOL> add = self . probability_computer ( states , weights ) <EOL> return states , weights , add <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' or name == '<STR_LIT>' : <EOL> return MAX_STATES <EOL> if name == '<STR_LIT>' : <EOL> return self . out_dim <EOL> if name == '<STR_LIT>' : <EOL> return <NUM_LIT:0> <EOL> return super ( FSTTransition , self ) . get_dim ( name ) <EOL> class ShallowFusionReadout ( Readout ) : <EOL> def __init__ ( self , lm_costs_name , lm_weight , <EOL> normalize_am_weights = False , <EOL> normalize_lm_weights = False , <EOL> normalize_tot_weights = True , <EOL> am_beta = <NUM_LIT:1.0> , <EOL> ** kwargs ) : <EOL> super ( ShallowFusionReadout , self ) . __init__ ( ** kwargs ) <EOL> self . lm_costs_name = lm_costs_name <EOL> self . lm_weight = lm_weight <EOL> self . normalize_am_weights = normalize_am_weights <EOL> self . normalize_lm_weights = normalize_lm_weights <EOL> self . normalize_tot_weights = normalize_tot_weights <EOL> self . am_beta = am_beta <EOL> self . softmax = NDimensionalSoftmax ( ) <EOL> self . children += [ self . softmax ] <EOL> @ application <EOL> def readout ( self , ** kwargs ) : <EOL> lm_costs = - kwargs . pop ( self . lm_costs_name ) <EOL> if self . normalize_lm_weights : <EOL> lm_costs = self . softmax . log_probabilities ( <EOL> lm_costs , extra_ndim = lm_costs . ndim - <NUM_LIT:2> ) <EOL> am_pre_softmax = self . am_beta * super ( ShallowFusionReadout , self ) . readout ( ** kwargs ) <EOL> if self . normalize_am_weights : <EOL> am_pre_softmax = self . softmax . log_probabilities ( <EOL> am_pre_softmax , extra_ndim = am_pre_softmax . ndim - <NUM_LIT:2> ) <EOL> x = am_pre_softmax + self . lm_weight * lm_costs <EOL> if self . normalize_tot_weights : <EOL> x = self . softmax . log_probabilities ( x , extra_ndim = x . ndim - <NUM_LIT:2> ) <EOL> return x <EOL> class LanguageModel ( SequenceGenerator ) : <EOL> def __init__ ( self , path , nn_char_map , no_transition_cost = <NUM_LIT> , ** kwargs ) : <EOL> fst = FST ( path ) <EOL> fst_char_map = dict ( fst . fst . isyms . items ( ) ) <EOL> del fst_char_map [ '<STR_LIT>' ] <EOL> if not len ( fst_char_map ) == len ( nn_char_map ) : <EOL> raise ValueError ( ) <EOL> remap_table = { nn_char_map [ character ] : fst_code <EOL> for character , fst_code in fst_char_map . items ( ) } <EOL> transition = FSTTransition ( fst , remap_table , no_transition_cost ) <EOL> dummy_readout = Readout ( <EOL> source_names = [ '<STR_LIT>' ] , readout_dim = len ( remap_table ) , <EOL> merge = Merge ( input_names = [ '<STR_LIT>' ] , prototype = Identity ( ) ) , <EOL> post_merge = Identity ( ) , <EOL> emitter = SoftmaxEmitter ( ) ) <EOL> super ( LanguageModel , self ) . __init__ ( <EOL> transition = transition , <EOL> fork = Fork ( output_names = [ name for name in transition . apply . sequences <EOL> if name != '<STR_LIT>' ] , <EOL> prototype = Identity ( ) ) , <EOL> readout = dummy_readout , ** kwargs ) <EOL> class SelectInEachRow ( Brick ) : <EOL> @ application ( inputs = [ '<STR_LIT>' , '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def apply ( self , matrix , indices ) : <EOL> return matrix [ tensor . arange ( matrix . shape [ <NUM_LIT:0> ] ) , indices ] <EOL> class SelectInEachSubtensor ( SelectInEachRow ) : <EOL> decorators = [ WithExtraDims ( ) ] <EOL> class LMEmitter ( AbstractEmitter ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' ] ) <EOL> def __init__ ( self , readout_dim , ** kwargs ) : <EOL> super ( LMEmitter , self ) . __init__ ( ** kwargs ) <EOL> self . readout_dim = readout_dim <EOL> self . select = SelectInEachSubtensor ( ) <EOL> self . children = [ self . select ] <EOL> @ application <EOL> def emit ( self , readouts ) : <EOL> return tensor . zeros_like ( readouts [ : , <NUM_LIT:0> ] , dtype = '<STR_LIT>' ) <EOL> @ application <EOL> def cost ( self , readouts , outputs ) : <EOL> return - self . select . apply ( <EOL> readouts , outputs , extra_ndim = readouts . ndim - <NUM_LIT:2> ) <EOL> @ application <EOL> def costs ( self , readouts ) : <EOL> return - readouts <EOL> @ application <EOL> def initial_outputs ( self , batch_size ) : <EOL> return tensor . zeros ( ( batch_size , ) , dtype = '<STR_LIT>' ) <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return <NUM_LIT:0> <EOL> return super ( LMEmitter , self ) . get_dim ( name ) </s>
