<s> """<STR_LIT>""" <EOL> from abc import ABCMeta , abstractmethod <EOL> from theano import tensor <EOL> from six import add_metaclass <EOL> from blocks . bricks import ( Brick , Initializable , Sequence , <EOL> Feedforward , Linear , Tanh ) <EOL> from blocks . bricks . base import lazy , application <EOL> from blocks . bricks . parallel import Parallel , Distribute <EOL> from blocks . bricks . recurrent import recurrent , BaseRecurrent <EOL> from blocks . utils import dict_union , dict_subset , pack <EOL> class AbstractAttention ( Brick ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def __init__ ( self , state_names , state_dims , attended_dim , ** kwargs ) : <EOL> self . state_names = state_names <EOL> self . state_dims = state_dims <EOL> self . attended_dim = attended_dim <EOL> super ( AbstractAttention , self ) . __init__ ( ** kwargs ) <EOL> @ application ( inputs = [ '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def preprocess ( self , attended ) : <EOL> """<STR_LIT>""" <EOL> return attended <EOL> @ abstractmethod <EOL> def take_glimpses ( self , attended , preprocessed_attended = None , <EOL> attended_mask = None , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def initial_glimpses ( self , batch_size , attended ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def get_dim ( self , name ) : <EOL> if name in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return self . attended_dim <EOL> if name in [ '<STR_LIT>' ] : <EOL> return <NUM_LIT:0> <EOL> return super ( AbstractAttention , self ) . get_dim ( name ) <EOL> class GenericSequenceAttention ( AbstractAttention ) : <EOL> """<STR_LIT>""" <EOL> @ application <EOL> def compute_weights ( self , energies , attended_mask ) : <EOL> """<STR_LIT>""" <EOL> energies = energies - energies . max ( axis = <NUM_LIT:0> ) <EOL> unnormalized_weights = tensor . exp ( energies ) <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 compute_weighted_averages ( self , weights , attended ) : <EOL> """<STR_LIT>""" <EOL> return ( tensor . shape_padright ( weights ) * attended ) . sum ( axis = <NUM_LIT:0> ) <EOL> class SequenceContentAttention ( GenericSequenceAttention , Initializable ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' ] ) <EOL> def __init__ ( self , match_dim , state_transformer = None , <EOL> attended_transformer = None , energy_computer = None , ** 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 = kwargs [ '<STR_LIT>' ] , <EOL> prototype = state_transformer , <EOL> name = "<STR_LIT>" ) <EOL> if not attended_transformer : <EOL> attended_transformer = Linear ( name = "<STR_LIT>" ) <EOL> if not energy_computer : <EOL> energy_computer = ShallowEnergyComputer ( name = "<STR_LIT>" ) <EOL> self . attended_transformer = attended_transformer <EOL> self . energy_computer = energy_computer <EOL> children = [ self . state_transformers , attended_transformer , <EOL> energy_computer ] + kwargs . get ( '<STR_LIT>' , [ ] ) <EOL> super ( SequenceContentAttention , self ) . __init__ ( children = children , <EOL> ** kwargs ) <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> @ application <EOL> def compute_energies ( self , attended , preprocessed_attended , 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> 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> @ application ( outputs = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def take_glimpses ( self , attended , preprocessed_attended = None , <EOL> attended_mask = None , ** states ) : <EOL> r"""<STR_LIT>""" <EOL> energies = self . compute_energies ( attended , preprocessed_attended , <EOL> states ) <EOL> weights = self . compute_weights ( energies , attended_mask ) <EOL> weighted_averages = self . compute_weighted_averages ( weights , attended ) <EOL> return weighted_averages , weights . T <EOL> @ take_glimpses . property ( '<STR_LIT>' ) <EOL> def take_glimpses_inputs ( self ) : <EOL> return ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] + <EOL> self . state_names ) <EOL> @ application ( outputs = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def initial_glimpses ( self , batch_size , attended ) : <EOL> return [ tensor . zeros ( ( batch_size , self . attended_dim ) ) , <EOL> tensor . zeros ( ( batch_size , attended . shape [ <NUM_LIT:0> ] ) ) ] <EOL> @ application ( inputs = [ '<STR_LIT>' ] , outputs = [ '<STR_LIT>' ] ) <EOL> def preprocess ( self , attended ) : <EOL> """<STR_LIT>""" <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>' ] : <EOL> return <NUM_LIT:0> <EOL> return super ( SequenceContentAttention , self ) . get_dim ( name ) <EOL> class ShallowEnergyComputer ( Sequence , Initializable , Feedforward ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( ) <EOL> def __init__ ( self , use_bias = False , ** kwargs ) : <EOL> super ( ShallowEnergyComputer , self ) . __init__ ( <EOL> [ Tanh ( ) . apply , Linear ( use_bias = use_bias ) . apply ] , ** kwargs ) <EOL> @ property <EOL> def input_dim ( self ) : <EOL> return self . children [ <NUM_LIT:1> ] . input_dim <EOL> @ input_dim . setter <EOL> def input_dim ( self , value ) : <EOL> self . children [ <NUM_LIT:1> ] . input_dim = value <EOL> @ property <EOL> def output_dim ( self ) : <EOL> return self . children [ <NUM_LIT:1> ] . output_dim <EOL> @ output_dim . setter <EOL> def output_dim ( self , value ) : <EOL> self . children [ <NUM_LIT:1> ] . output_dim = value <EOL> @ add_metaclass ( ABCMeta ) <EOL> class AbstractAttentionRecurrent ( BaseRecurrent ) : <EOL> """<STR_LIT>""" <EOL> @ abstractmethod <EOL> def apply ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def take_glimpses ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def compute_states ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class AttentionRecurrent ( AbstractAttentionRecurrent , Initializable ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , transition , attention , distribute = None , <EOL> add_contexts = True , <EOL> attended_name = None , attended_mask_name = None , <EOL> ** kwargs ) : <EOL> self . _sequence_names = list ( transition . apply . sequences ) <EOL> self . _state_names = list ( transition . apply . states ) <EOL> self . _context_names = list ( transition . apply . contexts ) <EOL> if add_contexts : <EOL> if not attended_name : <EOL> attended_name = '<STR_LIT>' <EOL> if not attended_mask_name : <EOL> attended_mask_name = '<STR_LIT>' <EOL> self . _context_names += [ attended_name , attended_mask_name ] <EOL> else : <EOL> attended_name = self . _context_names [ <NUM_LIT:0> ] <EOL> attended_mask_name = self . _context_names [ <NUM_LIT:1> ] <EOL> if not distribute : <EOL> normal_inputs = [ name for name in self . _sequence_names <EOL> if '<STR_LIT>' not in name ] <EOL> distribute = Distribute ( normal_inputs , <EOL> attention . take_glimpses . outputs [ <NUM_LIT:0> ] ) <EOL> self . transition = transition <EOL> self . attention = attention <EOL> self . distribute = distribute <EOL> self . add_contexts = add_contexts <EOL> self . attended_name = attended_name <EOL> self . attended_mask_name = attended_mask_name <EOL> self . preprocessed_attended_name = "<STR_LIT>" + self . attended_name <EOL> self . _glimpse_names = self . attention . take_glimpses . outputs <EOL> self . previous_glimpses_needed = [ <EOL> name for name in self . _glimpse_names <EOL> if name in self . attention . take_glimpses . inputs ] <EOL> children = [ self . transition , self . attention , self . distribute ] <EOL> children += kwargs . get ( '<STR_LIT>' , [ ] ) <EOL> super ( AttentionRecurrent , self ) . __init__ ( children = children , ** kwargs ) <EOL> def _push_allocation_config ( self ) : <EOL> self . attention . state_dims = self . transition . get_dims ( <EOL> self . attention . state_names ) <EOL> self . attention . attended_dim = self . get_dim ( self . attended_name ) <EOL> self . distribute . source_dim = self . attention . get_dim ( <EOL> self . distribute . source_name ) <EOL> self . distribute . target_dims = self . transition . get_dims ( <EOL> self . distribute . target_names ) <EOL> @ application <EOL> def take_glimpses ( self , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> states = dict_subset ( kwargs , self . _state_names , pop = True ) <EOL> glimpses = dict_subset ( kwargs , self . _glimpse_names , pop = True ) <EOL> glimpses_needed = dict_subset ( glimpses , self . previous_glimpses_needed ) <EOL> result = self . attention . take_glimpses ( <EOL> kwargs . pop ( self . attended_name ) , <EOL> kwargs . pop ( self . preprocessed_attended_name , None ) , <EOL> kwargs . pop ( self . attended_mask_name , None ) , <EOL> ** dict_union ( states , glimpses_needed ) ) <EOL> return result <EOL> @ take_glimpses . property ( '<STR_LIT>' ) <EOL> def take_glimpses_outputs ( self ) : <EOL> return self . _glimpse_names <EOL> @ application <EOL> def compute_states ( self , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> normal_inputs = [ name for name in self . _sequence_names <EOL> if '<STR_LIT>' not in name ] <EOL> sequences = dict_subset ( kwargs , normal_inputs , pop = True ) <EOL> glimpses = dict_subset ( kwargs , self . _glimpse_names , pop = True ) <EOL> if self . add_contexts : <EOL> kwargs . pop ( self . attended_name ) <EOL> kwargs . pop ( self . attended_mask_name , None ) <EOL> sequences . update ( self . distribute . apply ( <EOL> as_dict = True , ** dict_subset ( dict_union ( sequences , glimpses ) , <EOL> self . distribute . apply . inputs ) ) ) <EOL> current_states = self . transition . apply ( <EOL> iterate = False , as_list = True , <EOL> ** dict_union ( sequences , kwargs ) ) <EOL> return current_states <EOL> @ compute_states . property ( '<STR_LIT>' ) <EOL> def compute_states_outputs ( self ) : <EOL> return self . _state_names <EOL> @ recurrent <EOL> def do_apply ( self , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> attended = kwargs [ self . attended_name ] <EOL> preprocessed_attended = kwargs . pop ( self . preprocessed_attended_name ) <EOL> attended_mask = kwargs . get ( self . attended_mask_name ) <EOL> sequences = dict_subset ( kwargs , self . _sequence_names , pop = True , <EOL> must_have = False ) <EOL> states = dict_subset ( kwargs , self . _state_names , pop = True ) <EOL> glimpses = dict_subset ( kwargs , self . _glimpse_names , pop = True ) <EOL> current_glimpses = self . take_glimpses ( <EOL> as_dict = True , <EOL> ** dict_union ( <EOL> states , glimpses , <EOL> { self . attended_name : attended , <EOL> self . attended_mask_name : attended_mask , <EOL> self . preprocessed_attended_name : preprocessed_attended } ) ) <EOL> current_states = self . compute_states ( <EOL> as_list = True , <EOL> ** dict_union ( sequences , states , current_glimpses , kwargs ) ) <EOL> return current_states + list ( current_glimpses . values ( ) ) <EOL> @ do_apply . property ( '<STR_LIT>' ) <EOL> def do_apply_sequences ( self ) : <EOL> return self . _sequence_names <EOL> @ do_apply . property ( '<STR_LIT>' ) <EOL> def do_apply_contexts ( self ) : <EOL> return self . _context_names + [ self . preprocessed_attended_name ] <EOL> @ do_apply . property ( '<STR_LIT>' ) <EOL> def do_apply_states ( self ) : <EOL> return self . _state_names + self . _glimpse_names <EOL> @ do_apply . property ( '<STR_LIT>' ) <EOL> def do_apply_outputs ( self ) : <EOL> return self . _state_names + self . _glimpse_names <EOL> @ application <EOL> def apply ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> preprocessed_attended = self . attention . preprocess ( <EOL> kwargs [ self . attended_name ] ) <EOL> return self . do_apply ( <EOL> ** dict_union ( kwargs , <EOL> { self . preprocessed_attended_name : <EOL> preprocessed_attended } ) ) <EOL> @ apply . delegate <EOL> def apply_delegate ( self ) : <EOL> return self . do_apply . __get__ ( self , None ) <EOL> @ apply . property ( '<STR_LIT>' ) <EOL> def apply_contexts ( self ) : <EOL> return self . _context_names <EOL> @ application <EOL> def initial_states ( self , batch_size , ** kwargs ) : <EOL> return ( pack ( self . transition . initial_states ( <EOL> batch_size , ** kwargs ) ) + <EOL> pack ( self . attention . initial_glimpses ( <EOL> batch_size , kwargs [ self . attended_name ] ) ) ) <EOL> @ initial_states . property ( '<STR_LIT>' ) <EOL> def initial_states_outputs ( self ) : <EOL> return self . do_apply . states <EOL> def get_dim ( self , name ) : <EOL> if name in self . _glimpse_names : <EOL> return self . attention . get_dim ( name ) <EOL> if name == self . preprocessed_attended_name : <EOL> ( original_name , ) = self . attention . preprocess . outputs <EOL> return self . attention . get_dim ( original_name ) <EOL> if self . add_contexts : <EOL> if name == self . attended_name : <EOL> return self . attention . get_dim ( <EOL> self . attention . take_glimpses . inputs [ <NUM_LIT:0> ] ) <EOL> if name == self . attended_mask_name : <EOL> return <NUM_LIT:0> <EOL> return self . transition . get_dim ( name ) </s>
