connectx_virtusup_genetic_algorithm.py

# -*- coding: utf-8 -*-
"""ConnectX-VirtusUP.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/16F9nEjQY0mdhpHu39irrao94ZZw-Pgim

# Install kaggle-environments
"""

from kaggle_environments import evaluate, make, utils
import numpy as np
import random
from prettytable import PrettyTable

# Imports do algorítmo genético
import networkx as nx
import matplotlib.pyplot as plt
from matplotlib import lines

from ipywidgets import interact
import ipywidgets as widgets

from collections import deque
import math
import heapq

import time
import bisect

""" ----- Agente ----- """

weights = [2, 6, 100, 100]

class Log:
    moves = []
    
    def add_move(pieces, fit_list):
        move = [pieces, fit_list]
        self.moves.append(move)

    def show_log():
        t = PrettyTable(['Column', 1, 2, 3, 4, 5, 6, 7])
        for move in moves:
            t.add_row(['Pieces'].append(move[0]))
            t.add_row(['Fit list'].append(move[1]))
        print(t)

def show_board(board):
    for row in reversed(board):
        print (row)

def fill_piece_lists(piece_list, piece, row, r):
    temp = [i for i,x in enumerate(row) if x==piece]

    for c in temp:
        piece_list += [(r, c)]

def perception(board):
    ones, twos = [], []

    r = -1

    for row in (board):
        r += 1
        if (1 in row) or (2 in row):
            # Tem peças nessa linha e vamos registrar quais são elas

            fill_piece_lists(ones, 1, row, r)
            fill_piece_lists(twos, 2, row, r)

    return (ones, twos)

def find_seven_row(step, board, target):
    seven_row = [0, 0, 0, 0, 0, 0, 0]

    begin = (target[0] - step[0]*3, target[1] - step[1]*3)

    row_locations = map(lambda x: (begin[0]+step[0]*x, begin[1]+step[1]*x), range(7))

    for row_location, i in zip(row_locations, range(7)):
        if row_location[0] < 0 or row_location[1] < 0 or row_location[0] > 5 or row_location[1] > 6:
            seven_row[i] = -1
        else:
            seven_row[i] = board[row_location[0]][row_location[1]]

    return seven_row

def evaluate_seven_row(seven_row, player, non_valid=-1, fit=0): # Avaliação do seven row por meio de filtros de 4 espaços
    oponent = 2 if player == 1 else 1
    w = weights[0:4]
    
    for i in range(4):
        four_filter = seven_row[i:i+4]
        #print(four_filter, four_filter.count(1))

        if oponent in four_filter or non_valid in four_filter:
            pass
        elif four_filter.count(player) == 2:
            fit += w[0]
        elif four_filter.count(player) == 3:
            fit += w[1]
        elif four_filter.count(player) == 4:
            fit += w[2]
            if player == 1:
                fit += w[3]

    return fit

def alternate_evaluation(state_space, board, target, player=1, fit=0):
    possibilities = [(1, 0), (0, 1), (1, 1), (-1, 1)]

    for possibilitie in possibilities:
        # Linhas de 7 com o target no centro
        seven_row = find_seven_row(possibilitie, board, target)

        fit += evaluate_seven_row(seven_row, player)
        #print (seven_row)
    #print("-------")
    
    return fit

def my_agent(observation, configuration):
    columns, rows, inarow = configuration.columns, configuration.rows, configuration.inarow

    first_play = 1 not in observation.board

    board = [observation.board[:7]] + [observation.board[7:14]] + [observation.board[14:21]] + [observation.board[21:28]] + [observation.board[28:35]] + [observation.board[35:42]]
    board = list(reversed(board))

    state_space = perception(board)
    fit = []

    for c in range(columns):
        # Encontra a linha livre para a coluna selecionada
        r = 0
        while board[r][c] != 0:
            r += 1
            if r == rows:
                break

        if (r < rows):
            temp_fit = 0
            # Colocando uma peça no tabuleiro na coluna c
            state_space[0].append((r, c))
            board[r][c] = 1

            # Avaliar o quanto a jogada leva a vitória
            #temp_fit += evaluation(state_space, board)

            temp_fit += alternate_evaluation(state_space, board, (r, c))
            
            # Evitar que a jogada beneficie o oponente na próxima rodada checando se ele ganha por causa dela
            if r + 1 < 6: 
                board[r+1][c] = 2 
                temp_fit -= alternate_evaluation(state_space, board, (r+1, c), 2)
                board[r+1][c] = 0

            # Tirando peça
            state_space[0].pop()
            board[r][c] = 0

            # Colocando peça oponente
            board[r][c] = 2
            temp_fit += alternate_evaluation(state_space, board, (r, c), 2)

            # Tirando peça
            board[r][c] = 0

            fit.append(temp_fit)
        else:
            fit.append(-100)

    decision = fit.index(max(fit))

    if first_play:
        #decision = random.choice([2, 3, 4])
        decision = 2
        first_play = False

    #print ("---- Fit list ----")
    #print (observation.board)
    #print (fit)
    #print (decision)

    return decision

