added gomokubot

Gomoku_Bot/HumanVSAI.py

@@ -0,0 +1,54 @@
+import os
+import sys
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from .board import Board
+from .minmax import vct, cache_hits, minmax
+from .eval import FIVE, FOUR, performance
+
+
+class Game():
+    def __init__(self, firstRole=1):
+        self.board = Board(8, firstRole)
+        self.steps = []
+        self.step = 0
+        self.enableVCT = True  # 是否开启算杀, 算杀会在某些leaf节点加深搜索, 但是不一定会增加搜索时间
+
+    def human_input(self):
+        x, y = map(int, input('Your move: ').split())
+        return x, y
+
+    def start_play(self, human_first=False):
+        if not human_first:
+            while not self.board.isGameOver():
+                print(self.board.display())
+                if self.step % 2 == 1:
+                    x, y = self.human_input()
+                    while not self.board.put(x, y):
+                        x, y = self.human_input()
+                else:
+                    score = minmax(self.board, 1, 4, enableVCT=self.enableVCT)
+                    print(score)
+                    x, y = score[1]
+                    print("move at", x, y)
+                    self.board.put(x, y)
+                self.step += 1
+        else:
+            while not self.board.isGameOver():
+                print(self.board.display())
+                if self.step % 2 == 0:
+                    x, y = self.human_input()
+                    while not self.board.put(x, y):
+                        x, y = self.human_input()
+                else:
+                    score = minmax(self.board, -1, 4, enableVCT=self.enableVCT)
+                    print(score)
+                    x, y = score[1]
+                    self.board.put(x, y)
+                self.step += 1
+        print(self.board.display())
+
+
+if __name__ == '__main__':
+    game = Game()
+    game.start_play(True)

Gomoku_Bot/__init__.py

@@ -0,0 +1,2 @@
+from .board import Board
+from .gomoku_bot import Gomoku_bot

Gomoku_Bot/board.py

@@ -0,0 +1,216 @@
+from .zobrist import ZobristCache as Zobrist
+from .cache import Cache
+from .eval import Evaluate, FIVE
+from scipy import signal
+import pickle
+import os
+save_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'train_data/data', 'train_data.pkl')
+
+if 'numpy' not in globals():
+    import numpy as np
+
+
+class Board:
+    def __init__(self, size=15, firstRole=1):
+        self.size = size
+        self.board = [[0] * self.size for _ in range(self.size)]
+        self.firstRole = firstRole  # 1 for black, -1 for white
+        self.role = firstRole  # 1 for black, -1 for white
+        self.history = []
+        self.zobrist = Zobrist(self.size)
+        self.winnerCache = Cache()
+        self.gameoverCache = Cache()
+        self.evaluateCache = Cache()
+        self.valuableMovesCache = Cache()
+        self.evaluateTime = 0
+        self.evaluator = Evaluate(self.size)
+        self.available = [(i, j) for i in range(self.size) for j in range(self.size)]
+        self.patterns = [np.ones((1, 5)), np.ones((5, 1)), np.eye(5), np.fliplr(np.eye(5))]
+        self.train_data = {1:[], -1: []}
+        if os.path.exists(save_path):
+            with open(save_path, 'rb') as f:
+                self.train_data = pickle.load(f)
+
+    def isGameOver(self):
+        # Checked
+        hash = self.hash()
+        if self.gameoverCache.get(hash):
+            return self.gameoverCache.get(hash)
+        if self.getWinner() != 0:
+            self.gameoverCache.put(hash, True)
+            # save train data
+            # with open(save_path, 'wb') as f:
+            #     pickle.dump(self.train_data, f)
+            return True  # Someone has won
+        # Game is over when there is no empty space on the board or someone has won
+        if len(self.history) == self.size ** 2:
+            self.gameoverCache.put(hash, True)
+            return True
+        else:
+            self.gameoverCache.put(hash, False)
+            return False
+
+    def getWinner(self):
+        # Checked
+        hash = self.hash()
+        flag = True
+        if self.winnerCache.get(hash):
+            return self.winnerCache.get(hash)
+        directions = [[1, 0], [0, 1], [1, 1], [1, -1]]  # Horizontal, Vertical, Diagonal
+        for i in range(self.size):
+            for j in range(self.size):
+                if self.board[i][j] == 0:
+                    flag = False
+                    continue
+                for direction in directions:
+                    count = 0
+                    while (
+                            0 <= i + direction[0] * count < self.size and
+                            0 <= j + direction[1] * count < self.size and
+                            self.board[i + direction[0] * count][j + direction[1] * count] == self.board[i][j]
+                    ):
+                        count += 1
+                    if count >= 5:
+                        self.winnerCache.put(hash, self.board[i][j])
+                        return self.board[i][j]
+        if flag:
+            print("tie!!!")
+            return 0
+        self.winnerCache.put(hash, 0)
+        return 0
+
+    def getValidMoves(self):
+        return self.available
+
+    def put(self, i, j, role=None):
+        # Checked
+        if role is None:
+            role = self.role
+        if not isinstance(i, int) or not isinstance(j, int):
+            print("Invalid move: Not Number!", i, j)
+            return False
+        if self.board[i][j] != 0:
+            print("Invalid move!", i, j)
+            return False
+        self.board[i][j] = role
+        self.available.remove((i, j))
+        self.history.append({"i": i, "j": j, "role": role})
+        self.zobrist.togglePiece(i, j, role)
+        self.evaluator.move(i, j, role)
+        self.role *= -1  # Switch role
+        return True
+
+    def undo(self):
+        # Checked
+        if len(self.history) == 0:
+            print("No moves to undo!")
+            return False
+
+        lastMove = self.history.pop()
+        self.board[lastMove['i']][lastMove['j']] = 0  # Remove the piece from the board
+        self.role = lastMove['role']  # Switch back to the previous player
+        self.zobrist.togglePiece(lastMove['i'], lastMove['j'], lastMove['role'])
+        self.evaluator.undo(lastMove['i'], lastMove['j'])
+        self.available.append((lastMove['i'], lastMove['j']))
+        return True
+
+    def position2coordinate(self, position):
+        # checked
+        row = position // self.size
+        col = position % self.size
+        return [row, col]
+
+    def coordinate2position(self, coordinate):
+        # Checked
+        return coordinate[0] * self.size + coordinate[1]
+
+    def getValuableMoves(self, role, depth=0, onlyThree=False, onlyFour=False):
+        # Checked
+        hash = self.hash()
+        prev = self.valuableMovesCache.get(hash)
+        if prev:
+            if (prev["role"] == role and
+                    prev["depth"] == depth and
+                    prev["onlyThree"] == onlyThree
+                    and prev["onlyFour"] == onlyFour):
+                return prev["moves"]
+
+        moves, train_data = self.evaluator.getMoves(role, depth, onlyThree, onlyFour)
+        self.train_data[self.role].append(train_data)
+        # Handle a special case, if the center point is not occupied, add it by default
+
+        # 开局的时候随机走一步，增加开局的多样性
+        if not onlyThree and not onlyFour:
+            center = self.size // 2
+            if self.board[center][center] == 0:
+                moves.append((center, center))
+
+            # x_step = np.random.randint(-self.size // 2, self.size // 2)
+            # y_step = np.random.randint(-self.size // 2, self.size // 2)
+            # x = center + x_step
+            # y = center + y_step
+            # if 0 <= x < self.size and 0 <= y < self.size and self.board[x][y] == 0:
+            #     moves.append((x, y))
+
+        self.valuableMovesCache.put(hash, {
+            "role": role,
+            "moves": moves,
+            "depth": depth,
+            "onlyThree": onlyThree,
+            "onlyFour": onlyFour
+        })
+        return moves
+
+    def display(self, extraPoints=[]):
+        # Checked
+        extraPosition = [self.coordinate2position(point) for point in extraPoints]
+        result = ""
+        for i in range(self.size):
+            for j in range(self.size):
+                position = self.coordinate2position([i, j])
+                if position in extraPosition:
+                    result += "? "
+                    continue
+                value = self.board[i][j]
+                if value == 1:
+                    result += "B " # Black
+                elif value == -1:
+                    result += "W " # White
+                else:
+                    result += "- "
+            result += "\n"
+        return result
+
+    def hash(self):
+        # Checked
+        return self.zobrist.getHash()  # Return the hash value of the current board, used for caching
+
+    def evaluate(self, role):
+        # Checked
+        hash_key = self.hash()
+        prev = self.evaluateCache.get(hash_key)
+        if prev:
+            if prev["role"] == role:
+                return prev["score"]
+
+        winner = self.getWinner()
+        score = 0
+        if winner != 0:
+            score = FIVE * winner * role
+        else:
+            score = self.evaluator.evaluate(role)
+
+        self.evaluateCache.put(hash_key, {"role": role, "score": score})
+        return score
+
+    def reverse(self):
+        # Checked
+        new_board = Board(self.size, -self.firstRole)
+        for move in self.history:
+            x, y, role = move['i'], move['j'], move['role']
+            new_board.put(x, y, -role)
+        return new_board
+
+    def toString(self):
+        # Checked
+        return ''.join([''.join(map(str, row)) for row in self.board])

Gomoku_Bot/board_manuls.py

@@ -0,0 +1,117 @@
+wins = [
+  # O O O O O
+  # X X X X -
+  # - - - - -
+  # - - - - -
+  # - - - - -
+  [5, [0, 5, 1, 6, 2, 7, 3, 8, 4], 1],  # 横向五
+  # O O O O -
+  # X X X X X
+  # O - - - -
+  # - - - - -
+  # - - - - -
+  [5, [0, 5, 1, 6, 2, 7, 3, 8, 10, 9], -1],  # 白子横向五
+  # O O - O O
+  # X X X X -
+  # O - - - -
+  # - - - - -
+  # - - - - -
+  [5, [0, 5, 1, 6, 10, 7, 3, 8, 4], 0],  # 有一个空位
+  # O O O X O
+  # X X X X -
+  # O - - - -
+  # - - - - -
+  # - - - - -
+  [5, [0, 5, 1, 6, 2, 7, 10, 8, 4], 0],  # 有一个白子
+  # O X X X X
+  # O - - - -
+  # O - - - -
+  # O - - - -
+  # O - - - -
+  [5, [0, 1, 5, 2, 10, 3, 15, 4, 20], 1],  # 纵向五
+  # O X X X X
+  # O - - - -
+  # O - - - -
+  # - O - - -
+  # O - - - -
+  [5, [0, 1, 5, 2, 10, 3, 16, 4, 20], 0],  # 纵向五有一个空位
+  # O X X X X
+  # O - - - -
+  # O - - - -
+  # X O - - -
+  # O - - - -
+  [5, [0, 1, 5, 2, 10, 3, 16, 4, 20, 15], 0],  # 纵向五有一个白子
+  # O X X X X
+  # - O - - -
+  # - - O - -
+  # - - - O -
+  # - - - - O
+  [5, [0, 1, 6, 2, 12, 3, 18, 4, 24], 1],  # 斜线五
+  # O X X X X
+  # - O - - -
+  # - - - O -
+  # - - - O -
+  # - - - - O
+  [5, [0, 1, 6, 2, 12, 3, 19, 4, 24], 0],  # 斜线五有一个空的
+  # O X X X X
+  # - O - - -
+  # - - X O -
+  # - - - O -
+  # - - - - O
+  [5, [0, 1, 6, 2, 12, 3, 19, 4, 24, 18], 0],  # 斜线五有一个白子
+  # X X X X O
+  # - - - O -
+  # - - O - -
+  # - O - - -
+  # O - - - -
+  [5, [4, 0, 8, 1, 12, 2, 16, 3, 20], 1],  # 反斜线五
+  # X X X X O
+  # - - - O -
+  # - - O - -
+  # O - - - -
+  # O - - - -
+  [5, [4, 0, 8, 1, 12, 2, 15, 3, 20], 0],  # 反斜线五 有一个空位
+  # X X X - O
+  # - - - O -
+  # - - O - -
+  # - X - - -
+  # O - - - -
+  [5, [4, 0, 8, 1, 12, 2, 16, 20], 0],  # 反斜线五 有一个空位
+]
+
+# valid moves
+validMoves = [
+  # O - - 
+  # - - - 
+  # - - O 
+  [3, [0, 8], [1, 2, 3, 4, 5, 6, 7]],
+  # O - - - -
+  # - - - - -
+  # - - - - -
+  # - - - - -
+  # - - - - -
+  [5, [0], [1, 2, 5, 6, 7, 8, 10, 11, 12, 13, 16, 17, 18]],
+  # - - - - -
+  # - - - - -
+  # - O - - -
+  # - - - - -
+  # - - - - -
+  [5, [11], [0, 1, 2, 3, 5, 6, 7, 8, 10, 12, 13, 15, 16, 17, 18, 20, 21, 22, 23]],
+
+  # - - - - - - - -
+  # - - - - - - - -
+  # - - - - - - - -
+  # - - O - X - - -
+  # - - - - - - - -
+  # - - - - - - - -
+  # - - - - - - - -
+  # - - - - - - - -
+  [8, [26, 28], [
+    8, 9, 10, 11, 12, 13, 14,
+    16, 17, 18, 19, 20, 21, 22,
+    24, 25, 27, 29, 30,
+    32, 33, 34, 35, 36, 37, 38,
+    40, 41, 42, 43, 44, 45, 46,
+    ],
+  ],
+]

