<gh_stars>1-10
import numpy as np
import tensorflow as tf

np.random.seed(42)
tf.set_random_seed(42)


def weight_variable(w_name, shape, c_names):
    w = tf.get_variable(w_name, shape=shape, initializer=tf.truncated_normal_initializer(stddev=0.1),
                        collections=c_names)
    return tf.Variable(w)


def bias_variable(b_name, shape, c_names):
    b = tf.get_variable(b_name, shape=shape, initializer=tf.constant_initializer(0), collections=c_names)
    return tf.Variable(b)


def conv2d(x, w):
    return tf.nn.conv2d(x, w, strides=[1, 1, 1, 1], padding='VALID')


class DuelingDQN:
    def __init__(
            self,
            n_actions,
            n_features,
            learning_rate=1e-4,
            reward_decay=0.95,
            e_greedy=0.9,
            replace_target_iter=200,
            memory_size=500,
            batch_size=512,
            e_greedy_increment=None,
            start_epsilon=0,
            output_graph=False,
            sess=None,
    ):
        self.n_actions = n_actions
        self.n_features = n_features
        self.global_step = tf.Variable(0)  # count the number of steps taken.
        self.lr = learning_rate
        self.gamma = reward_decay
        self.epsilon_max = e_greedy
        self.replace_target_iter = replace_target_iter
        self.memory_size = memory_size
        self.batch_size = batch_size
        self.epsilon_increment = e_greedy_increment
        self.epsilon = start_epsilon if e_greedy_increment is not None else self.epsilon_max

        self.learn_step_counter = 0
        self.memory = np.zeros((self.memory_size, n_features*2+2))
        self.memory_index = 0
        self._build_net()
        t_params = tf.get_collection('target_net_params')
        e_params = tf.get_collection('eval_net_params')
        self.replace_target_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]

        if sess is None:
            self.sess = tf.Session()
            self.sess.run(tf.global_variables_initializer())
        else:
            self.sess = sess
        if output_graph:
            tf.summary.FileWriter("logs/", self.sess.graph)
        self.cost_his = []

    def _build_net(self):
        def build_layers(x, c_names):
            # n_l1, n_l2, n_l3, n_l4 = 64, 64, 128, 128

            def make_conv_layer(k, i, _id):
                w = weight_variable('w'+_id, k, c_names)
                b = bias_variable('b'+_id, [k[-1]], c_names)
                o = tf.nn.tanh(conv2d(i, w) + b)
                return o

            # first conv layer.
            conv_o1 = make_conv_layer([1, 1, 1, 64], x, "1")

            # second conv two-layer.
            conv_o11 = make_conv_layer([4, 1, 64, 128], conv_o1, "11")
            conv_o12 = make_conv_layer([1, 4, 64, 128], conv_o1, "12")

            # third flatten layer.
            flat = tf.concat([tf.reshape(conv_o11, [-1, 4 * 128]),
                              tf.reshape(conv_o12, [-1, 4 * 128])], axis=1)

            # fourth full-connected layer.
            with tf.variable_scope('l4_fc'):
                w4 = weight_variable('w4', [4*128*2, 256], c_names)
                b4 = bias_variable('b4', [256], c_names)
                l4 = tf.nn.tanh(tf.matmul(flat, w4) + b4)

            # fifth layer. Dueling DQN
            with tf.variable_scope('Value'):
                w5 = weight_variable('w5', [256, 1], c_names)
                b5 = bias_variable('b5', [1], c_names)
                self.V = tf.matmul(l4, w5) + b5

            with tf.variable_scope('Advantage'):
                w5 = weight_variable('w5', [256, self.n_actions], c_names)
                b5 = bias_variable('b5', [self.n_actions], c_names)
                self.A = tf.matmul(l4, w5) + b5

            with tf.variable_scope('Q'):
                out = self.V + (self.A - tf.reduce_mean(self.A, axis=1, keep_dims=True))

            return out

        # ------------------ build evaluate_net ------------------
        self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s')  # input
        self.q_target = tf.placeholder(tf.float32, [None, self.n_actions], name='Q_target')  # for calculating loss
        with tf.variable_scope('eval_net'):
            c_names = ['eval_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
            x = tf.reshape(self.s, [-1, 4, 4, 1])
            self.q_eval = build_layers(x, c_names)

        with tf.variable_scope('loss'):
            self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval))
        with tf.variable_scope('train'):
            self._train_op = tf.train.AdamOptimizer(self.lr).minimize(self.loss)

        # ------------------ build target_net ------------------
        self.s_ = tf.placeholder(tf.float32, [None, self.n_features], name='s_')    # input
        with tf.variable_scope('target_net'):
            c_names_ = ['target_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
            x_ = tf.reshape(self.s_, [-1, 4, 4, 1])
            self.q_next = build_layers(x_, c_names_)

    def store_transition(self, s, a, r, s_):
        transition = np.hstack((s, [a, r], s_))
        self.memory[self.memory_index, :] = transition
        self.memory_index += 1
        if self.memory_index == self.memory_size:
            self.memory_index = 0

    def choose_action(self, observation):
        observation = observation[np.newaxis, :]
        if np.random.uniform() < self.epsilon:  # choosing action
            actions_value = self.sess.run(self.q_eval, feed_dict={self.s: observation})
            action = np.argmax(actions_value)
        else:
            action = np.random.randint(0, self.n_actions)
        return action


    def learn(self):
        if self.learn_step_counter % self.replace_target_iter == 0:
            self.sess.run(self.replace_target_op)

        sample_index = np.random.choice(self.memory_size, size=self.batch_size)
        batch_memory = self.memory[sample_index, :]

        q_next = self.sess.run(self.q_next, feed_dict={self.s_: batch_memory[:, -self.n_features:]})  # next observation
        q_eval = self.sess.run(self.q_eval, {self.s: batch_memory[:, :self.n_features]})

        q_target = q_eval.copy()

        batch_index = np.arange(self.batch_size, dtype=np.int32)
        eval_act_index = batch_memory[:, self.n_features].astype(int)
        reward = batch_memory[:, self.n_features + 1]

        q_target[batch_index, eval_act_index] = reward + self.gamma * np.max(q_next, axis=1)

        _, self.cost = self.sess.run([self._train_op, self.loss],
                                     feed_dict={self.s: batch_memory[:, :self.n_features],
                                                self.q_target: q_target})
        self.cost_his.append(self.cost)

        self.epsilon = self.epsilon + self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max
        self.learn_step_counter += 1

import numpy as np
import tensorflow as tf

np.random.seed(42)
tf.set_random_seed(42)


def weight_variable(w_name, shape, c_names):
    w = tf.get_variable(w_name, shape=shape, initializer=tf.truncated_normal_initializer(stddev=0.1),
                        collections=c_names)
    return tf.Variable(w)


def bias_variable(b_name, shape, c_names):
    b = tf.get_variable(b_name, shape=shape, initializer=tf.constant_initializer(0), collections=c_names)
    return tf.Variable(b)


def conv2d(x, w):
    return tf.nn.conv2d(x, w, strides=[1, 1, 1, 1], padding='VALID')


class DuelingDQN:
    def __init__(
            self,
            n_actions,
            n_features,
            learning_rate=1e-4,
            reward_decay=0.95,
            e_greedy=0.9,
            replace_target_iter=200,
            memory_size=500,
            batch_size=512,
            e_greedy_increment=None,
            start_epsilon=0,
            output_graph=False,
            sess=None,
    ):
        self.n_actions = n_actions
        self.n_features = n_features
        self.global_step = tf.Variable(0)  # count the number of steps taken.
        self.lr = learning_rate
        self.gamma = reward_decay
        self.epsilon_max = e_greedy
        self.replace_target_iter = replace_target_iter
        self.memory_size = memory_size
        self.batch_size = batch_size
        self.epsilon_increment = e_greedy_increment
        self.epsilon = start_epsilon if e_greedy_increment is not None else self.epsilon_max

        self.learn_step_counter = 0
        self.memory = np.zeros((self.memory_size, n_features*2+2))
        self.memory_index = 0
        self._build_net()
        t_params = tf.get_collection('target_net_params')
        e_params = tf.get_collection('eval_net_params')
        self.replace_target_op = [tf.assign(t, e) for t, e in zip(t_params, e_params)]

        if sess is None:
            self.sess = tf.Session()
            self.sess.run(tf.global_variables_initializer())
        else:
            self.sess = sess
        if output_graph:
            tf.summary.FileWriter("logs/", self.sess.graph)
        self.cost_his = []

    def _build_net(self):
        def build_layers(x, c_names):
            # n_l1, n_l2, n_l3, n_l4 = 64, 64, 128, 128

            def make_conv_layer(k, i, _id):
                w = weight_variable('w'+_id, k, c_names)
                b = bias_variable('b'+_id, [k[-1]], c_names)
                o = tf.nn.tanh(conv2d(i, w) + b)
                return o

            # first conv layer.
            conv_o1 = make_conv_layer([1, 1, 1, 64], x, "1")

            # second conv two-layer.
            conv_o11 = make_conv_layer([4, 1, 64, 128], conv_o1, "11")
            conv_o12 = make_conv_layer([1, 4, 64, 128], conv_o1, "12")

            # third flatten layer.
            flat = tf.concat([tf.reshape(conv_o11, [-1, 4 * 128]),
                              tf.reshape(conv_o12, [-1, 4 * 128])], axis=1)

            # fourth full-connected layer.
            with tf.variable_scope('l4_fc'):
                w4 = weight_variable('w4', [4*128*2, 256], c_names)
                b4 = bias_variable('b4', [256], c_names)
                l4 = tf.nn.tanh(tf.matmul(flat, w4) + b4)

            # fifth layer. Dueling DQN
            with tf.variable_scope('Value'):
                w5 = weight_variable('w5', [256, 1], c_names)
                b5 = bias_variable('b5', [1], c_names)
                self.V = tf.matmul(l4, w5) + b5

            with tf.variable_scope('Advantage'):
                w5 = weight_variable('w5', [256, self.n_actions], c_names)
                b5 = bias_variable('b5', [self.n_actions], c_names)
                self.A = tf.matmul(l4, w5) + b5

            with tf.variable_scope('Q'):
                out = self.V + (self.A - tf.reduce_mean(self.A, axis=1, keep_dims=True))

            return out

        # ------------------ build evaluate_net ------------------
        self.s = tf.placeholder(tf.float32, [None, self.n_features], name='s')  # input
        self.q_target = tf.placeholder(tf.float32, [None, self.n_actions], name='Q_target')  # for calculating loss
        with tf.variable_scope('eval_net'):
            c_names = ['eval_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
            x = tf.reshape(self.s, [-1, 4, 4, 1])
            self.q_eval = build_layers(x, c_names)

        with tf.variable_scope('loss'):
            self.loss = tf.reduce_mean(tf.squared_difference(self.q_target, self.q_eval))
        with tf.variable_scope('train'):
            self._train_op = tf.train.AdamOptimizer(self.lr).minimize(self.loss)

        # ------------------ build target_net ------------------
        self.s_ = tf.placeholder(tf.float32, [None, self.n_features], name='s_')    # input
        with tf.variable_scope('target_net'):
            c_names_ = ['target_net_params', tf.GraphKeys.GLOBAL_VARIABLES]
            x_ = tf.reshape(self.s_, [-1, 4, 4, 1])
            self.q_next = build_layers(x_, c_names_)

    def store_transition(self, s, a, r, s_):
        transition = np.hstack((s, [a, r], s_))
        self.memory[self.memory_index, :] = transition
        self.memory_index += 1
        if self.memory_index == self.memory_size:
            self.memory_index = 0

    def choose_action(self, observation):
        observation = observation[np.newaxis, :]
        if np.random.uniform() < self.epsilon:  # choosing action
            actions_value = self.sess.run(self.q_eval, feed_dict={self.s: observation})
            action = np.argmax(actions_value)
        else:
            action = np.random.randint(0, self.n_actions)
        return action


    def learn(self):
        if self.learn_step_counter % self.replace_target_iter == 0:
            self.sess.run(self.replace_target_op)

        sample_index = np.random.choice(self.memory_size, size=self.batch_size)
        batch_memory = self.memory[sample_index, :]

        q_next = self.sess.run(self.q_next, feed_dict={self.s_: batch_memory[:, -self.n_features:]})  # next observation
        q_eval = self.sess.run(self.q_eval, {self.s: batch_memory[:, :self.n_features]})

        q_target = q_eval.copy()

        batch_index = np.arange(self.batch_size, dtype=np.int32)
        eval_act_index = batch_memory[:, self.n_features].astype(int)
        reward = batch_memory[:, self.n_features + 1]

        q_target[batch_index, eval_act_index] = reward + self.gamma * np.max(q_next, axis=1)

        _, self.cost = self.sess.run([self._train_op, self.loss],
                                     feed_dict={self.s: batch_memory[:, :self.n_features],
                                                self.q_target: q_target})
        self.cost_his.append(self.cost)

        self.epsilon = self.epsilon + self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max
        self.learn_step_counter += 1
# count the number of steps taken. # n_l1, n_l2, n_l3, n_l4 = 64, 64, 128, 128 # first conv layer. # second conv two-layer. # third flatten layer. # fourth full-connected layer. # fifth layer. Dueling DQN # ------------------ build evaluate_net ------------------ # input # for calculating loss # ------------------ build target_net ------------------ # input # choosing action # next observation