Melee_Notebook.py

# -*- coding: utf-8 -*-
"""Melee.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/16hAmV_0WaROPXu4afM-snBCbo01_7VeO
"""

!pip install tensorflow==2.3.1 keras-rl2 pygame

import pygame
import numpy as np
from tensorflow import keras
from tensorflow.keras.layers import Dense, Input, Flatten, Conv2D
from tensorflow.keras import layers
from tensorflow.keras.models import Model, Sequential
import tensorflow as tf
import random

class Player1(pygame.sprite.Sprite):
    def __init__(self, SCREEN_WIDTH, SCREEN_HEIGHT, model, is_cpu):
        super(Player1, self).__init__()
        self.SCREEN_WIDTH = SCREEN_WIDTH
        self.SCREEN_HEIGHT = SCREEN_HEIGHT
        self.model = model
        self.is_cpu = is_cpu
        self.surf = pygame.Surface((25, 25))
        self.cooldown = 15
        self.health = 10
        self.surf.fill((255, 0, 0))
        self.rect = self.surf.get_rect()

    # Move the sprite based on user keypresses
    def update(self, player, features, action=None):
        upperY = self.rect.top
        lowerY = self.rect.bottom
        upperX = self.rect.right
        lowerX = self.rect.left
        hit = 0

        if self.is_cpu:
            if features is not None:
                features = np.expand_dims(features, axis=0)
                res = self.model.predict(np.asarray(features))
                action = np.argmax(res)

        if action == 0 and upperY > 0:
            self.rect.move_ip(0, -10)
        if action == 1 and lowerY < self.SCREEN_HEIGHT:
            self.rect.move_ip(0, 10)
        if action == 2 and lowerX > 0:
            self.rect.move_ip(-10, 0)
        if action == 3 and upperX < self.SCREEN_WIDTH:
            self.rect.move_ip(10, 0)
        if action == 4 and self.cooldown == 15:
            self.cooldown = 0
            player_pos = self.rect.center
            other_player_pos = player.rect.center
            x_diff = abs(player_pos[0] - other_player_pos[0])
            y_diff = abs(player_pos[1] - other_player_pos[1])
            hit = 2
            if x_diff < 50 and y_diff < 50:
                player.health -= 1
                player.surf.fill((255,255,255))
                hit = 1

        if self.cooldown < 15:
            self.cooldown += 1
        return hit

class Player2(pygame.sprite.Sprite):
    def __init__(self, SCREEN_WIDTH, SCREEN_HEIGHT, model, is_cpu):
        super(Player2, self).__init__()
        self.SCREEN_WIDTH = SCREEN_WIDTH
        self.SCREEN_HEIGHT = SCREEN_HEIGHT
        self.model = model
        self.is_cpu = is_cpu
        self.surf = pygame.Surface((25, 25))
        self.cooldown = 15
        self.health = 10
        self.surf.fill((0, 0, 255))
        self.rect = self.surf.get_rect()
        self.rect.move_ip(SCREEN_WIDTH-25, 0)

    # Move the sprite based on user keypresses
    def update(self, player, features, action=None):
        upperY = self.rect.top
        lowerY = self.rect.bottom
        upperX = self.rect.right
        lowerX = self.rect.left
        hit = 0

        # action = random.choice([0, 1, 2, 3, 4, 5])
        # action = 4
        if self.is_cpu:
            if features is not None:
                features = np.expand_dims(features, axis=0)
                res = self.model.predict(np.asarray(features))
                action = np.argmax(res)
        
        if action == 0 and upperY > 0:
            self.rect.move_ip(0, -10)
        elif action == 1 and lowerY < self.SCREEN_HEIGHT:
            self.rect.move_ip(0, 10)
        elif action == 2 and upperX < self.SCREEN_WIDTH:
            self.rect.move_ip(10, 0)
        elif action == 3 and lowerX > 0:
            self.rect.move_ip(-10, 0)
        elif action == 4 and self.cooldown == 15:
            self.cooldown = 0
            player_pos = self.rect.center
            other_player_pos = player.rect.center
            x_diff = abs(player_pos[0] - other_player_pos[0])
            y_diff = abs(player_pos[1] - other_player_pos[1])
            hit = 2
            if x_diff < 50 and y_diff < 50:
                player.surf.fill((255,255,255))
                player.health -= 1
                hit = 1

        if self.cooldown < 15:
            self.cooldown += 1
        return hit