Gomoku_Bot/cache.py

@@ -0,0 +1,23 @@
+from cachetools import LRUCache
+from .config import config
+
+class Cache:
+    def __init__(self, capacity=1000000):
+        self.capacity = capacity
+        self.cache = LRUCache(maxsize=capacity)
+        self.enable_cache = config['enableCache']
+
+    def get(self, key):
+        if not self.enable_cache:
+            return None
+        return self.cache.get(key, None)
+
+    def put(self, key, value):
+        if not self.enable_cache:
+            return
+        self.cache[key] = value
+
+    def has(self, key):
+        if not self.enable_cache:
+            return False
+        return key in self.cache

Gomoku_Bot/config.py

@@ -0,0 +1,6 @@
+config = {
+  "enableCache": True,       # Whether to enable caching
+  "onlyInLine": False,       # Whether to search only on a single line, an optimization option
+  "inlineCount": 4,          # Number of recent points to consider for being on the same line
+  "inLineDistance": 5        # Maximum distance to determine if a point is on the same line
+}

Gomoku_Bot/eval.py

@@ -0,0 +1,588 @@
+from enum import Enum
+import math
+from .shape import shapes, get_shape_fast, is_five, is_four, get_all_shapes_of_point
+from .position import coordinate2Position, isLine, isAllInLine, hasInLine, position2Coordinate
+from .config import config
+from datetime import datetime
+import os
+
+dir_path = os.path.dirname(os.path.abspath(__file__))
+import numpy as np
+import torch
+# from .minimax_Net import BoardEvaluationNet as net
+
+# mini_max_net = net(board_size=15)
+# mini_max_net.load_state_dict(torch.load(os.path.join(dir_path, 'train_data/model', 'best_loss=609.3356355479785.pth')))
+# mini_max_net.eval()
+
+
+# Enum to represent different shapes
+class Shapes(Enum):
+    FIVE = 0
+    BLOCK_FIVE = 1
+    FOUR = 2
+    FOUR_FOUR = 3
+    FOUR_THREE = 4
+    THREE_THREE = 5
+    BLOCK_FOUR = 6
+    THREE = 7
+    BLOCK_THREE = 8
+    TWO_TWO = 9
+    TWO = 10
+    NONE = 11
+
+
+# Constants representing scores for each shape
+FIVE = 10000000
+BLOCK_FIVE = FIVE
+FOUR = 100000
+FOUR_FOUR = FOUR  # 双冲四
+FOUR_THREE = FOUR  # 冲四活三
+THREE_THREE = FOUR / 2  # 双活三
+BLOCK_FOUR = 1500
+THREE = 1000
+BLOCK_THREE = 150
+TWO_TWO = 200  # 双活二
+TWO = 100
+BLOCK_TWO = 15
+ONE = 10
+BLOCK_ONE = 1
+
+
+# Function to calculate the real shape score based on the shape
+def getRealShapeScore(shape: Shapes) -> int:
+    # Checked
+    if shape == shapes['FIVE']:
+        return FOUR
+    elif shape == shapes['BLOCK_FIVE']:
+        return BLOCK_FOUR
+    elif shape in [shapes['FOUR'], shapes['FOUR_FOUR'], shapes['FOUR_THREE']]:
+        return THREE
+    elif shape == shapes['BLOCK_FOUR']:
+        return BLOCK_THREE
+    elif shape == shapes['THREE']:
+        return TWO
+    elif shape == shapes['THREE_THREE']:
+        return math.floor(THREE_THREE / 10)
+    elif shape == shapes['BLOCK_THREE']:
+        return BLOCK_TWO
+    elif shape == shapes['TWO']:
+        return ONE
+    elif shape == shapes['TWO_TWO']:
+        return math.floor(TWO_TWO / 10)
+    else:
+        return 0
+
+
+# List of all directions
+allDirections = [
+    [0, 1],  # Horizontal
+    [1, 0],  # Vertical
+    [1, 1],  # Diagonal \
+    [1, -1]  # Diagonal /
+]
+
+
+# Function to get the index of a direction
+def direction2index(ox: int, oy: int) -> int:
+    # Checked
+    if ox == 0:
+        return 0  # |
+    elif oy == 0:
+        return 1  # -
+    elif ox == oy:
+        return 2  # \
+    elif ox != oy:
+        return 3  # /
+
+
+# Performance dictionary
+performance = {
+    "updateTime": 0,
+    "getPointsTime": 0
+}
+
+
+class Evaluate:
+    def __init__(self, size=15):
+        # Checked
+        self.size = size
+        self.board = [[2] * (size + 2) for _ in range(size + 2)]
+        for i in range(size + 2):
+            for j in range(size + 2):
+                if i == 0 or j == 0 or i == self.size + 1 or j == self.size + 1:
+                    self.board[i][j] = 2
+                else:
+                    self.board[i][j] = 0
+        self.blackScores = [[0] * self.size for _ in range(size)]
+        self.whiteScores = [[0] * self.size for _ in range(size)]
+        self.initPoints()
+        self.history = []  # List of [position, role]
+
+    def move(self, x, y, role):
+        # Checked
+        # Clear the cache first
+        for d in [0, 1, 2, 3]:
+            self.shapeCache[role][d][x][y] = 0
+            self.shapeCache[-role][d][x][y] = 0
+        self.blackScores[x][y] = 0
+        self.whiteScores[x][y] = 0
+        # Update the board
+        self.board[x + 1][y + 1] = role  ## Adjust for the added wall
+        self.updatePoint(x, y)
+        self.history.append([coordinate2Position(x, y, self.size), role])
+
+    def undo(self, x, y):
+        # Checked
+        self.board[x + 1][y + 1] = 0
+        self.updatePoint(x, y)
+        self.history.pop()
+
+    def initPoints(self):
+        # Checked
+        # Initialize the cache, avoid calculating the same points multiple times
+        self.shapeCache = {}
+        for role in [1, -1]:
+            self.shapeCache[role] = {}
+            for direction in [0, 1, 2, 3]:
+                self.shapeCache[role][direction] = [[0] * self.size for _ in range(self.size)]
+
+        self.pointsCache = {}
+        for role in [1, -1]:
+            self.pointsCache[role] = {}
+            for shape in shapes:
+                self.pointsCache[role][shape] = set()
+
+    def getPointsInLine(self, role):
+        # Checked
+        pointsInLine = {}
+        hasPointsInLine = False
+        for key in shapes:
+            pointsInLine[shapes[key]] = set()
+
+        last2Points = [position for position, role in self.history[-config['inlineCount']:]]
+        processed = {}
+        # 在last2Points中查找是否有点位在一条线上
+        for r in [role, -role]:
+            for point in last2Points:
+                x, y = position2Coordinate(point, self.size)
+                for ox, oy in allDirections:
+                    for sign in [1, -1]:
+                        for step in range(1, config['inLineDistance'] + 1):
+                            nx = x + sign * step * ox
+                            ny = y + sign * step * oy
+                            position = coordinate2Position(nx, ny, self.size)
+                            # 检测是否到达边界
+                            if nx < 0 or nx >= self.size or ny < 0 or ny >= self.size:
+                                break
+                            if self.board[nx + 1][ny + 1] != 0:
+                                continue
+                            if processed.get(position) == r:
+                                continue
+                            processed[position] = r
+                            for direction in [0, 1, 2, 3]:
+                                shape = self.shapeCache[r][direction][nx][ny]
+                                # 到达边界停止，但是注意到达对方棋子不能停止
+                                if shape:
+                                    pointsInLine[shape].add(coordinate2Position(nx, ny, self.size))
+                                    hasPointsInLine = True
+
+        if hasPointsInLine:
+            return pointsInLine
+        return False
+
+    def getPoints(self, role, depth, vct, vcf):
+        first = role if depth % 2 == 0 else -role  # 先手
+        start = datetime.now()
+
+        if config['onlyInLine'] and len(self.history) >= config['inlineCount']:
+            points_in_line = self.getPointsInLine(role)
+            if points_in_line:
+                performance['getPointsTime'] += (datetime.now() - start).total_seconds()
+                return points_in_line
+
+        points = {}  # 全部点位
+
+        for key in shapes.keys():
+            points[shapes[key]] = set()
+
+        last_points = [position for position, _ in self.history[-4:]]
+
+        for r in [role, -role]:
+            # 这里是直接遍历了这个棋盘上的所有点位，如果棋盘很大，这里会有性能问题；可以用神经网络来预测
+            for i in range(self.size):
+                for j in range(self.size):
+                    four_count = 0
+                    block_four_count = 0
+                    three_count = 0
+
+                    for direction in [0, 1, 2, 3]:
+                        if self.board[i + 1][j + 1] != 0:
+                            continue
+
+                        shape = self.shapeCache[r][direction][i][j]
+
+                        if not shape:
+                            continue
+
+                        point = i * self.size + j
+
+                        if vcf:
+                            if r == first and not is_four(shape) and not is_five(shape):
+                                continue
+                            if r == -first and is_five(shape):
+                                continue
+
+                        if vct:
+                            if depth % 2 == 0:
+                                if depth == 0 and r != first:
+                                    continue
+                                if shape != shapes['THREE'] and not is_four(shape) and not is_five(shape):
+                                    continue
+                                if shape == shapes['THREE'] and r != first:
+                                    continue
+                                if depth == 0 and r != first:
+                                    continue
+                                if depth > 0:
+                                    if shape == shapes['THREE'] and len(
+                                            get_all_shapes_of_point(self.shapeCache, i, j, r)) == 1:
+                                        continue
+                                    if shape == shapes['BLOCK_FOUR'] and len(
+                                            get_all_shapes_of_point(self.shapeCache, i, j, r)) == 1:
+                                        continue
+                            else:
+                                if shape != shapes['THREE'] and not is_four(shape) and not is_five(shape):
+                                    continue
+                                if shape == shapes['THREE'] and r == -first:
+                                    continue
+                                if depth > 1:
+                                    if shape == shapes['BLOCK_FOUR'] and len(
+                                            get_all_shapes_of_point(self.shapeCache, i, j)) == 1:
+                                        continue
+                                    if shape == shapes['BLOCK_FOUR'] and not hasInLine(point, last_points, self.size):
+                                        continue
+
+                        if vcf:
+                            if not is_four(shape) and not is_five(shape):
+                                continue
+
+                        if depth > 2 and (shape == shapes['TWO'] or shape == shapes['TWO_TWO'] or shape == shapes[
+                            'BLOCK_THREE']) and not hasInLine(point, last_points, self.size):
+                            continue
+
+                        points[shape].add(point)
+
+                        if shape == shapes['FOUR']:
+                            four_count += 1
+                        elif shape == shapes['BLOCK_FOUR']:
+                            block_four_count += 1
+                        elif shape == shapes['THREE']:
+                            three_count += 1
+
+                        union_shape = None
+
+                        if four_count >= 2:
+                            union_shape = shapes['FOUR_FOUR']
+                        elif block_four_count and three_count:
+                            union_shape = shapes['FOUR_THREE']
+                        elif three_count >= 2:
+                            union_shape = shapes['THREE_THREE']
+
+                        if union_shape:
+                            points[union_shape].add(point)
+
+        performance['getPointsTime'] += (datetime.now() - start).total_seconds()
+
+        return points
+
+    """
+      当一个位置发生变时候，要更新这个位置的四个方向上得分，更新规则是：
+        1. 如果这个位置是空的，那么就重新计算这个位置的得分
+        2. 如果碰到了边界或者对方的棋子，那么就停止计算
+        3. 如果超过2个空位，那么就停止计算
+        4. 要更新自己的和对方的得分
+    """
+
+    def updatePoint(self, x, y):
+        # Checked
+        start = datetime.now()
+        self.updateSinglePoint(x, y, 1)
+        self.updateSinglePoint(x, y, -1)
+
+        for ox, oy in allDirections:
+            for sign in [1, -1]:  # -1 for negative direction, 1 for positive direction
+                for step in range(1, 6):
+                    reachEdge = False
+                    for role in [1, -1]:
+                        nx = x + sign * step * ox + 1  # +1 to adjust for wall
+                        ny = y + sign * step * oy + 1  # +1 to adjust for wall
+                        # Stop if wall or opponent's piece is found
+                        if self.board[nx][ny] == 2:
+                            reachEdge = True
+                            break
+                        elif self.board[nx][ny] == -role:  # Change role if opponent's piece is found
+                            continue
+                        elif self.board[nx][ny] == 0:
+                            self.updateSinglePoint(nx - 1, ny - 1, role,
+                                                   [sign * ox, sign * oy])  # -1 to adjust back from wall
+                    if reachEdge:
+                        break
+        performance['updateTime'] += (datetime.now() - start).total_seconds()
+
+    """
+       计算单个点的得分
+       计算原理是：
+       在当前位置放一个当前角色的棋子，遍历四个方向，生成四个方向上的字符串，用patters来匹配字符串, 匹配到的话，就将对应的得分加到scores上
+       四个方向的字符串生成规则是：向两边都延伸5个位置，如果遇到边界或者对方的棋子，就停止延伸
+       在更新周围棋子时，只有一个方向需要更新，因此可以传入direction参数，只更新一个方向
+    """
+
+    def updateSinglePoint(self, x, y, role, direction=None):
+        # Checked
+        if self.board[x + 1][y + 1] != 0:
+            return  # Not an empty spot
+
+        # Temporarily place the piece
+        self.board[x + 1][y + 1] = role
+
+        directions = []
+        if direction:
+            directions.append(direction)
+        else:
+            directions = allDirections
+
+        shapeCache = self.shapeCache[role]
+
+        # Clear the cache first
+        for ox, oy in directions:
+            shapeCache[direction2index(ox, oy)][x][y] = shapes['NONE']
+
+        score = 0
+        blockFourCount = 0
+        threeCount = 0
+        twoCount = 0
+
+        # Calculate existing score
+        for intDirection in [0, 1, 2, 3]:
+            shape = shapeCache[intDirection][x][y]
+            if shape > shapes['NONE']:
+                score += getRealShapeScore(shape)
+                if shape == shapes['BLOCK_FOUR']:
+                    blockFourCount += 1
+                if shape == shapes['THREE']:
+                    threeCount += 1
+                if shape == shapes['TWO']:
+                    twoCount += 1
+
+        for ox, oy in directions:
+            intDirection = direction2index(ox, oy)
+            shape, selfCount = get_shape_fast(self.board, x, y, ox, oy, role)
+            if not shape:
+                continue
+            if shape:
+                # Note: Only cache single shapes, do not cache compound shapes like double threes, as they depend on two shapes
+                shapeCache[intDirection][x][y] = shape
+                if shape == shapes['BLOCK_FOUR']:
+                    blockFourCount += 1
+                if shape == shapes['THREE']:
+                    threeCount += 1
+                if shape == shapes['TWO']:
+                    twoCount += 1
+                if blockFourCount >= 2:
+                    shape = shapes['FOUR_FOUR']
+                elif blockFourCount and threeCount:
+                    shape = shapes['FOUR_THREE']
+                elif threeCount >= 2:
+                    shape = shapes['THREE_THREE']
+                elif twoCount >= 2:
+                    shape = shapes['TWO_TWO']
+                score += getRealShapeScore(shape)
+
+        self.board[x + 1][y + 1] = 0  # Remove the temporary piece
+
+        if role == 1:
+            self.blackScores[x][y] = score
+        else:
+            self.whiteScores[x][y] = score
+
+        return score
+
+    def evaluate(self, role):
+        # Checked
+        blackScore = 0
+        whiteScore = 0
+
+        for i in range(len(self.blackScores)):
+            for j in range(len(self.blackScores[i])):
+                blackScore += self.blackScores[i][j]
+
+        for i in range(len(self.whiteScores)):
+            for j in range(len(self.whiteScores[i])):
+                whiteScore += self.whiteScores[i][j]
+
+        score = blackScore - whiteScore if role == 1 else whiteScore - blackScore
+        return score
+
+    def getMoves(self, role, depth, onThree=False, onlyFour=False, use_net = False):
+        # Checked
+        train_data = 0
+        if use_net and role == 1:
+            # value_move_num = 6
+            # input = torch.Tensor(np.array(self.board)[1:-1, 1:-1]).unsqueeze(0)
+            # scores = mini_max_net(input)
+            # flattened_scores = scores.flatten()
+            #
+            # moves = (flattened_scores.argsort(descending=True)[:value_move_num]).tolist()
+            moves = 0
+            # print(moves)
+        else:
+            moves, model_train_maxtrix = self._getMoves(role, depth, onThree, onlyFour)
+            train_data = {"state": np.array(self.board)[1:-1, 1:-1], "scores": model_train_maxtrix}
+        moves = [(move // self.size, move % self.size) for move in moves]
+        # cut the self.board into normal size
+        print("moves", moves)
+
+        return moves, train_data
+
+    def _getMoves(self, role, depth, only_three=False, only_four=False):
+        """
+        Get possible moves based on the current game state.
+        """
+        points = self.getPoints(role, depth, only_three, only_four)
+        fives = points[shapes['FIVE']]
+        block_fives = points[shapes['BLOCK_FIVE']]
+
+        # To train the model, we need to get all these points's score and store it to board size matrix
+        # Then we can use this matrix to train the model, given a state, we want it to output the score of each point, then we can choose the highest score point
+        model_train_matrix = [[0] * self.size for _ in range(self.size)]
+
+        if fives and len(fives) > 0 or block_fives and len(block_fives) > 0:
+            for point in fives:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(FIVE, model_train_matrix[x][y])
+            for point in block_fives:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FIVE, model_train_matrix[x][y])
+
+            return set(list(fives) + list(block_fives)), model_train_matrix
+
+        fours = points[shapes['FOUR']]
+        block_fours = points[shapes['BLOCK_FOUR']]  # Block four is special, consider it in both four and three
+        if only_four or (fours and len(fours) > 0):
+            for point in fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(FOUR, model_train_matrix[x][y])
+
+            for point in block_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+
+            return set(list(fours) + list(block_fours)), model_train_matrix
+
+        four_fours = points[shapes['FOUR_FOUR']]
+        if four_fours and len(four_fours) > 0:
+            for point in four_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(FOUR_FOUR, model_train_matrix[x][y])
+
+            for point in block_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+
+            return set(list(four_fours) + list(block_fours)), model_train_matrix
+
+        # Double threes and active threes
+        threes = points[shapes['THREE']]
+        four_threes = points[shapes['FOUR_THREE']]
+        if four_threes and len(four_threes) > 0:
+            for point in four_threes:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(FOUR_THREE, model_train_matrix[x][y])
+
+            for point in block_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+
+            for point in threes:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(THREE, model_train_matrix[x][y])
+
+            return set(list(four_threes) + list(block_fours) + list(threes)), model_train_matrix
+
+        three_threes = points[shapes['THREE_THREE']]
+        if three_threes and len(three_threes) > 0:
+
+            for point in three_threes:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(THREE_THREE, model_train_matrix[x][y])
+
+            for point in block_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+
+            for point in threes:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(THREE, model_train_matrix[x][y])
+
+            return set(list(three_threes) + list(block_fours) + list(threes)), model_train_matrix
+
+        if only_three:
+            for point in threes:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(THREE, model_train_matrix[x][y])
+
+            for point in block_fours:
+                x = point // self.size
+                y = point % self.size
+                model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+            return set(list(block_fours) + list(threes)), model_train_matrix
+
+        block_threes = points[shapes['BLOCK_THREE']]
+        two_twos = points[shapes['TWO_TWO']]
+        twos = points[shapes['TWO']]
+
+        for point in block_threes:
+            x = point // self.size
+            y = point % self.size
+            model_train_matrix[x][y] = max(BLOCK_THREE, model_train_matrix[x][y])
+
+        for point in two_twos:
+            x = point // self.size
+            y = point % self.size
+            model_train_matrix[x][y] = max(TWO_TWO, model_train_matrix[x][y])
+
+        for point in twos:
+            x = point // self.size
+            y = point % self.size
+            model_train_matrix[x][y] = max(TWO, model_train_matrix[x][y])
+
+        for point in block_fours:
+            x = point // self.size
+            y = point % self.size
+            model_train_matrix[x][y] = max(BLOCK_FOUR, model_train_matrix[x][y])
+
+        for point in threes:
+            x = point // self.size
+            y = point % self.size
+            model_train_matrix[x][y] = max(THREE, model_train_matrix[x][y])
+
+        mid = list(block_fours) + list(threes) + list(block_threes) + list(two_twos) + list(twos)
+        res = set(mid[:5])
+        for i in range(len(model_train_matrix)):
+            for j in range(len(model_train_matrix)):
+                if (i * len(model_train_matrix) + j) not in res:
+                    model_train_matrix[i][j] = 0
+        return res, model_train_matrix

Gomoku_Bot/gomoku_bot.py

@@ -0,0 +1,23 @@
+from .minmax import *
+import time
+
+
+class Gomoku_bot:
+    def __init__(self, board, role, depth=4, enableVCT=True):
+        self.board = board
+        self.role = role
+        self.depth = depth
+        self.enableVCT = enableVCT
+
+    def get_action(self, return_time=True):
+        start = time.time()
+        score = minmax(self.board, self.role, self.depth, self.enableVCT)
+        end = time.time()
+        sim_time = end - start
+        move = score[1]
+        # turn tuple into an int
+        move = move[0] * self.board.size + move[1]
+        if return_time:
+            return move, sim_time
+        else:
+            return move

Gomoku_Bot/minimax_Net.py

@@ -0,0 +1,117 @@
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import pickle
+import os
+
+dir_path = os.path.dirname(os.path.realpath(__file__))
+# from tensorboardX import SummaryWriter
+from tqdm import tqdm
+import datetime
+from torch.utils.data import DataLoader, TensorDataset
+
+date = datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+
+
+class BoardEvaluationNet(nn.Module):
+    def __init__(self, board_size):
+        super(BoardEvaluationNet, self).__init__()
+        self.board_size = board_size
+
+        self.conv1 = nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1)
+        self.conv2 = nn.Conv2d(16, 32, kernel_size=3, stride=1, padding=1)
+        self.fc1 = nn.Linear(32 * board_size * board_size, 256)
+        self.fc2 = nn.Linear(256, board_size * board_size)
+
+    def forward(self, x):
+        x = x.unsqueeze(1)  # Add a channel dimension
+        x = F.relu(self.conv1(x))
+        x = F.relu(self.conv2(x))
+        x = x.view(-1, 32 * self.board_size * self.board_size)
+        x = F.relu(self.fc1(x))
+        x = self.fc2(x)
+        return x.view(-1, self.board_size, self.board_size)
+
+
+def normalize(t):
+    return t
+
+
+if __name__ == "__main__":
+
+    writer = SummaryWriter(os.path.join(dir_path, 'train_data/log', date), comment='BoardEvaluationNet')
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+    best = np.Inf
+    loss_fn = nn.CrossEntropyLoss()
+
+    # Example usage
+    BS = 15
+
+    net_for_black = BoardEvaluationNet(BS).to(device)
+    net_for_white = BoardEvaluationNet(BS).to(device)
+
+    net_for_black.load_state_dict(torch.load(os.path.join(dir_path, 'train_data/model', 'best_loss=680.5813717259707.pth')))
+
+    optimizer = torch.optim.Adam(net_for_black.parameters(), lr=1e-5, betas=(0.9, 0.99),
+                                 eps=1e-8)
+
+    data_path = os.path.join(dir_path, 'train_data/data', 'train_data.pkl')
+    with open(data_path, 'rb') as f:
+        datas = pickle.load(f)
+
+    train_data_for_black = datas[1][:int(len(datas[1]) * 1)]
+    test_data_for_black = datas[1][int(len(datas[1]) * 0.8):]
+    train_data_for_white = datas[-1]
+    epochs = 500
+    batch_size = 32
+    train_dataset = TensorDataset(torch.stack([torch.tensor(item['state'], dtype=torch.float) for item in train_data_for_black]),
+                                  torch.stack([normalize(torch.tensor(item['scores'], dtype=torch.float)) for item in train_data_for_black]))
+    train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
+
+    for epoch in range(epochs):
+        epoch_loss = 0
+        print('Epoch:', epoch)
+        for i, (states, scores) in tqdm(enumerate(train_dataloader), total=len(train_dataloader)):
+            states = states.to(device)
+            scores = scores.to(device)
+
+            # print(input_tensor.shape)
+            infer_start = datetime.datetime.now()
+            output_tensor = net_for_black(states)
+            infer_end = datetime.datetime.now()
+            loss = loss_fn(output_tensor, scores)
+            print(loss.item())
+            exit(0)
+            loss.backward()
+            optimizer.step()
+            optimizer.zero_grad()
+            epoch_loss += loss.item()
+
+            writer.add_scalar('train/infer_time', (infer_end - infer_start).microseconds,
+                              i + epoch * len(train_dataloader))
+
+        epoch_loss /= len(train_dataloader)
+        writer.add_scalar('train/epoch_loss', epoch_loss, epoch)
+        # test
+        with torch.no_grad():
+            test_loss = 0
+            net_for_black.eval()
+            for j, item in tqdm(enumerate(test_data_for_black), total=len(test_data_for_black)):
+                scores = normalize(torch.tensor(item['scores'], dtype=torch.float).to(device).unsqueeze(0))  # 将数据类型设为float
+                state = item['state']
+                input_tensor = torch.tensor(state, dtype=torch.float).to(device).unsqueeze(0)  # 将数据类型设为float，并转移到设备上
+                output_tensor = net_for_black(input_tensor).to(device)
+                loss = loss_fn(output_tensor, scores)
+                test_loss += loss.item()
+            test_loss /=len(test_data_for_black)
+            writer.add_scalar('test/loss', test_loss, epoch)
+            if best > test_loss:
+                best = test_loss
+                model_path = os.path.join(dir_path, 'train_data/model')
+                if not os.path.exists(model_path):
+                    os.makedirs(model_path)
+                torch.save(net_for_black.state_dict(),
+                           os.path.join(model_path, f'best_loss={best}.pth'))
+        net_for_black.train()

