Merge pull request #1 from Tesla2000/feature/optimization

Feature/optimization

Author: Tesla2000

.idea/Splendor.iml

@@ -0,0 +1,15 @@
+import random
+
+import numpy as np
+import torch
+
+random.seed(42)
+np.random.seed(42)
+torch.random.manual_seed(42)
+
+
+class Config:
+    min_n_points_to_finish = 15
+    n_simulations = 100
+    n_games = 1
+    n_players = 2

.idea/misc.xml

@@ -0,0 +1,44 @@
+from itertools import pairwise, starmap
+
+import numpy as np
+from torch import nn, Tensor
+
+
+class Agent(nn.Module):
+    _input_size_dictionary = {
+        2: 215,
+    }
+
+    def __init__(
+        self,
+        n_players: int,
+        hidden_sizes: tuple = (256, 128, 64, 32),
+        n_moves: int = 46,
+    ):
+        super().__init__()
+        self.relu = nn.ReLU()
+        self.tanh = nn.Tanh()
+        self.softmax = nn.Softmax(dim=1)
+        first_size = self._get_size(n_players)
+        sizes = first_size, *hidden_sizes
+        self.layers = tuple(starmap(nn.Linear, pairwise(sizes)))
+        for index, layer in enumerate(self.layers):
+            setattr(self, f"layer_{index}", layer)
+        self.fc_v = nn.Linear(hidden_sizes[-1], 1)
+        self.fc_p = nn.Linear(hidden_sizes[-1], n_moves)
+        self._n_moves = n_moves
+        self._trained = False
+
+    def _get_size(self, n_players: int) -> int:
+        return self._input_size_dictionary[n_players]
+
+    def forward(self, state: Tensor):
+        if not self.training and not self._trained:
+            return self.softmax(Tensor(np.random.random((1, self._n_moves)))), Tensor(
+                np.random.uniform(-1, 1, (1, 1))
+            )
+        self._trained = True
+        for layer in self.layers:
+            state = layer(state)
+            state = self.relu(state)
+        return self.softmax(self.fc_p(state)), self.tanh(self.fc_v(state))

.idea/vcs.xml

@@ -0,0 +1,87 @@
+from collections import defaultdict
+from dataclasses import astuple
+from math import sqrt
+
+import numpy as np
+from torch import nn, Tensor
+from tqdm import tqdm
+
+from Config import Config
+from src.Game import Game
+from .Agent import Agent
+
+
+def train_agent():
+    agent = Agent(Config.n_players)
+    agent.eval()
+    examples = []
+    examples_per_game = []
+    for i in range(Config.n_games):
+        game = Game(n_players=Config.n_players)
+        while True:
+            pi, action = policy(game, agent, 1, Config.n_simulations)
+            examples_per_game.append((game, pi, 0))
+            game = game.perform(action)
+            print(len(game.players[1].cards), game.players[1].points)
+            if game.is_terminal():
+                for example in examples_per_game:
+                    example[2] = game.get_state()
+                break
+        examples += examples_per_game
+        break
+    return examples
+
+
+def search(
+    game: Game,
+    agent: nn.Module,
+    c: float,
+    N: defaultdict,
+    visited: set,
+    P: defaultdict,
+    Q: defaultdict,
+):
+    state = game.get_state()
+    if game.is_terminal():
+        return game.get_results()[game.current_player]
+    if state not in visited:
+        visited.add(state)
+        move_scores, v = agent(Tensor([state]))
+        tuple(
+            P[state].__setitem__(move, move_scores[0, index])
+            for index, move in enumerate(game.all_moves)
+        )
+        return -v
+
+    action = max(
+        game.get_possible_actions(),
+        key=lambda action: Q[state].get(action, 1)
+        + c * P[state][action] * sqrt(sum(N[state].values())) / (1 + N[state][action]),
+    )
+
+    next_game_state = game.perform(action)
+    v = search(next_game_state, agent, c, N, visited, P, Q)
+
+    Q[state][action] = (N[state][action] * Q[state].get(action, 1) + v) / (
+        N[state][action] + 1
+    )
+    N[state][action] += 1
+    return -v
+
+
+def policy(
+    game: Game,
+    agent: nn.Module,
+    c: float,
+    n_simulations: int,
+):
+    N = defaultdict(lambda: defaultdict(int))
+    visited = set()
+    P = defaultdict(dict)
+    Q = defaultdict(dict)
+    initial_state = game.get_state()
+    all_moves = game.get_possible_actions()
+    for _ in tqdm(range(n_simulations)):
+        search(game, agent, c, N, visited, P, Q)
+    pi = [N[initial_state][a] for a in all_moves]
+    return pi, all_moves[np.argmax(pi)]

