added stockfish elo parameter
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@@ -7,4 +7,5 @@ __pycache__
|
||||
|
||||
# Chess Engines
|
||||
/stockfish/
|
||||
/lc0/
|
||||
/lc0/
|
||||
/sunfish-master/
|
||||
|
||||
@@ -5,6 +5,7 @@ from abc import ABC, abstractmethod
|
||||
from torch import distributions as dist
|
||||
import chess
|
||||
import chess.engine
|
||||
from stockfish import Stockfish
|
||||
|
||||
from chesspp.mcts.baysian_mcts import BayesianMcts
|
||||
from chesspp.mcts.classic_mcts import ClassicMcts
|
||||
@@ -147,12 +148,15 @@ class RandomEngine(Engine):
|
||||
|
||||
|
||||
class StockFishEngine(Engine):
|
||||
def __init__(self, board: chess.Board, color: chess, path="../stockfish/stockfish-ubuntu-x86-64-avx2"):
|
||||
def __init__(self, board: chess.Board, color: chess, stockfish_elo: int, path="../stockfish/stockfish-ubuntu-x86-64-avx2"):
|
||||
super().__init__(board, color, None)
|
||||
self.stockfish = chess.engine.SimpleEngine.popen_uci(path)
|
||||
self.stockfish = Stockfish(path)
|
||||
self.stockfish.set_elo_rating(stockfish_elo)
|
||||
|
||||
def play(self, board: chess.Board, limit: Limit) -> chess.engine.PlayResult:
|
||||
return self.stockfish.play(board, limit.translate_to_engine_limit())
|
||||
self.stockfish.set_fen_position(board.fen())
|
||||
m = chess.Move.from_uci(self.stockfish.get_best_move())
|
||||
return chess.engine.PlayResult(move=m, ponder=None)
|
||||
|
||||
@staticmethod
|
||||
def get_name() -> str:
|
||||
|
||||
@@ -29,7 +29,8 @@ class StrategyEnum(Enum):
|
||||
class EngineFactory:
|
||||
|
||||
@staticmethod
|
||||
def create_engine(engine_name: EngineEnum, strategy_name: StrategyEnum, color: chess.Color, stockfish_path: str, lc0_path: str, rollout_depth: int = 4) -> Engine:
|
||||
def create_engine(engine_name: EngineEnum, strategy_name: StrategyEnum, color: chess.Color, stockfish_path: str,
|
||||
lc0_path: str, stockfish_elo: int, rollout_depth: int = 4) -> Engine:
|
||||
match strategy_name:
|
||||
case StrategyEnum.Stockfish:
|
||||
strategy = EngineFactory._get_stockfish_strategy(stockfish_path, rollout_depth)
|
||||
@@ -50,14 +51,14 @@ class EngineFactory:
|
||||
return EngineFactory.bayesian_mcts(color, strategy)
|
||||
|
||||
case EngineEnum.Stockfish:
|
||||
return EngineFactory.stockfish_engine(color, stockfish_path)
|
||||
return EngineFactory.stockfish_engine(color, stockfish_path, stockfish_elo)
|
||||
|
||||
case EngineEnum.Lc0:
|
||||
return EngineFactory.lc0_engine(color, lc0_path)
|
||||
|
||||
@staticmethod
|
||||
def stockfish_engine(color: chess.Color, engine_path: str, board: chess.Board | None = chess.Board()) -> Engine:
|
||||
return StockFishEngine(board, color, engine_path)
|
||||
def stockfish_engine(color: chess.Color, engine_path: str, stockfish_elo: int, board: chess.Board | None = chess.Board()) -> Engine:
|
||||
return StockFishEngine(board, color, stockfish_elo, engine_path)
|
||||
|
||||
@staticmethod
|
||||
def lc0_engine(color: chess.Color, engine_path: str, board: chess.Board | None = chess.Board()) -> Engine:
|
||||
@@ -90,4 +91,3 @@ class EngineFactory:
|
||||
@staticmethod
|
||||
def _get_pesto_strategy(rollout_depth: int) -> IStrategy:
|
||||
return PestoStrategy(rollout_depth)
|
||||
|
||||
|
||||
@@ -229,6 +229,13 @@ def score(board: chess.Board) -> int:
|
||||
eg = [0, 0]
|
||||
game_phase = 0
|
||||
|
||||
if board.outcome() is not None:
|
||||
winner = board.outcome().winner
|
||||
if winner is not None:
|
||||
if winner == chess.WHITE:
|
||||
return 100_000
|
||||
else:
|
||||
return -100_000
|
||||
# evaluate each piece
|
||||
for sq in range(64):
|
||||
pc = board.piece_at(sq)
|
||||
|
||||
@@ -37,7 +37,7 @@ def simulate_game(white: Engine, black: Engine, limit: Limit, board: chess.Board
|
||||
|
||||
class Evaluation:
|
||||
def __init__(self, engine_a: EngineEnum, strategy_a, engine_b: EngineEnum, strategy_b, limit: Limit,
|
||||
stockfish_path: str, lc0_path: str):
|
||||
stockfish_path: str, lc0_path: str, stockfish_elo: int):
|
||||
self.engine_a = engine_a
|
||||
self.strategy_a = strategy_a
|
||||
self.engine_b = engine_b
|
||||
@@ -45,10 +45,11 @@ class Evaluation:
|
||||
self.stockfish_path = stockfish_path
|
||||
self.lc0_path = lc0_path
|
||||
self.limit = limit
|
||||
self.stockfish_elo = stockfish_elo
|
||||
|
||||
def run(self, n_games=100, proc=mp.cpu_count()) -> List[EvaluationResult]:
|
||||
proc = min(proc, mp.cpu_count())
|
||||
arg = (self.engine_a, self.strategy_a, self.engine_b, self.strategy_b, self.limit, self.stockfish_path, self.lc0_path)
|
||||
arg = (self.engine_a, self.strategy_a, self.engine_b, self.strategy_b, self.limit, self.stockfish_path, self.lc0_path, self.stockfish_elo)
|
||||
if proc > 1:
|
||||
with mp.Pool(proc) as pool:
|
||||
args = [arg for i in range(n_games)]
|
||||
@@ -59,18 +60,18 @@ class Evaluation:
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _test_simulate(arg: Tuple[EngineEnum, StrategyEnum, EngineEnum, StrategyEnum, Limit, str, str]) -> EvaluationResult:
|
||||
engine_a, strategy_a, engine_b, strategy_b, limit, stockfish_path, lc0_path = arg
|
||||
def _test_simulate(arg: Tuple[EngineEnum, StrategyEnum, EngineEnum, StrategyEnum, Limit, str, str, int]) -> EvaluationResult:
|
||||
engine_a, strategy_a, engine_b, strategy_b, limit, stockfish_path, lc0_path, stockfish_elo = arg
|
||||
flip_engines = bool(random.getrandbits(1))
|
||||
|
||||
if flip_engines:
|
||||
black, white = EngineFactory.create_engine(engine_a, strategy_a, chess.BLACK,
|
||||
stockfish_path, lc0_path), EngineFactory.create_engine(
|
||||
engine_b, strategy_b, chess.WHITE, stockfish_path, lc0_path)
|
||||
stockfish_path, lc0_path, stockfish_elo), EngineFactory.create_engine(
|
||||
engine_b, strategy_b, chess.WHITE, stockfish_path, lc0_path, stockfish_elo)
|
||||
else:
|
||||
white, black = EngineFactory.create_engine(engine_a, strategy_a, chess.WHITE,
|
||||
stockfish_path, lc0_path), EngineFactory.create_engine(
|
||||
engine_b, strategy_b, chess.BLACK, stockfish_path, lc0_path)
|
||||
stockfish_path, lc0_path, stockfish_elo), EngineFactory.create_engine(
|
||||
engine_b, strategy_b, chess.BLACK, stockfish_path, lc0_path, stockfish_elo)
|
||||
|
||||
game = simulate_game(white, black, limit, chess.Board())
|
||||
winner = game.end().board().outcome().winner
|
||||
|
||||
@@ -25,6 +25,7 @@ def load_index() -> str:
|
||||
|
||||
class Simulate:
|
||||
""" Run a simulation of two engines"""
|
||||
|
||||
def __init__(self, engine_white, engine_black):
|
||||
self.white = engine_white
|
||||
self.black = engine_black
|
||||
@@ -41,7 +42,8 @@ class Simulate:
|
||||
|
||||
|
||||
class WebInterface:
|
||||
def __init__(self, white_engine, black_engine, strategy1, strategy2, stockfish_path, lc0_path, limit: engine.Limit):
|
||||
def __init__(self, white_engine, black_engine, strategy1, strategy2, stockfish_path, lc0_path, limit: engine.Limit,
|
||||
stockfish_elo: int):
|
||||
self.white = white_engine
|
||||
self.black = black_engine
|
||||
self.strategy1 = strategy1
|
||||
@@ -49,19 +51,17 @@ class WebInterface:
|
||||
self.stockfish_path = stockfish_path
|
||||
self.lc0_path = lc0_path
|
||||
self.limit = limit
|
||||
|
||||
self.stockfish_elo = stockfish_elo
|
||||
|
||||
async def handle_index(self, request) -> web.Response:
|
||||
""" Entry point of webpage, returns the index html"""
|
||||
return web.Response(text=load_index(), content_type='text/html')
|
||||
|
||||
|
||||
async def handle_websocket(self, request):
|
||||
""" Handles a websocket connection to the frontend"""
|
||||
ws = web.WebSocketResponse()
|
||||
await ws.prepare(request)
|
||||
|
||||
|
||||
async def wait_msg():
|
||||
""" Handles messages from client """
|
||||
async for msg in ws:
|
||||
@@ -71,12 +71,14 @@ class WebInterface:
|
||||
elif msg.type == aiohttp.WSMsgType.ERROR:
|
||||
print(f'ws connection closed with exception {ws.exception()}')
|
||||
|
||||
|
||||
async def turns():
|
||||
""" Simulates the game and sends the response to the client """
|
||||
white = EngineFactory.create_engine(self.white, self.strategy1, chess.WHITE, self.stockfish_path, self.lc0_path)
|
||||
black = EngineFactory.create_engine(self.black, self.strategy2, chess.BLACK, self.stockfish_path, self.lc0_path)
|
||||
white = EngineFactory.create_engine(self.white, self.strategy1, chess.WHITE, self.stockfish_path,
|
||||
self.lc0_path, self.stockfish_elo)
|
||||
black = EngineFactory.create_engine(self.black, self.strategy2, chess.BLACK, self.stockfish_path,
|
||||
self.lc0_path, self.stockfish_elo)
|
||||
runner = Simulate(white, black).run(self.limit)
|
||||
|
||||
def sim():
|
||||
return next(runner, None)
|
||||
|
||||
@@ -85,12 +87,10 @@ class WebInterface:
|
||||
await ws.send_str(board.fen())
|
||||
board = await asyncio.to_thread(sim)
|
||||
|
||||
|
||||
async with asyncio.TaskGroup() as tg:
|
||||
tg.create_task(wait_msg())
|
||||
tg.create_task(turns())
|
||||
|
||||
|
||||
print('websocket connection closed')
|
||||
return ws
|
||||
|
||||
|
||||
18
main.py
18
main.py
@@ -90,10 +90,10 @@ def analyze_results(moves: dict):
|
||||
|
||||
|
||||
def test_evaluation():
|
||||
a, b, s1, s2, n, limit, stockfish_path, lc0_path, proc = read_arguments()
|
||||
limit = engine.Limit(time=limit)
|
||||
a, b, s1, s2, n, limit, stockfish_path, lc0_path, proc, nodes, stockfish_elo = read_arguments()
|
||||
limit = engine.Limit(time=limit) if limit != -1 else engine.Limit(nodes=nodes)
|
||||
|
||||
evaluator = simulation.Evaluation(a, s1, b, s2, limit, stockfish_path, lc0_path)
|
||||
evaluator = simulation.Evaluation(a, s1, b, s2, limit, stockfish_path, lc0_path, stockfish_elo)
|
||||
results = evaluator.run(n, proc)
|
||||
games_played = len(results)
|
||||
a_wins = len(list(filter(lambda x: x.winner == simulation.Winner.Engine_A, results)))
|
||||
@@ -130,10 +130,12 @@ def read_arguments():
|
||||
lc0_default = "lc0/lc0"
|
||||
|
||||
parser.add_argument("--proc", default=2, help="Number of processors to use for simulation, default=1")
|
||||
parser.add_argument("--time", default=0.5, help="Time limit for each simulation step, default=0.5")
|
||||
parser.add_argument("--time", default=-1, help="Time limit for each simulation step, default=-1")
|
||||
parser.add_argument("--nodes", default=-1, help="Node limit for each simulation step, default=-1")
|
||||
parser.add_argument("-n", default=100, help="Number of games to simulate, default=100")
|
||||
parser.add_argument("--stockfish_path", default=stockfish_default,
|
||||
help=f"Path for engine executable, default='{stockfish_default}'")
|
||||
parser.add_argument("--stockfish_elo", default=1500, help="Elo for stockfish engine, default=1500")
|
||||
parser.add_argument("--lc0_path", default=lc0_default,
|
||||
help=f"Path for engine executable, default='{stockfish_default}'")
|
||||
parser.add_argument("--engine1", "--e1", help="Engine A for the simulation", choices=engines.keys(), required=True)
|
||||
@@ -151,10 +153,10 @@ def read_arguments():
|
||||
strategy1 = strategies[args.strategy1]
|
||||
strategy2 = strategies[args.strategy2]
|
||||
|
||||
print(engine1, engine2, strategy1, strategy2, int(args.n), float(args.time), args.stockfish_path, args.lc0_path,
|
||||
int(args.proc))
|
||||
return engine1, engine2, strategy1, strategy2, int(args.n), float(
|
||||
args.time), args.stockfish_path, args.lc0_path, int(args.proc)
|
||||
print(engine1, engine2, strategy1, strategy2, int(args.n), args.time, args.stockfish_path, args.lc0_path,
|
||||
int(args.proc), int(args.nodes), int(args.stockfish_elo))
|
||||
return (engine1, engine2, strategy1, strategy2, int(args.n), float(args.time),
|
||||
args.stockfish_path, args.lc0_path, int(args.proc), int(args.nodes), int(args.stockfish_elo))
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
7
web.py
7
web.py
@@ -2,8 +2,7 @@ from chesspp import engine
|
||||
from chesspp import web
|
||||
from main import read_arguments
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
engine1, engine2, strategy1, strategy2, n_games, time, stockfish_path, lc0_path, n_proc = read_arguments()
|
||||
limit = engine.Limit(time=time)
|
||||
web.WebInterface(engine1, engine2, strategy1, strategy2, stockfish_path, lc0_path, limit).run_app()
|
||||
engine1, engine2, strategy1, strategy2, n_games, time, stockfish_path, lc0_path, n_proc, nodes, stockfish_elo = read_arguments()
|
||||
limit = engine.Limit(time=time) if time != -1 else engine.Limit(nodes=nodes)
|
||||
web.WebInterface(engine1, engine2, strategy1, strategy2, stockfish_path, lc0_path, limit, stockfish_elo).run_app()
|
||||
|
||||
Reference in New Issue
Block a user