train_bot.py

from keras.models import Sequential
from keras.layers import Dense
from keras.callbacks import EarlyStopping

import pickle
import numpy as np
import json

def get_relative_frame(frame, position, new_position=(0,0)):
    """ Rolls a frame such that position is moved to new_position
    Input:
        frame: 2-dimensional array
        position: tuple (row, col)
        new_position: tuple (row,col)
    Output:
        np.array
    e.g.
        a = np.arange(9).reshape(3,3)
        # array([[0,1,2],
                 [3,4,5],
                 [6,7,8]])
        get_relative_frame(a, (1,1), new_position=(0,0))
        # array([[4,5,3],
                 [7,8,6],
                 [1,2,0]])
    """
    r, c = position
    new_r, new_c = new_position
    diff_r, diff_c = new_r - r, new_c - c
    return np.roll(np.roll(frame, diff_r, axis=0), diff_c, axis=1)

def game_generator(height, width, player_name, game_dir, only_moves=True, first_n_frames=200, positions_per_frame=100, skip_first_n=0):
    game_ids = [game for game in os.listdir("{}/".format(game_dir)) if game.endswith(".hlt")]
    ct = 0

    while True:
        for game_id in game_ids:
            with open("{}/{}".format(game_dir,game_id), 'r') as f:
                game = json.loads(f.read())
                f.close()

            player = game["player_names"].index(player_name) + 1
            
            productions = game["productions"]
            productions = np.array(productions)
            productions = productions / 20.
            
            for idx in range(min(len(game["moves"]),first_n_frames)):
                frame = np.array(game["frames"][idx])
                moves = np.array(game["moves"][idx])
                
                owners = frame[:,:,0]
                player_mask = owners == player
                enemy_mask = (owners != player) & (owners != 0)
                open_mask = owners == 0
                
                strengths = frame[:,:,1]
                strengths = np.array(strengths)
                strengths = strengths / 255.
                
                player_strength = strengths * player_mask
                enemy_strength = strengths * enemy_mask
                open_strength = strengths * open_mask
                
                if np.sum(player_mask) == 0:
                    continue # Player not in this frame

                positions = np.where(player_mask == True)
                positions = list(zip(*positions))
                position_idxes = np.random.choice(range(len(positions)), size=min(len(positions),positions_per_frame), replace=False)
                
                for position_idx in position_idxes:
                    ct += 1

                    position = positions[position_idx]
                    r,c = position

                    # Ignore if not square not moved
                    if only_moves and moves[r][c] == 0:
                        continue
                    
                    # Skip if part of validation set
                    if ct <= skip_first_n:
                        continue

                    new_y = height//2
                    new_x = width//2
                    
                    relative_player_strength = get_relative_frame(player_strength, position, new_position=(new_y, new_x))[:height, :width].ravel()
                    relative_enemy_strength = get_relative_frame(enemy_strength, position, new_position=(new_y, new_x))[:height, :width].ravel()
                    relative_open_strength = get_relative_frame(open_strength, position, new_position=(new_y, new_x))[:height, :width].ravel()
                    relative_productions = get_relative_frame(productions, position, new_position=(new_y, new_x))[:height, :width].ravel()
                    relative_combined = np.hstack([relative_player_strength, relative_enemy_strength, relative_open_strength, relative_productions])

                    if relative_combined.size == width * height * 4:
                        # At least width * height
                        if only_moves:
                            one_hot_move = np.zeros(5)
                            one_hot_move[moves[r][c]] = 1
                        else:
                            one_hot_move = np.zeros(2)
                            if moves[r][c] == 0:
                                one_hot_move[0] = 1
                            else:
                                one_hot_move[1] = 1
                        yield((relative_combined.reshape(1,-1), one_hot_move.reshape(1,-1)))


height = 9
width = 9
player_name = "djma v3"
game_dir = "games/djma_v3"
len_val_data = 100000

# generator used for validation set
g=game_generator(height,width,player_name=player_name,game_dir=game_dir)
input_dim = height*width*4

validation_data = [next(g) for _ in range(len_val_data)]
X_val, y_val = zip(*validation_data)
y_val = np.array([y.ravel() for y in y_val])
X_val = np.array([X.ravel() for X in X_val])

# generator for game
g=game_generator(height,width,player_name=player_name,game_dir=game_dir,skip_first_n=len_val_data)

move_model = Sequential([Dense(512, activation='relu', input_dim=input_dim),
            Dense(512, activation='relu'),
            # Dense(128, activation='relu'),
            Dense(5, activation='softmax')])

move_model.compile('nadam','categorical_crossentropy', metrics=['accuracy'])

move_model.fit_generator(g, samples_per_epoch=1000, nb_epoch=1000,callbacks=[EarlyStopping(patience=10)], validation_data=[X_val,y_val])