<repo_name>GalDude33/SnakeRL<file_sep>/policies/policy_ConvDQN.py
from keras import Sequential
from keras.layers import Flatten, Dense, Conv2D, BatchNormalization

from policies.policy_BaseDQN import BaseDQN, SQUARE, NO, FLIP, SPATIAL, FULL


class ConvDQN(BaseDQN):
    """
    A Deep Q-Learning implementation with a Convolutional Network as the Q estimator
    """

    def _additional_args(self, policy_args):
        self.flatten = NO

        self.state_radius = 6
        self.step_forward = True
        self.state_rep = SQUARE
        self.doubleDQN = True
        self.gamma = 0.75

        # for self
        self.batch_norm = False
        return policy_args

    def _build_model(self):
        model = Sequential()
        model.add(Conv2D(4, 1, strides=1,
                         activation='relu',
                         input_shape=self.input_shape))
        model.add(Conv2D(16, 4, strides=2,
                         activation='relu', padding='same'))
        if self.batch_norm:
            model.add(BatchNormalization())
        model.add(Conv2D(32, 4, strides=2,
                         activation='relu', padding='same'))
        if self.batch_norm:
           model.add(BatchNormalization())
        model.add(Conv2D(64, 3, strides=1,
                         activation='relu'))
        if self.batch_norm:
            model.add(BatchNormalization())
        model.add(Flatten())
        model.add(Dense(128, activation='relu'))
        if self.batch_norm:
            model.add(BatchNormalization())
        model.add(Dense(3, activation=None))
        return model<file_sep>/policies/policy_RadarDQN.py
from keras import Sequential
from keras.engine import InputLayer
from keras.layers import Flatten, Dense, Conv2D, Dropout, LeakyReLU

from policies.policy_BaseDQN import BaseDQN, DIAMOND, SQUARE, RADAR, NO, FLIP, SPATIAL, FULL, NUM_PER_TYPE


class RadarDQN(BaseDQN):
    """
    Our attempt at implementing a network that uses only Radar State representation
    """

    def _additional_args(self, policy_args):
        #self.state_radius = 4
        #self.step_forward = True
        self.state_rep = RADAR

        self.doubleDQN = True

        # for self
        self.dropout_rate = 0.0
        self.activation = LeakyReLU
        return policy_args

    def _build_model(self):
        model = Sequential()

        model.add(InputLayer(self.input_shape))
        model.add(Conv2D(filters=64, kernel_size=1, data_format='channels_first'))
        model.add(self.activation())
        model.add(Conv2D(filters=64, kernel_size=1, data_format='channels_first'))
        model.add(self.activation())
        model.add(Conv2D(filters=16, kernel_size=1, data_format='channels_first'))
        model.add(self.activation())

        model.add(Flatten())

        model.add(Dense(units=256))
        model.add(self.activation())

        if self.dropout_rate > 0:
            model.add(Dropout(self.dropout_rate))

        model.add(Dense(units=256))
        model.add(self.activation())

        if self.dropout_rate > 0:
            model.add(Dropout(self.dropout_rate))

        model.add(Dense(units=128))
        model.add(self.activation())

        model.add(Dense(units=3))
        return model
<file_sep>/utils/ReplayBuffer.py
from collections import deque
import numpy as np


class ReplayBuffer:
    """
    Self-maintaining limited length replay-buffer, with option to append records and sample batches
    """

    def __init__(self, maxlen, prefer_new=True, new_range=300, new_rate=0.40):
        self.buffer = deque(maxlen=maxlen)

        # "prefer new" parameters
        self.prefer_new = prefer_new
        self.new_range = new_range
        self.new_rate = new_rate

    def __len__(self):
        return self.buffer.__len__()

    def record(self, prev, action, reward, new):
        self.buffer.append((prev, action, reward, new))

    def sample(self, batch_size):
        if self.buffer.__len__() > batch_size:
            if self.prefer_new:
                probs = np.ones(self.buffer.__len__())
                probs[-self.new_range:] *= (self.new_rate * self.buffer.__len__())/self.new_range
                probs /= probs.sum()
            else:
                probs = None
            batch = np.array(self.buffer)[np.random.choice(self.buffer.__len__(), batch_size, p=probs)]
        else:
            batch = np.array(list(self.buffer))

        prev = np.stack(batch[:, 0])
        actions = np.vstack(batch[:, 1]).squeeze()
        rewards = np.vstack(batch[:, 2]).squeeze()
        new = np.stack(batch[:, 3])
        return prev, actions, rewards, new
<file_sep>/policies/policy_ConvDuelDQN.py
from keras import Input, Model
from keras.layers import Flatten, Dense, Conv2D, Lambda, Add, BatchNormalization
import keras.backend as K

from policies.policy_BaseDQN import BaseDQN, SQUARE, NO, FLIP, SPATIAL, FULL
from policies.policy_PriorityBaseDQN import PriorBaseDQN


class ConvDuelDQN(PriorBaseDQN):
    """
    A Dueling Deep Q-Learning implementation with a Convolutional Network as the Q estimator
    """

    def _additional_args(self, policy_args):
        self.flatten = NO

        self.epsilon_decay = 0.98
        self.min_epsilon = 0.1

        self.state_radius = 6
        self.step_forward = True
        self.state_rep = SQUARE
        self.doubleDQN = True
        self.gamma = 0.75

        # for self
        self.batch_norm = False
        return policy_args

    def _build_model(self):
        inputs = Input(self.input_shape)
        net = Conv2D(4, 1, strides=1, activation='relu')(inputs)
        net = Conv2D(32, 4, strides=2, activation='relu', padding='same')(net)
        if self.batch_norm:
            net = BatchNormalization()(net)
        net = Conv2D(45, 4, strides=2, activation='relu')(net)
        if self.batch_norm:
            net = BatchNormalization()(net)
        net = Flatten()(net)

        advt = Dense(128, activation='relu')(net)
        if self.batch_norm:
            advt = BatchNormalization()(advt)
        Advantage = Dense(3)(advt)

        value = Dense(128, activation='relu')(net)
        if self.batch_norm:
            value = BatchNormalization()(value)
        Value = Dense(1)(value)

        advt = Lambda(lambda advt: advt - K.mean(advt, axis=-1, keepdims=True))(Advantage)
        value = Lambda(lambda value: K.tile(value, [1, 3]))(Value)
        Q_out = Add()([value, advt])

        return Model(inputs=[inputs], outputs=[Q_out])