''' ----- Genetic algorithm ----- '''

gene_pool = []
gene_pool.extend(range (51))

# Genetic algorithm configuration
max_population = 4
mutation_rate = 0.07

# Specific ConnectX configuration
weight_quantity = 4

# Evaluation of fitness
def mean_win_draw(rewards):
    return sum( 1 for r in rewards if (r[0] == 1 or r[0] == 0.)) / len(rewards)

num_ep = 4

def fitness_fn(individual_weights):
    weights = individual_weights
    # Run multiple episodes to estimate its performance.
    vs_random = mean_win_draw(evaluate("connectx", [my_agent, "random"], num_episodes=num_ep))
    vs_negamax = mean_win_draw(evaluate("connectx", [my_agent, "negamax"], num_episodes=num_ep))
    vs_rules = mean_win_draw(evaluate("connectx", [my_agent, "rules"], num_episodes=num_ep))
    vs_greedy = mean_win_draw(evaluate("connectx", [my_agent, "greedy"], num_episodes=num_ep))
    
    return vs_random + vs_negamax + vs_rules + vs_greedy

def init_population(pop_number, gene_pool, state_length):
    """
    pop_number  :  Number of individuals in population
    gene_pool   :  List of possible values for individuals
    state_length:  The length of each individual
    """
    g = len(gene_pool)
    population = []
    for i in range(pop_number):
        new_individual = [gene_pool[random.randrange(0, g)] for j in range(state_length)]
        population.append(new_individual)

    return population

def weighted_sampler(seq, weights):
    """Return a random-sample function that picks from seq weighted by weights."""
    totals = []
    for w in weights:
        totals.append(w + totals[-1] if totals else w)

    return lambda: seq[bisect.bisect(totals, random.uniform(0, totals[-1]))]

def select(r, population, fitness_fn):
    fitnesses = map(fitness_fn, population)
    sampler = weighted_sampler(population, fitnesses)
    return [sampler() for i in range(r)]

def recombine(x, y):
    n = len(x)
    c = random.randrange(0, n)
    return x[:c] + y[c:]

def mutate(x, gene_pool, pmut):
    if random.uniform(0, 1) >= pmut:
        return x

    n = len(x)
    g = len(gene_pool)
    c = random.randrange(0, n)
    r = random.randrange(0, g)

    new_gene = gene_pool[r]
    return x[:c] + [new_gene] + x[c+1:]

ngen = 100 # maximum number of generations
f_thres = 3.0 # maximum fitness is 4.0

argmax = max
argmin = min

def fitness_threshold(fitness_fn, f_thres, population):
    if not f_thres:
        return None

    fittest_individual = argmax(population, key=fitness_fn)
    if fitness_fn(fittest_individual) >= f_thres:
        return fittest_individual

    return None

def genetic_algorithm_stepwise(population, fitness_fn, gene_pool=[0, 1], f_thres=None, ngen=1200, pmut=0.1):
    fitness_history = []
    for generation in range(ngen):
        population = [mutate(recombine(*select(2, population, fitness_fn)), gene_pool, pmut) for i in range(len(population))]
        current_best = max(population, key=fitness_fn)
        print(f'Current best: {current_best}\t\tGeneration: {str(generation)}\t\tFitness: {fitness_fn(current_best)}\r', end='')
        
        # compare the fitness of the current best individual to f_thres
        fitness_history.append(fitness_fn(argmax(population, key=fitness_fn)))
        fittest_individual = fitness_threshold(fitness_fn, f_thres, population)

        # if fitness is greater than or equal to f_thres, we terminate the algorithm
        if fittest_individual:
            return fittest_individual, generation, fitness_history
    return max(population, key=fitness_fn) , generation, fitness_history,


population = init_population(max_population, gene_pool, weight_quantity)
solution, generations, fitness_history = genetic_algorithm_stepwise(population, fitness_fn, gene_pool, f_thres, ngen, mutation_rate)

print("Melhor solução: {}		Generation: {}		Fitness: {}".format(solution, generations, fitness_fn(solution)))
plt.plot(range(len(fitness_history)), fitness_history)