Gomoku_Bot/minmax.py

@@ -0,0 +1,119 @@
+from .cache import Cache
+from .eval import FIVE
+# Checked
+
+
+MAX = 1000000000
+cache_hits = {
+    "search": 0,
+    "total": 0,
+    "hit": 0
+}
+
+onlyThreeThreshold = 6
+cache = Cache()
+
+
+def factory(onlyThree=False, onlyFour=False):
+
+    def helper(board, role, depth, cDepth=0, path=(), alpha=-MAX, beta=MAX):
+        cache_hits["search"] += 1
+        if cDepth >= depth or board.isGameOver():
+            return [board.evaluate(role), None, (path)]
+        hash = board.hash()
+        prev = cache.get(hash)
+        if prev and prev["role"] == role:
+            if (
+                (abs(prev["value"]) >= FIVE or prev["depth"] >= depth - cDepth)
+                and prev["onlyThree"] == onlyThree
+                and prev["onlyFour"] == onlyFour
+            ):
+                cache_hits["hit"] += 1
+                return [prev["value"], prev["move"], path + prev["path"]]
+        value = -MAX
+        move = None
+        bestPath = path  # Copy the current path
+        bestDepth = 0
+        # points = board.getValuableMoves(role, cDepth, onlyThree or cDepth > onlyThreeThreshold, onlyFour)
+        points = board.getValuableMoves(role, cDepth, onlyThree or cDepth > onlyThreeThreshold, onlyFour)
+        if cDepth == 0:
+            print('points:', points)
+        if not len(points):
+            return [board.evaluate(role), None, path]
+        for d in range(cDepth + 1, depth + 1):
+            # 迭代加深过程中只找己方能赢的解，因此只搜索偶数层即可
+            if d % 2 != 0:
+                continue
+            breakAll = False
+            for point in points:
+                board.put(point[0], point[1], role)
+                newPath = tuple(list(path) + [point])  # Add current move to path
+                currentValue, currentMove, currentPath = helper(board, -role, d, cDepth + 1, tuple(newPath) , -beta, -alpha)
+                currentValue = -currentValue
+                board.undo()
+                ## 迭代加深的过程中，除了能赢的棋，其他都不要
+                ## 原因是：除了必胜的，其他评估不准。比如必输的棋，由于走的步数偏少，也会变成没有输，比如5
+                ### 步之后输了，但是1步肯定不会输，这时候1步的分数是不准确的，显然不能选择。
+                if currentValue >= FIVE or d == depth:
+                    # 必输的棋，也要挣扎一下，选择最长的路径
+                    if (
+                        currentValue > value
+                        or (currentValue <= -FIVE and value <= -FIVE and len(currentPath) > bestDepth)
+                    ):
+                        value = currentValue
+                        move = point
+                        bestPath = currentPath
+                        bestDepth = len(currentPath)
+                alpha = max(alpha, value)
+                if alpha >= FIVE:
+                    breakAll = True
+                    break
+                if alpha >= beta:
+                    break
+            if breakAll:
+                break
+        if (cDepth < onlyThreeThreshold or onlyThree or onlyFour) and (not prev or prev["depth"] < depth - cDepth):
+            cache_hits["total"] += 1
+            cache.put(hash, {
+                "depth": depth - cDepth,
+                "value": value,
+                "move": move,
+                "role": role,
+                "path": bestPath[cDepth:],
+                "onlyThree": onlyThree,
+                "onlyFour": onlyFour,
+            })
+        return [value, move, bestPath]
+    return helper
+
+
+_minmax = factory()
+vct = factory(True)
+vcf = factory(False, True)
+
+
+def minmax(board, role, depth=4, enableVCT=False):
+
+    if enableVCT:
+        vctDepth = depth + 8
+        value, move, bestPath = vct(board, role, vctDepth)
+        if value >= FIVE:
+            return [value, move, bestPath]
+        value, move, bestPath = _minmax(board, role, depth)
+        '''
+        // 假设对方有杀棋，先按自己的思路走，走完之后看对方是不是还有杀棋
+        // 如果对方没有了，那么就说明走的是对的
+        // 如果对方还是有，那么要对比对方的杀棋路径和自己没有走棋时的长短
+        // 如果走了棋之后路径变长了，说明走的是对的
+        // 如果走了棋之后，对方杀棋路径长度没变，甚至更短，说明走错了，此时就优先封堵对方
+        '''
+        board.put(move[0], move[1], role)
+        value2, move2, bestPath2 = vct(board.reverse(), role, vctDepth)
+        board.undo()
+        if value < FIVE and value2 == FIVE and len(bestPath2) > len(bestPath):
+            value3, move3, bestPath3 = vct(board.reverse(), role, vctDepth)
+            if len(bestPath2) <= len(bestPath3):
+                return [value, move2, bestPath2] # value2 是被挡住的，所以这里还是用value
+        return [value, move, bestPath]
+    else:
+        return _minmax(board, role, depth)