<file_sep>/nn_utils/losses.py
import tensorflow as tf


def huber_loss(y_true, y_pred):
    """huber loss - supposed to improve stability"""
    return tf.losses.huber_loss(y_true, y_pred)
<file_sep>/policies/policy_RadarPlusDQN.py
from keras import Input, Model
from keras.layers import Conv1D, Flatten, Dense, Dropout, LeakyReLU, Concatenate

from policies.policy_RadarBaseDQN import RadarBaseDQN, DIAMOND, SQUARE, RADAR, NO, FLIP, SPATIAL, FULL,\
    NUM_PER_TYPE, RADAR_PLUS


class RadarPlusDQN(RadarBaseDQN):
    """
    Our attempt at implementing a more advanced network that uses Radar and Square states together
    """

    def _additional_args(self, policy_args):
        self.flatten = SPATIAL
        self.state_radius = 6
        self.step_forward = True
        self.state_rep = RADAR_PLUS

        self.doubleDQN = True
        self.batch_size = 96

        self.epsilon_decay = 0.98
        self.min_epsilon = 0.2

        # for self
        self.dropout_rate = 0.0
        self.activation = LeakyReLU
        return policy_args

    def _build_model(self):
        input_layer = Input(self.input_shape[0])
        net = self.activation()(Conv1D(filters=4, kernel_size=1)(input_layer))
        net = Flatten()(net)
        net = self.dense_layer(net, units=128)

        input_layer_radar = Input(self.input_shape[1])
        net_radar = Flatten(input_shape=self.input_shape[1])(input_layer_radar)
        net_radar = self.dense_layer(net_radar, units=64)

        net = Concatenate()([net, net_radar])
        net = self.dense_layer(net, units=128)

        Q_out = self.dense_layer(net, units=3)

        return Model(inputs=[input_layer, input_layer_radar], outputs=[Q_out])

    def dense_layer(self, input_layer, units):
        net = Dense(units=units)(input_layer)
        net = self.activation()(net)  # apply activation
        if self.dropout_rate > 0:
            net = Dropout(self.dropout_rate)(net)
        return net
<file_sep>/policies/policy_linear.py
from keras import Sequential
from keras.layers import Dense, Flatten

from policies.policy_BaseDQN import BaseDQN, SQUARE, DIAMOND, NO, FLIP, SPATIAL, FULL, NUM_PER_TYPE


class LinearQL(BaseDQN):
    """
    A Deep Q-Learning implementation with a simple linear network as the Q estimator
    """

    def _additional_args(self, policy_args):
        self.flatten = FULL
        self.state_radius = 4
        self.step_forward = True
        self.state_rep = SQUARE
        self.batch_size = 96
        self.save_model_round = 200
        policy_args['min_epsilon'] = 0.3
        policy_args['epsilon'] = 1.0
        policy_args['gamma'] = 0.5
        return policy_args

    def _build_model(self):
        model = Sequential()
        model.add(Dense(input_shape=self.input_shape, units=3, activation=None, use_bias=True))
        return model
<file_sep>/state_utils/State.py
import numpy as np
from keras.utils import to_categorical
from policies import base_policy as bp

MAP_VALUES = np.arange(-1, 9 + 1)
NUM_VALUES = len(MAP_VALUES)

NO = 0
FLIP = 1
ALL = 2

SPATIAL = 1
FULL = 2

NUM_PER_TYPE = 2


def normalize(matrix, direction, normalize):
    """
    Normalize direction that snake is facing, either to North only for FULL or to North/East for FLIP, by
    rotating the board/matrix
    """
    if normalize == NO:
        return matrix
    rot = 0  # direction is already 'N'
    if direction == 'E':
        rot = 1 + (1 if normalize == FLIP else 0)
    elif direction == 'S':
        rot = 2
    elif direction == 'W':
        rot = (0 if normalize == FLIP else 3)
    return np.rot90(matrix, rot)


def augment_after_normaliztion(prev_state_repr, prev_state_dir, prev_action, reward, new_state_repr, new_state_dir,
                               radius, normalize=FLIP):
    """ Augment states to visit more states by exploiting symmetries, even though we didn't actually visit them"""
    def flip_state(s, dir):
        orig_shape = s.shape
        if normalize == ALL:
            flip_ax = 1
        else:
            if dir in {'E', 'W'}:
                flip_ax = 0
            else:  # if dir in {'N', 'S'}
                flip_ax = 1
        return np.flip(s.reshape((2*radius+1, 2*radius+1, NUM_VALUES)), axis=flip_ax).reshape(orig_shape)

    def flip_action(a):
        if bp.Policy.ACTIONS[a] == 'R':
            return bp.Policy.ACTIONS.index('L')
        elif bp.Policy.ACTIONS[a] == 'L':
            return bp.Policy.ACTIONS.index('R')
        return a

    return flip_state(prev_state_repr, prev_state_dir), \
           prev_action, \
           reward, \
           flip_state(new_state_repr, new_state_dir)


class SquareAroundHeadState:
    """
    Normalized square around head, l_inf radius
    """

    def __init__(self, radius, step_forward=True, flatten=FULL, normalize=FLIP):
        self.radius = radius
        self.step_forward = step_forward
        self.flatten = flatten
        self.normalize = normalize

    def get_area(self):
        return (self.radius * 2 + 1) ** 2

    def get_shape(self):
        if self.flatten == FULL:
            return (self.get_area() * NUM_VALUES,)
        elif self.flatten == SPATIAL:
            return (self.get_area(), NUM_VALUES)
        else:  # no flattening
            diameter = 1 + 2 * self.radius
            return (diameter, diameter, NUM_VALUES)

    def get_state_repr(self, state):
        board, head = state
        head_pos, direction = head

        if self.step_forward:
            head_pos = head_pos.move(bp.Policy.TURNS[direction]['F'])

        x, y = head_pos

        # slicing along axes
        board = np.take(board, range(x - self.radius, x + self.radius + 1), axis=0, mode='wrap')
        output_state = np.take(board, range(y - self.radius, y + self.radius + 1), axis=1, mode='wrap')

        # normalize
        output_state = normalize(output_state, direction, self.normalize)

        output_state = to_categorical(output_state + 1, num_classes=NUM_VALUES)

        if self.flatten == FULL:
            output_state = output_state.reshape(-1)
        elif self.flatten == SPATIAL:
            output_state = output_state.reshape(-1, NUM_VALUES)

        return output_state


