<s> """<STR_LIT>""" <EOL> from abc import ABCMeta , abstractmethod <EOL> from collections import OrderedDict <EOL> from six import add_metaclass <EOL> from theano import tensor <EOL> from blocks . bricks import Initializable , Random , Bias , NDimensionalSoftmax <EOL> from blocks . bricks . base import application , Brick , lazy <EOL> from blocks . bricks . parallel import Fork , Merge <EOL> from blocks . bricks . lookup import LookupTable <EOL> from blocks . bricks . recurrent import recurrent <EOL> from blocks . bricks . attention import ( <EOL> AbstractAttentionRecurrent , AttentionRecurrent ) <EOL> from blocks . roles import add_role , COST <EOL> from blocks . utils import dict_union , dict_subset <EOL> class BaseSequenceGenerator ( Initializable ) : <EOL> r"""<STR_LIT>""" <EOL> @ lazy ( ) <EOL> def __init__ ( self , readout , transition , fork , language_model = None , <EOL> ** kwargs ) : <EOL> super ( BaseSequenceGenerator , self ) . __init__ ( ** kwargs ) <EOL> self . readout = readout <EOL> self . transition = transition <EOL> self . fork = fork <EOL> self . language_model = language_model <EOL> self . children = [ self . readout , self . fork , self . transition ] <EOL> if self . language_model : <EOL> self . children . append ( self . language_model ) <EOL> @ property <EOL> def _state_names ( self ) : <EOL> return self . transition . compute_states . outputs <EOL> @ property <EOL> def _context_names ( self ) : <EOL> return self . transition . apply . contexts <EOL> @ property <EOL> def _glimpse_names ( self ) : <EOL> return self . transition . take_glimpses . outputs <EOL> @ property <EOL> def _lm_state_names ( self ) : <EOL> return ( [ '<STR_LIT>' + name for name in self . language_model . _state_names ] <EOL> if self . language_model <EOL> else [ ] ) <EOL> def _push_allocation_config ( self ) : <EOL> self . transition . push_allocation_config ( ) <EOL> transition_sources = ( self . _state_names + self . _context_names + <EOL> self . _glimpse_names ) <EOL> self . readout . source_dims = [ ] <EOL> for name in self . readout . source_names : <EOL> if name in transition_sources : <EOL> self . readout . source_dims . append ( self . transition . get_dim ( name ) ) <EOL> elif self . language_model and name in self . _lm_state_names : <EOL> self . readout . source_dims . append ( <EOL> self . get_dim ( name ) ) <EOL> else : <EOL> self . readout . get_dim ( name ) <EOL> self . readout . push_allocation_config ( ) <EOL> feedback_name , = self . readout . feedback . outputs <EOL> self . fork . input_dim = self . readout . get_dim ( feedback_name ) <EOL> self . fork . output_dims = self . transition . get_dims ( <EOL> self . fork . apply . outputs ) <EOL> @ application <EOL> def cost ( self , application_call , outputs , mask = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> costs = self . cost_matrix ( outputs , mask = mask , ** kwargs ) <EOL> cost = tensor . mean ( costs . sum ( axis = <NUM_LIT:0> ) ) <EOL> add_role ( cost , COST ) <EOL> application_call . add_auxiliary_variable ( <EOL> ( costs . sum ( ) / mask . sum ( ) ) if mask is not None else costs . mean ( ) , <EOL> name = '<STR_LIT>' ) <EOL> return cost <EOL> @ application <EOL> def evaluate ( self , application_call , outputs , mask = None , ** kwargs ) : <EOL> batch_size = outputs . shape [ <NUM_LIT:1> ] <EOL> states = dict_subset ( kwargs , self . _state_names , must_have = False ) <EOL> contexts = dict_subset ( kwargs , self . _context_names , must_have = False ) <EOL> feedback = self . readout . feedback ( outputs ) <EOL> inputs = self . fork . apply ( feedback , as_dict = True ) <EOL> results = self . transition . apply ( <EOL> mask = mask , return_initial_states = True , as_dict = True , <EOL> ** dict_union ( inputs , states , contexts ) ) <EOL> states = OrderedDict ( ( name , results [ name ] [ : - <NUM_LIT:1> ] ) for name in self . _state_names ) <EOL> glimpses = OrderedDict ( ( name , results [ name ] [ <NUM_LIT:1> : ] ) for name in self . _glimpse_names ) <EOL> feedback = tensor . roll ( feedback , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> feedback = tensor . set_subtensor ( <EOL> feedback [ <NUM_LIT:0> ] , <EOL> self . readout . feedback ( self . readout . initial_outputs ( batch_size ) ) ) <EOL> if self . language_model : <EOL> lm_states = self . language_model . evaluate ( <EOL> outputs = outputs , mask = mask , as_dict = True ) <EOL> lm_states = { '<STR_LIT>' + name : value for name , value <EOL> in lm_states . items ( ) } <EOL> else : <EOL> lm_states = { } <EOL> readouts = self . readout . readout ( <EOL> feedback = feedback , <EOL> ** dict_union ( lm_states , states , glimpses , contexts ) ) <EOL> costs = self . readout . cost ( readouts , outputs ) <EOL> if mask is not None : <EOL> costs *= mask <EOL> for name , variable in list ( glimpses . items ( ) ) + list ( states . items ( ) ) : <EOL> application_call . add_auxiliary_variable ( <EOL> variable . copy ( ) , name = name ) <EOL> for name in self . _state_names + self . _glimpse_names : <EOL> application_call . add_auxiliary_variable ( <EOL> results [ name ] [ - <NUM_LIT:1> ] . copy ( ) , name = name + "<STR_LIT>" ) <EOL> return [ costs ] + states . values ( ) + glimpses . values ( ) <EOL> @ evaluate . property ( '<STR_LIT>' ) <EOL> def evaluate_outputs ( self ) : <EOL> return [ '<STR_LIT>' ] + self . _state_names + self . _glimpse_names <EOL> @ application <EOL> def cost_matrix ( self , outputs , mask = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . evaluate ( outputs , mask = mask , ** kwargs ) [ <NUM_LIT:0> ] <EOL> @ recurrent <EOL> def generate ( self , outputs , dont_generate_new_outputs = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> states = dict_subset ( kwargs , self . _state_names ) <EOL> contexts = dict_subset ( kwargs , self . _context_names , must_have = False ) <EOL> glimpses = dict_subset ( kwargs , self . _glimpse_names ) <EOL> lm_states = dict_subset ( kwargs , self . _lm_state_names ) <EOL> next_glimpses = self . transition . take_glimpses ( <EOL> as_dict = True , <EOL> ** dict_union ( states , glimpses , contexts ) ) <EOL> next_readouts = self . readout . readout ( <EOL> feedback = self . readout . feedback ( outputs ) , <EOL> ** dict_union ( states , next_glimpses , contexts , lm_states ) ) <EOL> next_outputs = ( self . readout . emit ( next_readouts ) <EOL> if not dont_generate_new_outputs <EOL> else outputs ) <EOL> next_costs = self . readout . cost ( next_readouts , next_outputs ) <EOL> next_feedback = self . readout . feedback ( next_outputs ) <EOL> next_inputs = ( self . fork . apply ( next_feedback , as_dict = True ) <EOL> if self . fork else { '<STR_LIT>' : next_feedback } ) <EOL> next_states = self . transition . compute_states ( <EOL> as_list = True , <EOL> ** dict_union ( next_inputs , states , next_glimpses , contexts ) ) <EOL> next_lm_states = { } <EOL> if self . language_model : <EOL> unmangled_lm_states = { name [ <NUM_LIT:3> : ] : lm_states [ name ] <EOL> for name in lm_states } <EOL> next_lm_states = OrderedDict ( zip ( <EOL> self . _lm_state_names , self . language_model . generate ( <EOL> next_outputs , dont_generate_new_outputs = True , iterate = False , <EOL> ** unmangled_lm_states ) ) ) <EOL> return ( next_states + [ next_outputs ] + <EOL> list ( next_glimpses . values ( ) ) + list ( next_lm_states . values ( ) ) + <EOL> [ next_costs ] ) <EOL> @ generate . delegate <EOL> def generate_delegate ( self ) : <EOL> return self . transition . apply <EOL> @ generate . property ( '<STR_LIT>' ) <EOL> def generate_states ( self ) : <EOL> result = self . _state_names + [ '<STR_LIT>' ] + self . _glimpse_names <EOL> if self . language_model : <EOL> result . extend ( self . _lm_state_names ) <EOL> return result <EOL> @ generate . property ( '<STR_LIT>' ) <EOL> def generate_outputs ( self ) : <EOL> result = self . _state_names + [ '<STR_LIT>' ] + self . _glimpse_names <EOL> if self . language_model : <EOL> result . extend ( self . _lm_state_names ) <EOL> result . append ( '<STR_LIT>' ) <EOL> return result <EOL> def get_dim ( self , name ) : <EOL> if name in ( self . _state_names + self . _context_names + <EOL> self . _glimpse_names ) : <EOL> return self . transition . get_dim ( name ) <EOL> if name == '<STR_LIT>' : <EOL> return self . readout . get_dim ( name ) <EOL> if self . language_model and name in self . _lm_state_names : <EOL> return self . language_model . get_dim ( name [ <NUM_LIT:3> : ] ) <EOL> return super ( BaseSequenceGenerator , self ) . get_dim ( name ) <EOL> @ application <EOL> def initial_states ( self , batch_size , * args , ** kwargs ) : <EOL> state_dict = dict ( <EOL> self . transition . initial_states ( <EOL> batch_size , as_dict = True , * args , ** kwargs ) , <EOL> outputs = self . readout . initial_outputs ( batch_size ) ) <EOL> if self . language_model : <EOL> lm_initial_states = self . language_model . initial_states ( <EOL> batch_size , as_dict = True , * args , ** kwargs ) <EOL> state_dict = dict_union ( state_dict , <EOL> { "<STR_LIT>" + name : state for name , state <EOL> in lm_initial_states . items ( ) } ) <EOL> return [ state_dict [ state_name ] <EOL> for state_name in self . generate . states ] <EOL> @ initial_states . property ( '<STR_LIT>' ) <EOL> def initial_states_outputs ( self ) : <EOL> return self . generate . states <EOL> @ add_metaclass ( ABCMeta ) <EOL> class AbstractReadout ( Initializable ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def __init__ ( self , source_names , readout_dim , ** kwargs ) : <EOL> self . source_names = source_names <EOL> self . readout_dim = readout_dim <EOL> super ( AbstractReadout , self ) . __init__ ( ** kwargs ) <EOL> @ abstractmethod <EOL> def emit ( self , readouts ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def cost ( self , readouts , outputs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def costs ( self , readouts ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def initial_outputs ( self , batch_size ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def readout ( self , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def feedback ( self , outputs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Readout ( AbstractReadout ) : <EOL> r"""<STR_LIT>""" <EOL> def __init__ ( self , emitter = None , feedback_brick = None , <EOL> merge = None , merge_prototype = None , <EOL> post_merge = None , merged_dim = None , ** kwargs ) : <EOL> super ( Readout , self ) . __init__ ( ** kwargs ) <EOL> if not emitter : <EOL> emitter = TrivialEmitter ( self . readout_dim ) <EOL> if not feedback_brick : <EOL> feedback_brick = TrivialFeedback ( self . readout_dim ) <EOL> if not merge : <EOL> merge = Merge ( input_names = self . source_names , <EOL> prototype = merge_prototype ) <EOL> if not post_merge : <EOL> post_merge = Bias ( dim = self . readout_dim ) <EOL> if not merged_dim : <EOL> merged_dim = self . readout_dim <EOL> self . emitter = emitter <EOL> self . feedback_brick = feedback_brick <EOL> self . merge = merge <EOL> self . post_merge = post_merge <EOL> self . merged_dim = merged_dim <EOL> self . children = [ self . emitter , self . feedback_brick , <EOL> self . merge , self . post_merge ] <EOL> def _push_allocation_config ( self ) : <EOL> self . emitter . readout_dim = self . get_dim ( '<STR_LIT>' ) <EOL> self . feedback_brick . output_dim = self . get_dim ( '<STR_LIT>' ) <EOL> self . merge . input_names = self . source_names <EOL> self . merge . input_dims = self . source_dims <EOL> self . merge . output_dim = self . merged_dim <EOL> self . post_merge . input_dim = self . merged_dim <EOL> self . post_merge . output_dim = self . readout_dim <EOL> @ application <EOL> def readout ( self , ** kwargs ) : <EOL> merged = self . merge . apply ( ** { name : kwargs [ name ] <EOL> for name in self . merge . input_names } ) <EOL> merged = self . post_merge . apply ( merged ) <EOL> return merged <EOL> @ application <EOL> def emit ( self , readouts ) : <EOL> return self . emitter . emit ( readouts ) <EOL> @ application <EOL> def cost ( self , readouts , outputs ) : <EOL> return self . emitter . cost ( readouts , outputs ) <EOL> @ application <EOL> def costs ( self , readouts ) : <EOL> return self . emitter . costs ( readouts ) <EOL> @ application <EOL> def initial_outputs ( self , batch_size ) : <EOL> return self . emitter . initial_outputs ( batch_size ) <EOL> @ application ( outputs = [ '<STR_LIT>' ] ) <EOL> def feedback ( self , outputs ) : <EOL> return self . feedback_brick . feedback ( outputs ) <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . emitter . get_dim ( name ) <EOL> elif name == '<STR_LIT>' : <EOL> return self . feedback_brick . get_dim ( name ) <EOL> elif name == '<STR_LIT>' : <EOL> return self . readout_dim <EOL> return super ( Readout , self ) . get_dim ( name ) <EOL> @ add_metaclass ( ABCMeta ) <EOL> class AbstractEmitter ( Brick ) : <EOL> """<STR_LIT>""" <EOL> @ abstractmethod <EOL> def emit ( self , readouts ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def cost ( self , readouts , outputs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ abstractmethod <EOL> def initial_outputs ( self , batch_size ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ add_metaclass ( ABCMeta ) <EOL> class AbstractFeedback ( Brick ) : <EOL> """<STR_LIT>""" <EOL> @ abstractmethod <EOL> def feedback ( self , outputs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class TrivialEmitter ( AbstractEmitter ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' ] ) <EOL> def __init__ ( self , readout_dim , ** kwargs ) : <EOL> super ( TrivialEmitter , self ) . __init__ ( ** kwargs ) <EOL> self . readout_dim = readout_dim <EOL> @ application <EOL> def emit ( self , readouts ) : <EOL> return readouts <EOL> @ application <EOL> def cost ( self , readouts , outputs ) : <EOL> return tensor . zeros_like ( outputs ) <EOL> @ application <EOL> def initial_outputs ( self , batch_size ) : <EOL> return tensor . zeros ( ( batch_size , self . readout_dim ) ) <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . readout_dim <EOL> return super ( TrivialEmitter , self ) . get_dim ( name ) <EOL> class SoftmaxEmitter ( AbstractEmitter , Initializable , Random ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , initial_output = <NUM_LIT:0> , ** kwargs ) : <EOL> super ( SoftmaxEmitter , self ) . __init__ ( ** kwargs ) <EOL> self . initial_output = initial_output <EOL> self . softmax = NDimensionalSoftmax ( ) <EOL> self . children = [ self . softmax ] <EOL> @ application <EOL> def probs ( self , readouts ) : <EOL> return self . softmax . apply ( readouts , extra_ndim = readouts . ndim - <NUM_LIT:2> ) <EOL> @ application <EOL> def emit ( self , readouts ) : <EOL> probs = self . probs ( readouts ) <EOL> batch_size = probs . shape [ <NUM_LIT:0> ] <EOL> pvals_flat = probs . reshape ( ( batch_size , - <NUM_LIT:1> ) ) <EOL> generated = self . theano_rng . multinomial ( pvals = pvals_flat ) <EOL> return generated . reshape ( probs . shape ) . argmax ( axis = - <NUM_LIT:1> ) <EOL> @ application <EOL> def cost ( self , readouts , outputs ) : <EOL> return self . softmax . categorical_cross_entropy ( <EOL> outputs , readouts , extra_ndim = readouts . ndim - <NUM_LIT:2> ) <EOL> @ application <EOL> def costs ( self , readouts ) : <EOL> return - self . softmax . log_probabilities ( <EOL> readouts , extra_ndim = readouts . ndim - <NUM_LIT:2> ) <EOL> @ application <EOL> def initial_outputs ( self , batch_size ) : <EOL> return self . initial_output * tensor . ones ( ( batch_size , ) , dtype = '<STR_LIT>' ) <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return <NUM_LIT:0> <EOL> return super ( SoftmaxEmitter , self ) . get_dim ( name ) <EOL> class TrivialFeedback ( AbstractFeedback ) : <EOL> """<STR_LIT>""" <EOL> @ lazy ( allocation = [ '<STR_LIT>' ] ) <EOL> def __init__ ( self , output_dim , ** kwargs ) : <EOL> super ( TrivialFeedback , self ) . __init__ ( ** kwargs ) <EOL> self . output_dim = output_dim <EOL> @ application ( outputs = [ '<STR_LIT>' ] ) <EOL> def feedback ( self , outputs ) : <EOL> return outputs <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . output_dim <EOL> return super ( TrivialFeedback , self ) . get_dim ( name ) <EOL> class LookupFeedback ( AbstractFeedback , Initializable ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , num_outputs = None , feedback_dim = None , ** kwargs ) : <EOL> super ( LookupFeedback , self ) . __init__ ( ** kwargs ) <EOL> self . num_outputs = num_outputs <EOL> self . feedback_dim = feedback_dim <EOL> self . lookup = LookupTable ( num_outputs , feedback_dim , <EOL> weights_init = self . weights_init ) <EOL> self . children = [ self . lookup ] <EOL> def _push_allocation_config ( self ) : <EOL> self . lookup . length = self . num_outputs <EOL> self . lookup . dim = self . feedback_dim <EOL> @ application <EOL> def feedback ( self , outputs ) : <EOL> assert self . output_dim == <NUM_LIT:0> <EOL> return self . lookup . apply ( outputs ) <EOL> def get_dim ( self , name ) : <EOL> if name == '<STR_LIT>' : <EOL> return self . feedback_dim <EOL> return super ( LookupFeedback , self ) . get_dim ( name ) <EOL> class FakeAttentionRecurrent ( AbstractAttentionRecurrent , Initializable ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , transition , ** kwargs ) : <EOL> super ( FakeAttentionRecurrent , self ) . __init__ ( ** kwargs ) <EOL> self . transition = transition <EOL> self . state_names = transition . apply . states <EOL> self . context_names = transition . apply . contexts <EOL> self . glimpse_names = [ ] <EOL> self . children = [ self . transition ] <EOL> @ application <EOL> def apply ( self , * args , ** kwargs ) : <EOL> return self . transition . apply ( * args , ** kwargs ) <EOL> @ apply . delegate <EOL> def apply_delegate ( self ) : <EOL> return self . transition . apply <EOL> @ application <EOL> def compute_states ( self , * args , ** kwargs ) : <EOL> return self . transition . apply ( iterate = False , * args , ** kwargs ) <EOL> @ compute_states . delegate <EOL> def compute_states_delegate ( self ) : <EOL> return self . transition . apply <EOL> @ application ( outputs = [ ] ) <EOL> def take_glimpses ( self , * args , ** kwargs ) : <EOL> return None <EOL> @ application <EOL> def initial_states ( self , batch_size , * args , ** kwargs ) : <EOL> return self . transition . initial_states ( batch_size , <EOL> * args , ** kwargs ) <EOL> @ initial_states . property ( '<STR_LIT>' ) <EOL> def initial_states_outputs ( self ) : <EOL> return self . transition . apply . states <EOL> def get_dim ( self , name ) : <EOL> return self . transition . get_dim ( name ) <EOL> class SequenceGenerator ( BaseSequenceGenerator ) : <EOL> r"""<STR_LIT>""" <EOL> def __init__ ( self , readout , transition , attention = None , <EOL> add_contexts = True , ** kwargs ) : <EOL> normal_inputs = [ name for name in transition . apply . sequences <EOL> if '<STR_LIT>' not in name ] <EOL> kwargs . setdefault ( '<STR_LIT>' , Fork ( normal_inputs ) ) <EOL> if attention : <EOL> transition = AttentionRecurrent ( <EOL> transition , attention , <EOL> add_contexts = add_contexts , name = "<STR_LIT>" ) <EOL> else : <EOL> transition = FakeAttentionRecurrent ( transition , <EOL> name = "<STR_LIT>" ) <EOL> super ( SequenceGenerator , self ) . __init__ ( <EOL> readout , transition , ** kwargs ) </s>