Gomoku_Bot/position.py

@@ -0,0 +1,36 @@
+from typing import List
+from .config import config
+# Checked
+
+# Function to convert position to coordinate
+def position2Coordinate(position: int, size: int) -> List[int]:
+    return [position // size, position % size]
+
+# Function to convert coordinate to position
+def coordinate2Position(x: int, y: int, size: int) -> int:
+    return x * size + y
+
+# Check if points a and b are on the same line and the distance is less than maxDistance
+def isLine(a: int, b: int, size: int) -> bool:
+    maxDistance = config["inLineDistance"]
+    [x1, y1] = position2Coordinate(a, size)
+    [x2, y2] = position2Coordinate(b, size)
+    return (
+        (x1 == x2 and abs(y1 - y2) < maxDistance) or
+        (y1 == y2 and abs(x1 - x2) < maxDistance) or
+        (abs(x1 - x2) == abs(y1 - y2) and abs(x1 - x2) < maxDistance)
+    )
+
+# Check if all points in the array are on the same line as point p
+def isAllInLine(p: int, arr: List[int], size: int) -> bool:
+    for i in range(len(arr)):
+        if not isLine(p, arr[i], size):
+            return False
+    return True
+
+# Check if any point in the array is on the same line as point p
+def hasInLine(p: int, arr: List[int], size: int) -> bool:
+    for i in range(len(arr)):
+        if isLine(p, arr[i], size):
+            return True
+    return False

Gomoku_Bot/shape.py

@@ -0,0 +1,288 @@
+
+import re
+
+# Define patterns using regular expressions
+patterns = {
+    'five': re.compile('11111'),
+    'block_five': re.compile('211111|111112'),
+    'four': re.compile('011110'),
+    'block_four': re.compile('10111|11011|11101|211110|211101|211011|210111|011112|101112|110112|111012'),
+    'three': re.compile('011100|011010|010110|001110'),
+    'block_three': re.compile('211100|211010|210110|001112|010112|011012'),
+    'two': re.compile('001100|011000|000110|010100|001010'),
+}
+
+# Define shapes with associated scores
+shapes = {
+    'FIVE': 5,
+    'BLOCK_FIVE': 50,
+    'FOUR': 4,
+    'FOUR_FOUR': 44,  # Double four
+    'FOUR_THREE': 43,  # Four with an open three
+    'THREE_THREE': 33,  # Double three
+    'BLOCK_FOUR': 40,
+    'THREE': 3,
+    'BLOCK_THREE': 30,
+    'TWO_TWO': 22,  # Double two
+    'TWO': 2,
+    'NONE': 0
+}
+
+# Initialize a performance record
+performance = {
+    'five': 0,
+    'block_five': 0,
+    'four': 0,
+    'block_four': 0,
+    'three': 0,
+    'block_three': 0,
+    'two': 0,
+    'none': 0,
+    'total': 0
+}
+
+# Function to detect shapes on the board
+def get_shape(board, x, y, offset_x, offset_y, role):
+    """
+    Detect shape at a given board position.
+    :param board: The game board.
+    :param x: X-coordinate.
+    :param y: Y-coordinate.
+    :param offset_x: X-direction offset for scanning.
+    :param offset_y: Y-direction offset for scanning.
+    :param role: Current player's role.
+    :return: A tuple of shape, self count, opponent count, and empty count.
+    """
+    opponent = -role
+    empty_count = 0
+    self_count = 1
+    opponent_count = 0
+    shape = shapes['NONE']
+
+    # Skip empty nodes
+    if (
+        board[x + offset_x + 1][y + offset_y + 1] == 0
+        and board[x - offset_x + 1][y - offset_y + 1] == 0
+        and board[x + 2 * offset_x + 1][y + 2 * offset_y + 1] == 0
+        and board[x - 2 * offset_x + 1][y - 2 * offset_y + 1] == 0
+    ):
+        return [0, self_count, opponent_count, empty_count]
+    # Check for 'two' pattern
+    for i in range(-3, 4):
+        if i == 0:
+            continue
+        nx, ny = x + i * offset_x, y + i * offset_y
+        current_role = board.get((nx, ny))
+        if current_role is None:
+            continue
+        if current_role == 2:
+            opponent_count += 1
+        elif current_role == role:
+            self_count += 1
+        elif current_role == 0:
+            empty_count += 1
+
+    if self_count == 2:
+        if opponent_count == 0:
+            return shapes['TWO'], self_count, opponent_count, empty_count
+        else:
+            return shapes['NONE'], self_count, opponent_count, empty_count
+
+    # Reset counts and prepare string for pattern matching
+    empty_count, self_count, opponent_count = 0, 1, 0
+    result_string = '1'
+
+    # Build result string for pattern matching
+    for i in range(1, 6):
+        nx = x + i * offset_x + 1
+        ny = y + i * offset_y + 1
+        currentRole = board[nx][ny]
+        if currentRole == 2:
+            result_string += '2'
+        elif currentRole == 0:
+            result_string += '0'
+        else:
+            result_string += '1' if currentRole == role else '2'
+        if currentRole == 2 or currentRole == opponent:
+            opponent_count += 1
+            break
+        if currentRole == 0:
+            empty_count += 1
+        if currentRole == role:
+            self_count += 1
+    
+    for i in range(1, 6):
+        nx = x - i * offset_x + 1
+        ny = y - i * offset_y + 1
+        currentRole = board[nx][ny]
+        if currentRole == 2:
+            result_string = '2' + result_string
+        elif currentRole == 0:
+            result_string = '0' + result_string
+        else:
+            result_string = '1' if currentRole == role else '2' + result_string
+        if currentRole == 2 or currentRole == opponent:
+            opponent_count += 1
+            break
+        if currentRole == 0:
+            empty_count += 1
+        if currentRole == role:
+            self_count += 1
+
+    # Check patterns and update performance
+    for pattern_key, shape_key in [('five', 'FIVE'), ('four', 'FOUR'), ('block_four', 'BLOCK_FOUR'),
+                                   ('three', 'THREE'), ('block_three', 'BLOCK_THREE'), ('two', 'TWO')]:
+        if patterns[pattern_key].search(result_string):
+            shape = shapes[shape_key]
+            performance[pattern_key] += 1
+            performance['total'] += 1
+            break
+    ## 尽量减少多余字符串生成
+    if self_count <= 1 or len(result_string) < 5:
+        return shape, self_count, opponent_count, empty_count
+
+    return shape, self_count, opponent_count, empty_count
+
+def count_shape(board, x, y, offset_x, offset_y, role):
+    opponent = - role
+
+    inner_empty_count = 0  # Number of empty positions inside the player's stones
+    temp_empty_count = 0
+    self_count = 0  # Number of the player's stones in the shape
+    total_length = 0
+
+    side_empty_count = 0  # Number of empty positions on the side of the shape
+    no_empty_self_count = 0
+    one_empty_self_count = 0
+
+    # Right direction
+    for i in range(1, 6):
+        nx = x + i * offset_x + 1
+        ny = y + i * offset_y + 1
+        current_role = board[nx][ny]
+        if current_role == 2 or current_role == opponent:
+            break
+        if current_role == role:
+            self_count += 1
+            side_empty_count = 0
+            if temp_empty_count:
+                inner_empty_count += temp_empty_count
+                temp_empty_count = 0
+            if inner_empty_count == 0:
+                no_empty_self_count += 1
+                one_empty_self_count += 1
+            elif inner_empty_count == 1:
+                one_empty_self_count += 1
+        total_length += 1
+        if current_role == 0:
+            temp_empty_count += 1
+            side_empty_count += 1
+        if side_empty_count >= 2:
+            break
+
+    if not inner_empty_count:
+        one_empty_self_count = 0
+
+    return {
+        'self_count': self_count,
+        'total_length': total_length,
+        'no_empty_self_count': no_empty_self_count,
+        'one_empty_self_count': one_empty_self_count,
+        'inner_empty_count': inner_empty_count,
+        'side_empty_count': side_empty_count
+    }
+
+# Fast shape detection function
+def get_shape_fast(board, x, y, offsetX, offsetY, role):
+    if (
+        board[x + offsetX + 1][y + offsetY + 1] == 0
+        and board[x - offsetX + 1][y - offsetY + 1] == 0
+        and board[x + 2 * offsetX + 1][y + 2 * offsetY + 1] == 0
+        and board[x - 2 * offsetX + 1][y - 2 * offsetY + 1] == 0
+    ):
+        return [shapes['NONE'], 1]
+
+    selfCount = 1
+    totalLength = 1
+    shape = shapes['NONE']
+
+    leftEmpty = 0
+    rightEmpty = 0
+    noEmptySelfCount = 1
+    OneEmptySelfCount = 1
+
+    left = count_shape(board, x, y, -offsetX, -offsetY, role)
+    right = count_shape(board, x, y, offsetX, offsetY, role)
+
+    selfCount = left['self_count'] + right['self_count'] + 1
+    totalLength = left['total_length'] + right['total_length'] + 1
+    noEmptySelfCount = left['no_empty_self_count'] + right['no_empty_self_count'] + 1
+    OneEmptySelfCount = max(
+        left['one_empty_self_count'] + right['no_empty_self_count'],
+        left['no_empty_self_count'] + right['one_empty_self_count'],
+    ) + 1
+    rightEmpty = right['side_empty_count']
+    leftEmpty = left['side_empty_count']
+
+    if totalLength < 5:
+        return [shape, selfCount]
+
+    if noEmptySelfCount >= 5:
+        if rightEmpty > 0 and leftEmpty > 0:
+            return [shapes['FIVE'], selfCount]
+        else:
+            return [shapes['BLOCK_FIVE'], selfCount]
+
+    if noEmptySelfCount == 4:
+        if (
+            (rightEmpty >= 1 or right['one_empty_self_count'] > right['no_empty_self_count'])
+            and (leftEmpty >= 1 or left['one_empty_self_count'] > left['no_empty_self_count'])
+        ):
+            return [shapes['FOUR'], selfCount]
+        elif not (rightEmpty == 0 and leftEmpty == 0):
+            return [shapes['BLOCK_FOUR'], selfCount]
+
+    if OneEmptySelfCount == 4:
+        return [shapes['BLOCK_FOUR'], selfCount]
+
+    if noEmptySelfCount == 3:
+        if (rightEmpty >= 2 and leftEmpty >= 1) or (rightEmpty >= 1 and leftEmpty >= 2):
+            return [shapes['THREE'], selfCount]
+        else:
+            return [shapes['BLOCK_THREE'], selfCount]
+
+    if OneEmptySelfCount == 3:
+        if rightEmpty >= 1 and leftEmpty >= 1:
+            return [shapes['THREE'], selfCount]
+        else:
+            return [shapes['BLOCK_THREE'], selfCount]
+
+    if (noEmptySelfCount == 2 or OneEmptySelfCount == 2) and totalLength > 5:
+        shape = shapes['TWO']
+
+    return [shape, selfCount]
+
+# Helper functions to check for specific shapes
+def is_five(shape):
+    # Checked
+    return shape in [shapes['FIVE'], shapes['BLOCK_FIVE']]
+
+def is_four(shape):
+    # Checked
+    return shape in [shapes['FOUR'], shapes['BLOCK_FOUR']]
+
+# Function to get all shapes at a specific point
+def get_all_shapes_of_point(shape_cache, x, y, role = None):
+    # Checked
+    roles = [role] if role else [1, -1]
+    result = []
+    for r in roles:
+        for d in range(4):
+            shape = shape_cache[r][d][x][y]
+            if shape > 0:
+                result.append(shape)
+    return result
+
+
+if __name__ == "__main__":
+    pass

Gomoku_Bot/zobrist.py

@@ -0,0 +1,35 @@
+import random
+# Checked
+class ZobristCache:
+    def __init__(self, size):
+        self.size = size
+        self.zobristTable = self.initializeZobristTable(size)
+        self.hash = 0
+
+    def initializeZobristTable(self, size):
+        table = []
+        for i in range(size):
+            table.append([])
+            for j in range(size):
+                table[i].append({
+                    1: random.getrandbits(64),  # black
+                    -1: random.getrandbits(64)  # white
+                })
+        return table
+
+    def togglePiece(self, x, y, role):
+        self.hash ^= self.zobristTable[x][y][role]
+
+    def getHash(self):
+        return self.hash
+
+if __name__ == '__main__':
+    # Example usage
+    size = 8
+    cache = ZobristCache(size)
+    x = 3
+    y = 4
+    role = 1
+    cache.togglePiece(x, y, role)
+    hash_value = cache.getHash()
+    print(hash_value)

ai.py

@@ -0,0 +1,39 @@
+import numpy as np
+import torch
+from transformers import GPT2LMHeadModel
+
+
+def load_model(model_name: str = "snoop2head/Gomoku-GPT2") -> GPT2LMHeadModel:
+    gpt2 = GPT2LMHeadModel.from_pretrained(model_name)
+    return gpt2
+
+
+BOS_TOKEN_ID = 401
+PAD_TOKEN_ID = 402
+EOS_TOKEN_ID = 403
+
+
+def generate_gpt2(model: GPT2LMHeadModel, input_ids: torch.LongTensor,temperature = 0.7) -> list:
+    """
+    input_ids: [batch_size, seq_len] torch.LongTensor
+    output_ids: [seq_len] list
+    """
+    output_ids = model.generate(
+        input_ids,
+        max_length=128,
+        num_beams=5,
+        temperature= temperature,
+        pad_token_id=PAD_TOKEN_ID,
+        eos_token_id=EOS_TOKEN_ID,
+    )
+    return output_ids.squeeze().tolist()
+
+
+def change_to_1d_coordinate(board: np.ndarray, x: int, y: int) -> int:
+    """change 2d coordinate to 1d coordinate"""
+    return x * board.shape[1] + y
+
+
+def change_to_2d_coordinate(board: np.ndarray, coordinate: int) -> tuple:
+    """change 1d coordinate to 2d coordinate"""
+    return (coordinate // board.shape[1], coordinate % board.shape[1])

const.py

@@ -14,10 +14,13 @@ _BOARD_SIZE_1D = _BOARD_SIZE * _BOARD_SIZE
 _BLANK = 0
 _BLACK = 1
 _WHITE = 2
+_NEW = 3
 _PLAYER_SYMBOL = {
     _WHITE: "⚪",
     _BLANK: "➕",
     _BLACK: "⚫",
+    _NEW: "🔴",
+
 }
 _PLAYER_COLOR = {
     _WHITE: "AI",

pages/AI_VS_AI.py

@@ -0,0 +1,575 @@
+"""
+FileName: app.py
+Author: Benhao Huang
+Create Date: 2023/11/19
+Description: this file is used to display our project and add visualization elements to the game, using Streamlit
+"""
+
+import time
+import pandas as pd
+from copy import deepcopy
+import torch
+
+# import torch
+import numpy as np
+import streamlit as st
+from scipy.signal import convolve  # this is used to check if any player wins
+from streamlit import session_state
+from streamlit_server_state import server_state, server_state_lock
+from Gomoku_MCTS import MCTSpure, alphazero, Board, PolicyValueNet
+from Gomoku_Bot import Gomoku_bot
+from Gomoku_Bot import Board as Gomoku_bot_board
+import matplotlib.pyplot as plt
+
+
+
+from const import (
+    _BLACK,  # 1, for human
+    _WHITE,  # 2 , for AI
+    _BLANK,
+    _PLAYER_COLOR,
+    _PLAYER_SYMBOL,
+    _ROOM_COLOR,
+    _VERTICAL,
+    _NEW,
+    _HORIZONTAL,
+    _DIAGONAL_UP_LEFT,
+    _DIAGONAL_UP_RIGHT,
+    _BOARD_SIZE,
+    _BOARD_SIZE_1D,
+    _AI_AID_INFO
+)
+
+
+from ai import (
+    BOS_TOKEN_ID,
+    generate_gpt2,
+    load_model,
+)
+
+gpt2 = load_model()
+
+
+# Utils
+class Room:
+    def __init__(self, room_id) -> None:
+        self.ROOM_ID = room_id
+        # self.BOARD = np.zeros(shape=(_BOARD_SIZE, _BOARD_SIZE), dtype=int)
+        self.BOARD = Board(width=_BOARD_SIZE, height=_BOARD_SIZE, n_in_row=5, players=[_BLACK, _WHITE])
+        self.PLAYER = _BLACK
+        self.TURN = self.PLAYER
+        self.HISTORY = (0, 0)
+        self.WINNER = _BLANK
+        self.TIME = time.time()
+        self.gomoku_bot_board = Gomoku_bot_board(_BOARD_SIZE, 1)
+        self.MCTS_dict = {'Pure MCTS': MCTSpure(c_puct=5, n_playout=1000),
+                          'AlphaZero': alphazero(PolicyValueNet(_BOARD_SIZE, _BOARD_SIZE, 'Gomoku_MCTS/checkpoints/best_policy_8_8_5_2torch.pth').policy_value_fn, c_puct=5, n_playout=100),
+                          'Gomoku Bot': Gomoku_bot(self.gomoku_bot_board, -1)}
+        self.MCTS = self.MCTS_dict['AlphaZero']
+        self.last_mcts = self.MCTS
+        self.AID_MCTS = self.MCTS_dict['AlphaZero']
+        self.COORDINATE_1D = [BOS_TOKEN_ID]
+        self.current_move = -1
+        self.simula_time_list = []
+
+
+def change_turn(cur):
+    return cur % 2 + 1
+
+
+# Initialize the game
+if "ROOM" not in session_state:
+    session_state.ROOM = Room("local")
+if "OWNER" not in session_state:
+    session_state.OWNER = False
+if "USE_AIAID" not in session_state:
+    session_state.USE_AIAID = False
+
+# Check server health
+if "ROOMS" not in server_state:
+    with server_state_lock["ROOMS"]:
+        server_state.ROOMS = {}
+
+def handle_oppo_model_selection():
+    if st.session_state['selected_oppo_model'] == 'Gomoku Bot':
+        session_state.ROOM.last_mcts = session_state.ROOM.MCTS # since use different mechanism, store previous mcts first
+        session_state.ROOM.MCTS = session_state.ROOM.MCTS_dict['Gomoku Bot']
+        return
+    else:
+        TreeNode = session_state.ROOM.last_mcts.mcts._root
+        new_mct = session_state.ROOM.MCTS_dict[st.session_state['selected_oppo_model']]
+        new_mct.mcts._root = deepcopy(TreeNode)
+        session_state.ROOM.MCTS = new_mct
+        session_state.ROOM.last_mcts  = new_mct
+    return
+
+def handle_aid_model_selection():
+    if st.session_state['selected_aid_model'] == 'None':
+        session_state.USE_AIAID = False
+        return
+    session_state.USE_AIAID = True
+    TreeNode = session_state.ROOM.MCTS.mcts._root # use the same tree node
+    new_mct = session_state.ROOM.MCTS_dict[st.session_state['selected_aid_model']]
+    new_mct.mcts._root = deepcopy(TreeNode)
+    session_state.ROOM.AID_MCTS = new_mct
+    return
+
+if 'selected_oppo_model' not in st.session_state:
+    st.session_state['selected_oppo_model'] = 'AlphaZero'  # 默认值
+
+if 'selected_aid_model' not in st.session_state:
+    st.session_state['selected_aid_model'] = 'AlphaZero'  # 默认值
+
+# Layout
+TITLE = st.empty()
+Model_Switch = st.empty()
+
+TITLE.header("🤖 AI 3603 Gomoku")
+selected_oppo_option = Model_Switch.selectbox('Select Opponent Model', ['Pure MCTS', 'AlphaZero','Gomoku Bot'], index=1, key='oppo_model')
+
+if st.session_state['selected_oppo_model'] != selected_oppo_option:
+    st.session_state['selected_oppo_model'] = selected_oppo_option
+    handle_oppo_model_selection()
+
+ROUND_INFO = st.empty()
+st.markdown("<br>", unsafe_allow_html=True)
+BOARD_PLATE = [
+    [cell.empty() for cell in st.columns([1 for _ in range(_BOARD_SIZE)])] for _ in range(_BOARD_SIZE)
+]
+LOG = st.empty()
+
+# Sidebar
+SCORE_TAG = st.sidebar.empty()
+SCORE_PLATE = st.sidebar.columns(2)
+# History scores
+SCORE_TAG.subheader("Scores")
+
+PLAY_MODE_INFO = st.sidebar.container()
+MULTIPLAYER_TAG = st.sidebar.empty()
+with st.sidebar.container():
+    ANOTHER_ROUND = st.empty()
+    RESTART = st.empty()
+    AIAID = st.empty()
+    EXIT = st.empty()
+selected_aid_option = AIAID.selectbox('Select Assistant Model', ['None', 'Pure MCTS', 'AlphaZero'], index=0, key='aid_model')
+if st.session_state['selected_aid_model'] != selected_aid_option:
+    st.session_state['selected_aid_model'] = selected_aid_option
+    handle_aid_model_selection()
+
+GAME_INFO = st.sidebar.container()
+message = st.empty()
+PLAY_MODE_INFO.write("---\n\n**You are Black, AI agent is White.**")
+GAME_INFO.markdown(
+    """
+    ---
+    # <span style="color:black;">Freestyle Gomoku game. 🎲</span>
+    - no restrictions 🚫
+    - no regrets 😎
+    - no regrets 😎
+    - swap players after one round is over 🔁
+    Powered by an AlphaZero approach with our own improvements! 🚀 For the specific details, please check out our <a href="insert_report_link_here" style="color:blue;">report</a>.
+    ##### Adapted and improved by us! 🌟  <a href="https://github.com/Lijiaxin0111/AI_3603_BIGHOME" style="color:blue;">Our Github repo</a>
+    """,
+    unsafe_allow_html=True,
+)
+
+
+
+def restart() -> None:
+    """
+    Restart the game.
+    """
+    session_state.ROOM = Room(session_state.ROOM.ROOM_ID)
+    st.session_state['selected_oppo_model'] = 'AlphaZero'
+
+RESTART.button(
+    "Reset",
+    on_click=restart,
+    help="Clear the board as well as the scores",
+)
+
+
+# Draw the board
+def gomoku():
+    """
+    Draw the board.
+    Handle the main logic.
+    """
+
+    # Restart the game
+
+    # Continue new round
+    def another_round() -> None:
+        """
+        Continue new round.
+        """
+        session_state.ROOM = deepcopy(session_state.ROOM)
+        session_state.ROOM.BOARD = Board(width=_BOARD_SIZE, height=_BOARD_SIZE, n_in_row=5)
+        session_state.ROOM.gomoku_bot_board = Gomoku_bot_board(_BOARD_SIZE, 1)
+        session_state.ROOM.MCTS_dict = {'Pure MCTS': MCTSpure(c_puct=5, n_playout=1000),
+                          'AlphaZero': alphazero(PolicyValueNet(_BOARD_SIZE, _BOARD_SIZE, 'Gomoku_MCTS/checkpoints/best_policy_8_8_5_2torch.pth').policy_value_fn, c_puct=5, n_playout=100),
+                          'Gomoku Bot': Gomoku_bot(session_state.ROOM.gomoku_bot_board, -1)}
+        session_state.ROOM.MCTS = session_state.ROOM.MCTS_dict[st.session_state['selected_oppo_model']]
+        session_state.ROOM.last_mcts = session_state.ROOM.MCTS
+        session_state.ROOM.PLAYER = session_state.ROOM.PLAYER
+        session_state.ROOM.TURN = session_state.ROOM.PLAYER
+        session_state.ROOM.WINNER = _BLANK  # 0
+        session_state.ROOM.COORDINATE_1D = [BOS_TOKEN_ID]
+
+    # Room status sync
+    def sync_room() -> bool:
+        room_id = session_state.ROOM.ROOM_ID
+        if room_id not in server_state.ROOMS.keys():
+            session_state.ROOM = Room("local")
+            return False
+        elif server_state.ROOMS[room_id].TIME == session_state.ROOM.TIME:
+            return False
+        elif server_state.ROOMS[room_id].TIME < session_state.ROOM.TIME:
+            # Only acquire the lock when writing to the server state
+            with server_state_lock["ROOMS"]:
+                server_rooms = server_state.ROOMS
+                server_rooms[room_id] = session_state.ROOM
+                server_state.ROOMS = server_rooms
+            return True
+        else:
+            session_state.ROOM = server_state.ROOMS[room_id]
+            return True
+
+    # Check if winner emerge from move
+    def check_win() -> int:
+        """
+        Use convolution to check if any player wins.
+        """
+        vertical = convolve(
+            session_state.ROOM.BOARD.board_map,
+            _VERTICAL,
+            mode="same",
+        )
+        horizontal = convolve(
+            session_state.ROOM.BOARD.board_map,
+            _HORIZONTAL,
+            mode="same",
+        )
+        diagonal_up_left = convolve(
+            session_state.ROOM.BOARD.board_map,
+            _DIAGONAL_UP_LEFT,
+            mode="same",
+        )
+        diagonal_up_right = convolve(
+            session_state.ROOM.BOARD.board_map,
+            _DIAGONAL_UP_RIGHT,
+            mode="same",
+        )
+        if (
+                np.max(
+                    [
+                        np.max(vertical),
+                        np.max(horizontal),
+                        np.max(diagonal_up_left),
+                        np.max(diagonal_up_right),
+                    ]
+                )
+                == 5 * _BLACK
+        ):
+            winner = _BLACK
+        elif (
+                np.min(
+                    [
+                        np.min(vertical),
+                        np.min(horizontal),
+                        np.min(diagonal_up_left),
+                        np.min(diagonal_up_right),
+                    ]
+                )
+                == 5 * _WHITE
+        ):
+            winner = _WHITE
+        else:
+            winner = _BLANK
+        return winner
+
+    # Triggers the board response on click
+    def handle_click(x, y):
+        """
+        Controls whether to pass on / continue current board / may start new round
+        """
+        if session_state.ROOM.BOARD.board_map[x][y] != _BLANK:
+            pass
+        elif (
+                session_state.ROOM.ROOM_ID in server_state.ROOMS.keys()
+                and _ROOM_COLOR[session_state.OWNER]
+                != server_state.ROOMS[session_state.ROOM.ROOM_ID].TURN
+        ):
+            sync_room()
+
+        # normal play situation
+        elif session_state.ROOM.WINNER == _BLANK:
+            # session_state.ROOM = deepcopy(session_state.ROOM)
+            # print("View of human player: ", session_state.ROOM.BOARD.board_map)
+            move = session_state.ROOM.BOARD.location_to_move((x, y))
+            session_state.ROOM.current_move = move
+            session_state.ROOM.BOARD.do_move(move)
+            # Gomoku Bot BOARD
+            session_state.ROOM.MCTS_dict["Gomoku Bot"].board.put(move // _BOARD_SIZE, move % _BOARD_SIZE)
+            session_state.ROOM.BOARD.board_map[x][y] = session_state.ROOM.TURN
+            session_state.ROOM.COORDINATE_1D.append(x * _BOARD_SIZE + y)
+
+            session_state.ROOM.TURN = change_turn(session_state.ROOM.TURN)
+            win, winner = session_state.ROOM.BOARD.game_end()
+            if win:
+                session_state.ROOM.WINNER = winner
+            session_state.ROOM.HISTORY = (
+                session_state.ROOM.HISTORY[0]
+                + int(session_state.ROOM.WINNER == _WHITE),
+                session_state.ROOM.HISTORY[1]
+                + int(session_state.ROOM.WINNER == _BLACK),
+            )
+            session_state.ROOM.TIME = time.time()
+
+    def forbid_click(x, y):
+        # st.warning('This posistion has been occupied!!!!', icon="⚠️")
+        st.error("({}, {}) has been occupied!!)".format(x, y), icon="🚨")
+
+    # Draw board
+    def draw_board(response: bool):
+        """construct each buttons for all cells of the board"""
+        if session_state.USE_AIAID and session_state.ROOM.WINNER == _BLANK and session_state.ROOM.TURN == _BLACK:
+            if session_state.USE_AIAID:
+                copy_mcts = deepcopy(session_state.ROOM.AID_MCTS.mcts)
+                _, acts_aid, probs_aid, simul_mean_time_aid = copy_mcts.get_move_probs(session_state.ROOM.BOARD)
+                sorted_acts_probs = sorted(zip(acts_aid, probs_aid), key=lambda x: x[1], reverse=True)
+                top_five_acts = [act for act, prob in sorted_acts_probs[:5]]
+                top_five_probs = [prob for act, prob in sorted_acts_probs[:5]]
+        if response and session_state.ROOM.TURN == _BLACK:  # human turn
+            print("Your turn")
+            # construction of clickable buttons
+            cur_move = (session_state.ROOM.current_move // _BOARD_SIZE, session_state.ROOM.current_move % _BOARD_SIZE)
+            for i, row in enumerate(session_state.ROOM.BOARD.board_map):
+                # print("row:", row)
+                for j, cell in enumerate(row):
+                    if (
+                            i * _BOARD_SIZE + j
+                            in (session_state.ROOM.COORDINATE_1D)
+                    ):
+                        if i == cur_move[0] and j == cur_move[1]:
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[_NEW],
+                                key=f"{i}:{j}",
+                                args=(i, j),
+                                on_click=handle_click,
+                            )
+                        else:
+                            # disable click for GPT choices
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[cell],
+                                key=f"{i}:{j}",
+                                args=(i, j),
+                                on_click=forbid_click
+                            )
+                    else:
+                        if session_state.USE_AIAID and i * _BOARD_SIZE + j in top_five_acts:
+                            # enable click for other cells available for human choices
+                            prob = top_five_probs[top_five_acts.index(i * _BOARD_SIZE + j)]
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[cell] + f"({round(prob, 2)})",
+                                key=f"{i}:{j}",
+                                on_click=handle_click,
+                                args=(i, j),
+                            )
+                        else:
+                            # enable click for other cells available for human choices
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[cell],
+                                key=f"{i}:{j}",
+                                on_click=handle_click,
+                                args=(i, j),
+                            )
+
+
+        elif response and session_state.ROOM.TURN == _WHITE:  # AI turn
+            message.empty()
+            with st.spinner('🔮✨ Waiting for AI response... ⏳🚀'):
+                time.sleep(0.1)
+                print("AI's turn")
+                print("Below are current board under AI's view")
+                # print(session_state.ROOM.BOARD.board_map)
+                # move = _BOARD_SIZE * _BOARD_SIZE
+                # forbid = []
+                # step = 0.1
+                # tmp = 0.7
+                # while move >= _BOARD_SIZE * _BOARD_SIZE or move in session_state.ROOM.COORDINATE_1D:
+                #
+                #     gpt_predictions = generate_gpt2(
+                #         gpt2,
+                #         torch.tensor(session_state.ROOM.COORDINATE_1D).unsqueeze(0),
+                #         tmp
+                #     )
+                #     print(gpt_predictions)
+                #     move = gpt_predictions[len(session_state.ROOM.COORDINATE_1D)]
+                #     print(move)
+                #     tmp += step
+                #     # if move >= _BOARD_SIZE * _BOARD_SIZE:
+                #     #     forbid.append(move)
+                #     # else:
+                #     #     break
+                #
+                #
+                # gpt_response = move
+                # gpt_i, gpt_j = gpt_response // _BOARD_SIZE, gpt_response % _BOARD_SIZE
+                # print(gpt_i, gpt_j)
+                # # session_state.ROOM.BOARD[gpt_i][gpt_j] = session_state.ROOM.TURN
+                #
+                # simul_time = 0
+                if st.session_state['selected_oppo_model'] != 'Gomoku Bot':
+                    move, simul_time = session_state.ROOM.MCTS.get_action(session_state.ROOM.BOARD, return_time=True)
+                else:
+                    move, simul_time = session_state.ROOM.MCTS.get_action(return_time=True)
+                session_state.ROOM.simula_time_list.append(simul_time)
+                print("AI takes move: ", move)
+                session_state.ROOM.current_move = move
+                gpt_response = move
+                gpt_i, gpt_j = gpt_response // _BOARD_SIZE, gpt_response % _BOARD_SIZE
+                print("AI's move is located at ({}, {}) :".format(gpt_i, gpt_j))
+                move = session_state.ROOM.BOARD.location_to_move((gpt_i, gpt_j))
+                print("Location to move: ", move)
+                # print("Location to move: ", move)
+                # MCTS BOARD
+                session_state.ROOM.BOARD.do_move(move)
+                # Gomoku Bot BOARD
+                session_state.ROOM.MCTS_dict["Gomoku Bot"].board.put(move // _BOARD_SIZE, move % _BOARD_SIZE)
+                # session_state.ROOM.BOARD[gpt_i][gpt_j] = session_state.ROOM.TURN
+                session_state.ROOM.COORDINATE_1D.append(gpt_i * _BOARD_SIZE + gpt_j)
+
+                if not session_state.ROOM.BOARD.game_end()[0]:
+                    if session_state.USE_AIAID:
+                        copy_mcts = deepcopy(session_state.ROOM.AID_MCTS.mcts)
+                        _, acts_aid, probs_aid, simul_mean_time_aid = copy_mcts.get_move_probs(session_state.ROOM.BOARD)
+                        sorted_acts_probs = sorted(zip(acts_aid, probs_aid), key=lambda x: x[1], reverse=True)
+                        top_five_acts = [act for act, prob in sorted_acts_probs[:5]]
+                        top_five_probs = [prob for act, prob in sorted_acts_probs[:5]]
+                else:
+                    top_five_acts = []
+                    top_five_probs = []
+
+                # construction of clickable buttons
+                for i, row in enumerate(session_state.ROOM.BOARD.board_map):
+                    # print("row:", row)
+                    for j, cell in enumerate(row):
+                        if (
+                                i * _BOARD_SIZE + j
+                                in (session_state.ROOM.COORDINATE_1D)
+                        ):
+                            if i == gpt_i and j == gpt_j:
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[_NEW],
+                                    key=f"{i}:{j}",
+                                    args=(i, j),
+                                    on_click=handle_click,
+                                )
+                            else:
+                                # disable click for GPT choices
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[cell],
+                                    key=f"{i}:{j}",
+                                    args=(i, j),
+                                    on_click=forbid_click
+                                )
+                        else:
+                            if session_state.USE_AIAID and i * _BOARD_SIZE + j in top_five_acts and not session_state.ROOM.BOARD.game_end()[0]:
+                                # enable click for other cells available for human choices
+                                prob = top_five_probs[top_five_acts.index(i * _BOARD_SIZE + j)]
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[cell] + f"({round(prob, 2)})",
+                                    key=f"{i}:{j}",
+                                    on_click=handle_click,
+                                    args=(i, j),
+                                )
+                            else:
+                                # enable click for other cells available for human choices
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[cell],
+                                    key=f"{i}:{j}",
+                                    on_click=handle_click,
+                                    args=(i, j),
+                                )
+
+
+            message.markdown(
+                'AI agent has calculated its strategy, which takes <span style="color: blue; font-size: 20px;">{:.3e}</span>s per simulation.'.format(
+                    simul_time),
+                unsafe_allow_html=True
+            )
+            LOG.subheader("Logs")
+            # change turn
+            session_state.ROOM.TURN = change_turn(session_state.ROOM.TURN)
+            # session_state.ROOM.WINNER = check_win()
+
+            win, winner = session_state.ROOM.BOARD.game_end()
+            if win:
+                session_state.ROOM.WINNER = winner
+
+            session_state.ROOM.HISTORY = (
+                session_state.ROOM.HISTORY[0]
+                + int(session_state.ROOM.WINNER == _WHITE),
+                session_state.ROOM.HISTORY[1]
+                + int(session_state.ROOM.WINNER == _BLACK),
+            )
+            session_state.ROOM.TIME = time.time()
+
+        if not response or session_state.ROOM.WINNER != _BLANK:
+            if session_state.ROOM.WINNER != _BLANK:
+                print("Game over")
+            for i, row in enumerate(session_state.ROOM.BOARD.board_map):
+                for j, cell in enumerate(row):
+                    BOARD_PLATE[i][j].write(
+                        _PLAYER_SYMBOL[cell],
+                        # key=f"{i}:{j}",
+                    )
+
+    # Game process control
+    def game_control():
+        if session_state.ROOM.WINNER != _BLANK:
+            draw_board(False)
+        else:
+            draw_board(True)
+        if session_state.ROOM.WINNER != _BLANK or 0 not in session_state.ROOM.BOARD.board_map:
+            ANOTHER_ROUND.button(
+                "Play Next round!",
+                on_click=another_round,
+                help="Clear board and swap first player",
+            )
+
+    # Infos
+    def update_info() -> None:
+        # Additional information
+        SCORE_PLATE[0].metric("Gomoku-Agent", session_state.ROOM.HISTORY[0])
+        SCORE_PLATE[1].metric("Black", session_state.ROOM.HISTORY[1])
+        if session_state.ROOM.WINNER != _BLANK:
+            st.balloons()
+            ROUND_INFO.write(
+                f"#### **{_PLAYER_COLOR[session_state.ROOM.WINNER]} WIN!**\n**Click buttons on the left for more plays.**"
+            )
+
+        # elif 0 not in session_state.ROOM.BOARD.board_map:
+        #     ROUND_INFO.write("#### **Tie**")
+        # else:
+        #     ROUND_INFO.write(
+        #         f"#### **{_PLAYER_SYMBOL[session_state.ROOM.TURN]} {_PLAYER_COLOR[session_state.ROOM.TURN]}'s turn...**"
+        #     )
+
+        # draw the plot for simulation time
+        # 创建一个 DataFrame
+
+        # print(session_state.ROOM.simula_time_list)
+        st.markdown("<br>", unsafe_allow_html=True)
+        st.markdown("<br>", unsafe_allow_html=True)
+        chart_data = pd.DataFrame(session_state.ROOM.simula_time_list, columns=["Simulation Time"])
+        st.line_chart(chart_data)
+
+
+    game_control()
+    update_info()
+
+
+if __name__ == "__main__":
+    gomoku()