# '''
# l2 radius - euclidean distance
# '''
# class CircleAroundHeadState:
#
#     def __init__(self, radius, step_forward=True):
#         self.radius = radius
#         self.step_forward = step_forward
#
#     def get_state_repr(self, state):
#         board, head = state
#         head_pos, direction = head
#
#         n_x, n_y = board.shape
#
#         if self.step_forward:
#             head_pos = head_pos.move(bp.Policy.TURNS[direction]['F'])
#         x, y = head_pos
#
#         def get_dists(size, ind):
#             all_inds = np.arange(size)
#             return np.min([np.abs((all_inds - size) - ind),
#                            np.abs((all_inds + size) - ind),
#                            np.abs(all_inds - ind)],
#                           axis=0)
#
#         rows = get_dists(n_x, x)
#         cols = get_dists(n_y, y)
#         mask = (rows[:, None] ** 2 + cols ** 2 <= self.radius ** 2).astype(bool)
#         return to_categorical(board[mask] + 1, num_classes=NUM_VALUES).reshape(-1)


class DiamondAroundHeadState:
    """
    Normalized diamond around head, l_1 radius
    """

    def __init__(self, radius, step_forward=True, flatten=FULL, normalize=FLIP):
        self.radius = radius
        self.step_forward = step_forward
        self.flatten = flatten
        self.square_state = SquareAroundHeadState(radius=radius, step_forward=step_forward, flatten=NO,
                                                  normalize=normalize)

    def get_area(self):
        return round((self.radius * 2 + 1) ** 2 // 2 + 1)

    def get_shape(self):
        if self.flatten == FULL:
            return (self.get_area() * NUM_VALUES,)
        elif self.flatten == SPATIAL:
            return (self.get_area(), NUM_VALUES)

    def get_state_repr(self, state):
        board = self.square_state.get_state_repr(state)

        output_state = []
        for d in range(-self.radius, self.radius + 1):
            col_radius = np.abs(self.radius - np.abs(d))
            output_state += [board[d + self.radius][(self.radius) - col_radius: (self.radius + 1) + col_radius]]

        output_state = np.vstack(output_state).squeeze()

        if self.flatten == FULL:
            output_state = output_state.reshape(-1)
        elif self.flatten == SPATIAL:
            output_state = output_state.reshape(-1, NUM_VALUES)

        return output_state


def get_dists(size, ind):
    return np.abs(get_directions(size, ind))


def get_directions(size, ind):
    all_inds = np.arange(size)
    all_dirs = np.array([all_inds - size - ind, all_inds + size - ind, (all_inds - ind)])
    dirs = all_dirs[np.argmin(np.abs(all_dirs), axis=0), range(size)]
    return dirs


class RadarState:
    """
    Sort of a radar around the snake's head. Returns directions and distances to num_per_type objects of each type
    """
    def __init__(self, num_per_type, polars=True, normalize=FLIP):
        self.num_per_type = num_per_type
        self.polars = polars

        self.normalize = normalize

    def get_shape(self):
        return [NUM_VALUES, self.num_per_type, 3]

    def get_state_repr(self, state):
        board, head = state
        head_pos, direction = head

        board = normalize(board, direction, self.normalize)

        x, y = head_pos

        n_x, n_y = board.shape
        rows_dists = get_dists(n_x, x)
        cols_dists = get_dists(n_y, y)
        dists = (rows_dists[:, None] + cols_dists)

        rows_dirs = get_directions(n_x, x)
        cols_dirs = get_directions(n_y, y)

        output_state = np.ones(self.get_shape())
        for i, v in enumerate(MAP_VALUES):
            inds = np.where(board == v)
            min_inds = dists[inds].argsort()[:self.num_per_type]
            min_inds_x = inds[0][min_inds]
            min_inds_y = inds[1][min_inds]

            curr_dirs = np.stack([rows_dirs[min_inds_x], cols_dirs[min_inds_y]], axis=1) / max(n_x, n_y)
            curr_dirs = np.hstack([curr_dirs, np.linalg.norm(curr_dirs, axis=1)[..., np.newaxis]])

            # if self.polars:
            #     r = np.linalg.norm(curr_dirs, axis=1)
            #     t = np.arctan2(curr_dirs[..., 0], curr_dirs[..., 1])
            #     curr_dirs[..., 0] = r
            #     curr_dirs[..., 1] = t
            output_state[i, range(len(min_inds))] = curr_dirs

        return output_state


class DoubleStateWrapper:
    """
    State wrapper that encapsulates a combination of Square and Radar state representations
    """
    def __init__(self, state_square: SquareAroundHeadState, state_radar : RadarState):
        self.state_radar = state_radar
        self.state_square = state_square

    def get_shape(self):
        return (self.state_square.get_shape(), self.state_radar.get_shape())

    def get_state_repr(self, state):
        return [self.state_square.get_state_repr(state), self.state_radar.get_state_repr(state)]
<file_sep>/policies/policy_avoid.py
from policies import base_policy as bp
import numpy as np

EPSILON = 0.05

class Avoid(bp.Policy):
    """
    A policy which avoids collisions with obstacles and other snakes. It has an epsilon parameter which controls the
    percentag of actions which are randomly chosen.
    """

    def cast_string_args(self, policy_args):
        policy_args['epsilon'] = float(policy_args['epsilon']) if 'epsilon' in policy_args else EPSILON
        return policy_args

    def init_run(self):
        self.r_sum = 0

    def learn(self, round, prev_state, prev_action, reward, new_state, too_slow):

        try:
            if round % 100 == 0:
                if round > self.game_duration - self.score_scope:
                    self.log("Rewards in last 100 rounds which counts towards the score: " + str(self.r_sum), 'VALUE')
                else:
                    self.log("Rewards in last 100 rounds: " + str(self.r_sum), 'VALUE')
                self.r_sum = 0
            else:
                self.r_sum += reward

        except Exception as e:
            self.log("Something Went Wrong...", 'EXCEPTION')
            self.log(e, 'EXCEPTION')

    def act(self, round, prev_state, prev_action, reward, new_state, too_slow):

        board, head = new_state
        head_pos, direction = head

        if np.random.rand() < self.epsilon:
            return np.random.choice(bp.Policy.ACTIONS)

        else:
            for a in list(np.random.permutation(bp.Policy.ACTIONS)):

                # get a Position object of the position in the relevant direction from the head:
                next_position = head_pos.move(bp.Policy.TURNS[direction][a])
                r = next_position[0]
                c = next_position[1]

                # look at the board in the relevant position:
                if board[r, c] > 5 or board[r, c] < 0:
                    return a

            # if all positions are bad:
            return np.random.choice(bp.Policy.ACTIONS)

<file_sep>/README.md
# SnakeRL
Deep Reinforcement Learning implementation for the mighty game of Snake

### Agents Implemented:
[Linear Agent](policies/policy_linear.py)

[Basic DQN](policies/policy_DQN.py)

[Dueling DQN](policies/policy_DuelDQN.py)

[Convolutional DQN](policies/policy_ConvDQN.py)

[Convolutional Dueling DQN](policies/policy_ConvDuelDQN.py)

##### Important Paramters:
DoubleDQN - enables double DQN learning.

radius - control the radius of the agent field of view.

### Prioritized Experience Replay
Some experiences may be more important than others for our training, but might occur less frequently. Since the batch is sampled uniformly by default, these rich experiences that occur rarely have practically no chance to be selected. 

Changing sampling distribution by using a criterion to define the priority of each tuple of experience (Used the prediction mean absolute difference error) as a priority criterion to give the model more chance to learn from experiences on
which its’ predictin is bad, thus giving it a better chance to improve its’ predictions.

For usage:

Just change the base class for any of the above to [PriorBaseDQN](policies/policy_PriorityBaseDQN.py).

### State Representations
Tried multiple different implementations:

• Square(l∞ distance) of a certain radius around the snake’s head.

• Diamond(l1 distance) of a certain radius around the snake’s head.

• Circle(l2 distance) of a certain radius around the Snake’s head.

• Using sort of a “Radar”, which returned directions and distances to a certain amount of entities of each type from the snake’s head.

### Exploration-Exploitation trade-off: 
Used softmax sampling, with a temperature value that decays exponentially with a certain rate, in order to
explore more on early stage and then exploit the agent’s “knowledge” in later stage.

Also have option to use epsilon-greedy.



<file_sep>/Snake.py
import pickle
import time
import datetime
import multiprocessing as mp
import queue
import os
import argparse
import sys
import gzip
import subprocess as sp

import numpy as np
import scipy.signal as ss

from policies import base_policy
from policies import *

EMPTY_VAL = -1
MAX_PLAYERS = 5
OBSTACLE_VAL = 5
REGULAR_RENDER_MAP = {EMPTY_VAL: ' ', OBSTACLE_VAL: '+'}
FOOD_RENDER_MAP = {6: '*', 7: '$', 8: 'X'}
FOOD_VALUE_MAP = {6: 1, 7: 3, 8: 0}
FOOD_REWARD_MAP = {6: 2, 7: 5, 8: -1}
THE_DEATH_PENALTY = -5

ILLEGAL_MOVE = "Illegal Action: the default action was selected instead. Player tried action: "
NO_RESPONSE = "No Response: player took too long to respond with action. This is No Response #"
PLAYER_INIT_TIME = 60
UNRESPONSIVE_PLAYER = "Unresponsive Player: the player hasn't responded in too long... SOMETHING IS WRONG!!"

STATUS_SKIP = 100
TOO_SLOW_THRESHOLD = 3
UNRESPONSIVE_THRESHOLD = 50
LEARNING_TIME = 5


def clear_q(q):
    """
    given a queue, empty it.
    """
    while not q.empty():
        try:
            q.get_nowait()
        except queue.Empty:
            break


def days_hours_minutes_seconds(td):
    """
    parse time for logging.
    """
    return td.days, td.seconds // 3600, (td.seconds // 60) % 60, td.seconds % 60


def random_partition(num, max_part_size):
    parts = []
    while num > 0:
        parts.append(np.random.randint(1, min(max_part_size, num + 1)))
        num -= parts[-1]
    return parts


class Position():

    def __init__(self, position, board_size):
        self.pos = position
        self.board_size = board_size

    def __getitem__(self, key):
        return self.pos[key]

    def __add__(self, other):
        return Position(((self[0] + other[0]) % self.board_size[0],
                         (self[1] + other[1]) % self.board_size[1]),
                        self.board_size)

    def move(self, dir):
        if dir == 'E': return self + (0, 1)
        if dir == 'W': return self + (0, -1)
        if dir == 'N': return self + (-1, 0)
        if dir == 'S': return self + (1, 0)
        raise ValueError('unrecognized direction')


class Agent(object):
    SHUTDOWN_TIMEOUT = 60  # seconds until policy is considered unresponsive

    def __init__(self, id, policy, policy_args, board_size, logq, game_duration, score_scope):
        """
        Construct a new player
        :param id: the player id (the value of the player positions in the board
        :param policy: the class of the policy to be used by the player
        :param policy_args: string (name, value) pairs that the policy can parse to arguments
        :param board_size: the size of the game board (height, width)
        :param logq: a queue for message logging through the game
        :param game_duration: the expected duration of the game in turns
        :param score_scope: the amount of rounds at the end of the game which count towards the score
        """

        self.id = id
        self.len = 0
        self.policy_class = policy
        self.round = 0
        self.unresponsive_count = 0
        self.too_slow = False

        self.sq = mp.Queue()
        self.aq = mp.Queue()
        self.mq = mp.Queue()
        self.logq = logq
        self.policy = policy(policy_args, board_size, self.sq, self.aq, self.mq, logq, id, game_duration, score_scope)
        self.policy.daemon = True
        self.policy.start()

    def handle_state(self, round, prev_state, prev_action, reward, new_state):
        """
        given the new state and previous state-action-reward, pass the information
        to the policy for action selection and/or learning.
        """

        self.round = round
        clear_q(self.sq)  # remove previous states from queue if they weren't handled yet
        self.sq.put((round, prev_state, prev_action, reward, new_state, self.too_slow))

    def get_action(self):
        """
        get waiting action from the policy's action queue. if there is no action
        in the queue, pick 'F' and log the unresponsiveness error.
        :return: action from {'R','L','F'}.
        """
        try:
            round, action = self.aq.get_nowait()
            if round != self.round:
                raise queue.Empty()
            elif action not in base_policy.Policy.ACTIONS:
                self.logq.put((str(self.id), "ERROR", ILLEGAL_MOVE + str(action)))
                raise queue.Empty()
            else:
                self.too_slow = False
                self.unresponsive_count = 0

        except queue.Empty:
            self.unresponsive_count += 1
            action = base_policy.Policy.DEFAULT_ACTION
            if self.unresponsive_count <= UNRESPONSIVE_THRESHOLD:
                self.logq.put((str(self.id), "ERROR", NO_RESPONSE + str(self.unresponsive_count) + " in a row!"))
            else:
                self.logq.put((str(self.id), "ERROR", UNRESPONSIVE_PLAYER))
                self.unresponsive_count = TOO_SLOW_THRESHOLD
            if self.unresponsive_count > TOO_SLOW_THRESHOLD:
                self.too_slow = True

        clear_q(self.aq)  # clear the queue from unhandled actions
        return action

    def shutdown(self):
        """
        shutdown the agent in the end of the game. the function asks the agent
        to save it's model and returns the saved model, which needs to be a data
        structure that can be pickled.
        :return: the model data structure.
        """

        clear_q(self.sq)
        clear_q(self.aq)
        self.sq.put(None)  # shutdown signal
        self.policy.join()
        return


class Game(object):

    @staticmethod
    def log(q, file_name, on_screen=True):
        start_time = datetime.datetime.now()
        logfile = None
        if file_name:
            logfile = gzip.GzipFile(file_name,
                                    'w') if file_name.endswith(
                '.gz') else open(file_name, 'wb')
        for frm, type, msg in iter(q.get, None):
            td = datetime.datetime.now() - start_time
            msg = '%i::%i:%i:%i\t%s\t%s\t%s' % (
                    days_hours_minutes_seconds(td) + (frm, type, msg))
            if logfile: logfile.write((msg + '\n').encode('ascii'))
            if on_screen: print(msg)
        if logfile: logfile.close()

    def _find_empty_slot(self, shape=(1, 3)):
        is_empty = np.asarray(self.board == EMPTY_VAL, dtype=int)
        match = ss.convolve2d(is_empty, np.ones(shape), mode='same') == np.prod(shape)
        if not np.any(match): raise ValueError('no empty slots of requested shape')
        r = np.random.choice(np.nonzero(np.any(match, axis=1))[0])
        c = np.random.choice(np.nonzero(match[r, :])[0])
        return Position((r, c), self.board_size)

    def __init__(self, args):
        self.__dict__.update(args.__dict__)

        # check that the number of players is OK:
        self.n = len(self.policies)
        assert self.n <= MAX_PLAYERS, "Too Many Players!"

        self.round = 0

        # check for playback:
        self.is_playback = False
        if self.playback_from is not None:
            self.archive = open(self.playback_from, 'rb')
            dict = pickle.load(self.archive)
            self.__dict__.update(dict)
            self.is_playback = True
            self.record_to = None
            self.record = False
            self.to_render = args.to_render
            return

        # init logger
        self.logq = mp.Queue()
        to_screen = not self.to_render
        self.logger = mp.Process(target=self.log, args=(self.logq, self.log_file, to_screen))
        self.logger.start()

        # initialize the board:
        self.item_count = 0
        self.board = EMPTY_VAL * np.ones(self.board_size, dtype=int)
        self.previous_board = None

        # initialize obstacles:
        for size in random_partition(int(self.obstacle_density * np.prod(self.board_size)), np.min(self.board_size)):
            if np.random.rand(1) > 0.5:
                pos = self._find_empty_slot(shape=(size, 1))
                for i in range(size):
                    new_pos = pos + (i, 0)
                    self.board[new_pos[0], new_pos[1]] = OBSTACLE_VAL
            else:
                pos = self._find_empty_slot(shape=(1, size))
                for i in range(size):
                    new_pos = pos + (i, 0)
                    self.board[new_pos[0], new_pos[1]] = OBSTACLE_VAL
            self.item_count += size

        # initialize players:
        self.rewards, self.players, self.scores, self.directions, self.actions, self.growing, self.size, self.chains, self.previous_heads = \
            [], [], [], [], [], [], [], [], []
        for i, (policy, pargs) in enumerate(self.policies):
            self.rewards.append(0)
            self.actions.append(None)
            self.previous_heads.append(None)
            self.scores.append([0])
            self.players.append(
                Agent(i, policy, pargs, self.board_size, self.logq, self.game_duration, self.score_scope))
            chain, player_size, growing, direction = self.init_player()
            self.chains.append(chain)
            self.size.append(player_size)
            self.growing.append(growing)
            self.directions.append((direction))
            for position in chain:
                self.board[position[0], position[1]] = i

        # configure symbols for rendering
        self.render_map = {p.id: chr(ord('1') + p.id) for p in self.players}
        self.render_map.update(REGULAR_RENDER_MAP)
        self.render_map.update(FOOD_RENDER_MAP)

        # finally, if it's a recording, then record
        if self.record_to is not None and self.playback_from is None:
            self.archive = open(self.record_to, 'wb')
            dict = self.__dict__.copy()
            del dict['players']  # remove problematic objects that are irrelevant to playback.
            del dict['archive']
            del dict['logq']
            del dict['logger']
            del dict['playback_initial_round']
            del dict['playback_final_round']
            pickle.dump(dict, self.archive)
        self.record = self.record_to is not None

        # wait for player initialization (Keras loading time):
        time.sleep(self.player_init_time)

    def init_player(self):

        # initialize the position and direction of the player:
        dir = np.random.choice(list(base_policy.Policy.TURNS.keys()))
        shape = (1, 3) if dir in ['W', 'E'] else (3, 1)
        pos = self._find_empty_slot(shape)

        # gather stats about the player:
        chain = []
        chain.append(pos)
        chain.append(pos.move(dir))
        player_size = 2
        growing = self.init_player_size - 2

        return chain, player_size, growing, dir

    def reset_player(self, id):

        positions = np.array(np.where(self.board == id))
        for pos in range(positions.shape[1]):
            self.board[positions[0, pos], positions[1, pos]] = EMPTY_VAL

        # turn parts of the corpse into food:
        food_n = np.random.binomial(positions.shape[1], self.food_ratio)
        if self.item_count + food_n < self.max_item_density * np.prod(self.board_size):
            subidx = np.array(np.random.choice(positions.shape[1], size=food_n, replace=False))
            if len(subidx) > 0:
                randfood = np.random.choice(list(FOOD_VALUE_MAP.keys()), food_n)
                for i, idx in enumerate(subidx):
                    self.board[positions[0, idx], positions[1, idx]] = randfood[i]
                self.item_count += food_n
        return self.init_player()

    def randomize(self):
        if np.random.rand(1) < self.random_food_prob:
            if self.item_count < self.max_item_density * np.prod(self.board_size):
                randfood = np.random.choice(list(FOOD_VALUE_MAP.keys()), 1)
                slot = self._find_empty_slot((1, 1))
                self.board[slot[0], slot[1]] = randfood
                self.item_count += 1

    def move_snake(self, id, action):

        # delete the tail if the snake isn't growing:
        if self.growing[id] > 0:
            self.growing[id] -= 1
            self.size[id] += 1
        else:
            self.board[self.chains[id][0][0], self.chains[id][0][1]] = EMPTY_VAL
            del self.chains[id][0]

        # move the head:
        if action != 'F':  # turn in the relevant direction
            self.directions[id] = base_policy.Policy.TURNS[self.directions[id]][action]
        self.chains[id].append(self.chains[id][-1].move(self.directions[id]))
        self.board[self.chains[id][-1][0], self.chains[id][-1][1]] = id

    def play_a_round(self):

        # randomize the players:
        pperm = np.random.permutation([(i, p) for i, p in enumerate(self.players)])

        # distribute states and rewards on previous round
        for i, p in pperm:
            current_head = (self.chains[p.id][-1], self.directions[p.id])
            if self.previous_board is None:
                p.handle_state(self.round, None, self.actions[p.id], self.rewards[p.id], (self.board, current_head))
            else:
                p.handle_state(self.round, (self.previous_board, self.previous_heads[p.id]), self.actions[p.id],
                               self.rewards[p.id], (self.board, current_head))
            self.previous_heads[p.id] = current_head
        self.previous_board = np.copy(self.board)

        # wait and collect actions
        time.sleep(self.policy_action_time)
        actions = {p: p.get_action() for _, p in pperm}
        if self.round % LEARNING_TIME == 0 and self.round > 5:
            time.sleep(self.policy_learn_time)

        # get the interactions of the players with the board:
        for _, p in pperm:
            action = actions[p]
            self.actions[p.id] = action
            move_to = self.chains[p.id][-1].move(base_policy.Policy.TURNS[self.directions[p.id]][action])

            # reset the player if he died:
            if self.board[move_to[0], move_to[1]] != EMPTY_VAL and self.board[
                move_to[0], move_to[1]] not in FOOD_VALUE_MAP:
                chain, player_size, growing, dir = self.reset_player(p.id)
                self.chains[p.id] = chain
                self.size[p.id] = player_size
                self.growing[p.id] = growing
                self.directions[p.id] = dir
                self.rewards[p.id] = THE_DEATH_PENALTY
                self.scores[p.id].append(self.rewards[p.id])
                for pos in chain:
                    self.board[pos[0], pos[1]] = p.id

            # otherwise, move the player on the board:
            else:
                self.rewards[p.id] = 0
                if self.board[move_to[0], move_to[1]] in FOOD_VALUE_MAP.keys():
                    self.rewards[p.id] += FOOD_REWARD_MAP[self.board[move_to[0], move_to[1]]]
                    self.growing[p.id] += FOOD_VALUE_MAP[self.board[move_to[0], move_to[1]]]  # start growing
                    self.item_count -= 1

                self.move_snake(p.id, action)
                self.scores[p.id].append(self.rewards[p.id])

        # update the food on the board:
        self.randomize()
        self.round += 1

    def render(self, r):

        if os.name == 'nt':
            os.system('cls')  # clear screen for Windows
        else:
            print(chr(27) + "[2J")  # clear screen for linux

        # print the scores:
        print("Time Step: " + str(r) + "/" + str(self.game_duration))
        for i in range(len(self.scores)):
            scope = np.min([len(self.scores[i]), self.score_scope])
            print("Player " + str(i + 1) + ": " + str("{0:.4f}".format(np.mean(self.scores[i][-scope:]))))

        # print the board:
        horzline = '-' * (self.board.shape[1] + 2)
        board = [horzline]
        for r in range(self.board.shape[0]):
            board.append('|' + ''.join(self.render_map[self.board[r, c]] for c in range(self.board.shape[1])) + '|')
        board.append(horzline)
        print('\n'.join(board))

    def run(self):
        try:
            r = 0
            while r < self.game_duration:
                r += 1
                if self.to_render:
                    if not (self.is_playback and (r < self.playback_initial_round or r > self.playback_final_round)):
                        self.render(r)
                        time.sleep(self.render_rate)
                else:
                    if r % STATUS_SKIP == 0:
                        print("At Round " + str(r) + " the scores are:")
                        for i, s in enumerate(self.scores):
                            scope = np.min([len(self.scores[i]), self.score_scope])
                            print("Player " + str(i + 1) + ": " + str(
                                str("{0:.4f}".format(np.mean(self.scores[i][-scope:])))))
                if self.is_playback:
                    try:
                        idx, vals, self.scores = pickle.load(self.archive)
                        self.board[idx] = vals
                    except EOFError:
                        break
                    if r > self.playback_final_round:
                        break

                else:
                    if self.record:
                        prev = self.board.copy()
                    self.play_a_round()
                    if self.record:
                        idx = np.nonzero(self.board - prev != 0)
                        pickle.dump((idx, self.board[idx], self.scores), self.archive)
        finally:

            if self.record or self.is_playback:
                self.archive.close()

            if not self.is_playback:

                output = [','.join(['game_id', 'player_id', 'policy', 'score'])]
                game_id = str(abs(id(self)))
                for p, s in zip(self.players, self.scores):
                    p.shutdown()
                    pstr = str(p.policy).split('<')[1].split('(')[0]
                    scope = np.min([len(s), self.score_scope])
                    p_score = np.mean(s[-scope:])
                    oi = [game_id, str(p.id), pstr, str("{0:.4f}".format(p_score))]
                    output.append(','.join(oi))

                with open(self.output_file, 'w') as outfile:
                    outfile.write('\n'.join(output))
                self.logq.put(None)
                self.logger.join()


def parse_args():
    p = argparse.ArgumentParser()
    g = p.add_argument_group('I/O')
    g.add_argument('--record_to', '-rt', type=str, default=None, help="file path to which game will be recorded.")
    g.add_argument('--playback_from', '-p', type=str, default=None,
                   help='file path from which game will be played-back (overrides record_to)')
    g.add_argument('--playback_initial_round', '-pir', type=int, default=0,
                   help='round in which to start the playback')
    g.add_argument('--playback_final_round', '-pfr', type=int, default=1000,
                   help='round in which to end the playback')
    g.add_argument('--log_file', '-l', type=str, default=None,
                   help="a path to which game events are logged. default: game.log")
    g.add_argument('--output_file', '-o', type=str, default=None,
                   help="a path to a file in which game results are written. default: game.out")
    g.add_argument('--to_render', '-r', type=int, default=0, help="whether game should not be rendered")
    g.add_argument('--render_rate', '-rr', type=float, default=0.1,
                   help='frames per second, note that the policy_wait_time bounds on the rate')

    g = p.add_argument_group('Game')
    g.add_argument('--board_size', '-bs', type=str, default='(20,60)', help='a tuple of (height, width)')
    g.add_argument('--obstacle_density', '-od', type=float, default=.04, help='the density of obstacles on the board')
    g.add_argument('--policy_wait_time', '-pwt', type=float, default=0.01,
                   help='seconds to wait for policies to respond with actions')
    g.add_argument('--random_food_prob', '-fp', type=float, default=.2,
                   help='probability of a random food appearing in a round')
    g.add_argument('--max_item_density', '-mid', type=float, default=.25,
                   help='maximum item density in the board (not including the players)')
    g.add_argument('--food_ratio', '-fr', type=float, default=.2,
                   help='the ratio between a corpse and the number of food items it produces')
    g.add_argument('--game_duration', '-D', type=int, default=10000, help='number of rounds in the session')
    g.add_argument('--policy_action_time', '-pat', type=float, default=0.01,
                   help='seconds to wait for agents to respond with actions')
    g.add_argument('--policy_learn_time', '-plt', type=float, default=0.1,
                   help='seconds to wait for agents to improve policy')
    g.add_argument('--player_init_time', '-pit', type=float, default=PLAYER_INIT_TIME,
                   help='seconds to wait for agents to initialize in the beginning of the session')

    g = p.add_argument_group('Players')
    g.add_argument('--score_scope', '-s', type=int, default=1000,
                   help='The score is the average reward during the last score_scope rounds of the session')
    g.add_argument('--init_player_size', '-is', type=int, default=5, help='player length at start, minimum is 3')
    g.add_argument('--policies', '-P', type=str, default=None,
                   help='a string describing the policies to be used in the game, of the form: '
                        '<policy_name>(<arg=val>,*);+.\n'
                        'e.g. MyPolicy(layer1=100,layer2=20);YourPolicy(your_params=123)')

    args = p.parse_args()

    # set defaults
    code_path = os.path.split(os.path.abspath(__file__))[0] + os.path.sep
    if not args.record_to:
        args.__dict__['record_to'] = None
    if args.log_file is None:
        args.__dict__['log_file'] = code_path + 'game.log'
    if args.output_file is None:
        args.__dict__['output_file'] = code_path + 'game.out'
    if args.playback_from is not None:
        args.__dict__['record_to'] = None
        args.__dict__['output_file'] = None
        args.__dict__['log_file'] = None

    args.__dict__['board_size'] = [int(x) for x in args.board_size[1:-1].split(',')]
    plcs = []
    if args.policies is not None: plcs.extend(args.policies.split(';'))
    args.__dict__['policies'] = [base_policy.build(p) for p in plcs]

    return args


if __name__ == '__main__':
    g = Game(parse_args())
    g.run()
<file_sep>/policies/policy_PriorityBaseDQN.py
from abc import abstractmethod

import keras
import numpy as np
from keras import Model
from keras.optimizers import Adam
from sklearn.utils.extmath import softmax

from nn_utils.losses import huber_loss
from policies import base_policy as bp
from state_utils.State import SquareAroundHeadState, DiamondAroundHeadState, RadarState, \
    FULL, NUM_PER_TYPE
from utils.replay_buffer import PrioritizedReplayBuffer

EPSILON = 2.0
MIN_EPSILON = 0.33
GAMMA = 0.75
BUFFER_SIZE = 6000

SQUARE = 1
DIAMOND = 2
RADAR = 3


class PriorBaseDQN(bp.Policy):
    """
    A different version of the abstract base class that implements Deep Q-Learning and allows for customization -
    to be extended by other policies that we wrote. This version was changed to support our attempt at using
    prioritized experience replay buffer
    """

    def cast_string_args(self, policy_args):
        policy_args['epsilon'] = float(policy_args['epsilon']) if 'epsilon' in policy_args else EPSILON
        policy_args['gamma'] = float(policy_args['gamma']) if 'gamma' in policy_args else GAMMA
        self.huber_loss = False
        self.use_softmax_sampling = True

        self.epsilon_decay = 0.90
        self.min_epsilon = MIN_EPSILON

        self.learning_rate = 1e-4
        self.batch_size = 96

        self.state_radius = 5
        self.state_rep = SQUARE
        self.step_forward = True
        self.flatten = FULL

        self.doubleDQN = False

        self.save_model_round = 250

        self.augment_after_normaliztion = False
        policy_args = self._additional_args(policy_args)

        return policy_args

    def _save_model(self):
        self.old_model.set_weights(self.model.get_weights())

    def init_run(self):
        self.log("Starting init")
        self.r_sum = 0

        if self.state_rep == SQUARE:
            self.state_proc = SquareAroundHeadState(radius=self.state_radius,
                                                    step_forward=self.step_forward, flatten=self.flatten)
        elif self.state_rep == DIAMOND:
            self.state_proc = DiamondAroundHeadState(radius=self.state_radius,
                                                     step_forward=self.step_forward, flatten=self.flatten)
        elif self.state_rep == RADAR:
            self.state_proc = RadarState(num_per_type=NUM_PER_TYPE)

        self.input_shape = self.state_proc.get_shape()

        self.model = self._build_model()
        self.model.summary()

        if self.huber_loss:
            loss = huber_loss
        else:
            loss = 'mse'

        opt = Adam(self.learning_rate)
        self.model.compile(loss=loss, optimizer=opt, metrics=['mae'])

        self.old_model = keras.models.clone_model(self.model)
        self._save_model()

        self.memory = PrioritizedReplayBuffer(BUFFER_SIZE, alpha=0.4)
        self.log("Init finished!")

        self.num_of_samples = 0
        self.sum_of_loss = 0

    def learn(self, round, prev_state, prev_action, reward, new_state, too_slow):
        try:
            if round % 100 == 0:
                if round > self.game_duration - self.score_scope:
                    self.log("Rewards in last 100 rounds which counts towards the score: {}, eps={:.2f}, "
                             "db_size={}".format(str(self.r_sum), self.epsilon, len(self.memory)), 'VALUE')
                else:
                    total_loss = self.sum_of_loss / self.num_of_samples
                    self.num_of_samples = self.sum_of_loss = 0
                    self.log("Rewards in last 100 rounds: {}, eps={:.2f}, db_size={}, loss={:.3f}".format(
                        str(self.r_sum), self.epsilon, len(self.memory), total_loss), 'VALUE')
                self.r_sum = 0
            else:
                self.r_sum += reward
        except Exception as e:
            self.log("Something Went Wrong...", 'EXCEPTION')
            self.log(e, 'EXCEPTION')

        prev, actions, rewards, new, _, sample_weight, idx = self.memory.sample(self.batch_size, beta=0.6)

        if self.doubleDQN:
            target = rewards + self.gamma * self.old_model.predict(new)[range(len(new)),
                                                                        np.argmax(self.model.predict(new), axis=1)]
        else:
            target = rewards + self.gamma * np.amax(self.old_model.predict(new), axis=1)
        target_f = self.model.predict(prev)

        try:
            target_f[range(len(actions)), actions] = target
            hist = self.model.fit(prev, target_f, epochs=1, verbose=0, batch_size=len(prev), shuffle=False,
                                  sample_weight=sample_weight)
            self.sum_of_loss += np.sum(hist.history['loss'])
            self.num_of_samples += len(hist.history['loss'])

            mae = np.sum(np.abs(target_f - self.model.predict(prev, verbose=0, batch_size=len(prev))), axis=-1)
            prioritized_replay_eps = 1e-6
            self.memory.update_priorities(idx, mae + prioritized_replay_eps)
        except Exception as e:
            print(e)

        if round % self.save_model_round == 0 and round > 0:
            self._save_model()
        if round % 200 == 0 and round > 0 and self.epsilon > 0:
            self.epsilon = max(self.epsilon * self.epsilon_decay, self.min_epsilon)

    def act(self, round, prev_state, prev_action, reward, new_state, too_slow):
        if round > self.game_duration - self.score_scope:
            # cancel exploration during "money-time"
            self.use_softmax_sampling = False
            self.epsilon = 0

        new_state_repr = self.state_proc.get_state_repr(new_state)

        if prev_state is not None:
            prev_state_repr = self.state_proc.get_state_repr(prev_state)
            self.memory.add(prev_state_repr, bp.Policy.ACTIONS.index(prev_action), reward, new_state_repr, 0)

        if self.use_softmax_sampling:
            return np.random.choice(bp.Policy.ACTIONS,
                                    p=softmax(self.model.predict(new_state_repr[np.newaxis]) / self.epsilon).squeeze())
        else:  # use epsilon-greedy
            if np.random.rand() < self.epsilon:
                return np.random.choice(bp.Policy.ACTIONS)
            else:
                prediction = self.model.predict(new_state_repr[np.newaxis])[0]
                action = bp.Policy.ACTIONS[np.argmax(prediction)]
                return action

    @abstractmethod
    def _build_model(self) -> Model:
        raise NotImplementedError

    @abstractmethod
    def _additional_args(self, policy_args):
        raise NotImplementedError
<file_sep>/policies/policy_DuelDQN.py
from keras import Input, Model
from keras.layers import Conv1D, Flatten, Dense, Dropout, Lambda, Add, LeakyReLU
import keras.backend as K

from policies.policy_BaseDQN import BaseDQN, SQUARE, SPATIAL


class DuelDQN(BaseDQN):
    """
    A Dueling Deep Q-Learning implementation with a Multi-Layered Dense network as the Q estimator
    """

    def _additional_args(self, policy_args):
        self.flatten = SPATIAL
        self.state_radius = 5
        self.step_forward = True
        self.state_rep = SQUARE

        self.doubleDQN = True

        # for self
        self.dropout_rate = 0.0
        self.activation = LeakyReLU
        return policy_args

    def _build_model(self):
        input_layer = Input(self.input_shape)
        net = self.activation()(Conv1D(input_shape=self.input_shape, filters=4, kernel_size=1)(input_layer))
        net = Flatten()(net)

        net = self.dense_layer(net, units=128)

        streamA = self.dense_layer(net, units=128)
        streamV = self.dense_layer(net, units=128)

        Advantage = Dense(units=3)(streamA)
        Value = Dense(units=1)(streamV)

        advt = Lambda(lambda advt: advt - K.mean(advt, axis=-1, keepdims=True))(Advantage)
        value = Lambda(lambda value: K.tile(value, [1, 3]))(Value)
        Q_out = Add()([value, advt])

        return Model(inputs=[input_layer], outputs=[Q_out])

    def dense_layer(self, input_layer, units):
        net = Dense(units=units)(input_layer)
        net = self.activation()(net)  # apply activation
        if self.dropout_rate > 0:
            net = Dropout(self.dropout_rate)(net)
        return net<file_sep>/policies/policy_DQN.py
from keras import Sequential
from keras.layers import Conv1D, Flatten, Dense, Dropout, LeakyReLU

from policies.policy_BaseDQN import BaseDQN, SQUARE, SPATIAL


class DQN(BaseDQN):
    """
    A Deep Q-Learning implementation with a Multi-Layered Dense network as the Q estimator
    """

    def _additional_args(self, policy_args):
        self.flatten = SPATIAL
        self.state_radius = 6
        self.step_forward = True
        self.state_rep = SQUARE

        self.doubleDQN = True
        self.batch_size = 96

        self.epsilon_decay = 0.98
        self.min_epsilon = 0.1
        self.gamma = 0.75

        # for self
        self.dropout_rate = 0.0
        self.activation = LeakyReLU
        return policy_args

    def _build_model(self):
        model = Sequential()
        model.add(Conv1D(input_shape=self.input_shape, filters=4, kernel_size=1))
        model.add(self.activation())
        model.add(Flatten())

        model.add(Dense(units=128))
        model.add(self.activation())

        if self.dropout_rate > 0:
            model.add(Dropout(self.dropout_rate))

        model.add(Dense(units=128))
        model.add(self.activation())

        if self.dropout_rate > 0:
            model.add(Dropout(self.dropout_rate))

        model.add(Dense(units=128))
        model.add(self.activation())

        model.add(Dense(units=3))
        return model