Config.py

@@ -0,0 +1,3 @@
+class hashabledict(dict):
+    def __hash__(self):
+        return tuple(self.items()).__hash__()

agent/Agent.py

@@ -1,8 +1,4 @@
-import random
+from agent.train_agent import train_agent
 
-from src.Game import Game
-
-if __name__ == '__main__':
-    game = Game()
-    while not game.is_terminal():
-        game.perform(random.choice(game.get_possible_actions()))
+if __name__ == "__main__":
+    train_agent()

agent/__init__.py

@@ -1,15 +1,28 @@
-from dataclasses import astuple, fields, asdict, dataclass, field
+from dataclasses import fields, dataclass, field
 from itertools import combinations, starmap, product
-from typing import Self, Iterable, Any
-from dacite import from_dict
+from typing import Self, Type
 
+from Config import Config
+from .StateExtractor import StateExtractor
+from .entities.AllResources import AllResources
 from .entities.BasicResources import BasicResources
 from .entities.Board import Board
-from .entities.Card import empty_card
 from .entities.Player import Player
 from .entities.Tier import Tier
-from .moves import Move, GrabThreeResource, GrabTwoResource, BuildBoard, BuildReserve, ReserveVisible, ReserveTop, \
-    NullMove
+from .entities.extended_lists.Aristocrats import Aristocrats
+from .entities.extended_lists.PlayerAristocrats import PlayerAristocrats
+from .entities.extended_lists.PlayerCards import PlayerCards
+from .entities.extended_lists.PlayerReserve import PlayerReserve
+from .moves import (
+    Move,
+    GrabThreeResource,
+    GrabTwoResource,
+    BuildBoard,
+    BuildReserve,
+    ReserveVisible,
+    ReserveTop,
+    NullMove,
+)
 
 
 @dataclass(slots=True)
@@ -22,24 +35,24 @@ class Game:
     _turn_counter: int = 0
     _performed_the_last_move: dict = None
     _last_turn: bool = False
+    _state_extractor: Type[StateExtractor] = StateExtractor
 
     def __post_init__(self):
-        if not self.board or not self.players:
+        if not self.players:
             self.players = tuple(Player() for _ in range(self.n_players))
+        if not self.board:
             self.board = Board(self.n_players)
+        if not self._performed_the_last_move:
             self._performed_the_last_move = dict(
                 (player, False) for player in self.players
             )
-            self.is_blocked = dict(
-                (player, False) for player in self.players
-            )
-            self._last_turn = False
+            self.is_blocked = dict((player, False) for player in self.players)
         self.current_player = self.players[0]
 
     def perform(self, action: Move) -> Self:
-        action.perform(self)
-        self.next_turn()
-        return self
+        new_state = action.perform(self)
+        new_state.next_turn()
+        return new_state
 
     def next_turn(self) -> None:
         self.players = (*self.players[1:], self.players[0])
@@ -48,73 +61,76 @@ def next_turn(self) -> None:
                 self.current_player.aristocrats.append(
                     self.board.aristocrats.pop(index)
                 )
-        if self.current_player.points >= 15 or self._last_turn:
+        if self.current_player.points >= Config.min_n_points_to_finish or self._last_turn:
             self._last_turn = True
         self._performed_the_last_move[self.current_player] = self._last_turn
         self.current_player = self.players[0]
