<s> import os , sys , copy <EOL> sys . path . append ( os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , os . pardir , '<STR_LIT>' ) ) ) <EOL> import collections , random <EOL> import theano <EOL> import theano . tensor as T <EOL> from qnetwork import QNetwork , HiddenLayer , OutputLayer <EOL> import numpy as np <EOL> MAX_FEATURES_TEST = <NUM_LIT:1> <EOL> class MDP : <EOL> def startState ( self ) : raise NotImplementedError ( "<STR_LIT>" ) <EOL> def actions ( self , state ) : raise NotImplementedError ( "<STR_LIT>" ) <EOL> def succAndProbReward ( self , state , action ) : raise NotImplementedError ( "<STR_LIT>" ) <EOL> def discount ( self ) : raise NotImplementedError ( "<STR_LIT>" ) <EOL> def computeStates ( self ) : <EOL> self . states = set ( ) <EOL> queue = [ ] <EOL> self . states . add ( self . startState ( ) ) <EOL> queue . append ( self . startState ( ) ) <EOL> while len ( queue ) > <NUM_LIT:0> : <EOL> state = queue . pop ( ) <EOL> for action in self . actions ( state ) : <EOL> for newState , prob , reward in self . succAndProbReward ( state , action ) : <EOL> if newState not in self . states : <EOL> self . states . add ( newState ) <EOL> queue . append ( newState ) <EOL> class NumberLineMDP ( MDP ) : <EOL> def __init__ ( self , n = <NUM_LIT:5> ) : self . n = n <EOL> def startState ( self ) : return <NUM_LIT:0> <EOL> def isEnd ( self , state ) : return state == self . n <EOL> def actions ( self , state ) : return [ - <NUM_LIT:1> , + <NUM_LIT:1> ] <EOL> def succAndProbReward ( self , state , action ) : <EOL> return [ ( state , <NUM_LIT> , <NUM_LIT:0> ) , <EOL> ( min ( max ( state + action , - self . n ) , + self . n ) , <NUM_LIT> , state ) ] <EOL> def discount ( self ) : return <NUM_LIT> <EOL> class GridSearchMDP ( MDP ) : <EOL> def __init__ ( self , n = <NUM_LIT:10> ) : <EOL> self . n = n <EOL> def startState ( self ) : return ( random . randint ( <NUM_LIT:2> , self . n - <NUM_LIT:4> ) + <NUM_LIT:0.0> , random . randint ( <NUM_LIT:2> , self . n - <NUM_LIT:4> ) + <NUM_LIT:0.0> ) <EOL> def isEnd ( self , state ) : return state [ <NUM_LIT:0> ] == <NUM_LIT:0> or state [ <NUM_LIT:0> ] >= self . n or state [ <NUM_LIT:1> ] == <NUM_LIT:0> or state [ <NUM_LIT:1> ] >= self . n <EOL> def actions ( self , state ) : return [ ( <NUM_LIT:0> , <NUM_LIT:1> ) , ( <NUM_LIT:1> , <NUM_LIT:0> ) , ( <NUM_LIT:0> , - <NUM_LIT:1> ) , ( - <NUM_LIT:1> , <NUM_LIT:0> ) , ( <NUM_LIT:1> , <NUM_LIT:1> ) , ( - <NUM_LIT:1> , - <NUM_LIT:1> ) ] <EOL> def succAndProbReward ( self , state , action ) : <EOL> newState = ( state [ <NUM_LIT:0> ] + action [ <NUM_LIT:0> ] , state [ <NUM_LIT:1> ] + action [ <NUM_LIT:1> ] ) <EOL> nextReward = <NUM_LIT:0> <EOL> if self . isEnd ( newState ) : <EOL> if newState [ <NUM_LIT:0> ] == <NUM_LIT:0> or newState [ <NUM_LIT:1> ] == <NUM_LIT:0> or ( newState [ <NUM_LIT:0> ] == self . n and newState [ <NUM_LIT:1> ] != self . n ) : <EOL> nextReward = - <NUM_LIT:1> <EOL> elif newState [ <NUM_LIT:0> ] == self . n and newState [ <NUM_LIT:1> ] == self . n : <EOL> nextReward = <NUM_LIT:5> <EOL> elif newState [ <NUM_LIT:1> ] == self . n : <EOL> nextReward = <NUM_LIT:1> <EOL> return [ ( state , <NUM_LIT> , <NUM_LIT:0> ) , ( newState , <NUM_LIT> , nextReward ) ] <EOL> def discount ( self ) : return <NUM_LIT> <EOL> def sample ( probs ) : <EOL> target = random . random ( ) <EOL> accum = <NUM_LIT:0> <EOL> for i , prob in enumerate ( probs ) : <EOL> accum += prob <EOL> if accum >= target : return i <EOL> raise Exception ( "<STR_LIT>" % probs ) <EOL> def test_nnet_numberline_mdp ( n_episodes , exploration_prob = <NUM_LIT> , learning_rate = <NUM_LIT> , target_freeze_period = <NUM_LIT> ) : <EOL> reduce_explore = <NUM_LIT> <EOL> size = <NUM_LIT> <EOL> mdp = GridSearchMDP ( size ) <EOL> actions = mdp . actions ( mdp . startState ( ) ) <EOL> print actions <EOL> features = T . dvector ( '<STR_LIT>' ) <EOL> action = T . lscalar ( '<STR_LIT:action>' ) <EOL> reward = T . dscalar ( '<STR_LIT>' ) <EOL> next_features = T . dvector ( '<STR_LIT>' ) <EOL> n_vis = <NUM_LIT:2> <EOL> hidden_layer_1 = HiddenLayer ( n_vis = n_vis , n_hid = len ( actions ) , layer_name = '<STR_LIT>' , activation = '<STR_LIT>' ) <EOL> output_layer = OutputLayer ( layer_name = '<STR_LIT>' , activation = '<STR_LIT:relu>' ) <EOL> layers = [ hidden_layer_1 , output_layer ] <EOL> mlp = QNetwork ( layers , discount = mdp . discount ( ) , learning_rate = learning_rate ) <EOL> loss , updates = mlp . get_loss_and_updates ( features , action , reward , next_features ) <EOL> train_model = theano . function ( <EOL> [ theano . Param ( features , default = np . zeros ( MAX_FEATURES_TEST ) ) , <EOL> theano . Param ( action , default = <NUM_LIT:0> ) , <EOL> theano . Param ( reward , default = <NUM_LIT:0> ) , <EOL> theano . Param ( next_features , default = np . zeros ( MAX_FEATURES_TEST ) ) ] , <EOL> outputs = loss , <EOL> updates = updates , <EOL> mode = '<STR_LIT>' ) <EOL> rewards = [ ] <EOL> counter = <NUM_LIT:0> <EOL> for episode in xrange ( n_episodes ) : <EOL> curDiscount = mdp . discount ( ) <EOL> totalReward = <NUM_LIT:0> <EOL> cur_state = mdp . startState ( ) <EOL> print cur_state <EOL> while not mdp . isEnd ( cur_state ) : <EOL> counter += <NUM_LIT:1> <EOL> if counter % <NUM_LIT:1000> == <NUM_LIT:0> : <EOL> mlp . frozen_layers = copy . deepcopy ( mlp . layers ) <EOL> if ( counter % <NUM_LIT:100> == <NUM_LIT:0> ) : <EOL> print '<STR_LIT>' . format ( cur_state ) <EOL> if random . random ( ) < exploration_prob : <EOL> action = random . choice ( actions ) <EOL> action_index = actions . index ( action ) <EOL> else : <EOL> action_index = T . argmax ( mlp . fprop ( [ cur_state [ <NUM_LIT:0> ] / mdp . n , cur_state [ <NUM_LIT:1> ] / mdp . n ] ) ) . eval ( ) <EOL> action = actions [ action_index ] <EOL> if ( counter % <NUM_LIT:100> == <NUM_LIT:0> ) : <EOL> print '<STR_LIT>' . format ( action ) <EOL> transitions = mdp . succAndProbReward ( cur_state , action ) <EOL> if len ( transitions ) == <NUM_LIT:0> : <EOL> break <EOL> i = sample ( [ prob for newState , prob , reward in transitions ] ) <EOL> newState , prob , reward = transitions [ i ] <EOL> reward *= curDiscount <EOL> totalReward += reward <EOL> curDiscount *= mdp . discount ( ) <EOL> loss = train_model ( [ cur_state [ <NUM_LIT:0> ] / mdp . n , cur_state [ <NUM_LIT:1> ] / mdp . n ] , action_index , reward , [ newState [ <NUM_LIT:0> ] / mdp . n , newState [ <NUM_LIT:1> ] / mdp . n ] ) <EOL> cur_state = newState <EOL> exploration_prob -= reduce_explore <EOL> if ( exploration_prob < <NUM_LIT> ) : <EOL> exploration_prob = <NUM_LIT> <EOL> rewards . append ( totalReward ) <EOL> print ( '<STR_LIT:*>' * <NUM_LIT:30> ) <EOL> print ( '<STR_LIT>' . format ( episode , rewards [ - <NUM_LIT:1> ] ) ) <EOL> print ( '<STR_LIT>' . format ( np . mean ( rewards [ - <NUM_LIT> : ] ) ) ) <EOL> print ( '<STR_LIT>' . format ( exploration_prob ) ) <EOL> print ( '<STR_LIT:*>' * <NUM_LIT:30> ) <EOL> print ( '<STR_LIT:\n>' ) <EOL> return rewards <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_nnet_numberline_mdp ( <NUM_LIT> ) </s>