class Melee():
    def __init__(self, models, is_cpu):
        self.SCREEN_WIDTH = 500 // 2
        self.SCREEN_HEIGHT = 300 // 2
        self.is_game_over = False
        self.attack_range = 50
        self.models = models
        self.is_cpu = is_cpu

        self.past_frames = []

        self.player1 = Player1(self.SCREEN_WIDTH, self.SCREEN_HEIGHT, models[0], is_cpu[0])
        self.player2 = Player2(self.SCREEN_WIDTH, self.SCREEN_HEIGHT, models[1], is_cpu[1])

        self.all_sprites = pygame.sprite.Group()
        self.players = pygame.sprite.Group()
        self.all_sprites.add(self.player1)
        self.all_sprites.add(self.player2)
        self.players.add(self.player1)
        self.players.add(self.player2)

    def is_in_range(self):
        player_pos = self.player1.rect.center
        other_player_pos = self.player2.rect.center
        x_diff = abs(player_pos[0] - other_player_pos[0])
        y_diff = abs(player_pos[1] - other_player_pos[1])
        if x_diff < 50 and y_diff < 50:
            return 1
        return 0

    def game_over(self):
        return self.is_game_over

    def get_score(self):
        return self.player1.health

    def get_env(self):
        features = []
        features.append((self.player1.rect.centerx - self.player2.rect.centerx) / self.SCREEN_WIDTH)
        features.append((self.player1.rect.centery - self.player2.rect.centery) / self.SCREEN_HEIGHT)
        features.append(self.player1.rect.centerx / self.SCREEN_WIDTH)
        features.append(self.player2.rect.centerx / self.SCREEN_WIDTH)
        features.append(self.player1.rect.centery / self.SCREEN_HEIGHT)
        features.append(self.player2.rect.centery / self.SCREEN_HEIGHT)

        features.append(self.is_in_range())
        features.append(self.player1.cooldown / 15)
        features.append(self.player2.cooldown / 15)
        return features

    def get_features(self):
        features = self.get_env()
        num_past_frames = 8
        if len(self.past_frames) == num_past_frames:
            self.past_frames.pop(0)
        self.past_frames.append(features)
        if len(self.past_frames) == num_past_frames:
            return self.past_frames
        return None

    def step(self, action):
        reward = 0
        features = self.get_features()
        is_damaged = self.player1.update(self.player2, features)
        is_dealt = self.player2.update(self.player1, features, action)

        if is_dealt == 1:
          reward += 50
        elif is_damaged == 1:
          reward -= 50
        # elif self.is_in_range() == 1:
        #   reward += 0.01
        # else:
        #   reward -= 0.001

        if self.player1.health == 0 or self.player2.health == 0:
          self.is_game_over = True

        if self.player1.cooldown == 15:
            self.player2.surf.fill((0, 0, 255))
        if self.player2.cooldown == 15:
            self.player1.surf.fill((255, 0, 0))

        done = self.is_game_over 
        info = {}
        return self.get_env(), reward, done, info

    def reset(self):
        self.is_game_over = False
        self.past_frames = []

        for sprite in self.all_sprites:
          sprite.kill()

        self.player1 = Player1(self.SCREEN_WIDTH, self.SCREEN_HEIGHT, self.models[0], self.is_cpu[0])
        self.player2 = Player2(self.SCREEN_WIDTH, self.SCREEN_HEIGHT, self.models[1], self.is_cpu[1])
        self.all_sprites = pygame.sprite.Group()
        self.players = pygame.sprite.Group()
        self.all_sprites.add(self.player1)
        self.all_sprites.add(self.player2)
        self.players.add(self.player1)
        self.players.add(self.player2)
        return self.get_env()

    def render(self):
      pass

    def close(self):
      pass

del model, dqn, game

model = keras.models.load_model('gen_3')
models = [model, None]
is_cpu = [True, False]
game = Melee(models, is_cpu)

num_classes = 6
window_length = 8
num_features = 9

input = Input(shape=(window_length, num_features,))
x = input
x = Flatten()(x)
x = Dense(32, activation="relu")(x)
x = Dense(16, activation="relu")(x)
x = Dense(8, activation="relu")(x)
x = Dense(num_classes, activation="softmax")(x)
model = Model(inputs=input, outputs=x)

from rl.agents import DQNAgent, NAFAgent
from rl.memory import SequentialMemory
from rl.policy import LinearAnnealedPolicy, EpsGreedyQPolicy

policy = LinearAnnealedPolicy(EpsGreedyQPolicy(), attr='eps', value_max=1., value_min=0.1, value_test=.2, nb_steps=200000)
memory = SequentialMemory(limit=1000, window_length=window_length)
dqn = DQNAgent(model=model, memory=memory, policy=policy, nb_actions=num_classes, nb_steps_warmup=1000, enable_dueling_network=True, dueling_type='avg')
dqn.compile(tf.keras.optimizers.Adam(learning_rate=1e-4))
dqn.fit(game, nb_steps=200000, visualize=False, verbose=1)

scores = dqn.test(game, nb_episodes=5, visualize=False)
print(np.mean(scores.history['episode_reward']))

episodes = 5
model = keras.models.load_model('gen_1')
models = [model, None]
is_cpu = [True, False]
game = Melee(models, is_cpu)
for episode in range(1, episodes+1):
    state = game.reset()
    done = False
    score = 0 
    steps = 0
    while not done:
        game.render()
        action = random.choice([0,1,2,3,4,5])
        # action = 1
        n_state, reward, done, info = game.step(action)
        score+=reward
        steps += 1
    print('Episode:{} Score:{} Steps:{}'.format(episode, score, steps))

game.close()

model = dqn.model
model.save('SavedKerasWeights/gen_4')