-        self._turn_counter += 1
 
     def is_terminal(self) -> bool:
         return all(self._performed_the_last_move.values()) or (
             not self.get_possible_actions()
         )
 
-    def get_results(self) -> dict[Player, bool]:
+    def get_results(self) -> dict[Player, int]:
         results = {}
         for player in self.players:
-            if not all(self._performed_the_last_move.values()):
-                results[player] = player == max(self.players, key=lambda p: (p.points, -len(p.cards)))
-            else:
-                print("Finished game")
+            results[player] = (
+                1
+                if player
+                   == max(self.players, key=lambda p: (p.points, -len(p.cards)))
+                else -1
+            )
         return results
 
     def get_state(self) -> tuple:
-        tiers = self.board.tiers
-        self.board.tiers = list(Tier([], tier.visible) for tier in tiers)
-        state = self._flatter_recursively(astuple(self.board))
-        self.board.tiers = tiers
-        for player in self.players:
-            state += astuple(player.resources, tuple_factory=list)
-            state += astuple(player.production, tuple_factory=list)
-            if player != self.current_player:
-                state.append(sum(card != empty_card for card in player.reserve))
-            else:
-                state += self._flatter_recursively(map(astuple, self.current_player.reserve))
-            state.append(player.points)
-        return tuple(state)
+        return self._state_extractor.get_state(self)
 
     def copy(self) -> Self:
-        game = from_dict(Game, asdict(self))
+        game = Game(
+            players=tuple(
+                Player(
+                    resources=AllResources(
+                        (resources := player.resources).red,
+                        resources.green,
+                        resources.blue,
+                        resources.black,
+                        resources.white,
+                        resources.gold,
+                    ),
+                    cards=PlayerCards(player.cards),
+                    reserve=PlayerReserve(player.reserve),
+                    aristocrats=PlayerAristocrats(player.aristocrats),
+                )
+                for player in self.players
+            ),
+            board=Board(
+                n_players=(board := self.board).n_players,
+                tiers=list(Tier(list(tier.hidden), list(tier.visible)) for tier in board.tiers),
+                aristocrats=Aristocrats(board.aristocrats),
+                resources=AllResources(
+                    board.resources.red,
+                    board.resources.green,
+                    board.resources.blue,
+                    board.resources.black,
+                    board.resources.white,
+                    board.resources.gold,
+                ),
+            ),
+            n_players=self.n_players,
+        )
         game.current_player = game.players[0]
+        for player in game.players:
+            game.is_blocked[player] = next(
+                value for key, value in self.is_blocked.items() if key == player
+            )
+            game._performed_the_last_move[player] = next(
+                value for key, value in self._performed_the_last_move.items() if key == player
+            )
         return game
 
     def get_possible_actions(self) -> list[Move]:
         return list(move for move in self.all_moves if move.is_valid(self))
 
-    def _flatter_recursively(
-        self, iterable: Iterable, output: list = None, expected_length: int = None
-    ) -> list:
-        if output is None:
-            if expected_length:
-                output = expected_length * [None]
-        if not expected_length:
-            return list(self._get_flatten_elements(iterable))
-        index = 0
-        for index, item in enumerate(self._get_flatten_elements(iterable)):
-            if expected_length is None:
-                output[index] = item
-        if index != expected_length - 1:
-            raise ValueError
-        return output
-
-    def _get_flatten_elements(self, iterable: Iterable) -> Any:
-        for element in iterable:
-            if isinstance(element, Iterable):
-                for inner_element in self._get_flatten_elements(element):
-                    yield inner_element
-            else:
-                yield element
-
     combos = combinations([{field.name: 1} for field in fields(BasicResources)], 3)
     all_moves = list(
         GrabThreeResource(BasicResources(**res_1, **res_2, **res_3))
@@ -130,4 +146,3 @@ def _get_flatten_elements(self, iterable: Iterable) -> Any:
     all_moves += list(starmap(ReserveVisible, product(range(3), range(4))))
     all_moves += list(map(ReserveTop, range(3)))
     all_moves.append(NullMove())
-    all_moves = tuple(all_moves)