pages/Player_VS_AI.py

@@ -8,6 +8,7 @@ Description: this file is used to display our project and add visualization elem
 import time
 import pandas as pd
 from copy import deepcopy
+import torch
 
 # import torch
 import numpy as np
@@ -16,8 +17,12 @@ from scipy.signal import convolve  # this is used to check if any player wins
 from streamlit import session_state
 from streamlit_server_state import server_state, server_state_lock
 from Gomoku_MCTS import MCTSpure, alphazero, Board, PolicyValueNet
+from Gomoku_Bot import Gomoku_bot
+from Gomoku_Bot import Board as Gomoku_bot_board
 import matplotlib.pyplot as plt
 
+
+
 from const import (
     _BLACK,  # 1, for human
     _WHITE,  # 2 , for AI
@@ -26,6 +31,7 @@ from const import (
     _PLAYER_SYMBOL,
     _ROOM_COLOR,
     _VERTICAL,
+    _NEW,
     _HORIZONTAL,
     _DIAGONAL_UP_LEFT,
     _DIAGONAL_UP_RIGHT,
@@ -35,6 +41,13 @@ from const import (
 )
 
 
+from ai import (
+    BOS_TOKEN_ID,
+    generate_gpt2,
+    load_model,
+)
+
+gpt2 = load_model()
 
 
 # Utils
@@ -48,11 +61,14 @@ class Room:
         self.HISTORY = (0, 0)
         self.WINNER = _BLANK
         self.TIME = time.time()
-        self.MCTS_dict = {'Pure MCTS': MCTSpure(c_puct=5, n_playout=10),
-                          'AlphaZero': alphazero(PolicyValueNet(_BOARD_SIZE, _BOARD_SIZE, 'Gomoku_MCTS/checkpoints/best_policy_8_8_5_2torch.pth').policy_value_fn, c_puct=5, n_playout=100)}
+        self.gomoku_bot_board = Gomoku_bot_board(_BOARD_SIZE, 1)
+        self.MCTS_dict = {'Pure MCTS': MCTSpure(c_puct=5, n_playout=1000),
+                          'AlphaZero': alphazero(PolicyValueNet(_BOARD_SIZE, _BOARD_SIZE, 'Gomoku_MCTS/checkpoints/best_policy_8_8_5_2torch.pth').policy_value_fn, c_puct=5, n_playout=100),
+                          'Gomoku Bot': Gomoku_bot(self.gomoku_bot_board, -1)}
         self.MCTS = self.MCTS_dict['AlphaZero']
+        self.last_mcts = self.MCTS
         self.AID_MCTS = self.MCTS_dict['AlphaZero']
-        self.COORDINATE_1D = [_BOARD_SIZE_1D + 1]
+        self.COORDINATE_1D = [BOS_TOKEN_ID]
         self.current_move = -1
         self.simula_time_list = []
 
@@ -75,10 +91,16 @@ if "ROOMS" not in server_state:
         server_state.ROOMS = {}
 
 def handle_oppo_model_selection():
-    TreeNode = session_state.ROOM.MCTS.mcts._root
-    new_mct = session_state.ROOM.MCTS_dict[st.session_state['selected_oppo_model']]
-    new_mct.mcts._root = deepcopy(TreeNode)
-    session_state.ROOM.MCTS = new_mct
+    if st.session_state['selected_oppo_model'] == 'Gomoku Bot':
+        session_state.ROOM.last_mcts = session_state.ROOM.MCTS # since use different mechanism, store previous mcts first
+        session_state.ROOM.MCTS = session_state.ROOM.MCTS_dict['Gomoku Bot']
+        return
+    else:
+        TreeNode = session_state.ROOM.last_mcts.mcts._root
+        new_mct = session_state.ROOM.MCTS_dict[st.session_state['selected_oppo_model']]
+        new_mct.mcts._root = deepcopy(TreeNode)
+        session_state.ROOM.MCTS = new_mct
+        session_state.ROOM.last_mcts  = new_mct
     return
 
 def handle_aid_model_selection():
@@ -103,7 +125,7 @@ TITLE = st.empty()
 Model_Switch = st.empty()
 
 TITLE.header("🤖 AI 3603 Gomoku")
-selected_oppo_option = Model_Switch.selectbox('Select Opponent Model', ['Pure MCTS', 'AlphaZero'], index=1, key='oppo_model')
+selected_oppo_option = Model_Switch.selectbox('Select Opponent Model', ['Pure MCTS', 'AlphaZero','Gomoku Bot'], index=1, key='oppo_model')
 
 if st.session_state['selected_oppo_model'] != selected_oppo_option:
     st.session_state['selected_oppo_model'] = selected_oppo_option
@@ -158,7 +180,7 @@ def restart() -> None:
     Restart the game.
     """
     session_state.ROOM = Room(session_state.ROOM.ROOM_ID)
-
+    st.session_state['selected_oppo_model'] = 'AlphaZero'
 
 RESTART.button(
     "Reset",
@@ -183,10 +205,16 @@ def gomoku():
         """
         session_state.ROOM = deepcopy(session_state.ROOM)
         session_state.ROOM.BOARD = Board(width=_BOARD_SIZE, height=_BOARD_SIZE, n_in_row=5)
+        session_state.ROOM.gomoku_bot_board = Gomoku_bot_board(_BOARD_SIZE, 1)
+        session_state.ROOM.MCTS_dict = {'Pure MCTS': MCTSpure(c_puct=5, n_playout=1000),
+                          'AlphaZero': alphazero(PolicyValueNet(_BOARD_SIZE, _BOARD_SIZE, 'Gomoku_MCTS/checkpoints/best_policy_8_8_5_2torch.pth').policy_value_fn, c_puct=5, n_playout=100),
+                          'Gomoku Bot': Gomoku_bot(session_state.ROOM.gomoku_bot_board, -1)}
+        session_state.ROOM.MCTS = session_state.ROOM.MCTS_dict[st.session_state['selected_oppo_model']]
+        session_state.ROOM.last_mcts = session_state.ROOM.MCTS
         session_state.ROOM.PLAYER = session_state.ROOM.PLAYER
         session_state.ROOM.TURN = session_state.ROOM.PLAYER
         session_state.ROOM.WINNER = _BLANK  # 0
-        session_state.ROOM.COORDINATE_1D = [_BOARD_SIZE_1D + 1]
+        session_state.ROOM.COORDINATE_1D = [BOS_TOKEN_ID]
 
     # Room status sync
     def sync_room() -> bool:
@@ -281,6 +309,8 @@ def gomoku():
             move = session_state.ROOM.BOARD.location_to_move((x, y))
             session_state.ROOM.current_move = move
             session_state.ROOM.BOARD.do_move(move)
+            # Gomoku Bot BOARD
+            session_state.ROOM.MCTS_dict["Gomoku Bot"].board.put(move // _BOARD_SIZE, move % _BOARD_SIZE)
             session_state.ROOM.BOARD.board_map[x][y] = session_state.ROOM.TURN
             session_state.ROOM.COORDINATE_1D.append(x * _BOARD_SIZE + y)
 
@@ -299,7 +329,6 @@ def gomoku():
     def forbid_click(x, y):
         # st.warning('This posistion has been occupied!!!!', icon="⚠️")
         st.error("({}, {}) has been occupied!!)".format(x, y), icon="🚨")
-        print("asdas")
 
     # Draw board
     def draw_board(response: bool):
@@ -314,6 +343,7 @@ def gomoku():
         if response and session_state.ROOM.TURN == _BLACK:  # human turn
             print("Your turn")
             # construction of clickable buttons
+            cur_move = (session_state.ROOM.current_move // _BOARD_SIZE, session_state.ROOM.current_move % _BOARD_SIZE)
             for i, row in enumerate(session_state.ROOM.BOARD.board_map):
                 # print("row:", row)
                 for j, cell in enumerate(row):
@@ -321,13 +351,21 @@ def gomoku():
                             i * _BOARD_SIZE + j
                             in (session_state.ROOM.COORDINATE_1D)
                     ):
-                        # disable click for GPT choices
-                        BOARD_PLATE[i][j].button(
-                            _PLAYER_SYMBOL[cell],
-                            key=f"{i}:{j}",
-                            args=(i, j),
-                            on_click=forbid_click
-                        )
+                        if i == cur_move[0] and j == cur_move[1]:
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[_NEW],
+                                key=f"{i}:{j}",
+                                args=(i, j),
+                                on_click=handle_click,
+                            )
+                        else:
+                            # disable click for GPT choices
+                            BOARD_PLATE[i][j].button(
+                                _PLAYER_SYMBOL[cell],
+                                key=f"{i}:{j}",
+                                args=(i, j),
+                                on_click=forbid_click
+                            )
                     else:
                         if session_state.USE_AIAID and i * _BOARD_SIZE + j in top_five_acts:
                             # enable click for other cells available for human choices
@@ -355,7 +393,37 @@ def gomoku():
                 print("AI's turn")
                 print("Below are current board under AI's view")
                 # print(session_state.ROOM.BOARD.board_map)
-                move, simul_time = session_state.ROOM.MCTS.get_action(session_state.ROOM.BOARD, return_time=True)
+                # move = _BOARD_SIZE * _BOARD_SIZE
+                # forbid = []
+                # step = 0.1
+                # tmp = 0.7
+                # while move >= _BOARD_SIZE * _BOARD_SIZE or move in session_state.ROOM.COORDINATE_1D:
+                #
+                #     gpt_predictions = generate_gpt2(
+                #         gpt2,
+                #         torch.tensor(session_state.ROOM.COORDINATE_1D).unsqueeze(0),
+                #         tmp
+                #     )
+                #     print(gpt_predictions)
+                #     move = gpt_predictions[len(session_state.ROOM.COORDINATE_1D)]
+                #     print(move)
+                #     tmp += step
+                #     # if move >= _BOARD_SIZE * _BOARD_SIZE:
+                #     #     forbid.append(move)
+                #     # else:
+                #     #     break
+                #
+                #
+                # gpt_response = move
+                # gpt_i, gpt_j = gpt_response // _BOARD_SIZE, gpt_response % _BOARD_SIZE
+                # print(gpt_i, gpt_j)
+                # # session_state.ROOM.BOARD[gpt_i][gpt_j] = session_state.ROOM.TURN
+                #
+                # simul_time = 0
+                if st.session_state['selected_oppo_model'] != 'Gomoku Bot':
+                    move, simul_time = session_state.ROOM.MCTS.get_action(session_state.ROOM.BOARD, return_time=True)
+                else:
+                    move, simul_time = session_state.ROOM.MCTS.get_action(return_time=True)
                 session_state.ROOM.simula_time_list.append(simul_time)
                 print("AI takes move: ", move)
                 session_state.ROOM.current_move = move
@@ -364,7 +432,11 @@ def gomoku():
                 print("AI's move is located at ({}, {}) :".format(gpt_i, gpt_j))
                 move = session_state.ROOM.BOARD.location_to_move((gpt_i, gpt_j))
                 print("Location to move: ", move)
+                # print("Location to move: ", move)
+                # MCTS BOARD
                 session_state.ROOM.BOARD.do_move(move)
+                # Gomoku Bot BOARD
+                session_state.ROOM.MCTS_dict["Gomoku Bot"].board.put(move // _BOARD_SIZE, move % _BOARD_SIZE)
                 # session_state.ROOM.BOARD[gpt_i][gpt_j] = session_state.ROOM.TURN
                 session_state.ROOM.COORDINATE_1D.append(gpt_i * _BOARD_SIZE + gpt_j)
 
@@ -387,13 +459,21 @@ def gomoku():
                                 i * _BOARD_SIZE + j
                                 in (session_state.ROOM.COORDINATE_1D)
                         ):
-                            # disable click for GPT choices
-                            BOARD_PLATE[i][j].button(
-                                _PLAYER_SYMBOL[cell],
-                                key=f"{i}:{j}",
-                                args=(i, j),
-                                on_click=forbid_click
-                            )
+                            if i == gpt_i and j == gpt_j:
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[_NEW],
+                                    key=f"{i}:{j}",
+                                    args=(i, j),
+                                    on_click=handle_click,
+                                )
+                            else:
+                                # disable click for GPT choices
+                                BOARD_PLATE[i][j].button(
+                                    _PLAYER_SYMBOL[cell],
+                                    key=f"{i}:{j}",
+                                    args=(i, j),
+                                    on_click=forbid_click
+                                )
                         else:
                             if session_state.USE_AIAID and i * _BOARD_SIZE + j in top_five_acts and not session_state.ROOM.BOARD.game_end()[0]:
                                 # enable click for other cells available for human choices
@@ -480,7 +560,7 @@ def gomoku():
         # draw the plot for simulation time
         # 创建一个 DataFrame
 
-        print(session_state.ROOM.simula_time_list)
+        # print(session_state.ROOM.simula_time_list)
         st.markdown("<br>", unsafe_allow_html=True)
         st.markdown("<br>", unsafe_allow_html=True)
         chart_data = pd.DataFrame(session_state.ROOM.simula_time_list, columns=["Simulation Time"])

try.py

@@ -0,0 +1,42 @@
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+
+model_name_or_path = "TheBloke/Llama-2-7b-Chat-GPTQ"
+# To use a different branch, change revision
+# For example: revision="gptq-4bit-64g-actorder_True"
+model = AutoModelForCausalLM.from_pretrained(model_name_or_path,
+                                             device_map="auto",
+                                             trust_remote_code=False,
+                                             revision="main")
+
+tokenizer = AutoTokenizer.from_pretrained(model_name_or_path, use_fast=False)
+
+prompt = "Tell me about AI"
+prompt_template=f'''[INST] <<SYS>>
+You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe.  Your answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure that your responses are socially unbiased and positive in nature. If a question does not make any sense, or is not factually coherent, explain why instead of answering something not correct. If you don't know the answer to a question, please don't share false information.
+<</SYS>>
+{prompt}[/INST]
+
+'''
+
+print("\n\n*** Generate:")
+
+input_ids = tokenizer(prompt_template, return_tensors='pt').input_ids.cuda()
+output = model.generate(inputs=input_ids, temperature=0.7, do_sample=True, top_p=0.95, top_k=40, max_new_tokens=512)
+print(tokenizer.decode(output[0]))
+
+# Inference can also be done using transformers' pipeline
+
+print("*** Pipeline:")
+pipe = pipeline(
+    "text-generation",
+    model=model,
+    tokenizer=tokenizer,
+    max_new_tokens=512,
+    do_sample=True,
+    temperature=0.7,
+    top_p=0.95,
+    top_k=40,
+    repetition_penalty=1.1
+)
+
+print(pipe(prompt_template)[0]['generated_text'])