snake_nn2.py

'''

This is an initial test for the Skynet Bot program. It is using an initial game template for neural network
design from here: https://towardsdatascience.com/today-im-going-to-talk-about-a-small-practical-example-of-using-neural-networks-training-one-to-6b2cbd6efdb3

Attribution for this goes to Slava Korolev.

'''

from snake import SnakeGame
from random import randint
import numpy as np
import tflearn
import math
from tflearn.layers.core import input_data, fully_connected
from tflearn.layers.estimator import regression
from statistics import mean
from collections import Counter

class SnakeNN:
    def __init__(self, initial_games = 10000, test_games = 1000, goal_steps = 2000, lr = 0.0002, filename = 'snake_nn_2.tflearn'):
        self.initial_games = initial_games
        self.test_games = test_games
        self.goal_steps = goal_steps
        self.lr = lr
        self.filename = filename
        self.vectors_and_keys = [
                [[-1, 0], 0],
                [[0, 1], 1],
                [[1, 0], 2],
                [[0, -1], 3]
                ]

    def initial_population(self):
        training_data = []
        for _ in range(self.initial_games):
            game = SnakeGame()
            _, prev_score, snake, food, obstacles = game.start() # returns generate_observation from snake game.
            prev_observation = self.generate_observation(snake, food, obstacles)
            prev_food_distance = self.get_food_distance(snake, food)
            for _ in range(self.goal_steps):
                action, game_action = self.generate_action(snake)
                done, score, snake, food, obstacles  = game.step(game_action)
                #print(training_data)
                #input("test")
                if done:
                    training_data.append([self.add_action_to_observation(prev_observation, action), -1])
                    break
                else:
                    food_distance = self.get_food_distance(snake, food)
                    if score > prev_score or food_distance < prev_food_distance: # did you get closer to the objective?
                        training_data.append([self.add_action_to_observation(prev_observation, action), 1]) # label as good decision.
                    else:
                        training_data.append([self.add_action_to_observation(prev_observation, action), 0]) # label as bad decision.
                    prev_observation = self.generate_observation(snake, food, obstacles)
                    prev_food_distance = food_distance
                    '''
                    Later we will be using this "1" or "0" to provide estimates for each possible decision. 
                    '''
        return training_data

    def generate_action(self, snake):
        action = randint(0,2) - 1
        return action, self.get_game_action(snake, action)

    def get_game_action(self, snake, action):
        snake_direction = self.get_snake_direction_vector(snake)
        new_direction = snake_direction
        if action == -1:
            new_direction = self.turn_vector_to_the_left(snake_direction)
        elif action == 1:
            new_direction = self.turn_vector_to_the_right(snake_direction)
        for pair in self.vectors_and_keys:
            if pair[0] == new_direction.tolist():
                game_action = pair[1] # there is some weird scope stuff going on here. what the heck // test 32
        return game_action

    def generate_observation(self, snake, food, obstacles):
        snake_direction = self.get_snake_direction_vector(snake)
        food_direction = self.get_food_direction_vector(snake, food)
        barrier_left = self.is_direction_blocked(snake, self.turn_vector_to_the_left(snake_direction), obstacles)
        barrier_front = self.is_direction_blocked(snake, snake_direction, obstacles)
        barrier_right = self.is_direction_blocked(snake, self.turn_vector_to_the_right(snake_direction), obstacles)
        angle = self.get_angle(snake_direction, food_direction)
        return np.array([int(barrier_left), int(barrier_front), int(barrier_right), angle])

    def add_action_to_observation(self, observation, action):
        return np.append([action], observation)

    def get_snake_direction_vector(self, snake):
        return np.array(snake[0]) - np.array(snake[1])

    def get_food_direction_vector(self, snake, food):
        return np.array(food) - np.array(snake[0])

    def normalize_vector(self, vector):
        return vector / np.linalg.norm(vector)

    def get_food_distance(self, snake, food):
        return np.linalg.norm(self.get_food_direction_vector(snake, food))

    def is_direction_blocked(self, snake, direction, obstacles):
        point = np.array(snake[0]) + np.array(direction)
        return(point.tolist() in snake[:-1] or point[0] == 0 or point[1] == 0 or point[0] == 21 or point[1] == 21 or (point.tolist() in obstacles))

    def turn_vector_to_the_left(self, vector):
        return np.array([-vector[1], vector[0]])

    def turn_vector_to_the_right(self, vector):
        return np.array([vector[1], -vector[0]])

    def get_angle(self, a, b):
        a = self.normalize_vector(a)
        b = self.normalize_vector(b)
        return math.atan2(a[0] * b[1] - a[1] * b[0], a[0] * b[0] + a[1] * b[1]) / math.pi

    def model(self):
        network = input_data(shape=[None, 5, 1], name='input') # shape of data [left, forward, right, angle, 
        
        # Activation and Regularization inside a layer:
        network = fully_connected(network, 25, activation='relu') # hidden layer neurons
        # Equivalent to:
        network = fully_connected(network, 1, activation='linear') # hidden layer neurons
        
        # Optimizer, Objective and Metric:        
        network = regression(network, optimizer='adam', learning_rate=self.lr, loss='mean_square', name='target')
        model = tflearn.DNN(network, tensorboard_dir='log')
        return model

    def train_model(self, training_data, model):
        X = np.array([i[0] for i in training_data]).reshape(-1, 5, 1) # observations array
        y = np.array([i[1] for i in training_data]).reshape(-1, 1) # fitness value
        model.fit(X,y, n_epoch = 3, shuffle = True, run_id = self.filename)
        model.save(self.filename)
        return model

    def test_model(self, model):
        steps_arr = []
        scores_arr = []
        for _ in range(self.test_games):
            steps = 0
            game_memory = []
            game = SnakeGame()
            _, score, snake, food, obstacles = game.start()
            prev_observation = self.generate_observation(snake, food, obstacles)
            for _ in range(self.goal_steps):
                predictions = []
                for action in range(-1, 2): # iterate through each possible decision
                   predictions.append(model.predict(self.add_action_to_observation(prev_observation, action).reshape(-1, 5, 1)))
                action = np.argmax(np.array(predictions)) # choose decision with highest value (1)
                game_action = self.get_game_action(snake, action - 1) # perform action in the game. 
                done, score, snake, food, obstacles  = game.step(game_action)
                game_memory.append([prev_observation, action])
                print("PRINTING THE ACTION!")
                print(action)
                exit()
                if done:
                    print('-----')
                    print(steps)
                    print(snake)
                    print(food)
                    print(prev_observation)
                    print(predictions)
                    break
                else:
                    prev_observation = self.generate_observation(snake, food, obstacles)
                    steps += 1
            steps_arr.append(steps)
            scores_arr.append(score)
        print('Average steps:',mean(steps_arr))
        print(Counter(steps_arr))
        print('Average score:',mean(scores_arr))
        print(Counter(scores_arr))

    def visualize_game(self, model):
        game = SnakeGame(gui = True)
        _, _, snake, food, obstacles = game.start()
        prev_observation = self.generate_observation(snake, food, obstacles)
        for _ in range(self.goal_steps):
            predictions = []
            for action in range(-1, 2):
               predictions.append(model.predict(self.add_action_to_observation(prev_observation, action).reshape(-1, 5, 1)))
            action = np.argmax(np.array(predictions))
            game_action = self.get_game_action(snake, action - 1)
            done, _, snake, food, obstacles = game.step(game_action)
            if done:
                break
            else:
                prev_observation = self.generate_observation(snake, food, obstacles)

    def train(self):
        training_data = self.initial_population() # create data 
        #print(training_data)
        #input("hi")
        nn_model = self.model() # create empty model
        nn_model = self.train_model(training_data, nn_model) # train from data collected, store into model 
        self.test_model(nn_model)

    def visualize(self):
        nn_model = self.model()
        nn_model.load(self.filename)
        self.visualize_game(nn_model)

    def test(self):
        nn_model = self.model()
        nn_model.load(self.filename)
        self.test_model(nn_model)

if __name__ == "__main__":
    SnakeNN().test()
    SnakeNN().visualize()