<s> import theano <EOL> import numpy <EOL> import scipy <EOL> from theano import tensor <EOL> from blocks . bricks import Initializable , Linear <EOL> from blocks . bricks . parallel import Parallel <EOL> from blocks . bricks . base import lazy , application <EOL> from blocks . bricks . attention import ( <EOL> GenericSequenceAttention , SequenceContentAttention , <EOL> ShallowEnergyComputer ) <EOL> from blocks . utils import ( put_hook , ipdb_breakpoint , shared_floatx , <EOL> shared_floatx_nans ) <EOL> from lvsr . expressions import conv1d <EOL> floatX = theano . config . floatX <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Conv1D ( Initializable ) : <EOL> def __init__ ( self , num_filters , filter_length , ** kwargs ) : <EOL> self . num_filters = num_filters <EOL> self . filter_length = filter_length <EOL> super ( Conv1D , self ) . __init__ ( ** kwargs ) <EOL> def _allocate ( self ) : <EOL> self . parameters = [ shared_floatx_nans ( ( self . num_filters , self . filter_length ) , <EOL> name = "<STR_LIT>" ) ] <EOL> def _initialize ( self ) : <EOL> self . weights_init . initialize ( self . parameters [ <NUM_LIT:0> ] , self . rng ) <EOL> def apply ( self , input_ ) : <EOL> return conv1d ( input_ , self . parameters [ <NUM_LIT:0> ] , border_mode = "<STR_LIT>" ) <EOL> class SequenceContentAndConvAttention ( GenericSequenceAttention , Initializable ) : <EOL> @ lazy ( ) <EOL> def __init__ ( self , match_dim , conv_n , conv_num_filters = <NUM_LIT:1> , <EOL> state_transformer = None , <EOL> attended_transformer = None , energy_computer = None , <EOL> prior = None , energy_normalizer = None , ** kwargs ) : <EOL> super ( SequenceContentAndConvAttention , self ) . __init__ ( ** kwargs ) <EOL> if not state_transformer : <EOL> state_transformer = Linear ( use_bias = False ) <EOL> self . match_dim = match_dim <EOL> self . state_transformer = state_transformer <EOL> self . state_transformers = Parallel ( input_names = self . state_names , <EOL> prototype = state_transformer , <EOL> name = "<STR_LIT>" ) <EOL> if not attended_transformer : <EOL> attended_transformer = Linear ( name = "<STR_LIT>" ) <EOL> if not energy_normalizer : <EOL> energy_normalizer = '<STR_LIT>' <EOL> self . energy_normalizer = energy_normalizer <EOL> if not energy_computer : <EOL> energy_computer = ShallowEnergyComputer ( <EOL> name = "<STR_LIT>" , <EOL> use_bias = self . energy_normalizer != '<STR_LIT>' ) <EOL> self . filter_handler = Linear ( name = "<STR_LIT>" , use_bias = False ) <EOL> self . attended_transformer = attended_transformer <EOL> self . energy_computer = energy_computer <EOL> if not prior : <EOL> prior = dict ( type = '<STR_LIT>' , initial_begin = <NUM_LIT:0> , initial_end = <NUM_LIT> , <EOL> min_speed = <NUM_LIT:0> , max_speed = <NUM_LIT:0> ) <EOL> self . prior = prior <EOL> self . conv_n = conv_n <EOL> self . conv_num_filters = conv_num_filters <EOL> self . conv = Conv1D ( conv_num_filters , <NUM_LIT:2> * conv_n + <NUM_LIT:1> ) <EOL> self . children = [ self . state_transformers , self . attended_transformer , <EOL> self . energy_computer , self . filter_handler , self . conv ] <EOL> def _push_allocation_config ( self ) : <EOL> self . state_transformers . input_dims = self . state_dims <EOL> self . state_transformers . output_dims = [ self . match_dim <EOL> for name in self . state_names ] <EOL> self . attended_transformer . input_dim = self . attended_dim <EOL> self . attended_transformer . output_dim = self . match_dim <EOL> self . energy_computer . input_dim = self . match_dim <EOL> self . energy_computer . output_dim = <NUM_LIT:1> <EOL> self . filter_handler . input_dim = self . conv_num_filters <EOL> self . filter_handler . output_dim = self . match_dim <EOL> @ application <EOL> def compute_energies ( self , attended , preprocessed_attended , <EOL> previous_weights , states ) : <EOL> if not preprocessed_attended : <EOL> preprocessed_attended = self . preprocess ( attended ) <EOL> transformed_states = self . state_transformers . apply ( as_dict = True , <EOL> ** states ) <EOL> match_vectors = sum ( transformed_states . values ( ) , <EOL> preprocessed_attended ) <EOL> conv_result = self . conv . apply ( previous_weights ) <EOL> match_vectors += self . filter_handler . apply ( <EOL> conv_result [ : , : , self . conv_n : - self . conv_n ] <EOL> . dimshuffle ( <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) ) . dimshuffle ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> energies = self . energy_computer . apply ( match_vectors ) . reshape ( <EOL> match_vectors . shape [ : - <NUM_LIT:1> ] , ndim = match_vectors . ndim - <NUM_LIT:1> ) <EOL> return energies <EOL> @ staticmethod <EOL> def mask_row ( offset , length , empty_row ) : <EOL> return tensor . set_subtensor ( empty_row [ offset : offset + length ] , <NUM_LIT:1> ) <EOL> @ application ( outputs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def take_glimpses ( self , attended , preprocessed_attended = None , <EOL> attended_mask = None , weights = None , step = None , ** states ) : <EOL> p = self . prior <EOL> length = attended . shape [ <NUM_LIT:0> ] <EOL> prior_type = p . get ( '<STR_LIT:type>' , '<STR_LIT>' ) <EOL> if prior_type == '<STR_LIT>' : <EOL> begin = p [ '<STR_LIT>' ] + step [ <NUM_LIT:0> ] * p [ '<STR_LIT>' ] <EOL> end = p [ '<STR_LIT>' ] + step [ <NUM_LIT:0> ] * p [ '<STR_LIT>' ] <EOL> begin = tensor . maximum ( <NUM_LIT:0> , tensor . minimum ( length - <NUM_LIT:1> , begin ) ) <EOL> end = tensor . maximum ( <NUM_LIT:0> , tensor . minimum ( length , end ) ) <EOL> additional_mask = None <EOL> elif prior_type . startswith ( '<STR_LIT>' ) : <EOL> if prior_type == '<STR_LIT>' : <EOL> position_in_attended = tensor . arange ( length , dtype = floatX ) [ None , : ] <EOL> expected_last_source_pos = ( weights * position_in_attended ) . sum ( axis = <NUM_LIT:1> ) <EOL> elif prior_type == '<STR_LIT>' : <EOL> ali_to_05 = tensor . extra_ops . cumsum ( weights , axis = <NUM_LIT:1> ) - <NUM_LIT:0.5> <EOL> ali_to_05 = ( ali_to_05 >= <NUM_LIT:0> ) <EOL> ali_median_pos = ali_to_05 [ : , <NUM_LIT:1> : ] - ali_to_05 [ : , : - <NUM_LIT:1> ] <EOL> expected_last_source_pos = tensor . argmax ( ali_median_pos , axis = <NUM_LIT:1> ) <EOL> expected_last_source_pos = theano . gradient . disconnected_grad ( <EOL> expected_last_source_pos ) <EOL> else : <EOL> raise ValueError <EOL> begins = tensor . floor ( expected_last_source_pos - p [ '<STR_LIT>' ] ) <EOL> ends = tensor . ceil ( expected_last_source_pos + p [ '<STR_LIT>' ] ) <EOL> begin = tensor . maximum ( <NUM_LIT:0> , begins . min ( ) ) . astype ( '<STR_LIT>' ) <EOL> end = tensor . minimum ( length , ends . max ( ) ) . astype ( '<STR_LIT>' ) <EOL> position_in_attended_cut = tensor . arange ( <EOL> begin * <NUM_LIT:1.> , end * <NUM_LIT:1.> , <NUM_LIT:1.> , dtype = floatX ) [ None , : ] <EOL> additional_mask = ( ( position_in_attended_cut > begins [ : , None ] ) * <EOL> ( position_in_attended_cut < ends [ : , None ] ) ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" , prior_type ) <EOL> begin = tensor . floor ( begin ) . astype ( '<STR_LIT>' ) <EOL> end = tensor . ceil ( end ) . astype ( '<STR_LIT>' ) <EOL> attended_cut = attended [ begin : end ] <EOL> preprocessed_attended_cut = ( preprocessed_attended [ begin : end ] <EOL> if preprocessed_attended else None ) <EOL> attended_mask_cut = ( <EOL> ( attended_mask [ begin : end ] if attended_mask else None ) <EOL> * ( additional_mask . T if additional_mask else <NUM_LIT:1> ) ) <EOL> weights_cut = weights [ : , begin : end ] <EOL> energies_cut = self . compute_energies ( attended_cut , preprocessed_attended_cut , <EOL> weights_cut , states ) <EOL> weights_cut = self . compute_weights ( energies_cut , attended_mask_cut ) <EOL> weighted_averages = self . compute_weighted_averages ( weights_cut , attended_cut ) <EOL> new_weights = new_energies = tensor . zeros_like ( weights . T ) <EOL> new_weights = tensor . set_subtensor ( new_weights [ begin : end ] , <EOL> weights_cut ) <EOL> new_energies = tensor . set_subtensor ( new_energies [ begin : end ] , <EOL> energies_cut ) <EOL> return weighted_averages , new_weights . T , new_energies . T , step + <NUM_LIT:1> <EOL> @ take_glimpses . property ( '<STR_LIT>' ) <EOL> def take_glimpses_inputs ( self ) : <EOL> return ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] + <EOL> self . state_names ) <EOL> @ application <EOL> def compute_weights ( self , energies , attended_mask ) : <EOL> if self . energy_normalizer == '<STR_LIT>' : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> energies = energies - energies . max ( axis = <NUM_LIT:0> ) <EOL> unnormalized_weights = tensor . exp ( energies ) <EOL> elif self . energy_normalizer == '<STR_LIT>' : <EOL> logger . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> unnormalized_weights = tensor . nnet . sigmoid ( energies ) <EOL> elif self . energy_normalizer == '<STR_LIT:relu>' : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> unnormalized_weights = tensor . maximum ( energies / <NUM_LIT> , <NUM_LIT:0.0> ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" <EOL> . format ( self . energy_computer ) ) <EOL> if attended_mask : <EOL> unnormalized_weights *= attended_mask <EOL> normalization = ( unnormalized_weights . sum ( axis = <NUM_LIT:0> ) + <EOL> tensor . all ( <NUM_LIT:1> - attended_mask , axis = <NUM_LIT:0> ) ) <EOL> return unnormalized_weights / normalization <EOL> @ application <EOL> def initial_glimpses ( self , batch_size , attended ) : <EOL> return ( [ tensor . zeros ( ( batch_size , self . attended_dim ) ) ] <EOL> + <NUM_LIT:2> * [ tensor . concatenate ( [ <EOL> tensor . ones ( ( batch_size , <NUM_LIT:1> ) ) , <EOL> tensor . zeros ( ( batch_size , attended . shape [ <NUM_LIT:0> ] - <NUM_LIT:1> ) ) ] , <EOL> axis = <NUM_LIT:1> ) ] <EOL> + [ tensor . zeros ( ( batch_size , ) , dtype = '<STR_LIT>' ) ] ) <EOL> @ initial_glimpses . property ( '<STR_LIT>' ) <EOL> def initial_glimpses_outputs ( self ) : <EOL> return [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> @ application ( inputs = [ '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def preprocess ( self , attended ) : <EOL> return self . attended_transformer . apply ( attended ) <EOL> def get_dim ( self , name ) : <EOL> if name in [ '<STR_LIT>' ] : <EOL> return self . attended_dim <EOL> if name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return <NUM_LIT:0> <EOL> return super ( SequenceContentAndConvAttention , self ) . get_dim ( name ) </s>
