<s> import numpy <EOL> import theano <EOL> import logging <EOL> from theano import tensor <EOL> from blocks . bricks import ( <EOL> Bias , Identity , Initializable , MLP , Tanh ) <EOL> from blocks . bricks . attention import SequenceContentAttention <EOL> from blocks . bricks . base import application <EOL> from blocks . bricks . recurrent import ( <EOL> BaseRecurrent , RecurrentStack ) <EOL> from blocks . bricks . sequence_generators import ( <EOL> SequenceGenerator , Readout , <EOL> SoftmaxEmitter , LookupFeedback ) <EOL> from blocks . bricks . lookup import LookupTable <EOL> from blocks . graph import ComputationGraph <EOL> from blocks . model import Model <EOL> from blocks . filter import VariableFilter <EOL> from blocks . roles import OUTPUT <EOL> from blocks . search import BeamSearch <EOL> from blocks . serialization import load_parameters <EOL> from lvsr . bricks import ( <EOL> Encoder , OneOfNFeedback , InitializableSequence , RewardRegressionEmitter ) <EOL> from lvsr . bricks . attention import SequenceContentAndConvAttention <EOL> from lvsr . bricks . language_models import ( <EOL> LanguageModel , LMEmitter , ShallowFusionReadout ) <EOL> from lvsr . utils import global_push_initialization_config <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Bottom ( Initializable ) : <EOL> """<STR_LIT>""" <EOL> vector_input_sources = [ ] <EOL> discrete_input_sources = [ ] <EOL> def __init__ ( self , input_dims , input_num_chars , ** kwargs ) : <EOL> super ( Bottom , self ) . __init__ ( ** kwargs ) <EOL> self . input_dims = input_dims <EOL> self . input_num_chars = input_num_chars <EOL> class LookupBottom ( Bottom ) : <EOL> discrete_input_sources = [ '<STR_LIT>' ] <EOL> def __init__ ( self , dim , ** kwargs ) : <EOL> super ( LookupBottom , self ) . __init__ ( ** kwargs ) <EOL> self . dim = dim <EOL> self . mask = tensor . matrix ( '<STR_LIT>' ) <EOL> self . batch_inputs = { <EOL> '<STR_LIT>' : tensor . lmatrix ( '<STR_LIT>' ) } <EOL> self . single_inputs = { <EOL> '<STR_LIT>' : tensor . lvector ( '<STR_LIT>' ) } <EOL> self . children = [ LookupTable ( self . input_num_chars [ '<STR_LIT>' ] , self . dim ) ] <EOL> @ application ( inputs = [ '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def apply ( self , inputs ) : <EOL> return self . children [ <NUM_LIT:0> ] . apply ( inputs ) <EOL> def batch_size ( self , inputs ) : <EOL> return inputs . shape [ <NUM_LIT:1> ] <EOL> def num_time_steps ( self , inputs ) : <EOL> return inputs . shape [ <NUM_LIT:0> ] <EOL> def single_to_batch_inputs ( self , inputs ) : <EOL> inputs = { n : v [ : , None , : ] if v . ndim == <NUM_LIT:2> else v [ : , None ] <EOL> for ( n , v ) in inputs . items ( ) } <EOL> inputs_mask = tensor . ones ( ( self . num_time_steps ( ** inputs ) , <EOL> self . batch_size ( ** inputs ) ) ) <EOL> return inputs , inputs_mask <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . dim <EOL> return super ( LookupBottom , self ) . get_dim ( name ) <EOL> class SpeechBottom ( Bottom ) : <EOL> """<STR_LIT>""" <EOL> vector_input_sources = [ '<STR_LIT>' ] <EOL> def __init__ ( self , activation , dims = None , ** kwargs ) : <EOL> super ( SpeechBottom , self ) . __init__ ( ** kwargs ) <EOL> self . num_features = self . input_dims [ '<STR_LIT>' ] <EOL> if activation is None : <EOL> activation = Tanh ( ) <EOL> if dims : <EOL> child = MLP ( [ activation ] * len ( dims ) , <EOL> [ self . num_features ] + dims , <EOL> name = "<STR_LIT>" ) <EOL> self . output_dim = child . output_dim <EOL> else : <EOL> child = Identity ( name = '<STR_LIT>' ) <EOL> self . output_dim = self . num_features <EOL> self . children . append ( child ) <EOL> self . mask = tensor . matrix ( '<STR_LIT>' ) <EOL> self . batch_inputs = { <EOL> '<STR_LIT>' : tensor . tensor3 ( '<STR_LIT>' ) } <EOL> self . single_inputs = { <EOL> '<STR_LIT>' : tensor . matrix ( '<STR_LIT>' ) } <EOL> @ application ( inputs = [ '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def apply ( self , recordings ) : <EOL> return self . children [ <NUM_LIT:0> ] . apply ( recordings ) <EOL> def batch_size ( self , recordings ) : <EOL> return recordings . shape [ <NUM_LIT:1> ] <EOL> def num_time_steps ( self , recordings ) : <EOL> return recordings . shape [ <NUM_LIT:0> ] <EOL> def single_to_batch_inputs ( self , inputs ) : <EOL> inputs = { n : v [ : , None , : ] if v . ndim == <NUM_LIT:2> else v [ : , None ] <EOL> for ( n , v ) in inputs . items ( ) } <EOL> inputs_mask = tensor . ones ( ( self . num_time_steps ( ** inputs ) , <EOL> self . batch_size ( ** inputs ) ) ) <EOL> return inputs , inputs_mask <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . output_dim <EOL> return super ( SpeechBottom , self ) . get_dim ( name ) <EOL> class SpeechRecognizer ( Initializable ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> input_dims , <EOL> input_num_chars , <EOL> eos_label , <EOL> num_phonemes , <EOL> dim_dec , dims_bidir , <EOL> enc_transition , dec_transition , <EOL> use_states_for_readout , <EOL> attention_type , <EOL> criterion , <EOL> bottom , <EOL> lm = None , character_map = None , <EOL> bidir = True , <EOL> subsample = None , <EOL> dims_top = None , <EOL> prior = None , conv_n = None , <EOL> post_merge_activation = None , <EOL> post_merge_dims = None , <EOL> dim_matcher = None , <EOL> embed_outputs = True , <EOL> dim_output_embedding = None , <EOL> dec_stack = <NUM_LIT:1> , <EOL> conv_num_filters = <NUM_LIT:1> , <EOL> data_prepend_eos = True , <EOL> energy_normalizer = None , <EOL> max_decoded_length_scale = <NUM_LIT:1> , <EOL> ** kwargs ) : <EOL> if post_merge_activation is None : <EOL> post_merge_activation = Tanh ( ) <EOL> super ( SpeechRecognizer , self ) . __init__ ( ** kwargs ) <EOL> self . eos_label = eos_label <EOL> self . data_prepend_eos = data_prepend_eos <EOL> self . rec_weights_init = None <EOL> self . initial_states_init = None <EOL> self . enc_transition = enc_transition <EOL> self . dec_transition = dec_transition <EOL> self . dec_stack = dec_stack <EOL> self . criterion = criterion <EOL> self . max_decoded_length_scale = max_decoded_length_scale <EOL> post_merge_activation = post_merge_activation <EOL> if dim_matcher is None : <EOL> dim_matcher = dim_dec <EOL> bottom_class = bottom . pop ( '<STR_LIT>' ) <EOL> bottom = bottom_class ( <EOL> input_dims = input_dims , input_num_chars = input_num_chars , <EOL> name = '<STR_LIT>' , <EOL> ** bottom ) <EOL> if not subsample : <EOL> subsample = [ <NUM_LIT:1> ] * len ( dims_bidir ) <EOL> encoder = Encoder ( self . enc_transition , dims_bidir , <EOL> bottom . get_dim ( bottom . apply . outputs [ <NUM_LIT:0> ] ) , <EOL> subsample , bidir = bidir ) <EOL> dim_encoded = encoder . get_dim ( encoder . apply . outputs [ <NUM_LIT:0> ] ) <EOL> if dims_top : <EOL> top = MLP ( [ Tanh ( ) ] , <EOL> [ dim_encoded ] + dims_top + [ dim_encoded ] , name = "<STR_LIT>" ) <EOL> else : <EOL> top = Identity ( name = '<STR_LIT>' ) <EOL> if dec_stack == <NUM_LIT:1> : <EOL> transition = self . dec_transition ( <EOL> dim = dim_dec , activation = Tanh ( ) , name = "<STR_LIT>" ) <EOL> else : <EOL> transitions = [ self . dec_transition ( dim = dim_dec , <EOL> activation = Tanh ( ) , <EOL> name = "<STR_LIT>" . format ( trans_level ) ) <EOL> for trans_level in xrange ( dec_stack ) ] <EOL> transition = RecurrentStack ( transitions = transitions , <EOL> skip_connections = True ) <EOL> if attention_type == "<STR_LIT:content>" : <EOL> attention = SequenceContentAttention ( <EOL> state_names = transition . apply . states , <EOL> attended_dim = dim_encoded , match_dim = dim_matcher , <EOL> name = "<STR_LIT>" ) <EOL> elif attention_type == "<STR_LIT>" : <EOL> attention = SequenceContentAndConvAttention ( <EOL> state_names = transition . apply . states , <EOL> conv_n = conv_n , <EOL> conv_num_filters = conv_num_filters , <EOL> attended_dim = dim_encoded , match_dim = dim_matcher , <EOL> prior = prior , <EOL> energy_normalizer = energy_normalizer , <EOL> name = "<STR_LIT>" ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> . format ( attention_type ) ) <EOL> if embed_outputs : <EOL> feedback = LookupFeedback ( num_phonemes + <NUM_LIT:1> , <EOL> dim_dec if <EOL> dim_output_embedding is None <EOL> else dim_output_embedding ) <EOL> else : <EOL> feedback = OneOfNFeedback ( num_phonemes + <NUM_LIT:1> ) <EOL> if criterion [ '<STR_LIT:name>' ] == '<STR_LIT>' : <EOL> emitter = SoftmaxEmitter ( initial_output = num_phonemes , name = "<STR_LIT>" ) <EOL> if lm : <EOL> emitter = LMEmitter ( ) <EOL> elif criterion [ '<STR_LIT:name>' ] . startswith ( '<STR_LIT>' ) : <EOL> emitter = RewardRegressionEmitter ( <EOL> criterion [ '<STR_LIT:name>' ] , eos_label , num_phonemes , <EOL> criterion . get ( '<STR_LIT>' , - <NUM_LIT:1.0> ) , <EOL> name = "<STR_LIT>" ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" . format ( criterion [ '<STR_LIT:name>' ] ) ) <EOL> readout_config = dict ( <EOL> readout_dim = num_phonemes , <EOL> source_names = ( transition . apply . states if use_states_for_readout else [ ] ) <EOL> + [ attention . take_glimpses . outputs [ <NUM_LIT:0> ] ] , <EOL> emitter = emitter , <EOL> feedback_brick = feedback , <EOL> name = "<STR_LIT>" ) <EOL> if post_merge_dims : <EOL> readout_config [ '<STR_LIT>' ] = post_merge_dims [ <NUM_LIT:0> ] <EOL> readout_config [ '<STR_LIT>' ] = InitializableSequence ( [ <EOL> Bias ( post_merge_dims [ <NUM_LIT:0> ] ) . apply , <EOL> post_merge_activation . apply , <EOL> MLP ( [ post_merge_activation ] * ( len ( post_merge_dims ) - <NUM_LIT:1> ) + [ Identity ( ) ] , <EOL> [ d // getattr ( post_merge_activation , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> for d in post_merge_dims ] + [ num_phonemes ] ) . apply , <EOL> ] , <EOL> name = '<STR_LIT>' ) <EOL> readout = Readout ( ** readout_config ) <EOL> language_model = None <EOL> if lm and lm . get ( '<STR_LIT:path>' ) : <EOL> lm_weight = lm . pop ( '<STR_LIT>' , <NUM_LIT:0.0> ) <EOL> normalize_am_weights = lm . pop ( '<STR_LIT>' , True ) <EOL> normalize_lm_weights = lm . pop ( '<STR_LIT>' , False ) <EOL> normalize_tot_weights = lm . pop ( '<STR_LIT>' , False ) <EOL> am_beta = lm . pop ( '<STR_LIT>' , <NUM_LIT:1.0> ) <EOL> if normalize_am_weights + normalize_lm_weights + normalize_tot_weights < <NUM_LIT:1> : <EOL> logger . warn ( "<STR_LIT>" ) <EOL> language_model = LanguageModel ( nn_char_map = character_map , ** lm ) <EOL> readout = ShallowFusionReadout ( lm_costs_name = '<STR_LIT>' , <EOL> lm_weight = lm_weight , <EOL> normalize_am_weights = normalize_am_weights , <EOL> normalize_lm_weights = normalize_lm_weights , <EOL> normalize_tot_weights = normalize_tot_weights , <EOL> am_beta = am_beta , <EOL> ** readout_config ) <EOL> generator = SequenceGenerator ( <EOL> readout = readout , transition = transition , attention = attention , <EOL> language_model = language_model , <EOL> name = "<STR_LIT>" ) <EOL> self . encoder = encoder <EOL> self . bottom = bottom <EOL> self . top = top <EOL> self . generator = generator <EOL> self . children = [ encoder , top , bottom , generator ] <EOL> self . inputs = self . bottom . batch_inputs <EOL> self . inputs_mask = self . bottom . mask <EOL> self . labels = tensor . lmatrix ( '<STR_LIT>' ) <EOL> self . labels_mask = tensor . matrix ( "<STR_LIT>" ) <EOL> self . single_inputs = self . bottom . single_inputs <EOL> self . single_labels = tensor . lvector ( '<STR_LIT>' ) <EOL> self . n_steps = tensor . lscalar ( '<STR_LIT>' ) <EOL> def push_initialization_config ( self ) : <EOL> super ( SpeechRecognizer , self ) . push_initialization_config ( ) <EOL> if self . rec_weights_init : <EOL> rec_weights_config = { '<STR_LIT>' : self . rec_weights_init , <EOL> '<STR_LIT>' : self . rec_weights_init } <EOL> global_push_initialization_config ( self , <EOL> rec_weights_config , <EOL> BaseRecurrent ) <EOL> if self . initial_states_init : <EOL> global_push_initialization_config ( self , <EOL> { '<STR_LIT>' : self . initial_states_init } ) <EOL> @ application <EOL> def cost ( self , ** kwargs ) : <EOL> inputs_mask = kwargs . pop ( '<STR_LIT>' ) <EOL> labels = kwargs . pop ( '<STR_LIT>' ) <EOL> labels_mask = kwargs . pop ( '<STR_LIT>' ) <EOL> bottom_processed = self . bottom . apply ( ** kwargs ) <EOL> encoded , encoded_mask = self . encoder . apply ( <EOL> input_ = bottom_processed , <EOL> mask = inputs_mask ) <EOL> encoded = self . top . apply ( encoded ) <EOL> return self . generator . cost_matrix ( <EOL> labels , labels_mask , <EOL> attended = encoded , attended_mask = encoded_mask ) <EOL> @ application <EOL> def generate ( self , ** kwargs ) : <EOL> inputs_mask = kwargs . pop ( '<STR_LIT>' ) <EOL> n_steps = kwargs . pop ( '<STR_LIT>' ) <EOL> encoded , encoded_mask = self . encoder . apply ( <EOL> input_ = self . bottom . apply ( ** kwargs ) , <EOL> mask = inputs_mask ) <EOL> encoded = self . top . apply ( encoded ) <EOL> return self . generator . generate ( <EOL> n_steps = n_steps if n_steps is not None else self . n_steps , <EOL> batch_size = encoded . shape [ <NUM_LIT:1> ] , <EOL> attended = encoded , <EOL> attended_mask = encoded_mask , <EOL> as_dict = True ) <EOL> def load_params ( self , path ) : <EOL> generated = self . get_generate_graph ( ) <EOL> with open ( path , '<STR_LIT:r>' ) as src : <EOL> param_values = load_parameters ( src ) <EOL> Model ( generated [ '<STR_LIT>' ] ) . set_parameter_values ( param_values ) <EOL> def get_generate_graph ( self , use_mask = True , n_steps = None ) : <EOL> inputs_mask = None <EOL> if use_mask : <EOL> inputs_mask = self . inputs_mask <EOL> bottom_inputs = self . inputs <EOL> return self . generate ( n_steps = n_steps , <EOL> inputs_mask = inputs_mask , <EOL> ** bottom_inputs ) <EOL> def get_cost_graph ( self , batch = True , <EOL> prediction = None , prediction_mask = None ) : <EOL> if batch : <EOL> inputs = self . inputs <EOL> inputs_mask = self . inputs_mask <EOL> groundtruth = self . labels <EOL> groundtruth_mask = self . labels_mask <EOL> else : <EOL> inputs , inputs_mask = self . bottom . single_to_batch_inputs ( <EOL> self . single_inputs ) <EOL> groundtruth = self . single_labels [ : , None ] <EOL> groundtruth_mask = None <EOL> if not prediction : <EOL> prediction = groundtruth <EOL> if not prediction_mask : <EOL> prediction_mask = groundtruth_mask <EOL> cost = self . cost ( inputs_mask = inputs_mask , <EOL> labels = prediction , <EOL> labels_mask = prediction_mask , <EOL> ** inputs ) <EOL> cost_cg = ComputationGraph ( cost ) <EOL> if self . criterion [ '<STR_LIT:name>' ] . startswith ( "<STR_LIT>" ) : <EOL> placeholder , = VariableFilter ( theano_name = '<STR_LIT>' ) ( cost_cg ) <EOL> cost_cg = cost_cg . replace ( { placeholder : groundtruth } ) <EOL> return cost_cg <EOL> def analyze ( self , inputs , groundtruth , prediction = None ) : <EOL> """<STR_LIT>""" <EOL> input_values_dict = dict ( inputs ) <EOL> input_values_dict [ '<STR_LIT>' ] = groundtruth <EOL> if prediction is not None : <EOL> input_values_dict [ '<STR_LIT>' ] = prediction <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> input_variables = list ( self . single_inputs . values ( ) ) <EOL> input_variables . append ( self . single_labels . copy ( name = '<STR_LIT>' ) ) <EOL> prediction_variable = tensor . lvector ( '<STR_LIT>' ) <EOL> if prediction is not None : <EOL> input_variables . append ( prediction_variable ) <EOL> cg = self . get_cost_graph ( <EOL> batch = False , prediction = prediction_variable [ : , None ] ) <EOL> else : <EOL> cg = self . get_cost_graph ( batch = False ) <EOL> cost = cg . outputs [ <NUM_LIT:0> ] <EOL> weights , = VariableFilter ( <EOL> bricks = [ self . generator ] , name = "<STR_LIT>" ) ( cg ) <EOL> energies = VariableFilter ( <EOL> bricks = [ self . generator ] , name = "<STR_LIT>" ) ( cg ) <EOL> energies_output = [ energies [ <NUM_LIT:0> ] [ : , <NUM_LIT:0> , : ] if energies <EOL> else tensor . zeros_like ( weights ) ] <EOL> states , = VariableFilter ( <EOL> applications = [ self . encoder . apply ] , roles = [ OUTPUT ] , <EOL> name = "<STR_LIT>" ) ( cg ) <EOL> ctc_matrix_output = [ ] <EOL> self . _analyze = theano . function ( <EOL> input_variables , <EOL> [ cost [ : , <NUM_LIT:0> ] , weights [ : , <NUM_LIT:0> , : ] ] + energies_output + ctc_matrix_output , <EOL> on_unused_input = '<STR_LIT>' ) <EOL> return self . _analyze ( ** input_values_dict ) <EOL> def init_beam_search ( self , beam_size ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( self , '<STR_LIT>' ) and self . beam_size == beam_size : <EOL> return <EOL> self . beam_size = beam_size <EOL> generated = self . get_generate_graph ( use_mask = False , n_steps = <NUM_LIT:3> ) <EOL> cg = ComputationGraph ( generated . values ( ) ) <EOL> samples , = VariableFilter ( <EOL> applications = [ self . generator . generate ] , name = "<STR_LIT>" ) ( cg ) <EOL> self . _beam_search = BeamSearch ( beam_size , samples ) <EOL> self . _beam_search . compile ( ) <EOL> def beam_search ( self , inputs , ** kwargs ) : <EOL> self . init_beam_search ( self . beam_size ) <EOL> inputs = dict ( inputs ) <EOL> max_length = int ( self . bottom . num_time_steps ( ** inputs ) / <EOL> self . max_decoded_length_scale ) <EOL> search_inputs = { } <EOL> for var in self . inputs . values ( ) : <EOL> search_inputs [ var ] = inputs . pop ( var . name ) [ : , numpy . newaxis , ... ] <EOL> if inputs : <EOL> raise Exception ( <EOL> '<STR_LIT>' . format ( <EOL> inputs . keys ( ) ) ) <EOL> outputs , search_costs = self . _beam_search . search ( <EOL> search_inputs , self . eos_label , <EOL> max_length , <EOL> ignore_first_eol = self . data_prepend_eos , <EOL> ** kwargs ) <EOL> return outputs , search_costs <EOL> def init_generate ( self ) : <EOL> generated = self . get_generate_graph ( use_mask = False ) <EOL> cg = ComputationGraph ( generated [ '<STR_LIT>' ] ) <EOL> self . _do_generate = cg . get_theano_function ( ) <EOL> def sample ( self , inputs , n_steps = None ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . init_generate ( ) <EOL> batch , unused_mask = self . bottom . single_to_batch_inputs ( inputs ) <EOL> batch [ '<STR_LIT>' ] = n_steps if n_steps is not None else int ( self . bottom . num_time_steps ( ** batch ) / <EOL> self . max_decoded_length_scale ) <EOL> return self . _do_generate ( ** batch ) [ <NUM_LIT:0> ] <EOL> def __getstate__ ( self ) : <EOL> state = dict ( self . __dict__ ) <EOL> for attr in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> state . pop ( attr , None ) <EOL> return state <EOL> def __setstate__ ( self , state ) : <EOL> self . __dict__ . update ( state ) <EOL> try : <EOL> emitter = self . generator . readout . emitter <EOL> del emitter . _theano_rng <EOL> except : <EOL> pass </s>
