Play chess against large language models.
#!/usr/bin/env python3

## Copyright (C) 2023, Nicholas Carlini <nicholas@carlini.com>.
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see <http://www.gnu.org/licenses/>.

import requests
import json

import os
import chess
import chess.engine
import chess.pgn
import random
import pickle
import sys

class ChessLLM:
    def __init__(self, api_key, config, **override):
        self.config = config
        for k,v in override.items():
            config[k] = v
        if os.path.exists("cache.p"):
            with open("cache.p", "rb") as f:
                self.cache = pickle.load(f)
        else:
            self.cache = {}
        print("Loading cache with", len(self.cache), "entries")
        self.api_key = api_key

    def get_query_pgn(self, board):
        pgn = str(chess.pgn.Game().from_board(board))

        if board.outcome() is None:
            pgn = pgn[:-1].strip()
        else:
            print("Game is over; no moves valid")
            return None

        if board.turn == chess.WHITE:
            if board.fullmove_number == 1:
                pgn = pgn + "\n\n1."
            else:
                pgn += ' '+str(board.fullmove_number)+"."

        with_header = f"""[White "Magnus Carlsen"]\n[Black "Garry Kasparov"]\n[WhiteElo "2900"]\n[BlackElo "2800"]\n\n"""+pgn.split("\n\n")[1]

        return with_header

    def try_moves(self, board, next_text):
        board = board.copy()
        moves = next_text.split()
        ok_moves = []
        for move in moves:
            if '.' in move:
                continue
            try:
                board.push_san(move)
                ok_moves.append(move)
            except:
                break

        return ok_moves
    
    def get_best_move(self, board, num_tokens=None, conversation=None):
        if num_tokens is None:
            num_tokens = self.config['num_lookahead_tokens']
        assert num_tokens >= 9, "A single move might take as many as 9 tokens (3 for the number + 6 for, e.g., 'N3xg5+)."

        if board.fen() in self.cache:
            out = self.cache[board.fen()]
            if conversation:
                if board.ply() > 0:
                    conversation.send_message("player", f"You played a move already in my cache (because I predicted it or someone already played it)! Returning {out}.")
                    conversation.send_message("spectator", f"Player played a move already in my cache (because I predicted it or someone already played it). Returning {out}.")
            return out

        pgn_to_query = self.get_query_pgn(board)

        if conversation:
            conversation.send_message("player", f"Querying {self.config['model']} with ... {pgn_to_query.split(']')[-1][-90:]}")
            conversation.send_message("spectator", f"Querying {self.config['model']} with ... {pgn_to_query.split(']')[-1][-90:]}")
        
        next_text = self.make_request(pgn_to_query, num_tokens)
        if next_text[:2] == "-O":
            next_text = self.make_request(pgn_to_query+" ", num_tokens)

        if conversation:
            conversation.send_message("spectator", f"Received reply of '{next_text}'")

        next_moves = self.try_moves(board, next_text)

        if len(next_moves) == 0:
            conversation.send_message("player", f"Tried to make an invalid move.")
            conversation.send_message("spectator", f"Tried to make an invalid move.")
            return None

        if conversation:
            conversation.send_message("player", f"Received reply and making move {next_moves[0]}.")
        
        new_board = board.copy()
        for move in next_moves:
            self.cache[new_board.fen()] = move
            new_board.push_san(move)

        with open("cache.p", "wb") as f:
            pickle.dump(self.cache, f)
        return next_moves[0]

    def make_request(self, content, num_tokens):
        url = "https://api.openai.com/v1/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        data = {
            "model": self.config['model'],
            "prompt": content,
            "temperature": self.config['temperature'],
            "max_tokens": num_tokens,
        }
    

        #sys.stderr.write(repr(data)+"\n")
        response = requests.post(url, headers=headers, data=json.dumps(data))
        response = response.json()['choices'][0]['text']
        #sys.stderr.write(response+"\n")

        return response
    


chessllm.py
carlini/chess-llm
## Copyright (C) 2023, Nicholas Carlini <nicholas@carlini.com>.
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see <http://www.gnu.org/licenses/>.

import requests
import time
import chess
import chess.pgn
import io
import csv
import os
import re
import pickle


def download_and_decompress(url):
    os.popen("wget " + url).read()
    compressed_filename = os.path.basename(url)
    time.sleep(1)
    os.popen("zstd -d " + compressed_filename).read()
    os.remove(compressed_filename)

    
def get_data():
    download_and_decompress("https://database.lichess.org/standard/lichess_db_standard_rated_2016-02.pgn.zst")
    download_and_decompress("https://database.lichess.org/lichess_db_puzzle.csv.zst")


import re

def generate_mapping(filename):
    mapping = {}

    # Regular expression pattern to match the game ID from the Site URL
    site_pattern = re.compile(r'\[Site "https://lichess.org/([a-zA-Z0-9]+)"]')

    # Open the file in binary mode to compute byte offsets
    with open(filename, 'rb') as f:
        line = f.readline()
        while line:
            match = site_pattern.search(line.decode('utf-8'))
            if match:
                current_game_id = match.group(1)
                mapping[current_game_id] = f.tell() - len(line)  # Get the starting byte offset of this line
            line = f.readline()

    return mapping

def fetch_game_moves(filename, game_id, offset):
    moves = []
    with open(filename, 'r') as f:
        f.seek(offset)
        pgn = f.read(10000).split("[Event")[0]
        

    return pgn

def convert_pgn_to_game(pgn_moves):
    pgn = io.StringIO(pgn_moves)
    game = chess.pgn.read_game(pgn)
    if len(game.errors) > 0:
        return None
    return game

def process_puzzles(csv_filename, games_filename, mapping):

    extracted_puzzles = []

    with open(csv_filename, 'r') as f:
        reader = csv.reader(f)
        next(reader)
        for row in reader:
            game_url, uci_moves = row[8], row[2].split()
            game_id = game_url.split('.org/')[1]
            move_num = int(game_url.split('#')[-1])
            game_id = game_id.split("/")[0].split("#")[0]
            rating = int(row[3])

            if game_id in mapping:
                pgn = fetch_game_moves(games_filename, game_id, mapping[game_id])
                game = convert_pgn_to_game(pgn)
                if game is None: continue

                board = game.board()

                for move in list(game.mainline_moves())[:move_num]:
                    board.push(move)

                new_board = board.copy()

                try:
                    solution = []
                    for move in uci_moves[1:]:
                        m = chess.Move.from_uci(move)
                        solution.append(new_board.san(m))
                        new_board.push(m)
                except:
                    print("Board import failed")
                    continue

                extracted_puzzles.append((row[0],
                                          rating,
                                         board,
                                         solution,
                ))
                print(len(extracted_puzzles))

    with open("pgn_puzzles.csv", "w") as f:
        writer = csv.writer(f)
        for uid, rating, board, solution in extracted_puzzles:
            writer.writerow((uid, rating,
                             str(chess.pgn.Game().from_board(board)).split("\n")[-1][:-2],
                             " ".join(solution)))



if __name__ == "__main__":
    get_data()

    filename = "lichess_db_standard_rated_2016-02.pgn"
    mapping = generate_mapping(filename)

    with open('mapping.pickle', 'wb') as f:
        pickle.dump(mapping, f)

    process_puzzles("lichess_db_puzzle.csv", filename, pickle.load(open("mapping.pickle","rb")))

generate_pgn_puzzles.py
carlini/chess-llm
"""Automatically updates the lichess-bot version."""
import yaml
import datetime
import os

with open("versioning.yml") as version_file:
    versioning_info = yaml.safe_load(version_file)

current_version = versioning_info["lichess_bot_version"]

utc_datetime = datetime.datetime.utcnow()
new_version = f"{utc_datetime.year}.{utc_datetime.month}.{utc_datetime.day}."
if current_version.startswith(new_version):
    current_version_list = current_version.split(".")
    current_version_list[-1] = str(int(current_version_list[-1]) + 1)
    new_version = ".".join(current_version_list)
else:
    new_version += "1"

versioning_info["lichess_bot_version"] = new_version

with open("versioning.yml", "w") as version_file:
    yaml.dump(versioning_info, version_file, sort_keys=False)

with open(os.environ['GITHUB_OUTPUT'], 'a') as fh:
    print(f"new_version={new_version}", file=fh)

lichess-bot/.github/workflows/update_version.py
carlini/chess-llm
"""Code related to the config that lichess-bot uses."""
from __future__ import annotations
import yaml
import os
import os.path
import logging
import math
from abc import ABCMeta
from enum import Enum
from typing import Any, Union
CONFIG_DICT_TYPE = dict[str, Any]

logger = logging.getLogger(__name__)


class FilterType(str, Enum):
    """What to do if the opponent declines our challenge."""

    NONE = "none"
    """Will still challenge the opponent."""
    COARSE = "coarse"
    """Won't challenge the opponent again."""
    FINE = "fine"
    """
    Won't challenge the opponent to a game of the same mode, speed, and variant
    based on the reason for the opponent declining the challenge.
    """


class Configuration:
    """The config or a sub-config that the bot uses."""

    def __init__(self, parameters: CONFIG_DICT_TYPE) -> None:
        """:param parameters: A `dict` containing the config for the bot."""
        self.config = parameters

    def __getattr__(self, name: str) -> Any:
        """
        Enable the use of `config.key1.key2`.

        :param name: The key to get its value.
        :return: The value of the key.
        """
        return self.lookup(name)

    def lookup(self, name: str) -> Any:
        """
        Get the value of a key.

        :param name: The key to get its value.
        :return: `Configuration` if the value is a `dict` else returns the value.
        """
        data = self.config.get(name)
        return Configuration(data) if isinstance(data, dict) else data

    def items(self) -> Any:
        """:return: All the key-value pairs in this config."""
        return self.config.items()

    def keys(self) -> list[str]:
        """:return: All of the keys in this config."""
        return list(self.config.keys())

    def __or__(self, other: Union[Configuration, CONFIG_DICT_TYPE]) -> Configuration:
        """Create a copy of this configuration that is updated with values from the parameter."""
        other_dict = other.config if isinstance(other, Configuration) else other
        return Configuration(self.config | other_dict)

    def __bool__(self) -> bool:
        """Whether `self.config` is empty."""
        return bool(self.config)

    def __getstate__(self) -> CONFIG_DICT_TYPE:
        """Get `self.config`."""
        return self.config

    def __setstate__(self, d: CONFIG_DICT_TYPE) -> None:
        """Set `self.config`."""
        self.config = d


def config_assert(assertion: bool, error_message: str) -> None:
    """Raise an exception if an assertion is false."""
    if not assertion:
        raise Exception(error_message)


def check_config_section(config: CONFIG_DICT_TYPE, data_name: str, data_type: ABCMeta, subsection: str = "") -> None:
    """
    Check the validity of a config section.

    :param config: The config section.
    :param data_name: The key to check its value.
    :param data_type: The expected data type.
    :param subsection: The subsection of the key.
    """
    config_part = config[subsection] if subsection else config
    sub = f"`{subsection}` sub" if subsection else ""
    data_location = f"`{data_name}` subsection in `{subsection}`" if subsection else f"Section `{data_name}`"
    type_error_message = {str: f"{data_location} must be a string wrapped in quotes.",
                          dict: f"{data_location} must be a dictionary with indented keys followed by colons."}
    config_assert(data_name in config_part, f"Your config.yml does not have required {sub}section `{data_name}`.")
    config_assert(isinstance(config_part[data_name], data_type), type_error_message[data_type])


def set_config_default(config: CONFIG_DICT_TYPE, *sections: str, key: str, default: Any,
                       force_empty_values: bool = False) -> CONFIG_DICT_TYPE:
    """
    Fill a specific config key with the default value if it is missing.

    :param config: The bot's config.
    :param sections: The sections that the key is in.
    :param key: The key to set.
    :param default: The default value.
    :param force_empty_values: Whether an empty value should be replaced with the default value.
    :return: The new config with the default value inserted if needed.
    """
    subconfig = config
    for section in sections:
        subconfig = subconfig.setdefault(section, {})
        if not isinstance(subconfig, dict):
            raise Exception(f"The {section} section in {sections} should hold a set of key-value pairs, not a value.")
    if force_empty_values:
        if subconfig.get(key) in [None, ""]:
            subconfig[key] = default
    else:
        subconfig.setdefault(key, default)
    return subconfig


def change_value_to_list(config: CONFIG_DICT_TYPE, *sections: str, key: str) -> None:
    """
    Change a single value to a list. e.g. 60 becomes [60]. Used to maintain backwards compatibility.

    :param config: The bot's config.
    :param sections: The sections that the key is in.
    :param key: The key to set.
    """
    subconfig = set_config_default(config, *sections, key=key, default=[])

    if subconfig[key] is None:
        subconfig[key] = []

    if not isinstance(subconfig[key], list):
        subconfig[key] = [subconfig[key]]


def insert_default_values(CONFIG: CONFIG_DICT_TYPE) -> None:
    """
    Insert the default values of most keys to the config if they are missing.

    :param CONFIG: The bot's config.
    """
    set_config_default(CONFIG, key="abort_time", default=20)
    set_config_default(CONFIG, key="move_overhead", default=1000)
    set_config_default(CONFIG, key="rate_limiting_delay", default=0)
    set_config_default(CONFIG, key="pgn_file_grouping", default="game", force_empty_values=True)
    set_config_default(CONFIG, "engine", key="working_dir", default=os.getcwd(), force_empty_values=True)
    set_config_default(CONFIG, "engine", key="silence_stderr", default=False)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="offer_draw_enabled", default=False)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="offer_draw_for_egtb_zero", default=True)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="resign_enabled", default=False)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="resign_for_egtb_minus_two", default=True)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="resign_moves", default=3)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="resign_score", default=-1000)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="offer_draw_moves", default=5)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="offer_draw_score", default=0)
    set_config_default(CONFIG, "engine", "draw_or_resign", key="offer_draw_pieces", default=10)
    set_config_default(CONFIG, "engine", "online_moves", key="max_out_of_book_moves", default=10)
    set_config_default(CONFIG, "engine", "online_moves", key="max_retries", default=2, force_empty_values=True)
    set_config_default(CONFIG, "engine", "online_moves", "online_egtb", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "online_moves", "online_egtb", key="source", default="lichess")
    set_config_default(CONFIG, "engine", "online_moves", "online_egtb", key="min_time", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "online_egtb", key="max_pieces", default=7)
    set_config_default(CONFIG, "engine", "online_moves", "online_egtb", key="move_quality", default="best")
    set_config_default(CONFIG, "engine", "online_moves", "chessdb_book", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "online_moves", "chessdb_book", key="min_time", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "chessdb_book", key="move_quality", default="good")
    set_config_default(CONFIG, "engine", "online_moves", "chessdb_book", key="min_depth", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="min_time", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="move_quality", default="best")
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="min_depth", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="min_knodes", default=0)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="max_score_difference", default=50)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="min_time", default=20)
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="source", default="masters")
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="player_name", default="")
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="sort", default="winrate")
    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="min_games", default=10)
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="max_pieces", default=7)
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="move_quality", default="best")
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "gaviota", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "gaviota", key="max_pieces", default=5)
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "gaviota", key="move_quality", default="best")
    set_config_default(CONFIG, "engine", "lichess_bot_tbs", "gaviota", key="min_dtm_to_consider_as_wdl_1", default=120)
    set_config_default(CONFIG, "engine", "polyglot", key="enabled", default=False)
    set_config_default(CONFIG, "engine", "polyglot", key="max_depth", default=8)
    set_config_default(CONFIG, "engine", "polyglot", key="selection", default="weighted_random")
    set_config_default(CONFIG, "engine", "polyglot", key="min_weight", default=1)
    set_config_default(CONFIG, "challenge", key="concurrency", default=1)
    set_config_default(CONFIG, "challenge", key="sort_by", default="best")
    set_config_default(CONFIG, "challenge", key="accept_bot", default=False)
    set_config_default(CONFIG, "challenge", key="only_bot", default=False)
    set_config_default(CONFIG, "challenge", key="max_increment", default=180)
    set_config_default(CONFIG, "challenge", key="min_increment", default=0)
    set_config_default(CONFIG, "challenge", key="max_base", default=math.inf)
    set_config_default(CONFIG, "challenge", key="min_base", default=0)
    set_config_default(CONFIG, "challenge", key="max_days", default=math.inf)
    set_config_default(CONFIG, "challenge", key="min_days", default=1)
    set_config_default(CONFIG, "challenge", key="block_list", default=[], force_empty_values=True)
    set_config_default(CONFIG, "challenge", key="allow_list", default=[], force_empty_values=True)
    set_config_default(CONFIG, "correspondence", key="checkin_period", default=600)
    set_config_default(CONFIG, "correspondence", key="move_time", default=60, force_empty_values=True)
    set_config_default(CONFIG, "correspondence", key="disconnect_time", default=300)
    set_config_default(CONFIG, "matchmaking", key="challenge_timeout", default=30, force_empty_values=True)
    CONFIG["matchmaking"]["challenge_timeout"] = max(CONFIG["matchmaking"]["challenge_timeout"], 1)
    set_config_default(CONFIG, "matchmaking", key="block_list", default=[], force_empty_values=True)
    default_filter = (CONFIG.get("matchmaking") or {}).get("delay_after_decline") or FilterType.NONE.value
    set_config_default(CONFIG, "matchmaking", key="challenge_filter", default=default_filter, force_empty_values=True)
    set_config_default(CONFIG, "matchmaking", key="allow_matchmaking", default=False)
    set_config_default(CONFIG, "matchmaking", key="challenge_initial_time", default=[None], force_empty_values=True)
    change_value_to_list(CONFIG, "matchmaking", key="challenge_initial_time")
    set_config_default(CONFIG, "matchmaking", key="challenge_increment", default=[None], force_empty_values=True)
    change_value_to_list(CONFIG, "matchmaking", key="challenge_increment")
    set_config_default(CONFIG, "matchmaking", key="challenge_days", default=[None], force_empty_values=True)
    change_value_to_list(CONFIG, "matchmaking", key="challenge_days")
    set_config_default(CONFIG, "matchmaking", key="opponent_min_rating", default=600, force_empty_values=True)
    set_config_default(CONFIG, "matchmaking", key="opponent_max_rating", default=4000, force_empty_values=True)
    set_config_default(CONFIG, "matchmaking", key="opponent_allow_tos_violation", default=True)
    set_config_default(CONFIG, "matchmaking", key="challenge_variant", default="random")
    set_config_default(CONFIG, "matchmaking", key="challenge_mode", default="random")
    set_config_default(CONFIG, "matchmaking", key="overrides", default={}, force_empty_values=True)
    for override_config in CONFIG["matchmaking"]["overrides"].values():
        for parameter in ["challenge_initial_time", "challenge_increment", "challenge_days"]:
            if parameter in override_config:
                set_config_default(override_config, key=parameter, default=[None], force_empty_values=True)
                change_value_to_list(override_config, key=parameter)

    for section in ["engine", "correspondence"]:
        for ponder in ["ponder", "uci_ponder"]:
            set_config_default(CONFIG, section, key=ponder, default=False)

    for type in ["hello", "goodbye"]:
        for target in ["", "_spectators"]:
            set_config_default(CONFIG, "greeting", key=type + target, default="", force_empty_values=True)


def log_config(CONFIG: CONFIG_DICT_TYPE) -> None:
    """
    Log the config to make debugging easier.

    :param CONFIG: The bot's config.
    """
    logger_config = CONFIG.copy()
    logger_config["token"] = "logger"
    logger.debug(f"Config:\n{yaml.dump(logger_config, sort_keys=False)}")
    logger.debug("====================")


def validate_config(CONFIG: CONFIG_DICT_TYPE) -> None:
    """Check if the config is valid."""
    check_config_section(CONFIG, "token", str)
    check_config_section(CONFIG, "url", str)
    check_config_section(CONFIG, "engine", dict)
    check_config_section(CONFIG, "challenge", dict)
    check_config_section(CONFIG, "dir", str, "engine")
    check_config_section(CONFIG, "name", str, "engine")

    config_assert(os.path.isdir(CONFIG["engine"]["dir"]),
                  f'Your engine directory `{CONFIG["engine"]["dir"]}` is not a directory.')

    working_dir = CONFIG["engine"].get("working_dir")
    config_assert(not working_dir or os.path.isdir(working_dir),
                  f"Your engine's working directory `{working_dir}` is not a directory.")

    engine = os.path.join(CONFIG["engine"]["dir"], CONFIG["engine"]["name"])
    config_assert(os.path.isfile(engine) or CONFIG["engine"]["protocol"] == "homemade",
                  f"The engine {engine} file does not exist.")
    config_assert(os.access(engine, os.X_OK) or CONFIG["engine"]["protocol"] == "homemade",
                  f"The engine {engine} doesn't have execute (x) permission. Try: chmod +x {engine}")

    if CONFIG["engine"]["protocol"] == "xboard":
        for section, subsection in (("online_moves", "online_egtb"),
                                    ("lichess_bot_tbs", "syzygy"),
                                    ("lichess_bot_tbs", "gaviota")):
            online_section = (CONFIG["engine"].get(section) or {}).get(subsection) or {}
            config_assert(online_section.get("move_quality") != "suggest" or not online_section.get("enabled"),
                          f"XBoard engines can't be used with `move_quality` set to `suggest` in {subsection}.")

    valid_pgn_grouping_options = ["game", "opponent", "all"]
    config_pgn_choice = CONFIG["pgn_file_grouping"]
    config_assert(config_pgn_choice in valid_pgn_grouping_options,
                  f"The `pgn_file_grouping` choice of `{config_pgn_choice}` is not valid. "
                  f"Please choose from {valid_pgn_grouping_options}.")

    matchmaking = CONFIG.get("matchmaking") or {}
    matchmaking_enabled = matchmaking.get("allow_matchmaking") or False
    # `, []` is there only for mypy. It isn't used.
    matchmaking_has_values = (matchmaking.get("challenge_initial_time", [])[0] is not None
                              and matchmaking.get("challenge_increment", [])[0] is not None
                              or matchmaking.get("challenge_days", [])[0] is not None)
    config_assert(not matchmaking_enabled or matchmaking_has_values,
                  "The time control to challenge other bots is not set. Either lists of challenge_initial_time and "
                  "challenge_increment is required, or a list of challenge_days, or both.")

    filter_option = "challenge_filter"
    filter_type = (CONFIG.get("matchmaking") or {}).get(filter_option)
    config_assert(filter_type is None or filter_type in FilterType.__members__.values(),
                  f"{filter_type} is not a valid value for {filter_option} (formerly delay_after_decline) parameter. "
                  f"Choices are: {', '.join(FilterType)}.")

    selection_choices = {"polyglot": ["weighted_random", "uniform_random", "best_move"],
                         "chessdb_book": ["all", "good", "best"],
                         "lichess_cloud_analysis": ["good", "best"],
                         "online_egtb": ["best", "suggest"]}
    for db_name, valid_selections in selection_choices.items():
        is_online = db_name != "polyglot"
        db_section = (CONFIG["engine"].get("online_moves") or {}) if is_online else CONFIG["engine"]
        db_config = db_section.get(db_name)
        select_key = "selection" if db_name == "polyglot" else "move_quality"
        selection = db_config.get(select_key)
        select = f"{'online_moves:' if is_online else ''}{db_name}:{select_key}"
        config_assert(selection in valid_selections,
                      f"`{selection}` is not a valid `engine:{select}` value. "
                      f"Please choose from {valid_selections}.")

    lichess_tbs_config = CONFIG["engine"].get("lichess_bot_tbs") or {}
    quality_selections = ["best", "suggest"]
    for tb in ["syzygy", "gaviota"]:
        selection = (lichess_tbs_config.get(tb) or {}).get("move_quality")
        config_assert(selection in quality_selections,
                      f"`{selection}` is not a valid choice for `engine:lichess_bot_tbs:{tb}:move_quality`. "
                      f"Please choose from {quality_selections}.")

    explorer_choices = {"source": ["lichess", "masters", "player"],
                        "sort": ["winrate", "games_played"]}
    explorer_config = (CONFIG["engine"].get("online_moves") or {}).get("lichess_opening_explorer")
    if explorer_config:
        for parameter, choice_list in explorer_choices.items():
            explorer_choice = explorer_config.get(parameter)
            config_assert(explorer_choice in choice_list,
                          f"`{explorer_choice}` is not a valid"
                          f" `engine:online_moves:lichess_opening_explorer:{parameter}`"
                          f" value. Please choose from {choice_list}.")


def load_config(config_file: str) -> Configuration:
    """
    Read the config.

    :param config_file: The filename of the config (usually `config.yml`).
    :return: A `Configuration` object containing the config.
    """
    with open(config_file) as stream:
        try:
            CONFIG = yaml.safe_load(stream)
        except Exception:
            logger.exception("There appears to be a syntax problem with your config.yml")
            raise

    log_config(CONFIG)

    if "LICHESS_BOT_TOKEN" in os.environ:
        CONFIG["token"] = os.environ["LICHESS_BOT_TOKEN"]

    insert_default_values(CONFIG)
    log_config(CONFIG)
    validate_config(CONFIG)

    return Configuration(CONFIG)

lichess-bot/config.py
carlini/chess-llm
"""Allows lichess-bot to send messages to the chat."""
from __future__ import annotations
import logging
import model
from engine_wrapper import EngineWrapper
from lichess import Lichess
from collections.abc import Sequence
from timer import seconds
MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]

logger = logging.getLogger(__name__)


class Conversation:
    """Enables the bot to communicate with its opponent and the spectators."""

    def __init__(self, game: model.Game, engine: EngineWrapper, li: Lichess, version: str,
                 challenge_queue: MULTIPROCESSING_LIST_TYPE) -> None:
        """
        Communication between lichess-bot and the game chats.

        :param game: The game that the bot will send messages to.
        :param engine: The engine playing the game.
        :param li: A class that is used for communication with lichess.
        :param version: The lichess-bot version.
        :param challenge_queue: The active challenges the bot has.
        """
        self.game = game
        self.engine = engine
        self.li = li
        self.version = version
        self.challengers = challenge_queue

    command_prefix = "!"

    def react(self, line: ChatLine) -> None:
        """
        React to a received message.

        :param line: Information about the message.
        """
        logger.info(f'*** {self.game.url()} [{line.room}] {line.username}: {line.text}')
        if line.text[0] == self.command_prefix:
            self.command(line, line.text[1:].lower())

    def command(self, line: ChatLine, cmd: str) -> None:
        """
        Reacts to the specific commands in the chat.

        :param line: Information about the message.
        :param cmd: The command to react to.
        """
        from_self = line.username == self.game.username
        if cmd == "commands" or cmd == "help":
            self.send_reply(line, "Supported commands: !wait (wait a minute for my first move), !name, !howto, !eval, !queue")
        elif cmd == "wait" and self.game.is_abortable():
            self.game.ping(seconds(60), seconds(120), seconds(120))
            self.send_reply(line, "Waiting 60 seconds...")
        elif cmd == "name":
            name = self.game.me.name
            self.send_reply(line, f"{name} running {self.engine.name()} (lichess-bot v{self.version})")
        elif cmd == "howto":
            self.send_reply(line, "How to run: Check out 'Lichess Bot API'")
        elif cmd == "eval" and (from_self or line.room == "spectator"):
            stats = self.engine.get_stats(for_chat=True)
            self.send_reply(line, ", ".join(stats))
        elif cmd == "eval":
            self.send_reply(line, "I don't tell that to my opponent, sorry.")
        elif cmd == "queue":
            if self.challengers:
                challengers = ", ".join([f"@{challenger.challenger.name}" for challenger in reversed(self.challengers)])
                self.send_reply(line, f"Challenge queue: {challengers}")
            else:
                self.send_reply(line, "No challenges queued.")

    def send_reply(self, line: ChatLine, reply: str) -> None:
        """
        Send the reply to the chat.

        :param line: Information about the original message that we reply to.
        :param reply: The reply to send.
        """
        logger.info(f'*** {self.game.url()} [{line.room}] {self.game.username}: {reply}')
        self.li.chat(self.game.id, line.room, reply)

    def send_message(self, room: str, message: str) -> None:
        """Send the message to the chat."""
        if message:
            self.send_reply(ChatLine({"room": room, "username": "", "text": ""}), message)


class ChatLine:
    """Information about the message."""

    def __init__(self, message_info: dict[str, str]) -> None:
        """Information about the message."""
        self.room = message_info["room"]
        """Whether the message was sent in the chat room or in the spectator room."""
        self.username = message_info["username"]
        """The username of the account that sent the message."""
        self.text = message_info["text"]
        """The message sent."""

lichess-bot/conversation.py
carlini/chess-llm
"""Provides communication with the engine."""
from __future__ import annotations
import os
import chess.engine
import chess.polyglot
import chess.syzygy
import chess.gaviota
import chess
import subprocess
import logging
import datetime
import time
import random
from collections import Counter
from collections.abc import Generator, Callable
from contextlib import contextmanager
import config
import model
import lichess
from config import Configuration
from timer import Timer, msec, seconds, msec_str, sec_str, to_seconds
from typing import Any, Optional, Union
OPTIONS_TYPE = dict[str, Any]
MOVE_INFO_TYPE = dict[str, Any]
COMMANDS_TYPE = list[str]
LICHESS_EGTB_MOVE = dict[str, Any]
CHESSDB_EGTB_MOVE = dict[str, Any]
MOVE = Union[chess.engine.PlayResult, list[chess.Move]]

logger = logging.getLogger(__name__)

out_of_online_opening_book_moves: Counter[str] = Counter()


@contextmanager
def create_engine(engine_config: config.Configuration) -> Generator[EngineWrapper, None, None]:
    """
    Create the engine.

    Use in a with-block to automatically close the engine when exiting the game.

    :param engine_config: The options for the engine.
    :return: An engine. Either UCI, XBoard, or Homemade.
    """
    cfg = engine_config.engine
    engine_path = os.path.abspath(os.path.join(cfg.dir, cfg.name))
    engine_type = cfg.protocol
    commands = [engine_path]
    if cfg.engine_options:
        for k, v in cfg.engine_options.items():
            commands.append(f"--{k}={v}" if v is not None else f"--{k}")

    stderr = None if cfg.silence_stderr else subprocess.DEVNULL

    Engine: Union[type[UCIEngine], type[XBoardEngine], type[MinimalEngine]]
    if engine_type == "xboard":
        Engine = XBoardEngine
    elif engine_type == "uci":
        Engine = UCIEngine
    elif engine_type == "homemade":
        Engine = getHomemadeEngine(cfg.name)
    else:
        raise ValueError(
            f"    Invalid engine type: {engine_type}. Expected xboard, uci, or homemade.")
    options = remove_managed_options(cfg.lookup(f"{engine_type}_options") or config.Configuration({}))
    logger.debug(f"Starting engine: {commands}")
    engine = Engine(commands, options, stderr, cfg.draw_or_resign, cwd=cfg.working_dir)
    try:
        yield engine
    finally:
        engine.ping()
        engine.quit()


def remove_managed_options(config: config.Configuration) -> OPTIONS_TYPE:
    """Remove the options managed by python-chess."""
    def is_managed(key: str) -> bool:
        return chess.engine.Option(key, "", None, None, None, None).is_managed()

    return {name: value for (name, value) in config.items() if not is_managed(name)}


PONDERPV_CHARACTERS = 6  # The length of ", Pv: ".


class EngineWrapper:
    """A wrapper used by all engines (UCI, XBoard, Homemade)."""

    def __init__(self, options: OPTIONS_TYPE, draw_or_resign: config.Configuration) -> None:
        """
        Initialize the values of the wrapper used by all engines (UCI, XBoard, Homemade).

        :param options: The options to send to the engine.
        :param draw_or_resign: Options on whether the bot should resign or offer draws.
        """
        self.engine: Union[chess.engine.SimpleEngine, FillerEngine]
        self.scores: list[chess.engine.PovScore] = []
        self.draw_or_resign = draw_or_resign
        self.go_commands = config.Configuration(options.pop("go_commands", {}) or {})
        self.move_commentary: list[MOVE_INFO_TYPE] = []
        self.comment_start_index = -1

    def play_move(self,
                  board: chess.Board,
                  game: model.Game,
                  li: lichess.Lichess,
                  setup_timer: Timer,
                  move_overhead: datetime.timedelta,
                  can_ponder: bool,
                  is_correspondence: bool,
                  correspondence_move_time: datetime.timedelta,
                  engine_cfg: config.Configuration,
                  min_time: datetime.timedelta,
                  conversation) -> None:
        """
        Play a move.

        :param board: The current position.
        :param game: The game that the bot is playing.
        :param li: Provides communication with lichess.org.
        :param start_time: The time that the bot received the move.
        :param move_overhead: The time it takes to communicate between the engine and lichess.org.
        :param can_ponder: Whether the engine is allowed to ponder.
        :param is_correspondence: Whether this is a correspondence or unlimited game.
        :param correspondence_move_time: The time the engine will think if `is_correspondence` is true.
        :param engine_cfg: Options for external moves (e.g. from an opening book), and for engine resignation and draw offers.
        :param min_time: Minimum time to spend, in seconds.
        :return: The move to play.
        """
        polyglot_cfg = engine_cfg.polyglot
        online_moves_cfg = engine_cfg.online_moves
        draw_or_resign_cfg = engine_cfg.draw_or_resign
        lichess_bot_tbs = engine_cfg.lichess_bot_tbs

        best_move: MOVE
        best_move = get_book_move(board, game, polyglot_cfg)

        if best_move.move is None:
            best_move = get_egtb_move(board,
                                      game,
                                      lichess_bot_tbs,
                                      draw_or_resign_cfg)

        if not isinstance(best_move, list) and best_move.move is None:
            best_move = get_online_move(li,
                                        board,
                                        game,
                                        online_moves_cfg,
                                        draw_or_resign_cfg)

        if isinstance(best_move, list) or best_move.move is None:
            draw_offered = check_for_draw_offer(game)

            if len(board.move_stack) < 2:
                time_limit = first_move_time(game)
                can_ponder = False  # No pondering after the first move since a new clock starts afterwards.
            elif is_correspondence:
                time_limit = single_move_time(board, game, correspondence_move_time, setup_timer, move_overhead)
            else:
                time_limit = game_clock_time(board, game, setup_timer, move_overhead)

            best_move = self.search(board, time_limit, can_ponder, draw_offered, best_move,
                                    conversation)

        # Heed min_time
        elapsed = setup_timer.time_since_reset()
        if elapsed < min_time:
            time.sleep(to_seconds(min_time - elapsed))

        self.add_comment(best_move, board)
        self.print_stats()
        if best_move.resigned and len(board.move_stack) >= 2:
            li.resign(game.id)
        else:
            li.make_move(game.id, best_move)

    def add_go_commands(self, time_limit: chess.engine.Limit) -> chess.engine.Limit:
        """Add extra commands to send to the engine. For example, to search for 1000 nodes or up to depth 10."""
        movetime_cfg = self.go_commands.movetime
        if movetime_cfg is not None:
            movetime = msec(movetime_cfg)
            if time_limit.time is None or seconds(time_limit.time) > movetime:
                time_limit.time = to_seconds(movetime)
        time_limit.depth = self.go_commands.depth
        time_limit.nodes = self.go_commands.nodes
        return time_limit

    def offer_draw_or_resign(self, result: chess.engine.PlayResult, board: chess.Board) -> chess.engine.PlayResult:
        """Offer draw or resign depending on the score of the engine."""
        def actual(score: chess.engine.PovScore) -> int:
            return score.relative.score(mate_score=40000)

        can_offer_draw = self.draw_or_resign.offer_draw_enabled
        draw_offer_moves = self.draw_or_resign.offer_draw_moves
        draw_score_range: int = self.draw_or_resign.offer_draw_score
        draw_max_piece_count = self.draw_or_resign.offer_draw_pieces
        pieces_on_board = chess.popcount(board.occupied)
        enough_pieces_captured = pieces_on_board <= draw_max_piece_count
        if can_offer_draw and len(self.scores) >= draw_offer_moves and enough_pieces_captured:
            scores = self.scores[-draw_offer_moves:]

            def score_near_draw(score: chess.engine.PovScore) -> bool:
                return abs(actual(score)) <= draw_score_range
            if len(scores) == len(list(filter(score_near_draw, scores))):
                result.draw_offered = True

        resign_enabled = self.draw_or_resign.resign_enabled
        min_moves_for_resign = self.draw_or_resign.resign_moves
        resign_score: int = self.draw_or_resign.resign_score
        if resign_enabled and len(self.scores) >= min_moves_for_resign:
            scores = self.scores[-min_moves_for_resign:]

            def score_near_loss(score: chess.engine.PovScore) -> bool:
                return actual(score) <= resign_score
            if len(scores) == len(list(filter(score_near_loss, scores))):
                result.resigned = True
        return result

    def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool,
               root_moves: MOVE) -> chess.engine.PlayResult:
        """
        Tell the engine to search.

        :param board: The current position.
        :param time_limit: Conditions for how long the engine can search (e.g. we have 10 seconds and search up to depth 10).
        :param ponder: Whether the engine can ponder.
        :param draw_offered: Whether the bot was offered a draw.
        :param root_moves: If it is a list, the engine will only play a move that is in `root_moves`.
        :return: The move to play.
        """
        time_limit = self.add_go_commands(time_limit)
        result = self.engine.play(board,
                                  time_limit,
                                  info=chess.engine.INFO_ALL,
                                  ponder=ponder,
                                  draw_offered=draw_offered,
                                  root_moves=root_moves if isinstance(root_moves, list) else None)
        # Use null_score to have no effect on draw/resign decisions
        null_score = chess.engine.PovScore(chess.engine.Mate(1), board.turn)
        self.scores.append(result.info.get("score", null_score))
        result = self.offer_draw_or_resign(result, board)
        return result

    def comment_index(self, move_stack_index: int) -> int:
        """
        Get the index of a move for use in `comment_for_board_index`.

        :param move_stack_index: The move number.
        :return: The index of the move in `self.move_commentary`.
        """
        if self.comment_start_index < 0:
            return -1
        else:
            return move_stack_index - self.comment_start_index

    def comment_for_board_index(self, index: int) -> MOVE_INFO_TYPE:
        """
        Get the engine comments for a specific move.

        :param index: The move number.
        :return: The move comments.
        """
        no_info: MOVE_INFO_TYPE = {}
        comment_index = self.comment_index(index)
        if comment_index < 0 or comment_index % 2 != 0:
            return no_info

        try:
            return self.move_commentary[comment_index // 2]
        except IndexError:
            return no_info

    def add_comment(self, move: chess.engine.PlayResult, board: chess.Board) -> None:
        """
        Store the move's comments.

        :param move: The move. Contains the comments in `move.info`.
        :param board: The current position.
        """
        if self.comment_start_index < 0:
            self.comment_start_index = len(board.move_stack)
        move_info: MOVE_INFO_TYPE = dict(move.info.copy()) if move.info else {}
        if "pv" in move_info:
            move_info["ponderpv"] = board.variation_san(move.info["pv"])
        if "refutation" in move_info:
            move_info["refutation"] = board.variation_san(move.info["refutation"])
        if "currmove" in move_info:
            move_info["currmove"] = board.san(move.info["currmove"])
        self.move_commentary.append(move_info)

    def print_stats(self) -> None:
        """Print the engine stats."""
        for line in self.get_stats():
            logger.info(line)

    def readable_score(self, relative_score: chess.engine.PovScore) -> str:
        """Convert the score to a more human-readable format."""
        score = relative_score.relative
        cp_score = score.score()
        if cp_score is None:
            str_score = f"#{score.mate()}"
        else:
            str_score = str(round(cp_score / 100, 2))
        return str_score

    def readable_wdl(self, wdl: chess.engine.PovWdl) -> str:
        """Convert the WDL score to a percentage, so it is more human-readable."""
        wdl_percentage = round(wdl.relative.expectation() * 100, 1)
        return f"{wdl_percentage}%"

    def readable_time(self, number: int) -> str:
        """Convert time given as a number into minutes and seconds, so it is more human-readable. e.g. 123 -> 2m 3s."""
        minutes, seconds = divmod(number, 60)
        if minutes >= 1:
            return f"{minutes:0.0f}m {seconds:0.1f}s"
        else:
            return f"{seconds:0.1f}s"

    def readable_number(self, number: int) -> str:
        """Convert number to a more human-readable format. e.g. 123456789 -> 123M."""
        if number >= 1e9:
            return f"{round(number / 1e9, 1)}B"
        elif number >= 1e6:
            return f"{round(number / 1e6, 1)}M"
        elif number >= 1e3:
            return f"{round(number / 1e3, 1)}K"
        return str(number)

    def to_readable_value(self, stat: str, info: MOVE_INFO_TYPE) -> str:
        """Change a value to a more human-readable format."""
        readable: dict[str, Callable[[Any], str]] = {"Evaluation": self.readable_score, "Winrate": self.readable_wdl,
                                                     "Hashfull": lambda x: f"{round(x / 10, 1)}%",
                                                     "Nodes": self.readable_number,
                                                     "Speed": lambda x: f"{self.readable_number(x)}nps",
                                                     "Tbhits": self.readable_number,
                                                     "Cpuload": lambda x: f"{round(x / 10, 1)}%",
                                                     "Movetime": self.readable_time}

        def identity(x: Any) -> str:
            return str(x)

        return str(readable.get(stat, identity)(info[stat]))

    def get_stats(self, for_chat: bool = False) -> list[str]:
        """
        Get the stats returned by the engine.

        :param for_chat: Whether the stats will be sent to the game chat, which has a 140 character limit.
        """
        can_index = self.move_commentary and self.move_commentary[-1]
        info: MOVE_INFO_TYPE = self.move_commentary[-1].copy() if can_index else {}

        def to_readable_item(stat: str, value: Any) -> tuple[str, Any]:
            readable = {"wdl": "winrate", "ponderpv": "PV", "nps": "speed", "score": "evaluation", "time": "movetime"}
            stat = readable.get(stat, stat)
            if stat == "string" and value.startswith("lichess-bot-source:"):
                stat = "source"
                value = value.split(":", 1)[1]
            return stat.title(), value

        info = dict(to_readable_item(key, value) for (key, value) in info.items())
        if "Source" not in info:
            info["Source"] = "Engine"

        stats = ["Source", "Evaluation", "Winrate", "Depth", "Nodes", "Speed", "Pv"]
        if for_chat and "Pv" in info:
            bot_stats = [f"{stat}: {self.to_readable_value(stat, info)}"
                         for stat in stats if stat in info and stat != "Pv"]
            len_bot_stats = len(", ".join(bot_stats)) + PONDERPV_CHARACTERS
            ponder_pv = info["Pv"].split()
            try:
                while len(" ".join(ponder_pv)) + len_bot_stats > lichess.MAX_CHAT_MESSAGE_LEN:
                    ponder_pv.pop()
                if ponder_pv[-1].endswith("."):
                    ponder_pv.pop()
                info["Pv"] = " ".join(ponder_pv)
            except IndexError:
                pass
            if not info["Pv"]:
                info.pop("Pv")
        return [f"{stat}: {self.to_readable_value(stat, info)}" for stat in stats if stat in info]

    def get_opponent_info(self, game: model.Game) -> None:
        """Get the opponent's information and sends it to the engine."""
        opponent = chess.engine.Opponent(name=game.opponent.name,
                                         title=game.opponent.title,
                                         rating=game.opponent.rating,
                                         is_engine=game.opponent.is_bot)
        self.engine.send_opponent_information(opponent=opponent, engine_rating=game.me.rating)

    def name(self) -> str:
        """Get the name of the engine."""
        engine_info: dict[str, str] = dict(self.engine.id)
        name: str = engine_info["name"]
        return name

    def get_pid(self) -> str:
        """Get the pid of the engine."""
        pid = "?"
        if self.engine.transport is not None:
            pid = str(self.engine.transport.get_pid())
        return pid

    def ping(self) -> None:
        """Ping the engine."""
        self.engine.ping()

    def send_game_result(self, game: model.Game, board: chess.Board) -> None:
        """
        Inform engine of the game ending.

        :param game: The final game state from lichess.
        :param board: The final board state.
        """
        termination = game.state.get("status")
        winner = game.state.get("winner")
        winning_color = chess.WHITE if winner == "white" else chess.BLACK

        if termination == model.Termination.MATE:
            self.engine.send_game_result(board)
        elif termination == model.Termination.RESIGN:
            resigner = "White" if winner == "black" else "Black"
            self.engine.send_game_result(board, winning_color, f"{resigner} resigned")
        elif termination == model.Termination.ABORT:
            self.engine.send_game_result(board, None, "Game aborted", False)
        elif termination == model.Termination.DRAW:
            draw_reason = None if board.is_game_over(claim_draw=True) else "Draw by agreement"
            self.engine.send_game_result(board, None, draw_reason)
        elif termination == model.Termination.TIMEOUT:
            if winner:
                self.engine.send_game_result(board, winning_color, "Time forfeiture")
            else:
                self.engine.send_game_result(board, None, "Time out with insufficient material")
        else:
            self.engine.send_game_result(board, None, termination)

    def quit(self) -> None:
        """Close the engine."""
        self.engine.quit()
        self.engine.close()


class UCIEngine(EngineWrapper):
    """The class used to communicate with UCI engines."""

    def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optional[int],
                 draw_or_resign: config.Configuration, **popen_args: str) -> None:
        """
        Communicate with UCI engines.

        :param commands: The engine path and commands to send to the engine. e.g. ["engines/engine.exe", "--option1=value1"]
        :param options: The options to send to the engine.
        :param stderr: Whether we should silence the stderr.
        :param draw_or_resign: Options on whether the bot should resign or offer draws.
        :param popen_args: The cwd of the engine.
        """
        super().__init__(options, draw_or_resign)
        self.engine = chess.engine.SimpleEngine.popen_uci(commands, timeout=10., debug=False, setpgrp=False, stderr=stderr,
                                                          **popen_args)
        self.engine.configure(options)


class XBoardEngine(EngineWrapper):
    """The class used to communicate with XBoard engines."""

    def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optional[int],
                 draw_or_resign: config.Configuration, **popen_args: str) -> None:
        """
        Communicate with XBoard engines.

        :param commands: The engine path and commands to send to the engine. e.g. ["engines/engine.exe", "--option1=value1"]
        :param options: The options to send to the engine.
        :param stderr: Whether we should silence the stderr.
        :param draw_or_resign: Options on whether the bot should resign or offer draws.
        :param popen_args: The cwd of the engine.
        """
        super().__init__(options, draw_or_resign)
        self.engine = chess.engine.SimpleEngine.popen_xboard(commands, timeout=10., debug=False, setpgrp=False,
                                                             stderr=stderr, **popen_args)
        egt_paths = options.pop("egtpath", {}) or {}
        features = self.engine.protocol.features if isinstance(self.engine.protocol, chess.engine.XBoardProtocol) else {}
        egt_features = features.get("egt", "")
        if isinstance(egt_features, str):
            egt_types_from_engine = egt_features.split(",")
            egt_type: str
            for egt_type in filter(None, egt_types_from_engine):
                if egt_type in egt_paths:
                    options[f"egtpath {egt_type}"] = egt_paths[egt_type]
                else:
                    logger.debug(f"No paths found for egt type: {egt_type}.")
        self.engine.configure(options)


class MinimalEngine(EngineWrapper):
    """
    Subclass this to prevent a few random errors.

    Even though MinimalEngine extends EngineWrapper,
    you don't have to actually wrap an engine.

    At minimum, just implement `search`,
    however you can also change other methods like
    `notify`, etc.
    """

    def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optional[int],
                 draw_or_resign: Configuration, name: Optional[str] = None, **popen_args: str) -> None:
        """
        Initialize the values of the engine that all homemade engines inherit.

        :param options: The options to send to the engine.
        :param draw_or_resign: Options on whether the bot should resign or offer draws.
        """
        super().__init__(options, draw_or_resign)

        self.engine_name = self.__class__.__name__ if name is None else name

        self.engine = FillerEngine(self, name=self.engine_name)

    def get_pid(self) -> str:
        """Homemade engines don't have a pid, so we return a question mark."""
        return "?"

    def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool,
               root_moves: MOVE) -> chess.engine.PlayResult:
        """
        Choose a move.

        The method to be implemented in your homemade engine.
        NOTE: This method must return an instance of "chess.engine.PlayResult"
        """
        raise NotImplementedError("The search method is not implemented")

    def notify(self, method_name: str, *args: Any, **kwargs: Any) -> None:
        """
        Enable the use of `self.engine.option1`.

        The EngineWrapper class sometimes calls methods on "self.engine".

        "self.engine" is a filler property that notifies <self>
        whenever an attribute is called.

        Nothing happens unless the main engine does something.

        Simply put, the following code is equivalent
        self.engine.<method_name>(<*args>, <**kwargs>)
        self.notify(<method_name>, <*args>, <**kwargs>)
        """
        pass


class FillerEngine:
    """
    Not meant to be an actual engine.

    This is only used to provide the property "self.engine"
    in "MinimalEngine" which extends "EngineWrapper"
    """

    def __init__(self, main_engine: MinimalEngine, name: str = "") -> None:
        """:param name: The name to send to the chat."""
        self.id: dict[str, str] = {
            "name": name
        }
        self.name = name
        self.main_engine = main_engine

    def __getattr__(self, method_name: str) -> Any:
        """Provide the property `self.engine`."""
        main_engine = self.main_engine

        def method(*args: Any, **kwargs: Any) -> Any:
            nonlocal main_engine
            nonlocal method_name
            return main_engine.notify(method_name, *args, **kwargs)

        return method


def getHomemadeEngine(name: str) -> type[MinimalEngine]:
    """
    Get the homemade engine with name `name`. e.g. If `name` is `RandomMove` then we will return `strategies.RandomMove`.

    :param name: The name of the homemade engine.
    :return: The engine with this name.
    """
    import strategies
    engine: type[MinimalEngine] = getattr(strategies, name)
    return engine


def single_move_time(board: chess.Board, game: model.Game, search_time: datetime.timedelta,
                     setup_timer: Timer, move_overhead: datetime.timedelta) -> chess.engine.Limit:
    """
    Calculate time to search in correspondence games.

    :param board: The current positions.
    :param game: The game that the bot is playing.
    :param search_time: How long the engine should search.
    :param start_time: The time we have left.
    :param move_overhead: The time it takes to communicate between the engine and lichess-bot.
    :return: The time to choose a move.
    """
    pre_move_time = setup_timer.time_since_reset()
    overhead = pre_move_time + move_overhead
    wb = "w" if board.turn == chess.WHITE else "b"
    clock_time = max(msec(0), msec(game.state[f"{wb}time"]) - overhead)
    search_time = min(search_time, clock_time)
    logger.info(f"Searching for time {sec_str(search_time)} seconds for game {game.id}")
    return chess.engine.Limit(time=to_seconds(search_time), clock_id="correspondence")


def first_move_time(game: model.Game) -> chess.engine.Limit:
    """
    Determine time limit for the first move in the game.

    :param game: The game that the bot is playing.
    :return: The time to choose the first move.
    """
    # Need to hardcode first movetime since Lichess has 30 sec limit.
    search_time = seconds(10)
    logger.info(f"Searching for time {sec_str(search_time)} seconds for game {game.id}")
    return chess.engine.Limit(time=to_seconds(search_time), clock_id="first move")


def game_clock_time(board: chess.Board,
                    game: model.Game,
                    setup_timer: Timer,
                    move_overhead: datetime.timedelta) -> chess.engine.Limit:
    """
    Get the time to play by the engine in realtime games.

    :param board: The current positions.
    :param game: The game that the bot is playing.
    :param start_time: The time we have left.
    :param move_overhead: The time it takes to communicate between the engine and lichess-bot.
    :return: The time to play a move.
    """
    pre_move_time = setup_timer.time_since_reset()
    overhead = pre_move_time + move_overhead
    times = {side: msec(game.state[side]) for side in ["wtime", "btime"]}
    wb = "w" if board.turn == chess.WHITE else "b"
    times[f"{wb}time"] = max(msec(0), times[f"{wb}time"] - overhead)
    logger.info(f"Searching for wtime {msec_str(times['wtime'])} btime {msec_str(times['btime'])} for game {game.id}")
    return chess.engine.Limit(white_clock=to_seconds(times["wtime"]),
                              black_clock=to_seconds(times["btime"]),
                              white_inc=to_seconds(msec(game.state["winc"])),
                              black_inc=to_seconds(msec(game.state["binc"])),
                              clock_id="real time")


def check_for_draw_offer(game: model.Game) -> bool:
    """Check if the bot was offered a draw."""
    return game.state.get(f"{game.opponent_color[0]}draw", False)


def get_book_move(board: chess.Board, game: model.Game,
                  polyglot_cfg: config.Configuration) -> chess.engine.PlayResult:
    """Get a move from an opening book."""
    no_book_move = chess.engine.PlayResult(None, None)
    use_book = polyglot_cfg.enabled
    max_game_length = polyglot_cfg.max_depth * 2 - 1
    if not use_book or len(board.move_stack) > max_game_length:
        return no_book_move

    if board.chess960:
        variant = "chess960"
    else:
        variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant)

    config.change_value_to_list(polyglot_cfg.config, "book", key=variant)
    books = polyglot_cfg.book.lookup(variant)

    for book in books:
        with chess.polyglot.open_reader(book) as reader:
            try:
                selection = polyglot_cfg.selection
                min_weight = polyglot_cfg.min_weight
                if selection == "weighted_random":
                    move = reader.weighted_choice(board).move
                elif selection == "uniform_random":
                    move = reader.choice(board, minimum_weight=min_weight).move
                elif selection == "best_move":
                    move = reader.find(board, minimum_weight=min_weight).move
            except IndexError:
                # python-chess raises "IndexError" if no entries found.
                move = None

        if move is not None:
            logger.info(f"Got move {move} from book {book} for game {game.id}")
            return chess.engine.PlayResult(move, None, {"string": "lichess-bot-source:Opening Book"})

    return no_book_move


def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, online_moves_cfg: config.Configuration,
                    draw_or_resign_cfg: config.Configuration) -> Union[chess.engine.PlayResult, list[chess.Move]]:
    """
    Get a move from an online source.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    online_egtb_cfg = online_moves_cfg.online_egtb
    chessdb_cfg = online_moves_cfg.chessdb_book
    lichess_cloud_cfg = online_moves_cfg.lichess_cloud_analysis
    opening_explorer_cfg = online_moves_cfg.lichess_opening_explorer
    max_out_of_book_moves = online_moves_cfg.max_out_of_book_moves
    offer_draw = False
    resign = False
    best_move, wdl, comment = get_online_egtb_move(li, board, game, online_egtb_cfg)
    if best_move is not None and comment is not None:  # `and comment is not None` is there only for mypy.
        can_offer_draw = draw_or_resign_cfg.offer_draw_enabled
        offer_draw_for_zero = draw_or_resign_cfg.offer_draw_for_egtb_zero
        if can_offer_draw and offer_draw_for_zero and wdl == 0:
            offer_draw = True

        can_resign = draw_or_resign_cfg.resign_enabled
        resign_on_egtb_loss = draw_or_resign_cfg.resign_for_egtb_minus_two
        if can_resign and resign_on_egtb_loss and wdl == -2:
            resign = True

        wdl_to_score = {2: 9900, 1: 500, 0: 0, -1: -500, -2: -9900}
        comment["score"] = chess.engine.PovScore(chess.engine.Cp(wdl_to_score[wdl]), board.turn)
    elif out_of_online_opening_book_moves[game.id] < max_out_of_book_moves:
        best_move, comment = get_chessdb_move(li, board, game, chessdb_cfg)

    if best_move is None and out_of_online_opening_book_moves[game.id] < max_out_of_book_moves:
        best_move, comment = get_lichess_cloud_move(li, board, game, lichess_cloud_cfg)

    if best_move is None and out_of_online_opening_book_moves[game.id] < max_out_of_book_moves:
        best_move, comment = get_opening_explorer_move(li, board, game, opening_explorer_cfg)

    if best_move:
        if isinstance(best_move, str):
            return chess.engine.PlayResult(chess.Move.from_uci(best_move),
                                           None,
                                           comment,
                                           draw_offered=offer_draw,
                                           resigned=resign)
        return [chess.Move.from_uci(move) for move in best_move]
    out_of_online_opening_book_moves[game.id] += 1
    used_opening_books = chessdb_cfg.enabled or lichess_cloud_cfg.enabled or opening_explorer_cfg.enabled
    if out_of_online_opening_book_moves[game.id] == max_out_of_book_moves and used_opening_books:
        logger.info(f"Will stop using online opening books for game {game.id}.")
    return chess.engine.PlayResult(None, None)


def get_chessdb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
                     chessdb_cfg: config.Configuration) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
    """Get a move from chessdb.cn's opening book."""
    wb = "w" if board.turn == chess.WHITE else "b"
    use_chessdb = chessdb_cfg.enabled
    time_left = msec(game.state[f"{wb}time"])
    min_time = seconds(chessdb_cfg.min_time)
    if not use_chessdb or time_left < min_time or board.uci_variant != "chess":
        return None, None

    move = None
    comment: chess.engine.InfoDict = {}
    site = "https://www.chessdb.cn/cdb.php"
    quality = chessdb_cfg.move_quality
    action = {"best": "querypv",
              "good": "querybest",
              "all": "query"}
    try:
        params = {"action": action[quality],
                  "board": board.fen(),
                  "json": 1}
        data = li.online_book_get(site, params=params)
        if data["status"] == "ok":
            if quality == "best":
                depth = data["depth"]
                if depth >= chessdb_cfg.min_depth:
                    score = data["score"]
                    move = data["pv"][0]
                    comment["score"] = chess.engine.PovScore(chess.engine.Cp(score), board.turn)
                    comment["depth"] = data["depth"]
                    comment["pv"] = list(map(chess.Move.from_uci, data["pv"]))
                    comment["string"] = "lichess-bot-source:ChessDB"
                    logger.info(f"Got move {move} from chessdb.cn (depth: {depth}, score: {score}) for game {game.id}")
            else:
                move = data["move"]
                logger.info(f"Got move {move} from chessdb.cn for game {game.id}")
    except Exception:
        pass

    return move, comment


def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
                           lichess_cloud_cfg: config.Configuration) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
    """Get a move from the lichess's cloud analysis."""
    wb = "w" if board.turn == chess.WHITE else "b"
    time_left = msec(game.state[f"{wb}time"])
    min_time = seconds(lichess_cloud_cfg.min_time)
    use_lichess_cloud = lichess_cloud_cfg.enabled
    if not use_lichess_cloud or time_left < min_time:
        return None, None

    move = None
    comment: chess.engine.InfoDict = {}

    quality = lichess_cloud_cfg.move_quality
    multipv = 1 if quality == "best" else 5
    variant = "standard" if board.uci_variant == "chess" else board.uci_variant

    try:
        data = li.online_book_get("https://lichess.org/api/cloud-eval",
                                  params={"fen": board.fen(),
                                          "multiPv": multipv,
                                          "variant": variant})
        if "error" not in data:
            depth = data["depth"]
            knodes = data["knodes"]
            min_depth = lichess_cloud_cfg.min_depth
            min_knodes = lichess_cloud_cfg.min_knodes
            if depth >= min_depth and knodes >= min_knodes:
                if quality == "best":
                    pv = data["pvs"][0]
                else:
                    best_eval = data["pvs"][0]["cp"]
                    pvs = data["pvs"]
                    max_difference = lichess_cloud_cfg.max_score_difference
                    if wb == "w":
                        pvs = list(filter(lambda pv: pv["cp"] >= best_eval - max_difference, pvs))
                    else:
                        pvs = list(filter(lambda pv: pv["cp"] <= best_eval + max_difference, pvs))
                    pv = random.choice(pvs)
                move = pv["moves"].split()[0]
                score = pv["cp"] if wb == "w" else -pv["cp"]
                comment["score"] = chess.engine.PovScore(chess.engine.Cp(score), board.turn)
                comment["depth"] = data["depth"]
                comment["nodes"] = data["knodes"] * 1000
                comment["pv"] = list(map(chess.Move.from_uci, pv["moves"].split()))
                comment["string"] = "lichess-bot-source:Lichess Cloud Analysis"
                logger.info(f"Got move {move} from lichess cloud analysis (depth: {depth}, score: {score}, knodes: {knodes})"
                            f" for game {game.id}")
    except Exception:
        pass

    return move, comment


def get_opening_explorer_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
                              opening_explorer_cfg: config.Configuration
                              ) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
    """Get a move from lichess's opening explorer."""
    wb = "w" if board.turn == chess.WHITE else "b"
    time_left = msec(game.state[f"{wb}time"])
    min_time = seconds(opening_explorer_cfg.min_time)
    source = opening_explorer_cfg.source
    if not opening_explorer_cfg.enabled or time_left < min_time or source == "master" and board.uci_variant != "chess":
        return None, None

    move = None
    comment: Optional[chess.engine.InfoDict] = None
    variant = "standard" if board.uci_variant == "chess" else board.uci_variant
    try:
        if source == "masters":
            params = {"fen": board.fen(), "moves": 100}
            response = li.online_book_get("https://explorer.lichess.ovh/masters", params)
            comment = {"string": "lichess-bot-source:Lichess Opening Explorer (Masters)"}
        elif source == "player":
            player = opening_explorer_cfg.player_name
            if not player:
                player = game.username
            params = {"player": player, "fen": board.fen(), "moves": 100, "variant": variant,
                      "recentGames": 0, "color": "white" if wb == "w" else "black"}
            response = li.online_book_get("https://explorer.lichess.ovh/player", params, True)
            comment = {"string": "lichess-bot-source:Lichess Opening Explorer (Player)"}
        else:
            params = {"fen": board.fen(), "moves": 100, "variant": variant, "topGames": 0, "recentGames": 0}
            response = li.online_book_get("https://explorer.lichess.ovh/lichess", params)
            comment = {"string": "lichess-bot-source:Lichess Opening Explorer (Lichess)"}
        moves = []
        for possible_move in response["moves"]:
            games_played = possible_move["white"] + possible_move["black"] + possible_move["draws"]
            winrate = (possible_move["white"] + possible_move["draws"] * .5) / games_played
            if games_played >= opening_explorer_cfg.min_games:
                # We add both winrate and games_played to the tuple, so that if 2 moves are tied on the first metric,
                # the second one will be used.
                moves.append((winrate if opening_explorer_cfg.sort == "winrate" else games_played,
                              games_played if opening_explorer_cfg.sort == "winrate" else winrate, possible_move["uci"]))
        moves.sort(reverse=True)
        move = moves[0][2]
        logger.info(f"Got move {move} from lichess opening explorer ({opening_explorer_cfg.sort}: {moves[0][0]})"
                    f" for game {game.id}")
    except Exception:
        pass

    return move, comment


def get_online_egtb_move(li: lichess.Lichess, board: chess.Board, game: model.Game, online_egtb_cfg: config.Configuration
                         ) -> tuple[Union[str, list[str], None], int, Optional[chess.engine.InfoDict]]:
    """
    Get a move from an online egtb (either by lichess or chessdb).

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    use_online_egtb = online_egtb_cfg.enabled
    wb = "w" if board.turn == chess.WHITE else "b"
    pieces = chess.popcount(board.occupied)
    source = online_egtb_cfg.source
    minimum_time = seconds(online_egtb_cfg.min_time)
    if (not use_online_egtb
            or msec(game.state[f"{wb}time"]) < minimum_time
            or board.uci_variant not in ["chess", "antichess", "atomic"]
            and source == "lichess"
            or board.uci_variant != "chess"
            and source == "chessdb"
            or pieces > online_egtb_cfg.max_pieces
            or board.castling_rights):

        return None, -3, None

    quality = online_egtb_cfg.move_quality
    variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant)

    try:
        if source == "lichess":
            return get_lichess_egtb_move(li, game, board, quality, variant)
        elif source == "chessdb":
            return get_chessdb_egtb_move(li, game, board, quality)
    except Exception:
        pass

    return None, -3, None


def get_egtb_move(board: chess.Board, game: model.Game, lichess_bot_tbs: config.Configuration,
                  draw_or_resign_cfg: config.Configuration) -> Union[chess.engine.PlayResult, list[chess.Move]]:
    """
    Get a move from a local egtb.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    best_move, wdl = get_syzygy(board, game, lichess_bot_tbs.syzygy)
    source = "lichess-bot-source:Syzygy EGTB"
    if best_move is None:
        best_move, wdl = get_gaviota(board, game, lichess_bot_tbs.gaviota)
        source = "lichess-bot-source:Gaviota EGTB"
    if best_move:
        can_offer_draw = draw_or_resign_cfg.offer_draw_enabled
        offer_draw_for_zero = draw_or_resign_cfg.offer_draw_for_egtb_zero
        offer_draw = bool(can_offer_draw and offer_draw_for_zero and wdl == 0)

        can_resign = draw_or_resign_cfg.resign_enabled
        resign_on_egtb_loss = draw_or_resign_cfg.resign_for_egtb_minus_two
        resign = bool(can_resign and resign_on_egtb_loss and wdl == -2)
        wdl_to_score = {2: 9900, 1: 500, 0: 0, -1: -500, -2: -9900}
        comment: chess.engine.InfoDict = {"score": chess.engine.PovScore(chess.engine.Cp(wdl_to_score[wdl]), board.turn),
                                          "string": source}
        if isinstance(best_move, chess.Move):
            return chess.engine.PlayResult(best_move, None, comment, draw_offered=offer_draw, resigned=resign)
        return best_move
    return chess.engine.PlayResult(None, None)


def get_lichess_egtb_move(li: lichess.Lichess, game: model.Game, board: chess.Board, quality: str,
                          variant: str) -> tuple[Union[str, list[str], None], int, Optional[chess.engine.InfoDict]]:
    """
    Get a move from lichess's egtb.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    name_to_wld = {"loss": -2,
                   "maybe-loss": -1,
                   "blessed-loss": -1,
                   "draw": 0,
                   "cursed-win": 1,
                   "maybe-win": 1,
                   "win": 2}
    pieces = chess.popcount(board.occupied)
    max_pieces = 7 if board.uci_variant == "chess" else 6
    if pieces <= max_pieces:
        data = li.online_book_get(f"http://tablebase.lichess.ovh/{variant}",
                                  params={"fen": board.fen()})
        if quality == "best":
            move = data["moves"][0]["uci"]
            wdl = name_to_wld[data["moves"][0]["category"]] * -1
            dtz = data["moves"][0]["dtz"] * -1
            dtm = data["moves"][0]["dtm"]
            if dtm:
                dtm *= -1
            logger.info(f"Got move {move} from tablebase.lichess.ovh (wdl: {wdl}, dtz: {dtz}, dtm: {dtm}) for game {game.id}")
        else:  # quality == "suggest":
            best_wdl = name_to_wld[data["moves"][0]["category"]]

            def good_enough(possible_move: LICHESS_EGTB_MOVE) -> bool:
                return name_to_wld[possible_move["category"]] == best_wdl

            possible_moves = list(filter(good_enough, data["moves"]))
            if len(possible_moves) > 1:
                move = [move["uci"] for move in possible_moves]
                wdl = best_wdl * -1
                logger.info(f"Suggesting moves from tablebase.lichess.ovh (wdl: {wdl}) for game {game.id}")
            else:
                best_move = possible_moves[0]
                move = best_move["uci"]
                wdl = name_to_wld[best_move["category"]] * -1
                dtz = best_move["dtz"] * -1
                dtm = best_move["dtm"]
                if dtm:
                    dtm *= -1
                logger.info(f"Got move {move} from tablebase.lichess.ovh (wdl: {wdl}, dtz: {dtz}, dtm: {dtm})"
                            f" for game {game.id}")

        return move, wdl, {"string": "lichess-bot-source:Lichess EGTB"}
    return None, -3, None


def get_chessdb_egtb_move(li: lichess.Lichess, game: model.Game, board: chess.Board,
                          quality: str) -> tuple[Union[str, list[str], None], int, Optional[chess.engine.InfoDict]]:
    """
    Get a move from chessdb's egtb.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    def score_to_wdl(score: int) -> int:
        return piecewise_function([(-20001, 2),
                                   (-1, -1),
                                   (0, 0),
                                   (20000, 1)], 2, score)

    def score_to_dtz(score: int) -> int:
        return piecewise_function([(-20001, -30000 - score),
                                   (-1, -20000 - score),
                                   (0, 0),
                                   (20000, 20000 - score)], 30000 - score, score)

    action = "querypv" if quality == "best" else "queryall"
    data = li.online_book_get("https://www.chessdb.cn/cdb.php",
                              params={"action": action, "board": board.fen(), "json": 1})
    if data["status"] == "ok":
        if quality == "best":
            score = data["score"]
            move = data["pv"][0]
            wdl = score_to_wdl(score)
            dtz = score_to_dtz(score)
            logger.info(f"Got move {move} from chessdb.cn (wdl: {wdl}, dtz: {dtz}) for game {game.id}")
        else:  # quality == "suggest"
            best_wdl = score_to_wdl(data["moves"][0]["score"])

            def good_enough(move: CHESSDB_EGTB_MOVE) -> bool:
                return score_to_wdl(move["score"]) == best_wdl

            possible_moves = list(filter(good_enough, data["moves"]))
            if len(possible_moves) > 1:
                wdl = score_to_wdl(possible_moves[0]["score"])
                move = [move["uci"] for move in possible_moves]
                logger.info(f"Suggesting moves from from chessdb.cn (wdl: {wdl}) for game {game.id}")
            else:
                best_move = possible_moves[0]
                score = best_move["score"]
                move = best_move["uci"]
                wdl = score_to_wdl(score)
                dtz = score_to_dtz(score)
                logger.info(f"Got move {move} from chessdb.cn (wdl: {wdl}, dtz: {dtz}) for game {game.id}")

        return move, wdl, {"string": "lichess-bot-source:ChessDB EGTB"}
    return None, -3, None


def get_syzygy(board: chess.Board, game: model.Game,
               syzygy_cfg: config.Configuration) -> tuple[Union[chess.Move, list[chess.Move], None], int]:
    """
    Get a move from local syzygy egtbs.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    if (not syzygy_cfg.enabled
            or chess.popcount(board.occupied) > syzygy_cfg.max_pieces
            or board.uci_variant not in ["chess", "antichess", "atomic"]):
        return None, -3
    move: Union[chess.Move, list[chess.Move]]
    move_quality = syzygy_cfg.move_quality
    with chess.syzygy.open_tablebase(syzygy_cfg.paths[0]) as tablebase:
        for path in syzygy_cfg.paths[1:]:
            tablebase.add_directory(path)

        try:
            moves = score_syzygy_moves(board, dtz_scorer, tablebase)

            best_wdl = max(map(dtz_to_wdl, moves.values()))
            good_moves = [(move, dtz) for move, dtz in moves.items() if dtz_to_wdl(dtz) == best_wdl]
            if move_quality == "suggest" and len(good_moves) > 1:
                move = [chess_move for chess_move, dtz in good_moves]
                logger.info(f"Suggesting moves from syzygy (wdl: {best_wdl}) for game {game.id}")
                return move, best_wdl
            else:
                # There can be multiple moves with the same dtz.
                best_dtz = min([dtz for chess_move, dtz in good_moves])
                best_moves = [chess_move for chess_move, dtz in good_moves if dtz == best_dtz]
                move = random.choice(best_moves)
                logger.info(f"Got move {move.uci()} from syzygy (wdl: {best_wdl}, dtz: {best_dtz}) for game {game.id}")
                return move, best_wdl
        except KeyError:
            # Attempt to only get the WDL score. It returns moves of quality="suggest", even if quality is set to "best".
            try:
                moves = score_syzygy_moves(board, lambda tablebase, b: -tablebase.probe_wdl(b), tablebase)
                best_wdl = int(max(moves.values()))  # int is there only for mypy.
                good_chess_moves = [chess_move for chess_move, wdl in moves.items() if wdl == best_wdl]
                logger.debug("Found moves using 'move_quality'='suggest'. We didn't find an '.rtbz' file for this endgame."
                             if move_quality == "best" else "")
                if len(good_chess_moves) > 1:
                    move = good_chess_moves
                    logger.info(f"Suggesting moves from syzygy (wdl: {best_wdl}) for game {game.id}")
                else:
                    move = good_chess_moves[0]
                    logger.info(f"Got move {move.uci()} from syzygy (wdl: {best_wdl}) for game {game.id}")
                return move, best_wdl
            except KeyError:
                return None, -3


def dtz_scorer(tablebase: chess.syzygy.Tablebase, board: chess.Board) -> Union[int, float]:
    """
    Score a position based on a syzygy DTZ egtb.

    For a zeroing move (capture or pawn move), a DTZ of +/-0.5 is returned.
    """
    dtz: Union[int, float] = -tablebase.probe_dtz(board)
    dtz = dtz if board.halfmove_clock else .5 * (1 if dtz > 0 else -1)
    return dtz + (1 if dtz > 0 else -1) * board.halfmove_clock * (0 if dtz == 0 else 1)


def dtz_to_wdl(dtz: Union[int, float]) -> int:
    """Convert DTZ scores to syzygy WDL scores."""
    return piecewise_function([(-100, -1), (-.1, -2), (0, 0), (99, 2)], 1, dtz)


def get_gaviota(board: chess.Board, game: model.Game,
                gaviota_cfg: config.Configuration) -> tuple[Union[chess.Move, list[chess.Move], None], int]:
    """
    Get a move from local gaviota egtbs.

    If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
    """
    if (not gaviota_cfg.enabled
            or chess.popcount(board.occupied) > gaviota_cfg.max_pieces
            or board.uci_variant != "chess"):
        return None, -3
    move: Union[chess.Move, list[chess.Move]]
    move_quality = gaviota_cfg.move_quality
    # Since gaviota TBs use dtm and not dtz, we have to put a limit where after it the position are considered to have
    # a syzygy wdl=1/-1, so the positions are draws under the 50 move rule. We use min_dtm_to_consider_as_wdl_1 as a
    # second limit, because if a position has 5 pieces and dtm=110 it may take 98 half-moves, to go down to 4 pieces and
    # another 12 to mate, so this position has a syzygy wdl=2/-2. To be safe, the first limit is 100 moves, which
    # guarantees that all moves have a syzygy wdl=2/-2. Setting min_dtm_to_consider_as_wdl_1 to 100 will disable it
    # because dtm >= dtz, so if abs(dtm) < 100 => abs(dtz) < 100, so wdl=2/-2.
    min_dtm_to_consider_as_wdl_1 = gaviota_cfg.min_dtm_to_consider_as_wdl_1
    with chess.gaviota.open_tablebase(gaviota_cfg.paths[0]) as tablebase:
        for path in gaviota_cfg.paths[1:]:
            tablebase.add_directory(path)

        try:
            moves = score_gaviota_moves(board, dtm_scorer, tablebase)

            best_wdl = max(map(dtm_to_gaviota_wdl, moves.values()))
            good_moves = [(move, dtm) for move, dtm in moves.items() if dtm_to_gaviota_wdl(dtm) == best_wdl]
            best_dtm = min([dtm for move, dtm in good_moves])

            pseudo_wdl = dtm_to_wdl(best_dtm, min_dtm_to_consider_as_wdl_1)
            if move_quality == "suggest":
                best_moves = good_enough_gaviota_moves(good_moves, best_dtm, min_dtm_to_consider_as_wdl_1)
                if len(best_moves) > 1:
                    move = [chess_move for chess_move, dtm in best_moves]
                    logger.info(f"Suggesting moves from gaviota (pseudo wdl: {pseudo_wdl}) for game {game.id}")
                else:
                    move, dtm = random.choice(best_moves)
                    logger.info(f"Got move {move.uci()} from gaviota (pseudo wdl: {pseudo_wdl}, dtm: {dtm})"
                                f" for game {game.id}")
            else:
                # There can be multiple moves with the same dtm.
                best_moves = [(move, dtm) for move, dtm in good_moves if dtm == best_dtm]
                move, dtm = random.choice(best_moves)
                logger.info(f"Got move {move.uci()} from gaviota (pseudo wdl: {pseudo_wdl}, dtm: {dtm}) for game {game.id}")
            return move, pseudo_wdl
        except KeyError:
            return None, -3


def dtm_scorer(tablebase: Union[chess.gaviota.NativeTablebase, chess.gaviota.PythonTablebase], board: chess.Board) -> int:
    """Score a position based on a gaviota DTM egtb."""
    dtm = -tablebase.probe_dtm(board)
    return dtm + (1 if dtm > 0 else -1) * board.halfmove_clock * (0 if dtm == 0 else 1)


def dtm_to_gaviota_wdl(dtm: int) -> int:
    """Convert DTM scores to gaviota WDL scores."""
    return piecewise_function([(-1, -1), (0, 0)], 1, dtm)


def dtm_to_wdl(dtm: int, min_dtm_to_consider_as_wdl_1: int) -> int:
    """Convert DTM scores to syzygy WDL scores."""
    # We use 100 and not min_dtm_to_consider_as_wdl_1, because we want to play it safe and not resign in a
    # position where dtz=-102 (only if resign_for_egtb_minus_two is enabled).
    return piecewise_function([(-100, -1), (-1, -2), (0, 0), (min_dtm_to_consider_as_wdl_1 - 1, 2)], 1, dtm)


def good_enough_gaviota_moves(good_moves: list[tuple[chess.Move, int]], best_dtm: int,
                              min_dtm_to_consider_as_wdl_1: int) -> list[tuple[chess.Move, int]]:
    """
    Get the moves that are good enough to consider.

    :param good_moves: All the moves to choose from.
    :param best_dtm: The best DTM score of a move.
    :param min_dtm_to_consider_as_wdl_1: The minimum DTM score to consider as WDL=1.
    :return: A list of the moves that are good enough to consider.
    """
    if best_dtm < 100:
        # If a move had wdl=2 and dtz=98, but halfmove_clock is 4 then the real wdl=1 and dtz=102, so we
        # want to avoid these positions, if there is a move where even when we add the halfmove_clock the
        # dtz is still <100.
        return [(move, dtm) for move, dtm in good_moves if dtm < 100]
    elif best_dtm < min_dtm_to_consider_as_wdl_1:
        # If a move had wdl=2 and dtz=98, but halfmove_clock is 4 then the real wdl=1 and dtz=102, so we
        # want to avoid these positions, if there is a move where even when we add the halfmove_clock the
        # dtz is still <100.
        return [(move, dtm) for move, dtm in good_moves if dtm < min_dtm_to_consider_as_wdl_1]
    elif best_dtm <= -min_dtm_to_consider_as_wdl_1:
        # If a move had wdl=-2 and dtz=-98, but halfmove_clock is 4 then the real wdl=-1 and dtz=-102, so we
        # want to only choose between the moves where the real wdl=-1.
        return [(move, dtm) for move, dtm in good_moves if dtm <= -min_dtm_to_consider_as_wdl_1]
    elif best_dtm <= -100:
        # If a move had wdl=-2 and dtz=-98, but halfmove_clock is 4 then the real wdl=-1 and dtz=-102, so we
        # want to only choose between the moves where the real wdl=-1.
        return [(move, dtm) for move, dtm in good_moves if dtm <= -100]
    else:
        return good_moves


def piecewise_function(range_definitions: list[tuple[Union[int, float], int]], last_value: int,
                       position: Union[int, float]) -> int:
    """
    Return a value according to a position argument.

    This function is meant to replace if-elif-else blocks that turn ranges into discrete values. For
    example, `piecewise_function([(-20001, 2), (-1, -1), (0, 0), (20000, 1)], 2, score)` is equivalent to:

    if score < -20000:
        return -2
    elif score < 0:
        return -1
    elif score == 0:
        return 0
    elif score <= 20000:
        return 1
    else:
        return 2
    Note: We use -20001 and not -20000, because we use <= and not <.

    Arguments:
    range_definitions:
        A list of tuples with the first element being the inclusive right border of region and the second
        element being the associated value. An element of this list (a, b) corresponds to
        if x <= a:
            return b
        where x is the value of the position argument. This argument should be sorted by the first element
        for correct operation.
    last_value:
        If the position argument is greater than all of the borders in the range_definition argument,
        return this value.
    position:
        The value that will be compared to the first element of the range_definitions tuples.

    """
    for border, value in range_definitions:
        if position <= border:
            return value
    return last_value


def score_syzygy_moves(board: chess.Board,
                       scorer: Union[Callable[[chess.syzygy.Tablebase, chess.Board], int],
                                     Callable[[chess.syzygy.Tablebase, chess.Board], Union[int, float]]],
                       tablebase: chess.syzygy.Tablebase) -> dict[chess.Move, Union[int, float]]:
    """Score all the moves using syzygy egtbs."""
    moves = {}
    for move in board.legal_moves:
        board_copy = board.copy()
        board_copy.push(move)
        moves[move] = scorer(tablebase, board_copy)
    return moves


def score_gaviota_moves(board: chess.Board,
                        scorer: Callable[[Union[chess.gaviota.NativeTablebase, chess.gaviota.PythonTablebase],
                                          chess.Board], int],
                        tablebase: Union[chess.gaviota.NativeTablebase, chess.gaviota.PythonTablebase]
                        ) -> dict[chess.Move, int]:
    """Score all the moves using gaviota egtbs."""
    moves = {}
    for move in board.legal_moves:
        board_copy = board.copy()
        board_copy.push(move)
        moves[move] = scorer(tablebase, board_copy)
    return moves

lichess-bot/engine_wrapper.py
carlini/chess-llm
"""The main module that controls lichess-bot."""
import argparse
import chess
import chess.pgn
from chess.variant import find_variant
import engine_wrapper
import model
import json
import lichess
import logging
import logging.handlers
import multiprocessing
import matchmaking
import signal
import time
import datetime
import backoff
import os
import io
import copy
import math
import sys
import yaml
import traceback
from config import load_config, Configuration
from conversation import Conversation, ChatLine
from timer import Timer, seconds, msec, hours, to_seconds
from requests.exceptions import ChunkedEncodingError, ConnectionError, HTTPError, ReadTimeout
from asyncio.exceptions import TimeoutError as MoveTimeout
from rich.logging import RichHandler
from collections import defaultdict
from collections.abc import Iterator, MutableSequence
from http.client import RemoteDisconnected
from queue import Queue
from multiprocessing.pool import Pool
from typing import Any, Optional, Union
USER_PROFILE_TYPE = dict[str, Any]
EVENT_TYPE = dict[str, Any]
PLAY_GAME_ARGS_TYPE = dict[str, Any]
EVENT_GETATTR_GAME_TYPE = dict[str, Any]
GAME_EVENT_TYPE = dict[str, Any]
CONTROL_QUEUE_TYPE = Queue[EVENT_TYPE]
CORRESPONDENCE_QUEUE_TYPE = Queue[str]
LOGGING_QUEUE_TYPE = Queue[logging.LogRecord]
MULTIPROCESSING_LIST_TYPE = MutableSequence[model.Challenge]
POOL_TYPE = Pool

logger = logging.getLogger(__name__)

with open("versioning.yml") as version_file:
    versioning_info = yaml.safe_load(version_file)

__version__ = versioning_info["lichess_bot_version"]

terminated = False
restart = True


def disable_restart() -> None:
    """Disable restarting lichess-bot when errors occur. Used during testing."""
    global restart
    restart = False


def signal_handler(signal: int, frame: Any) -> None:
    """Terminate lichess-bot."""
    global terminated
    logger.debug("Received SIGINT. Terminating client.")
    terminated = True


signal.signal(signal.SIGINT, signal_handler)


def is_final(exception: Exception) -> bool:
    """If `is_final` returns True then we won't retry."""
    return isinstance(exception, HTTPError) and exception.response.status_code < 500


def upgrade_account(li: lichess.Lichess) -> bool:
    """Upgrade the account to a BOT account."""
    if li.upgrade_to_bot_account() is None:
        return False

    logger.info("Successfully upgraded to Bot Account!")
    return True


def watch_control_stream(control_queue: CONTROL_QUEUE_TYPE, li: lichess.Lichess) -> None:
    """Put the events in a queue."""
    error = None
    while not terminated:
        try:
            response = li.get_event_stream()
            lines = response.iter_lines()
            for line in lines:
                if line:
                    event = json.loads(line.decode("utf-8"))
                    control_queue.put_nowait(event)
                else:
                    control_queue.put_nowait({"type": "ping"})
        except Exception:
            error = traceback.format_exc()
            break

    control_queue.put_nowait({"type": "terminated", "error": error})


def do_correspondence_ping(control_queue: CONTROL_QUEUE_TYPE, period: datetime.timedelta) -> None:
    """
    Tell the engine to check the correspondence games.

    :param period: How many seconds to wait before sending a correspondence ping.
    """
    while not terminated:
        time.sleep(to_seconds(period))
        control_queue.put_nowait({"type": "correspondence_ping"})


def handle_old_logs(auto_log_filename: str) -> None:
    """Remove old logs."""
    directory = os.path.dirname(auto_log_filename)
    old_path = os.path.join(directory, "old.log")
    if os.path.exists(old_path):
        os.remove(old_path)
    if os.path.exists(auto_log_filename):
        os.rename(auto_log_filename, old_path)


def logging_configurer(level: int, filename: Optional[str], auto_log_filename: Optional[str], delete_old_logs: bool) -> None:
    """
    Configure the logger.

    :param level: The logging level. Either `logging.INFO` or `logging.DEBUG`.
    :param filename: The filename to write the logs to. If it is `None` then the logs aren't written to a file.
    :param auto_log_filename: The filename for the automatic logger. If it is `None` then the logs aren't written to a file.
    """
    console_handler = RichHandler()
    console_formatter = logging.Formatter("%(message)s")
    console_handler.setFormatter(console_formatter)
    console_handler.setLevel(level)
    all_handlers: list[logging.Handler] = [console_handler]

    if filename:
        file_handler = logging.FileHandler(filename, delay=True, encoding="utf-8")
        FORMAT = "%(asctime)s %(name)s (%(filename)s:%(lineno)d) %(levelname)s %(message)s"
        file_formatter = logging.Formatter(FORMAT)
        file_handler.setFormatter(file_formatter)
        file_handler.setLevel(level)
        all_handlers.append(file_handler)

    if auto_log_filename:
        os.makedirs(os.path.dirname(auto_log_filename), exist_ok=True)

        # Clear old logs.
        if delete_old_logs:
            handle_old_logs(auto_log_filename)

        # Set up automatic logging.
        auto_file_handler = logging.FileHandler(auto_log_filename, delay=True, encoding="utf-8")
        auto_file_handler.setLevel(logging.DEBUG)

        FORMAT = "%(asctime)s %(name)s (%(filename)s:%(lineno)d) %(levelname)s %(message)s"
        file_formatter = logging.Formatter(FORMAT)
        auto_file_handler.setFormatter(file_formatter)
        all_handlers.append(auto_file_handler)

    logging.basicConfig(level=logging.DEBUG,
                        handlers=all_handlers,
                        force=True)


def logging_listener_proc(queue: LOGGING_QUEUE_TYPE, level: int, log_filename: Optional[str],
                          auto_log_filename: Optional[str]) -> None:
    """
    Handle events from the logging queue.

    This allows the logs from inside a thread to be printed.
    They are added to the queue, so they are printed outside the thread.
    """
    logging_configurer(level, log_filename, auto_log_filename, False)
    logger = logging.getLogger()
    while not terminated:
        task = queue.get()
        try:
            logger.handle(task)
        except Exception:
            pass
        queue.task_done()


def thread_logging_configurer(queue: Union[CONTROL_QUEUE_TYPE, LOGGING_QUEUE_TYPE]) -> None:
    """Configure the game logger."""
    h = logging.handlers.QueueHandler(queue)
    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(h)
    root.setLevel(logging.DEBUG)


def game_error_handler(error: BaseException) -> None:
    """Handle game errors."""
    logger.exception("Game ended due to error:", exc_info=error)


def start(li: lichess.Lichess, user_profile: USER_PROFILE_TYPE, config: Configuration, logging_level: int,
          log_filename: Optional[str], auto_log_filename: Optional[str], one_game: bool = False) -> None:
    """
    Start lichess-bot.

    :param li: Provides communication with lichess.org.
    :param user_profile: Information on our bot.
    :param config: The config that the bot will use.
    :param logging_level: The logging level. Either `logging.INFO` or `logging.DEBUG`.
    :param log_filename: The filename to write the logs to. If it is `None` then the logs aren't written to a file.
    :param auto_log_filename: The filename for the automatic logger. If it is `None` then the logs aren't written to a file.
    :param one_game: Whether the bot should play only one game. Only used in `test_bot/test_bot.py` to test lichess-bot.
    """
    logger.info(f"You're now connected to {config.url} and awaiting challenges.")
    manager = multiprocessing.Manager()
    challenge_queue: MULTIPROCESSING_LIST_TYPE = manager.list()
    control_queue: CONTROL_QUEUE_TYPE = manager.Queue()
    control_stream = multiprocessing.Process(target=watch_control_stream, args=(control_queue, li))
    control_stream.start()
    correspondence_pinger = multiprocessing.Process(target=do_correspondence_ping,
                                                    args=(control_queue,
                                                          seconds(config.correspondence.checkin_period)))
    correspondence_pinger.start()
    correspondence_queue: CORRESPONDENCE_QUEUE_TYPE = manager.Queue()

    logging_queue = manager.Queue()
    logging_listener = multiprocessing.Process(target=logging_listener_proc,
                                               args=(logging_queue,
                                                     logging_level,
                                                     log_filename,
                                                     auto_log_filename))
    logging_listener.start()
    thread_logging_configurer(logging_queue)

    try:
        lichess_bot_main(li,
                         user_profile,
                         config,
                         challenge_queue,
                         control_queue,
                         correspondence_queue,
                         logging_queue,
                         one_game)
    finally:
        control_stream.terminate()
        control_stream.join()
        correspondence_pinger.terminate()
        correspondence_pinger.join()
        logging_configurer(logging_level, log_filename, auto_log_filename, False)
        logging_listener.terminate()
        logging_listener.join()


def log_proc_count(change: str, active_games: set[str]) -> None:
    """
    Log the number of active games and their IDs.

    :param change: Either "Freed", "Used", or "Queued".
    :param active_games: A set containing the IDs of the active games.
    """
    symbol = "+++" if change == "Freed" else "---"
    logger.info(f"{symbol} Process {change}. Count: {len(active_games)}. IDs: {active_games or None}")


def lichess_bot_main(li: lichess.Lichess,
                     user_profile: USER_PROFILE_TYPE,
                     config: Configuration,
                     challenge_queue: MULTIPROCESSING_LIST_TYPE,
                     control_queue: CONTROL_QUEUE_TYPE,
                     correspondence_queue: CORRESPONDENCE_QUEUE_TYPE,
                     logging_queue: LOGGING_QUEUE_TYPE,
                     one_game: bool) -> None:
    """
    Handle all the games and challenges.

    :param li: Provides communication with lichess.org.
    :param user_profile: Information on our bot.
    :param config: The config that the bot will use.
    :param challenge_queue: The queue containing the challenges.
    :param control_queue: The queue containing all the events.
    :param correspondence_queue: The queue containing the correspondence games.
    :param logging_queue: The logging queue. Used by `logging_listener_proc`.
    :param one_game: Whether the bot should play only one game. Only used in `test_bot/test_bot.py` to test lichess-bot.
    """
    global restart

    max_games = config.challenge.concurrency

    one_game_completed = False

    all_games = li.get_ongoing_games()
    startup_correspondence_games = [game["gameId"]
                                    for game in all_games
                                    if game["speed"] == "correspondence"]
    active_games = set(game["gameId"]
                       for game in all_games
                       if game["gameId"] not in startup_correspondence_games)
    low_time_games: list[EVENT_GETATTR_GAME_TYPE] = []

    last_check_online_time = Timer(hours(1))
    matchmaker = matchmaking.Matchmaking(li, config, user_profile)
    matchmaker.show_earliest_challenge_time()

    play_game_args = {"li": li,
                      "control_queue": control_queue,
                      "user_profile": user_profile,
                      "config": config,
                      "challenge_queue": challenge_queue,
                      "correspondence_queue": correspondence_queue,
                      "logging_queue": logging_queue}

    recent_bot_challenges: defaultdict[str, list[Timer]] = defaultdict(list)

    with multiprocessing.pool.Pool(max_games + 1) as pool:
        while not (terminated or (one_game and one_game_completed) or restart):
            event = next_event(control_queue)
            if not event:
                continue

            if event["type"] == "terminated":
                restart = True
                logger.debug(f"Terminating exception:\n{event['error']}")
                control_queue.task_done()
                break
            elif event["type"] in ["local_game_done", "gameFinish"]:
                game_id = event["game"]["id"]
                if game_id in active_games:
                    active_games.discard(game_id)
                    matchmaker.game_done()
                    log_proc_count("Freed", active_games)
                save_pgn_record(event, config)
                one_game_completed = True
            elif event["type"] == "challenge":
                handle_challenge(event, li, challenge_queue, config.challenge, user_profile, matchmaker, recent_bot_challenges)
            elif event["type"] == "challengeDeclined":
                matchmaker.declined_challenge(event)
            elif event["type"] == "gameStart":
                matchmaker.accepted_challenge(event)
                start_game(event,
                           pool,
                           play_game_args,
                           config,
                           matchmaker,
                           startup_correspondence_games,
                           correspondence_queue,
                           active_games,
                           low_time_games)

            start_low_time_games(low_time_games, active_games, max_games, pool, play_game_args)
            check_in_on_correspondence_games(pool,
                                             event,
                                             correspondence_queue,
                                             challenge_queue,
                                             play_game_args,
                                             active_games,
                                             max_games)
            accept_challenges(li, challenge_queue, active_games, max_games)
            matchmaker.challenge(active_games, challenge_queue)
            check_online_status(li, user_profile, last_check_online_time)

            control_queue.task_done()

    logger.info("Terminated")


def next_event(control_queue: CONTROL_QUEUE_TYPE) -> EVENT_TYPE:
    """Get the next event from the control queue."""
    try:
        event: EVENT_TYPE = control_queue.get()
    except InterruptedError:
        return {}

    if "type" not in event:
        logger.warning("Unable to handle response from lichess.org:")
        logger.warning(event)
        control_queue.task_done()
        return {}

    if event.get("type") != "ping":
        logger.debug(f"Event: {event}")

    return event


correspondence_games_to_start = 0


def check_in_on_correspondence_games(pool: POOL_TYPE,
                                     event: EVENT_TYPE,
                                     correspondence_queue: CORRESPONDENCE_QUEUE_TYPE,
                                     challenge_queue: MULTIPROCESSING_LIST_TYPE,
                                     play_game_args: PLAY_GAME_ARGS_TYPE,
                                     active_games: set[str],
                                     max_games: int) -> None:
    """Start correspondence games."""
    global correspondence_games_to_start

    if event["type"] == "correspondence_ping":
        correspondence_games_to_start = correspondence_queue.qsize()
    elif event["type"] != "local_game_done":
        return

    if challenge_queue:
        return

    while len(active_games) < max_games and correspondence_games_to_start > 0:
        game_id = correspondence_queue.get_nowait()
        correspondence_games_to_start -= 1
        correspondence_queue.task_done()
        start_game_thread(active_games, game_id, play_game_args, pool)


def start_low_time_games(low_time_games: list[EVENT_GETATTR_GAME_TYPE], active_games: set[str], max_games: int,
                         pool: POOL_TYPE, play_game_args: PLAY_GAME_ARGS_TYPE) -> None:
    """Start the games based on how much time we have left."""
    low_time_games.sort(key=lambda g: g.get("secondsLeft", math.inf))
    while low_time_games and len(active_games) < max_games:
        game_id = low_time_games.pop(0)["id"]
        start_game_thread(active_games, game_id, play_game_args, pool)


def accept_challenges(li: lichess.Lichess, challenge_queue: MULTIPROCESSING_LIST_TYPE, active_games: set[str],
                      max_games: int) -> None:
    """Accept a challenge."""
    while len(active_games) < max_games and challenge_queue:
        chlng = challenge_queue.pop(0)
        if chlng.from_self:
            continue

        try:
            logger.info(f"Accept {chlng}")
            li.accept_challenge(chlng.id)
            active_games.add(chlng.id)
            log_proc_count("Queued", active_games)
        except (HTTPError, ReadTimeout) as exception:
            if isinstance(exception, HTTPError) and exception.response.status_code == 404:
                logger.info(f"Skip missing {chlng}")


def check_online_status(li: lichess.Lichess, user_profile: USER_PROFILE_TYPE, last_check_online_time: Timer) -> None:
    """Check if lichess.org thinks the bot is online or not. If it isn't, we restart it."""
    global restart

    if last_check_online_time.is_expired():
        try:
            if not li.is_online(user_profile["id"]):
                logger.info("Will restart lichess-bot")
                restart = True
            last_check_online_time.reset()
        except (HTTPError, ReadTimeout):
            pass


def sort_challenges(challenge_queue: MULTIPROCESSING_LIST_TYPE, challenge_config: Configuration) -> None:
    """
    Sort the challenges.

    They can be sorted either by rating (the best challenger is accepted first),
    or by time (the first challenger is accepted first).
    """
    if challenge_config.sort_by == "best":
        list_c = list(challenge_queue)
        list_c.sort(key=lambda c: -c.score())
        challenge_queue[:] = list_c


def start_game_thread(active_games: set[str], game_id: str, play_game_args: PLAY_GAME_ARGS_TYPE, pool: POOL_TYPE) -> None:
    """Start a game thread."""
    active_games.add(game_id)
    log_proc_count("Used", active_games)
    play_game_args["game_id"] = game_id
    pool.apply_async(play_game,
                     kwds=play_game_args,
                     error_callback=game_error_handler)


def start_game(event: EVENT_TYPE,
               pool: POOL_TYPE,
               play_game_args: PLAY_GAME_ARGS_TYPE,
               config: Configuration,
               matchmaker: matchmaking.Matchmaking,
               startup_correspondence_games: list[str],
               correspondence_queue: CORRESPONDENCE_QUEUE_TYPE,
               active_games: set[str],
               low_time_games: list[EVENT_GETATTR_GAME_TYPE]) -> None:
    """
    Start a game.

    :param event: The gameStart event.
    :param pool: The thread pool that the game is added to, so they can be run asynchronously.
    :param play_game_args: The args passed to `play_game`.
    :param config: The config the bot will use.
    :param matchmaker: The matchmaker that challenges other bots.
    :param startup_correspondence_games: A list of correspondence games that have to be started.
    :param correspondence_queue: The queue that correspondence games are added to, to be started.
    :param active_games: A set of all the games that aren't correspondence games.
    :param low_time_games: A list of games, in which we don't have much time remaining.
    """
    game_id = event["game"]["id"]
    if matchmaker.challenge_id == game_id:
        matchmaker.challenge_id = ""
    if game_id in startup_correspondence_games:
        if enough_time_to_queue(event, config):
            logger.info(f'--- Enqueue {config.url + game_id}')
            correspondence_queue.put_nowait(game_id)
        else:
            logger.info(f'--- Will start {config.url + game_id} as soon as possible')
            low_time_games.append(event["game"])
        startup_correspondence_games.remove(game_id)
    else:
        start_game_thread(active_games, game_id, play_game_args, pool)


def enough_time_to_queue(event: EVENT_TYPE, config: Configuration) -> bool:
    """Check whether the correspondence must be started now or if it can wait."""
    corr_cfg = config.correspondence
    minimum_time = (corr_cfg.checkin_period + corr_cfg.move_time) * 10
    game = event["game"]
    return not game["isMyTurn"] or game.get("secondsLeft", math.inf) > minimum_time


def handle_challenge(event: EVENT_TYPE, li: lichess.Lichess, challenge_queue: MULTIPROCESSING_LIST_TYPE,
                     challenge_config: Configuration, user_profile: USER_PROFILE_TYPE,
                     matchmaker: matchmaking.Matchmaking, recent_bot_challenges: defaultdict[str, list[Timer]]) -> None:
    """Handle incoming challenges. It either accepts, declines, or queues them to accept later."""
    chlng = model.Challenge(event["challenge"], user_profile)
    is_supported, decline_reason = chlng.is_supported(challenge_config, recent_bot_challenges)
    if is_supported:
        challenge_queue.append(chlng)
        sort_challenges(challenge_queue, challenge_config)
        time_window = challenge_config.recent_bot_challenge_age
        if time_window is not None:
            recent_bot_challenges[chlng.challenger.name].append(Timer(seconds(time_window)))
    elif chlng.id != matchmaker.challenge_id:
        li.decline_challenge(chlng.id, reason="tooSlow" if decline_reason == "timeControl" else decline_reason)


@backoff.on_exception(backoff.expo, BaseException, max_time=600, giveup=is_final,  # type: ignore[arg-type]
                      on_backoff=lichess.backoff_handler)
def play_game(li: lichess.Lichess,
              game_id: str,
              control_queue: CONTROL_QUEUE_TYPE,
              user_profile: USER_PROFILE_TYPE,
              config: Configuration,
              challenge_queue: MULTIPROCESSING_LIST_TYPE,
              correspondence_queue: CORRESPONDENCE_QUEUE_TYPE,
              logging_queue: LOGGING_QUEUE_TYPE) -> None:
    """
    Play a game.

    :param li: Provides communication with lichess.org.
    :param game_id: The id of the game.
    :param control_queue: The control queue that contains events (adds `local_game_done` to the queue).
    :param user_profile: Information on our bot.
    :param config: The config that the bot will use.
    :param challenge_queue: The queue containing the challenges.
    :param correspondence_queue: The queue containing the correspondence games.
    :param logging_queue: The logging queue. Used by `logging_listener_proc`.
    """
    thread_logging_configurer(logging_queue)
    logger = logging.getLogger(__name__)

    response = li.get_game_stream(game_id)
    lines = response.iter_lines()

    # Initial response of stream will be the full game info. Store it.
    initial_state = json.loads(next(lines).decode("utf-8"))
    logger.debug(f"Initial state: {initial_state}")
    abort_time = seconds(config.abort_time)
    game = model.Game(initial_state, user_profile["username"], li.baseUrl, abort_time)

    with engine_wrapper.create_engine(config) as engine:
        engine.get_opponent_info(game)
        logger.debug(f"The engine for game {game_id} has pid={engine.get_pid()}")
        conversation = Conversation(game, engine, li, __version__, challenge_queue)

        logger.info(f"+++ {game}")

        is_correspondence = game.speed == "correspondence"
        correspondence_cfg = config.correspondence
        correspondence_move_time = seconds(correspondence_cfg.move_time)
        correspondence_disconnect_time = seconds(correspondence_cfg.disconnect_time)

        engine_cfg = config.engine
        ponder_cfg = correspondence_cfg if is_correspondence else engine_cfg
        can_ponder = ponder_cfg.uci_ponder or ponder_cfg.ponder
        move_overhead = msec(config.move_overhead)
        delay = msec(config.rate_limiting_delay)

        keyword_map: defaultdict[str, str] = defaultdict(str, me=game.me.name, opponent=game.opponent.name)
        hello = get_greeting("hello", config.greeting, keyword_map)
        goodbye = get_greeting("goodbye", config.greeting, keyword_map)
        hello_spectators = get_greeting("hello_spectators", config.greeting, keyword_map)
        goodbye_spectators = get_greeting("goodbye_spectators", config.greeting, keyword_map)

        disconnect_time = correspondence_disconnect_time if not game.state.get("moves") else seconds(0)
        prior_game = None
        board = chess.Board()
        upd: dict[str, Any] = game.state
        while not terminated:
            move_attempted = False
            try:
                upd = upd or next_update(lines)
                u_type = upd["type"] if upd else "ping"
                if u_type == "chatLine":
                    conversation.react(ChatLine(upd))
                elif u_type == "gameState":
                    game.state = upd
                    board = setup_board(game)
                    if not is_game_over(game) and is_engine_move(game, prior_game, board):
                        disconnect_time = correspondence_disconnect_time
                        say_hello(conversation, hello, hello_spectators, board)
                        setup_timer = Timer()
                        print_move_number(board)
                        move_attempted = True
                        engine.play_move(board,
                                         game,
                                         li,
                                         setup_timer,
                                         move_overhead,
                                         can_ponder,
                                         is_correspondence,
                                         correspondence_move_time,
                                         engine_cfg,
                                         fake_think_time(config, board, game),
                                         conversation)
                        time.sleep(to_seconds(delay))
                    elif is_game_over(game):
                        tell_user_game_result(game, board)
                        engine.send_game_result(game, board)
                        conversation.send_message("player", goodbye)
                        conversation.send_message("spectator", goodbye_spectators)

                    wb = "w" if board.turn == chess.WHITE else "b"
                    terminate_time = msec(upd[f"{wb}time"]) + msec(upd[f"{wb}inc"]) + seconds(60)
                    game.ping(abort_time, terminate_time, disconnect_time)
                    prior_game = copy.deepcopy(game)
                elif u_type == "ping" and should_exit_game(board, game, prior_game, li, is_correspondence):
                    break
            except (HTTPError,
                    ReadTimeout,
                    RemoteDisconnected,
                    ChunkedEncodingError,
                    ConnectionError,
                    StopIteration,
                    MoveTimeout) as e:
                stopped = isinstance(e, StopIteration)
                is_ongoing = game.id in (ongoing_game["gameId"] for ongoing_game in li.get_ongoing_games())
                if stopped or (not move_attempted and not is_ongoing):
                    break
            finally:
                upd = {}

        pgn_record = try_get_pgn_game_record(li, config, game, board, engine)
    final_queue_entries(control_queue, correspondence_queue, game, is_correspondence, pgn_record)


def get_greeting(greeting: str, greeting_cfg: Configuration, keyword_map: defaultdict[str, str]) -> str:
    """Get the greeting to send to the chat."""
    greeting_text: str = greeting_cfg.lookup(greeting)
    return greeting_text.format_map(keyword_map)


def say_hello(conversation: Conversation, hello: str, hello_spectators: str, board: chess.Board) -> None:
    """Send the greetings to the chat rooms."""
    if len(board.move_stack) < 2:
        conversation.send_message("player", hello)
        conversation.send_message("spectator", hello_spectators)

        #conversation.send_message("player", "For more information see https://nicholas.carlini.com/writing/2023/chess-llm.html")
        #conversation.send_message("spectator", "For more information see https://nicholas.carlini.com/writing/2023/chess-llm.html")
        

def fake_think_time(config: Configuration, board: chess.Board, game: model.Game) -> datetime.timedelta:
    """Calculate how much time we should wait for fake_think_time."""
    sleep = seconds(0.0)

    if config.fake_think_time and len(board.move_stack) > 9:
        remaining = max(seconds(0), game.my_remaining_time() - msec(config.move_overhead))
        delay = remaining * 0.025
        accel = 0.99 ** (len(board.move_stack) - 10)
        sleep = delay * accel

    return sleep


def print_move_number(board: chess.Board) -> None:
    """Log the move number."""
    logger.info("")
    logger.info(f"move: {len(board.move_stack) // 2 + 1}")


def next_update(lines: Iterator[bytes]) -> GAME_EVENT_TYPE:
    """Get the next game state."""
    binary_chunk = next(lines)
    upd: GAME_EVENT_TYPE = json.loads(binary_chunk.decode("utf-8")) if binary_chunk else {}
    if upd:
        logger.debug(f"Game state: {upd}")
    return upd


def setup_board(game: model.Game) -> chess.Board:
    """Set up the board."""
    if game.variant_name.lower() == "chess960":
        board = chess.Board(game.initial_fen, chess960=True)
    elif game.variant_name == "From Position":
        board = chess.Board(game.initial_fen)
    else:
        VariantBoard = find_variant(game.variant_name)
        board = VariantBoard()

    for move in game.state["moves"].split():
        try:
            board.push_uci(move)
        except ValueError:
            logger.exception(f"Ignoring illegal move {move} on board {board.fen()}")

    return board


def is_engine_move(game: model.Game, prior_game: Optional[model.Game], board: chess.Board) -> bool:
    """Check whether it is the engine's turn."""
    return game_changed(game, prior_game) and game.is_white == (board.turn == chess.WHITE)


def is_game_over(game: model.Game) -> bool:
    """Check whether the game is over."""
    status: str = game.state["status"]
    return status != "started"


def should_exit_game(board: chess.Board, game: model.Game, prior_game: Optional[model.Game], li: lichess.Lichess,
                     is_correspondence: bool) -> bool:
    """Whether we should exit a game."""
    if (is_correspondence
            and not is_engine_move(game, prior_game, board)
            and game.should_disconnect_now()):
        return True
    elif game.should_abort_now():
        logger.info(f"Aborting {game.url()} by lack of activity")
        li.abort(game.id)
        return True
    elif game.should_terminate_now():
        logger.info(f"Terminating {game.url()} by lack of activity")
        if game.is_abortable():
            li.abort(game.id)
        return True
    else:
        return False


def final_queue_entries(control_queue: CONTROL_QUEUE_TYPE, correspondence_queue: CORRESPONDENCE_QUEUE_TYPE,
                        game: model.Game, is_correspondence: bool, pgn_record: str) -> None:
    """
    Log the game that ended or we disconnected from, and sends a `local_game_done` for the game.

     If this is an unfinished correspondence game, put it in a queue to resume later.
    """
    if is_correspondence and not is_game_over(game):
        logger.info(f"--- Disconnecting from {game.url()}")
        correspondence_queue.put_nowait(game.id)
    else:
        logger.info(f"--- {game.url()} Game over")

    control_queue.put_nowait({"type": "local_game_done", "game": {"id": game.id, "pgn": pgn_record, "state": game}})


def game_changed(current_game: model.Game, prior_game: Optional[model.Game]) -> bool:
    """Check whether the current game state is different from the previous game state."""
    if prior_game is None:
        return True

    current_game_moves_str: str = current_game.state["moves"]
    prior_game_moves_str: str = prior_game.state["moves"]
    return current_game_moves_str != prior_game_moves_str


def tell_user_game_result(game: model.Game, board: chess.Board) -> None:
    """Log the game result."""
    winner = game.state.get("winner")
    termination = game.state.get("status")

    winning_name = game.white.name if winner == "white" else game.black.name
    losing_name = game.white.name if winner == "black" else game.black.name

    if winner is not None:
        logger.info(f"{winning_name} won!")
    elif termination in [model.Termination.DRAW, model.Termination.TIMEOUT]:
        logger.info("Game ended in a draw.")
    else:
        logger.info("Game adjourned.")

    simple_endings = {model.Termination.MATE: "Game won by checkmate.",
                      model.Termination.RESIGN: f"{losing_name} resigned.",
                      model.Termination.ABORT: "Game aborted."}

    if termination in simple_endings:
        logger.info(simple_endings[termination])
    elif termination == model.Termination.DRAW:
        draw_results = [(board.is_fifty_moves(), "Game drawn by 50-move rule."),
                        (board.is_repetition(), "Game drawn by threefold repetition."),
                        (board.is_insufficient_material(), "Game drawn from insufficient material."),
                        (board.is_stalemate(), "Game drawn by stalemate."),
                        (True, "Game drawn by agreement.")]
        messages = [draw_message for is_result, draw_message in draw_results if is_result]
        logger.info(messages[0])
    elif termination == model.Termination.TIMEOUT:
        if winner:
            logger.info(f"{losing_name} forfeited on time.")
        else:
            timeout_name = game.white.name if game.state.get("wtime") == 0 else game.black.name
            other_name = game.white.name if timeout_name == game.black.name else game.black.name
            logger.info(f"{timeout_name} ran out of time, but {other_name} did not have enough material to mate.")
    elif termination:
        logger.info(f"Game ended by {termination}")


def try_get_pgn_game_record(li: lichess.Lichess, config: Configuration, game: model.Game, board: chess.Board,
                            engine: engine_wrapper.EngineWrapper) -> str:
    """
    Call `print_pgn_game_record` to write the game to a PGN file and handle errors raised by it.

    :param li: Provides communication with lichess.org.
    :param config: The config that the bot will use.
    :param game: Contains information about the game (e.g. the players' names).
    :param board: The board. Contains the moves.
    :param engine: The engine. Contains information about the moves (e.g. eval, PV, depth).
    """
    if board is None:
        return

    try:
        return pgn_game_record(li, config, game, board, engine)
    except Exception:
        logger.exception("Error writing game record:")
        return ""


def pgn_game_record(li: lichess.Lichess, config: Configuration, game: model.Game, board: chess.Board,
                    engine: engine_wrapper.EngineWrapper) -> str:
    """
    Return the text of the game's PGN.

    :param li: Provides communication with lichess.org.
    :param config: The config that the bot will use.
    :param game: Contains information about the game (e.g. the players' names).
    :param board: The board. Contains the moves.
    :param engine: The engine. Contains information about the moves (e.g. eval, PV, depth).
    """
    if not config.pgn_directory:
        return ""

    lichess_game_record = chess.pgn.read_game(io.StringIO(li.get_game_pgn(game.id))) or chess.pgn.Game()
    try:
        # Recall previously written PGN file to retain engine evaluations.
        previous_game_path = get_game_file_path(config, game, True)
        with open(previous_game_path) as game_data:
            game_record = chess.pgn.read_game(game_data) or lichess_game_record
        game_record.headers.update(lichess_game_record.headers)
    except FileNotFoundError:
        game_record = lichess_game_record

    fill_missing_pgn_headers(game_record, game)

    current_node: Union[chess.pgn.Game, chess.pgn.ChildNode] = game_record.game()
    lichess_node: Union[chess.pgn.Game, chess.pgn.ChildNode] = lichess_game_record.game()
    for index, move in enumerate(board.move_stack):
        next_node = current_node.next()
        if next_node is None or next_node.move != move:
            current_node = current_node.add_main_variation(move)
        else:
            current_node = next_node

        next_lichess_node = lichess_node.next()
        if next_lichess_node:
            lichess_node = next_lichess_node
            current_node.set_clock(lichess_node.clock())
            if current_node.comment != lichess_node.comment:
                current_node.comment = f"{current_node.comment} {lichess_node.comment}".strip()

        commentary = engine.comment_for_board_index(index)
        pv_node = current_node.parent.add_line(commentary["pv"]) if "pv" in commentary else current_node
        pv_node.set_eval(commentary.get("score"), commentary.get("depth"))

    pgn_writer = chess.pgn.StringExporter()
    return game_record.accept(pgn_writer)


def get_game_file_path(config: Configuration, game: model.Game, force_single: bool = False) -> str:
    """Return the path of the file where the game record will be written."""
    if config.pgn_file_grouping == "game" or not is_game_over(game) or force_single:
        game_file_name = f"{game.white.name} vs {game.black.name} - {game.id}.pgn"
    elif config.pgn_file_grouping == "opponent":
        game_file_name = f"{game.me.name} games vs. {game.opponent.name}.pgn"
    else:  # config.pgn_file_grouping == "all"
        game_file_name = f"{game.me.name} games.pgn"
    game_file_name = "".join(c for c in game_file_name if c not in '<>:"/\\|?*')
    return os.path.join(config.pgn_directory, game_file_name)


def fill_missing_pgn_headers(game_record: chess.pgn.Game, game: model.Game) -> None:
    """
    Fill in any missing headers in the PGN record provided by lichess.org with information from `game`.

    :param game_record: A `chess.pgn.Game` object containing information about the game lichess.org's PGN file.
    :param game: Contains information about the game (e.g. the players' names), which is used to get the local headers.
    """
    local_headers = get_headers(game)
    for header, game_value in local_headers.items():
        record_value = game_record.headers.get(header)
        if not record_value or record_value.startswith("?") or (header == "Result" and record_value == "*"):
            game_record.headers[header] = str(game_value)


def get_headers(game: model.Game) -> dict[str, Union[str, int]]:
    """
    Create local headers to be written in the PGN file.

    :param game: Contains information about the game (e.g. the players' names).
    :return: The headers in a dict.
    """
    headers: dict[str, Union[str, int]] = {}
    headers["Event"] = game.pgn_event()
    headers["Site"] = game.short_url()
    headers["Date"] = game.game_start.strftime("%Y.%m.%d")
    headers["White"] = game.white.name or str(game.white)
    headers["Black"] = game.black.name or str(game.black)
    headers["Result"] = game.result()

    if game.black.rating:
        headers["BlackElo"] = game.black.rating
    if game.black.title:
        headers["BlackTitle"] = game.black.title

    if game.perf_name != "correspondence":
        headers["TimeControl"] = game.time_control()

    headers["UTCDate"] = headers["Date"]
    headers["UTCTime"] = game.game_start.strftime("%H:%M:%S")
    if game.variant_name not in ["Standard", "From Position"]:
        headers["Variant"] = game.variant_name

    if game.white.rating:
        headers["WhiteElo"] = game.white.rating
    if game.white.title:
        headers["WhiteTitle"] = game.white.title

    return headers


def save_pgn_record(event: EVENT_TYPE, config: Configuration) -> None:
    """Write the game PGN record to a file."""
    if (not config.pgn_directory
            or event["type"] != "local_game_done"
            or not event["game"]["pgn"]):
        return

    os.makedirs(config.pgn_directory, exist_ok=True)
    game = event["game"]["state"]
    game_path = get_game_file_path(config, game)
    single_game_path = get_game_file_path(config, game, True)
    write_mode = "w" if game_path == single_game_path else "a"
    logger.debug(f"Writing PGN game record to: {game_path}")
    with open(game_path, write_mode) as game_file:
        game_file.write(event["game"]["pgn"] + "\n\n")

    if os.path.exists(single_game_path) and game_path != single_game_path:
        os.remove(single_game_path)


def intro() -> str:
    """Return the intro string."""
    return fr"""
    .   _/|
    .  // o\
    .  || ._)  lichess-bot {__version__}
    .  //__\
    .  )___(   Play on Lichess with a bot
    """


def start_lichess_bot() -> None:
    """Parse arguments passed to lichess-bot.py and starts lichess-bot."""
    parser = argparse.ArgumentParser(description="Play on Lichess with a bot")
    parser.add_argument("-u", action="store_true", help="Upgrade your account to a bot account.")
    parser.add_argument("-v", action="store_true", help="Make output more verbose. Include all communication with lichess.")
    parser.add_argument("--config", help="Specify a configuration file (defaults to ./config.yml).")
    parser.add_argument("-l", "--logfile", help="Record all console output to a log file.", default=None)
    parser.add_argument("--disable_auto_logging", action="store_true", help="Disable automatic logging.")
    args = parser.parse_args()

    logging_level = logging.DEBUG if args.v else logging.INFO
    auto_log_filename = None
    if not args.disable_auto_logging:
        auto_log_filename = "./lichess_bot_auto_logs/recent.log"
    logging_configurer(logging_level, args.logfile, auto_log_filename, True)
    logger.info(intro(), extra={"highlighter": None})

    CONFIG = load_config(args.config or "./config.yml")
    logger.info("Checking engine configuration ...")
    with engine_wrapper.create_engine(CONFIG):
        pass
    logger.info("Engine configuration OK")

    max_retries = CONFIG.engine.online_moves.max_retries
    check_python_version()
    li = lichess.Lichess(CONFIG.token, CONFIG.url, __version__, logging_level, max_retries)

    user_profile = li.get_profile()
    username = user_profile["username"]
    is_bot = user_profile.get("title") == "BOT"
    logger.info(f"Welcome {username}!")

    if args.u and not is_bot:
        is_bot = upgrade_account(li)

    if is_bot:
        start(li, user_profile, CONFIG, logging_level, args.logfile, auto_log_filename)
    else:
        logger.error(f"{username} is not a bot account. Please upgrade it to a bot account!")
    logging.shutdown()


def check_python_version() -> None:
    """Raise a warning or an exception if the version isn't supported or is deprecated."""
    def version_numeric(version_str: str) -> list[int]:
        return [int(n) for n in version_str.split(".")]

    python_deprecated_version = version_numeric(versioning_info["deprecated_python_version"])
    python_good_version = version_numeric(versioning_info["minimum_python_version"])
    version_change_date = versioning_info["deprecation_date"]
    this_python_version = list(sys.version_info[0:2])

    def version_str(version: list[int]) -> str:
        return f"Python {'.'.join(str(n) for n in version)}"

    upgrade_request = (f"You are currently running {version_str(this_python_version)}. "
                       f"Please upgrade to {version_str(python_good_version)} or newer")
    out_of_date_error = RuntimeError("A newer version of Python is required "
                                     f"to run this version of lichess-bot. {upgrade_request}.")
    out_of_date_warning = ("A newer version of Python will be required "
                           f"on {version_change_date} to run lichess-bot. {upgrade_request} before then.")

    this_lichess_bot_version = version_numeric(__version__)
    lichess_bot_breaking_version = list(version_change_date.timetuple()[0:3])

    if this_python_version < python_deprecated_version:
        raise out_of_date_error

    if this_python_version == python_deprecated_version:
        if this_lichess_bot_version < lichess_bot_breaking_version:
            logger.warning(out_of_date_warning)
        else:
            raise out_of_date_error


if __name__ == "__main__":
    multiprocessing.set_start_method('spawn')
    try:
        while restart:
            restart = False
            start_lichess_bot()
            time.sleep(10 if restart else 0)
    except Exception:
        logger.exception("Quitting lichess-bot due to an error:")

lichess-bot/lichess-bot.py
carlini/chess-llm
"""Communication with APIs."""
import json
import requests
from urllib.parse import urljoin
from requests.exceptions import ConnectionError, HTTPError, ReadTimeout
from http.client import RemoteDisconnected
import backoff
import logging
import traceback
from collections import defaultdict
import datetime
from timer import Timer, seconds, sec_str
from typing import Optional, Union, Any
import chess.engine
JSON_REPLY_TYPE = dict[str, Any]
REQUESTS_PAYLOAD_TYPE = dict[str, Any]

ENDPOINTS = {
    "profile": "/api/account",
    "playing": "/api/account/playing",
    "stream": "/api/bot/game/stream/{}",
    "stream_event": "/api/stream/event",
    "move": "/api/bot/game/{}/move/{}",
    "chat": "/api/bot/game/{}/chat",
    "abort": "/api/bot/game/{}/abort",
    "accept": "/api/challenge/{}/accept",
    "decline": "/api/challenge/{}/decline",
    "upgrade": "/api/bot/account/upgrade",
    "resign": "/api/bot/game/{}/resign",
    "export": "/game/export/{}",
    "online_bots": "/api/bot/online",
    "challenge": "/api/challenge/{}",
    "cancel": "/api/challenge/{}/cancel",
    "status": "/api/users/status",
    "public_data": "/api/user/{}",
    "token_test": "/api/token/test"
}


logger = logging.getLogger(__name__)

MAX_CHAT_MESSAGE_LEN = 140  # The maximum characters in a chat message.


class RateLimited(RuntimeError):
    """Exception raised when we are rate limited (status code 429)."""

    pass


def is_new_rate_limit(response: requests.models.Response) -> bool:
    """Check if the status code is 429, which means that we are rate limited."""
    return response.status_code == 429


def is_final(exception: Exception) -> bool:
    """If `is_final` returns True then we won't retry."""
    return isinstance(exception, HTTPError) and exception.response.status_code < 500


def backoff_handler(details: Any) -> None:
    """Log exceptions inside functions with the backoff decorator."""
    logger.debug("Backing off {wait:0.1f} seconds after {tries} tries "
                 "calling function {target} with args {args} and kwargs {kwargs}".format(**details))
    logger.debug(f"Exception: {traceback.format_exc()}")


# Docs: https://lichess.org/api.
class Lichess:
    """Communication with lichess.org (and chessdb.cn for getting moves)."""

    def __init__(self, token: str, url: str, version: str, logging_level: int, max_retries: int) -> None:
        """
        Communication with lichess.org (and chessdb.cn for getting moves).

        :param token: The bot's token.
        :param url: The base url (lichess.org).
        :param version: The lichess-bot version running.
        :param logging_level: The logging level (logging.INFO or logging.DEBUG).
        :param max_retries: The maximum amount of retries for online moves (e.g. chessdb's opening book).
        """
        self.version = version
        self.header = {
            "Authorization": f"Bearer {token}"
        }
        self.baseUrl = url
        self.session = requests.Session()
        self.session.headers.update(self.header)
        self.other_session = requests.Session()
        self.set_user_agent("?")
        self.logging_level = logging_level
        self.max_retries = max_retries
        self.rate_limit_timers: defaultdict[str, Timer] = defaultdict(Timer)

        # Confirm that the OAuth token has the proper permission to play on lichess
        token_info = self.api_post("token_test", data=token)[token]

        if not token_info:
            raise RuntimeError("Token in config file is not recognized by lichess. "
                               "Please check that it was copied correctly into your configuration file.")

        scopes = token_info["scopes"]
        if "bot:play" not in scopes.split(","):
            raise RuntimeError("Please use an API access token for your bot that "
                               'has the scope "Play games with the bot API (bot:play)". '
                               f"The current token has: {scopes}.")

    @backoff.on_exception(backoff.constant,
                          (RemoteDisconnected, ConnectionError, HTTPError, ReadTimeout),
                          max_time=60,
                          interval=0.1,
                          giveup=is_final,
                          on_backoff=backoff_handler,
                          backoff_log_level=logging.DEBUG,
                          giveup_log_level=logging.DEBUG)
    def api_get(self, endpoint_name: str, *template_args: str,
                params: Optional[dict[str, str]] = None,
                stream: bool = False, timeout: int = 2) -> requests.Response:
        """
        Send a GET to lichess.org.

        :param endpoint_name: The name of the endpoint.
        :param template_args: The values that go in the url (e.g. the challenge id if `endpoint_name` is `accept`).
        :param params: Parameters sent to lichess.org.
        :param stream: Whether the data returned from lichess.org should be streamed.
        :param timeout: The amount of time in seconds to wait for a response.
        :return: lichess.org's response.
        """
        logging.getLogger("backoff").setLevel(self.logging_level)
        path_template = self.get_path_template(endpoint_name)
        url = urljoin(self.baseUrl, path_template.format(*template_args))
        response = self.session.get(url, params=params, timeout=timeout, stream=stream)

        if is_new_rate_limit(response):
            delay = seconds(1 if endpoint_name == "move" else 60)
            self.set_rate_limit_delay(path_template, delay)

        response.raise_for_status()
        response.encoding = "utf-8"
        return response

    def api_get_json(self, endpoint_name: str, *template_args: str,
                     params: Optional[dict[str, str]] = None) -> JSON_REPLY_TYPE:
        """
        Send a GET to the lichess.org endpoints that return a JSON.

        :param endpoint_name: The name of the endpoint.
        :param template_args: The values that go in the url (e.g. the challenge id if `endpoint_name` is `accept`).
        :param params: Parameters sent to lichess.org.
        :return: lichess.org's response in a dict.
        """
        response = self.api_get(endpoint_name, *template_args, params=params)
        json_response: JSON_REPLY_TYPE = response.json()
        return json_response

    def api_get_list(self, endpoint_name: str, *template_args: str,
                     params: Optional[dict[str, str]] = None) -> list[JSON_REPLY_TYPE]:
        """
        Send a GET to the lichess.org endpoints that return a list containing JSON.

        :param endpoint_name: The name of the endpoint.
        :param template_args: The values that go in the url (e.g. the challenge id if `endpoint_name` is `accept`).
        :param params: Parameters sent to lichess.org.
        :return: lichess.org's response in a list of dicts.
        """
        response = self.api_get(endpoint_name, *template_args, params=params)
        json_response: list[JSON_REPLY_TYPE] = response.json()
        return json_response

    def api_get_raw(self, endpoint_name: str, *template_args: str,
                    params: Optional[dict[str, str]] = None, ) -> str:
        """
        Send a GET to lichess.org that returns plain text (UTF-8).

        :param endpoint_name: The name of the endpoint.
        :param template_args: The values that go in the url (e.g. the challenge id if `endpoint_name` is `accept`).
        :param params: Parameters sent to lichess.org.
        :return: The text of lichess.org's response.
        """
        response = self.api_get(endpoint_name, *template_args, params=params)
        return response.text

    @backoff.on_exception(backoff.constant,
                          (RemoteDisconnected, ConnectionError, HTTPError, ReadTimeout),
                          max_time=60,
                          interval=0.1,
                          giveup=is_final,
                          on_backoff=backoff_handler,
                          backoff_log_level=logging.DEBUG,
                          giveup_log_level=logging.DEBUG)
    def api_post(self,
                 endpoint_name: str,
                 *template_args: Any,
                 data: Union[str, dict[str, str], None] = None,
                 headers: Optional[dict[str, str]] = None,
                 params: Optional[dict[str, str]] = None,
                 payload: Optional[REQUESTS_PAYLOAD_TYPE] = None,
                 raise_for_status: bool = True) -> JSON_REPLY_TYPE:
        """
        Send a POST to lichess.org.

        :param endpoint_name: The name of the endpoint.
        :param template_args: The values that go in the url (e.g. the challenge id if `endpoint_name` is `accept`).
        :param data: Data sent to lichess.org.
        :param headers: The headers for the request.
        :param params: Parameters sent to lichess.org.
        :param payload: Payload sent to lichess.org.
        :param raise_for_status: Whether to raise an exception if the response contains an error code.
        :return: lichess.org's response in a dict.
        """
        logging.getLogger("backoff").setLevel(self.logging_level)
        path_template = self.get_path_template(endpoint_name)
        url = urljoin(self.baseUrl, path_template.format(*template_args))
        response = self.session.post(url, data=data, headers=headers, params=params, json=payload, timeout=2)

        if is_new_rate_limit(response):
            self.set_rate_limit_delay(path_template, seconds(60))

        if raise_for_status:
            response.raise_for_status()

        json_response: JSON_REPLY_TYPE = response.json()
        return json_response

    def get_path_template(self, endpoint_name: str) -> str:
        """
        Get the path template given the endpoint name. Will raise an exception if the path template is rate limited.

        :param endpoint_name: The name of the endpoint.
        :return: The path template.
        """
        path_template = ENDPOINTS[endpoint_name]
        if self.is_rate_limited(path_template):
            raise RateLimited(f"{path_template} is rate-limited. "
                              f"Will retry in {sec_str(self.rate_limit_time_left(path_template))} seconds.")
        return path_template

    def set_rate_limit_delay(self, path_template: str, delay_time: datetime.timedelta) -> None:
        """
        Set a delay to a path template if it was rate limited.

        :param path_template: The path template.
        :param delay_time: How long we won't call this endpoint.
        """
        logger.warning(f"Endpoint {path_template} is rate limited. Waiting {delay_time} seconds until next request.")
        self.rate_limit_timers[path_template] = Timer(delay_time)

    def is_rate_limited(self, path_template: str) -> bool:
        """Check if a path template is rate limited."""
        return not self.rate_limit_timers[path_template].is_expired()

    def rate_limit_time_left(self, path_template: str) -> datetime.timedelta:
        """How much time is left until we can use the path template normally."""
        return self.rate_limit_timers[path_template].time_until_expiration()

    def upgrade_to_bot_account(self) -> JSON_REPLY_TYPE:
        """Upgrade the account to a BOT account."""
        return self.api_post("upgrade")

    def make_move(self, game_id: str, move: chess.engine.PlayResult) -> JSON_REPLY_TYPE:
        """
        Make a move.

        :param game_id: The id of the game.
        :param move: The move to make.
        """
        return self.api_post("move", game_id, move.move,
                             params={"offeringDraw": str(move.draw_offered).lower()})

    def chat(self, game_id: str, room: str, text: str) -> JSON_REPLY_TYPE:
        """
        Send a message to the chat.

        :param game_id: The id of the game.
        :param room: The room (either chat or spectator room).
        :param text: The text to send.
        """
        if len(text) > MAX_CHAT_MESSAGE_LEN:
            logger.warning(f"This chat message is {len(text)} characters, which is longer "
                           f"than the maximum of {MAX_CHAT_MESSAGE_LEN}. It will not be sent.")
            logger.warning(f"Message: {text}")
            return {}

        payload = {"room": room, "text": text}
        return self.api_post("chat", game_id, data=payload)

    def abort(self, game_id: str) -> JSON_REPLY_TYPE:
        """Aborts a game."""
        return self.api_post("abort", game_id)

    def get_event_stream(self) -> requests.models.Response:
        """Get a stream of the events (e.g. challenge, gameStart)."""
        return self.api_get("stream_event", stream=True, timeout=15)

    def get_game_stream(self, game_id: str) -> requests.models.Response:
        """Get  stream of the in-game events (e.g. moves by the opponent)."""
        return self.api_get("stream", game_id, stream=True, timeout=15)

    def accept_challenge(self, challenge_id: str) -> JSON_REPLY_TYPE:
        """Accept a challenge."""
        return self.api_post("accept", challenge_id)

    def decline_challenge(self, challenge_id: str, reason: str = "generic") -> JSON_REPLY_TYPE:
        """Decline a challenge."""
        try:
            return self.api_post("decline", challenge_id,
                                 data=f"reason={reason}",
                                 headers={"Content-Type":
                                          "application/x-www-form-urlencoded"},
                                 raise_for_status=False)
        except Exception:
            return {}

    def get_profile(self) -> JSON_REPLY_TYPE:
        """Get the bot's profile (e.g. username)."""
        profile = self.api_get_json("profile")
        self.set_user_agent(profile["username"])
        return profile

    def get_ongoing_games(self) -> list[dict[str, Any]]:
        """Get the bot's ongoing games."""
        ongoing_games: list[dict[str, Any]] = []
        try:
            ongoing_games = self.api_get_json("playing")["nowPlaying"]
        except Exception:
            pass
        return ongoing_games

    def resign(self, game_id: str) -> None:
        """Resign a game."""
        self.api_post("resign", game_id)

    def set_user_agent(self, username: str) -> None:
        """Set the user agent for communication with lichess.org."""
        self.header.update({"User-Agent": f"lichess-bot/{self.version} user:{username}"})
        self.session.headers.update(self.header)

    def get_game_pgn(self, game_id: str) -> str:
        """Get the PGN (Portable Game Notation) record of a game."""
        try:
            return self.api_get_raw("export", game_id)
        except Exception:
            return ""

    def get_online_bots(self) -> list[dict[str, Any]]:
        """Get a list of bots that are online."""
        try:
            online_bots_str = self.api_get_raw("online_bots")
            online_bots = list(filter(bool, online_bots_str.split("\n")))
            return list(map(json.loads, online_bots))
        except Exception:
            return []

    def challenge(self, username: str, payload: REQUESTS_PAYLOAD_TYPE) -> JSON_REPLY_TYPE:
        """Create a challenge."""
        return self.api_post("challenge", username, payload=payload, raise_for_status=False)

    def cancel(self, challenge_id: str) -> JSON_REPLY_TYPE:
        """Cancel a challenge."""
        return self.api_post("cancel", challenge_id, raise_for_status=False)

    def online_book_get(self, path: str, params: Optional[dict[str, Any]] = None, stream: bool = False) -> JSON_REPLY_TYPE:
        """Get an external move from online sources (chessdb or lichess.org)."""
        @backoff.on_exception(backoff.constant,
                              (RemoteDisconnected, ConnectionError, HTTPError, ReadTimeout),
                              max_time=60,
                              max_tries=self.max_retries,
                              interval=0.1,
                              giveup=is_final,
                              on_backoff=backoff_handler,
                              backoff_log_level=logging.DEBUG,
                              giveup_log_level=logging.DEBUG)
        def online_book_get() -> JSON_REPLY_TYPE:
            json_response: JSON_REPLY_TYPE = self.other_session.get(path, timeout=2, params=params, stream=stream).json()
            return json_response
        return online_book_get()

    def is_online(self, user_id: str) -> bool:
        """Check if lichess.org thinks the bot is online or not."""
        user = self.api_get_list("status", params={"ids": user_id})
        return bool(user and user[0].get("online"))

    def get_public_data(self, user_name: str) -> JSON_REPLY_TYPE:
        """Get the public data of a bot."""
        return self.api_get_json("public_data", user_name)

lichess-bot/lichess.py
carlini/chess-llm
"""Challenge other bots."""
import random
import logging
import model
from timer import Timer, seconds, minutes, days
from collections import defaultdict
from collections.abc import Sequence
import lichess
import datetime
from config import Configuration, FilterType
from typing import Any, Optional
USER_PROFILE_TYPE = dict[str, Any]
EVENT_TYPE = dict[str, Any]
MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]
DAILY_TIMERS_TYPE = list[Timer]

logger = logging.getLogger(__name__)

daily_challenges_file_name = "daily_challenge_times.txt"
timestamp_format = "%Y-%m-%d %H:%M:%S\n"


def read_daily_challenges() -> DAILY_TIMERS_TYPE:
    """Read the challenges we have created in the past 24 hours from a text file."""
    timers: DAILY_TIMERS_TYPE = []
    try:
        with open(daily_challenges_file_name) as file:
            for line in file:
                timers.append(Timer(days(1), datetime.datetime.strptime(line, timestamp_format)))
    except FileNotFoundError:
        pass

    return [timer for timer in timers if not timer.is_expired()]


def write_daily_challenges(daily_challenges: DAILY_TIMERS_TYPE) -> None:
    """Write the challenges we have created in the past 24 hours to a text file."""
    with open(daily_challenges_file_name, "w") as file:
        for timer in daily_challenges:
            file.write(timer.starting_timestamp(timestamp_format))


class Matchmaking:
    """Challenge other bots."""

    def __init__(self, li: lichess.Lichess, config: Configuration, user_profile: USER_PROFILE_TYPE) -> None:
        """Initialize values needed for matchmaking."""
        self.li = li
        self.variants = list(filter(lambda variant: variant != "fromPosition", config.challenge.variants))
        self.matchmaking_cfg = config.matchmaking
        self.user_profile = user_profile
        self.last_challenge_created_delay = Timer(seconds(25))  # Challenges expire after 20 seconds.
        self.last_game_ended_delay = Timer(minutes(self.matchmaking_cfg.challenge_timeout))
        self.last_user_profile_update_time = Timer(minutes(5))
        self.min_wait_time = seconds(60)  # Wait before new challenge to avoid api rate limits.
        self.challenge_id: str = ""
        self.daily_challenges: DAILY_TIMERS_TYPE = read_daily_challenges()

        # (opponent name, game aspect) --> other bot is likely to accept challenge
        # game aspect is the one the challenged bot objects to and is one of:
        #   - game speed (bullet, blitz, etc.)
        #   - variant (standard, horde, etc.)
        #   - casual/rated
        #   - empty string (if no other reason is given or self.filter_type is COARSE)
        self.challenge_type_acceptable: defaultdict[tuple[str, str], bool] = defaultdict(lambda: True)
        self.challenge_filter = self.matchmaking_cfg.challenge_filter

        for name in self.matchmaking_cfg.block_list:
            self.add_to_block_list(name)

    def should_create_challenge(self) -> bool:
        """Whether we should create a challenge."""
        matchmaking_enabled = self.matchmaking_cfg.allow_matchmaking
        time_has_passed = self.last_game_ended_delay.is_expired()
        challenge_expired = self.last_challenge_created_delay.is_expired() and self.challenge_id
        min_wait_time_passed = self.last_challenge_created_delay.time_since_reset() > self.min_wait_time
        if challenge_expired:
            self.li.cancel(self.challenge_id)
            logger.info(f"Challenge id {self.challenge_id} cancelled.")
            self.challenge_id = ""
            self.show_earliest_challenge_time()
        return bool(matchmaking_enabled and (time_has_passed or challenge_expired) and min_wait_time_passed)

    def create_challenge(self, username: str, base_time: int, increment: int, days: int, variant: str,
                         mode: str) -> str:
        """Create a challenge."""
        params = {"rated": mode == "rated", "variant": variant}

        if days:
            params["days"] = days
        elif base_time or increment:
            params["clock.limit"] = base_time
            params["clock.increment"] = increment
        else:
            logger.error("At least one of challenge_days, challenge_initial_time, or challenge_increment "
                         "must be greater than zero in the matchmaking section of your config file.")
            return ""

        try:
            self.update_daily_challenge_record()
            self.last_challenge_created_delay.reset()
            response = self.li.challenge(username, params)
            challenge_id: str = response.get("challenge", {}).get("id", "")
            if not challenge_id:
                logger.error(response)
                self.add_to_block_list(username)
                self.show_earliest_challenge_time()
            return challenge_id
        except Exception as e:
            logger.warning("Could not create challenge")
            logger.debug(e, exc_info=e)
            self.show_earliest_challenge_time()
            return ""

    def update_daily_challenge_record(self) -> None:
        """
        Record timestamp of latest challenge and update minimum wait time.

        As the number of challenges in a day increase, the minimum wait time between challenges increases.
        0   -  49 challenges --> 1 minute
        50  -  99 challenges --> 2 minutes
        100 - 149 challenges --> 3 minutes
        etc.
        """
        self.daily_challenges = [timer for timer in self.daily_challenges if not timer.is_expired()]
        self.daily_challenges.append(Timer(days(1)))
        self.min_wait_time = seconds(60) * ((len(self.daily_challenges) // 50) + 1)
        write_daily_challenges(self.daily_challenges)

    def perf(self) -> dict[str, dict[str, Any]]:
        """Get the bot's rating in every variant. Bullet, blitz, rapid etc. are considered different variants."""
        user_perf: dict[str, dict[str, Any]] = self.user_profile["perfs"]
        return user_perf

    def username(self) -> str:
        """Our username."""
        username: str = self.user_profile["username"]
        return username

    def update_user_profile(self) -> None:
        """Update our user profile data, to get our latest rating."""
        if self.last_user_profile_update_time.is_expired():
            self.last_user_profile_update_time.reset()
            try:
                self.user_profile = self.li.get_profile()
            except Exception:
                pass

    def choose_opponent(self) -> tuple[Optional[str], int, int, int, str, str]:
        """Choose an opponent."""
        override_choice = random.choice(self.matchmaking_cfg.overrides.keys() + [None])
        logger.info(f"Using the {override_choice or 'default'} matchmaking configuration.")
        override = {} if override_choice is None else self.matchmaking_cfg.overrides.lookup(override_choice)
        match_config = self.matchmaking_cfg | override

        variant = self.get_random_config_value(match_config, "challenge_variant", self.variants)
        mode = self.get_random_config_value(match_config, "challenge_mode", ["casual", "rated"])

        base_time = random.choice(match_config.challenge_initial_time)
        increment = random.choice(match_config.challenge_increment)
        days = random.choice(match_config.challenge_days)

        play_correspondence = [bool(days), not bool(base_time or increment)]
        if random.choice(play_correspondence):
            base_time = 0
            increment = 0
        else:
            days = 0

        game_type = game_category(variant, base_time, increment, days)

        min_rating = match_config.opponent_min_rating
        max_rating = match_config.opponent_max_rating
        rating_diff = match_config.opponent_rating_difference
        bot_rating = self.perf().get(game_type, {}).get("rating", 0)
        if rating_diff is not None and bot_rating > 0:
            min_rating = bot_rating - rating_diff
            max_rating = bot_rating + rating_diff
        logger.info(f"Seeking {game_type} game with opponent rating in [{min_rating}, {max_rating}] ...")
        allow_tos_violation = match_config.opponent_allow_tos_violation

        def is_suitable_opponent(bot: USER_PROFILE_TYPE) -> bool:
            perf = bot.get("perfs", {}).get(game_type, {})
            return (bot["username"] != self.username()
                    and not self.in_block_list(bot["username"])
                    and not bot.get("disabled")
                    and (allow_tos_violation or not bot.get("tosViolation"))  # Terms of Service violation.
                    and perf.get("games", 0) > 0
                    and min_rating <= perf.get("rating", 0) <= max_rating)

        online_bots = self.li.get_online_bots()
        online_bots = list(filter(is_suitable_opponent, online_bots))

        def ready_for_challenge(bot: USER_PROFILE_TYPE) -> bool:
            aspects = [variant, game_type, mode] if self.challenge_filter == FilterType.FINE else []
            return all(self.should_accept_challenge(bot["username"], aspect) for aspect in aspects)

        ready_bots = list(filter(ready_for_challenge, online_bots))
        online_bots = ready_bots or online_bots
        bot_username = None

        try:
            bot = random.choice(online_bots)
            bot_profile = self.li.get_public_data(bot["username"])
            if bot_profile.get("blocking"):
                self.add_to_block_list(bot["username"])
            else:
                bot_username = bot["username"]
        except Exception:
            if online_bots:
                logger.exception("Error:")
            else:
                logger.error("No suitable bots found to challenge.")

        return bot_username, base_time, increment, days, variant, mode

    def get_random_config_value(self, config: Configuration, parameter: str, choices: list[str]) -> str:
        """Choose a random value from `choices` if the parameter value in the config is `random`."""
        value: str = config.lookup(parameter)
        return value if value != "random" else random.choice(choices)

    def challenge(self, active_games: set[str], challenge_queue: MULTIPROCESSING_LIST_TYPE) -> None:
        """
        Challenge an opponent.

        :param active_games: The games that the bot is playing.
        :param challenge_queue: The queue containing the challenges.
        """
        if active_games or challenge_queue or not self.should_create_challenge():
            return

        logger.info("Challenging a random bot")
        self.update_user_profile()
        bot_username, base_time, increment, days, variant, mode = self.choose_opponent()
        logger.info(f"Will challenge {bot_username} for a {variant} game.")
        challenge_id = self.create_challenge(bot_username, base_time, increment, days, variant, mode) if bot_username else ""
        logger.info(f"Challenge id is {challenge_id if challenge_id else 'None'}.")
        self.challenge_id = challenge_id

    def game_done(self) -> None:
        """Reset the timer for when the last game ended, and prints the earliest that the next challenge will be created."""
        self.last_game_ended_delay.reset()
        self.show_earliest_challenge_time()

    def show_earliest_challenge_time(self) -> None:
        """Show the earliest that the next challenge will be created."""
        if self.matchmaking_cfg.allow_matchmaking:
            postgame_timeout = self.last_game_ended_delay.time_until_expiration()
            time_to_next_challenge = self.min_wait_time - self.last_challenge_created_delay.time_since_reset()
            time_left = max(postgame_timeout, time_to_next_challenge)
            earliest_challenge_time = datetime.datetime.now() + time_left
            challenges = "challenge" + ("" if len(self.daily_challenges) == 1 else "s")
            logger.info(f"Next challenge will be created after {earliest_challenge_time.strftime('%X')} "
                        f"({len(self.daily_challenges)} {challenges} in last 24 hours)")

    def add_to_block_list(self, username: str) -> None:
        """Add a bot to the blocklist."""
        self.add_challenge_filter(username, "")

    def in_block_list(self, username: str) -> bool:
        """Check if an opponent is in the block list to prevent future challenges."""
        return not self.should_accept_challenge(username, "")

    def add_challenge_filter(self, username: str, game_aspect: str) -> None:
        """
        Prevent creating another challenge when an opponent has decline a challenge.

        :param username: The name of the opponent.
        :param game_aspect: The aspect of a game (time control, chess variant, etc.)
        that caused the opponent to decline a challenge. If the parameter is empty,
        that is equivalent to adding the opponent to the block list.
        """
        self.challenge_type_acceptable[(username, game_aspect)] = False

    def should_accept_challenge(self, username: str, game_aspect: str) -> bool:
        """
        Whether a bot is likely to accept a challenge to a game.

        :param username: The name of the opponent.
        :param game_aspect: A category of the challenge type (time control, chess variant, etc.) to test for acceptance.
        If game_aspect is empty, this is equivalent to checking if the opponent is in the block list.
        """
        return self.challenge_type_acceptable[(username, game_aspect)]

    def accepted_challenge(self, event: EVENT_TYPE) -> None:
        """
        Set the challenge id to an empty string, if the challenge was accepted.

        Otherwise, we would attempt to cancel the challenge later.
        """
        if self.challenge_id == event["game"]["id"]:
            self.challenge_id = ""

    def declined_challenge(self, event: EVENT_TYPE) -> None:
        """
        Handle a challenge that was declined by the opponent.

        Depends on whether `FilterType` is `NONE`, `COARSE`, or `FINE`.
        """
        challenge = model.Challenge(event["challenge"], self.user_profile)
        opponent = challenge.opponent
        reason = event["challenge"]["declineReason"]
        logger.info(f"{opponent} declined {challenge}: {reason}")
        if self.challenge_id == challenge.id:
            self.challenge_id = ""
        if not challenge.from_self or self.challenge_filter == FilterType.NONE:
            return

        mode = "rated" if challenge.rated else "casual"
        decline_details: dict[str, str] = {"generic": "",
                                           "later": "",
                                           "nobot": "",
                                           "toofast": challenge.speed,
                                           "tooslow": challenge.speed,
                                           "timecontrol": challenge.speed,
                                           "rated": mode,
                                           "casual": mode,
                                           "standard": challenge.variant,
                                           "variant": challenge.variant}

        reason_key = event["challenge"]["declineReasonKey"].lower()
        if reason_key not in decline_details:
            logger.warning(f"Unknown decline reason received: {reason_key}")
        game_problem = decline_details.get(reason_key, "") if self.challenge_filter == FilterType.FINE else ""
        self.add_challenge_filter(opponent.name, game_problem)
        logger.info(f"Will not challenge {opponent} to another {game_problem}".strip() + " game.")

        self.show_earliest_challenge_time()


def game_category(variant: str, base_time: int, increment: int, days: int) -> str:
    """
    Get the game type (e.g. bullet, atomic, classical). Lichess has one rating for every variant regardless of time control.

    :param variant: The game's variant.
    :param base_time: The base time in seconds.
    :param increment: The increment in seconds.
    :param days: If the game is correspondence, we have some days to play the move.
    :return: The game category.
    """
    game_duration = base_time + increment * 40
    if variant != "standard":
        return variant
    elif days:
        return "correspondence"
    elif game_duration < 179:
        return "bullet"
    elif game_duration < 479:
        return "blitz"
    elif game_duration < 1499:
        return "rapid"
    else:
        return "classical"

lichess-bot/matchmaking.py
carlini/chess-llm
"""Store information about a challenge, game or player in a class."""
import math
from urllib.parse import urljoin
import logging
import datetime
from enum import Enum
from timer import Timer, msec, seconds, sec_str, to_msec, to_seconds, years
from config import Configuration
from typing import Any
from collections import defaultdict

logger = logging.getLogger(__name__)


class Challenge:
    """Store information about a challenge."""

    def __init__(self, challenge_info: dict[str, Any], user_profile: dict[str, Any]) -> None:
        """:param user_profile: Information about our bot."""
        self.id = challenge_info["id"]
        self.rated = challenge_info["rated"]
        self.variant = challenge_info["variant"]["key"]
        self.perf_name = challenge_info["perf"]["name"]
        self.speed = challenge_info["speed"]
        self.increment: int = challenge_info.get("timeControl", {}).get("increment")
        self.base: int = challenge_info.get("timeControl", {}).get("limit")
        self.days: int = challenge_info.get("timeControl", {}).get("daysPerTurn")
        self.challenger = Player(challenge_info.get("challenger") or {})
        self.opponent = Player(challenge_info.get("destUser") or {})
        self.from_self = self.challenger.name == user_profile["username"]

    def is_supported_variant(self, challenge_cfg: Configuration) -> bool:
        """Check whether the variant is supported."""
        return self.variant in challenge_cfg.variants

    def is_supported_time_control(self, challenge_cfg: Configuration) -> bool:
        """Check whether the time control is supported."""
        speeds = challenge_cfg.time_controls
        increment_max: int = challenge_cfg.max_increment
        increment_min: int = challenge_cfg.min_increment
        base_max: int = challenge_cfg.max_base
        base_min: int = challenge_cfg.min_base
        days_max: int = challenge_cfg.max_days
        days_min: int = challenge_cfg.min_days

        if self.speed not in speeds:
            return False

        require_non_zero_increment = (self.challenger.is_bot
                                      and self.speed == "bullet"
                                      and challenge_cfg.bullet_requires_increment)
        increment_min = max(increment_min, 1 if require_non_zero_increment else 0)

        if self.base is not None and self.increment is not None:
            # Normal clock game
            return (increment_min <= self.increment <= increment_max
                    and base_min <= self.base <= base_max)
        elif self.days is not None:
            # Correspondence game
            return days_min <= self.days <= days_max
        else:
            # Unlimited game
            return days_max == math.inf

    def is_supported_mode(self, challenge_cfg: Configuration) -> bool:
        """Check whether the mode is supported."""
        return ("rated" if self.rated else "casual") in challenge_cfg.modes

    def is_supported_recent(self, config: Configuration, recent_bot_challenges: defaultdict[str, list[Timer]]) -> bool:
        """Check whether we have played a lot of games with this opponent recently. Only used when the oppoennt is a BOT."""
        # Filter out old challenges
        recent_bot_challenges[self.challenger.name] = [timer for timer
                                                       in recent_bot_challenges[self.challenger.name]
                                                       if not timer.is_expired()]
        max_recent_challenges = config.max_recent_bot_challenges
        return (not self.challenger.is_bot
                or max_recent_challenges is None
                or len(recent_bot_challenges[self.challenger.name]) < max_recent_challenges)

    def decline_due_to(self, requirement_met: bool, decline_reason: str) -> str:
        """
        Get the reason lichess-bot declined an incoming challenge.

        :param requirement_met: Whether a requirement is met.
        :param decline_reason: The reason we declined the challenge if the requirement wasn't met.
        :return: `decline_reason` if `requirement_met` is false else returns an empty string.
        """
        return "" if requirement_met else decline_reason

    def is_supported(self, config: Configuration,
                     recent_bot_challenges: defaultdict[str, list[Timer]]) -> tuple[bool, str]:
        """Whether the challenge is supported."""
        try:
            if self.from_self:
                return True, ""

            allowed_opponents: list[str] = list(filter(None, config.allow_list)) or [self.challenger.name]
            decline_reason = (self.decline_due_to(config.accept_bot or not self.challenger.is_bot, "noBot")
                              or self.decline_due_to(not config.only_bot or self.challenger.is_bot, "onlyBot")
                              or self.decline_due_to(self.is_supported_time_control(config), "timeControl")
                              or self.decline_due_to(self.is_supported_variant(config), "variant")
                              or self.decline_due_to(self.is_supported_mode(config), "casual" if self.rated else "rated")
                              or self.decline_due_to(self.challenger.name not in config.block_list, "generic")
                              or self.decline_due_to(self.challenger.name in allowed_opponents, "generic")
                              or self.decline_due_to(self.is_supported_recent(config, recent_bot_challenges), "later"))

            return not decline_reason, decline_reason

        except Exception:
            logger.exception(f"Error while checking challenge {self.id}:")
            return False, "generic"

    def score(self) -> int:
        """Give a rating estimate to the opponent."""
        rated_bonus = 200 if self.rated else 0
        challenger_master_title = self.challenger.title if not self.challenger.is_bot else None
        titled_bonus = 200 if challenger_master_title else 0
        challenger_rating_int = self.challenger.rating or 0
        return challenger_rating_int + rated_bonus + titled_bonus

    def mode(self) -> str:
        """Get the mode of the challenge (rated or casual)."""
        return "rated" if self.rated else "casual"

    def __str__(self) -> str:
        """Get a string representation of `Challenge`."""
        return f"{self.perf_name} {self.mode()} challenge from {self.challenger} ({self.id})"

    def __repr__(self) -> str:
        """Get a string representation of `Challenge`."""
        return self.__str__()


class Termination(str, Enum):
    """The possible game terminations."""

    MATE = "mate"
    TIMEOUT = "outoftime"
    RESIGN = "resign"
    ABORT = "aborted"
    DRAW = "draw"


class Game:
    """Store information about a game."""

    def __init__(self, game_info: dict[str, Any], username: str, base_url: str, abort_time: datetime.timedelta) -> None:
        """:param abort_time: How long to wait before aborting the game."""
        self.username = username
        self.id: str = game_info["id"]
        self.speed = game_info.get("speed")
        clock = game_info.get("clock") or {}
        ten_years_in_ms = to_msec(years(10))
        self.clock_initial = msec(clock.get("initial", ten_years_in_ms))
        self.clock_increment = msec(clock.get("increment", 0))
        self.perf_name = (game_info.get("perf") or {}).get("name", "{perf?}")
        self.variant_name = game_info["variant"]["name"]
        self.mode = "rated" if game_info.get("rated") else "casual"
        self.white = Player(game_info["white"])
        self.black = Player(game_info["black"])
        self.initial_fen = game_info.get("initialFen")
        self.state: dict[str, Any] = game_info["state"]
        self.is_white = (self.white.name or "").lower() == username.lower()
        self.my_color = "white" if self.is_white else "black"
        self.opponent_color = "black" if self.is_white else "white"
        self.me = self.white if self.is_white else self.black
        self.opponent = self.black if self.is_white else self.white
        self.base_url = base_url
        self.game_start = datetime.datetime.fromtimestamp(to_seconds(msec(game_info["createdAt"])),
                                                          tz=datetime.timezone.utc)
        self.abort_time = Timer(abort_time)
        self.terminate_time = Timer(self.clock_initial + self.clock_increment + abort_time + seconds(60))
        self.disconnect_time = Timer(seconds(0))

    def url(self) -> str:
        """Get the url of the game."""
        return f"{self.short_url()}/{self.my_color}"

    def short_url(self) -> str:
        """Get the short url of the game."""
        return urljoin(self.base_url, self.id)

    def pgn_event(self) -> str:
        """Get the event to write in the PGN file."""
        if self.variant_name in ["Standard", "From Position"]:
            return f"{self.mode.title()} {self.perf_name.title()} game"
        else:
            return f"{self.mode.title()} {self.variant_name} game"

    def time_control(self) -> str:
        """Get the time control of the game."""
        return f"{sec_str(self.clock_initial)}+{sec_str(self.clock_increment)}"

    def is_abortable(self) -> bool:
        """Whether the game can be aborted."""
        # Moves are separated by spaces. A game is abortable when less
        # than two moves (one from each player) have been played.
        return " " not in self.state["moves"]

    def ping(self, abort_in: datetime.timedelta, terminate_in: datetime.timedelta, disconnect_in: datetime.timedelta) -> None:
        """
        Tell the bot when to abort, terminate, and disconnect from a game.

        :param abort_in: How many seconds to wait before aborting.
        :param terminate_in: How many seconds to wait before terminating.
        :param disconnect_in: How many seconds to wait before disconnecting.
        """
        if self.is_abortable():
            self.abort_time = Timer(abort_in)
        self.terminate_time = Timer(terminate_in)
        self.disconnect_time = Timer(disconnect_in)

    def should_abort_now(self) -> bool:
        """Whether we should abort the game."""
        return self.is_abortable() and self.abort_time.is_expired()

    def should_terminate_now(self) -> bool:
        """Whether we should terminate the game."""
        return self.terminate_time.is_expired()

    def should_disconnect_now(self) -> bool:
        """Whether we should disconnect form the game."""
        return self.disconnect_time.is_expired()

    def my_remaining_time(self) -> datetime.timedelta:
        """How many seconds we have left."""
        wtime = msec(self.state["wtime"])
        btime = msec(self.state["btime"])
        return wtime if self.is_white else btime

    def result(self) -> str:
        """Get the result of the game."""
        class GameEnding(str, Enum):
            WHITE_WINS = "1-0"
            BLACK_WINS = "0-1"
            DRAW = "1/2-1/2"
            INCOMPLETE = "*"

        winner = self.state.get("winner")
        termination = self.state.get("status")

        if winner == "white":
            result = GameEnding.WHITE_WINS
        elif winner == "black":
            result = GameEnding.BLACK_WINS
        elif termination in [Termination.DRAW, Termination.TIMEOUT]:
            result = GameEnding.DRAW
        else:
            result = GameEnding.INCOMPLETE

        return result.value

    def __str__(self) -> str:
        """Get a string representation of `Game`."""
        return f"{self.url()} {self.perf_name} vs {self.opponent} ({self.id})"

    def __repr__(self) -> str:
        """Get a string representation of `Game`."""
        return self.__str__()


class Player:
    """Store information about a player."""

    def __init__(self, player_info: dict[str, Any]) -> None:
        """:param player_info: Contains information about a player."""
        self.title = player_info.get("title")
        self.rating = player_info.get("rating")
        self.provisional = player_info.get("provisional")
        self.aiLevel = player_info.get("aiLevel")
        self.is_bot = self.title == "BOT" or self.aiLevel is not None
        self.name: str = f"AI level {self.aiLevel}" if self.aiLevel else player_info.get("name", "")

    def __str__(self) -> str:
        """Get a string representation of `Player`."""
        if self.aiLevel:
            return self.name
        else:
            rating = f'{self.rating}{"?" if self.provisional else ""}'
            return f'{self.title or ""} {self.name} ({rating})'.strip()

    def __repr__(self) -> str:
        """Get a string representation of `Player`."""
        return self.__str__()

lichess-bot/model.py
carlini/chess-llm
"""
Some example strategies for people who want to create a custom, homemade bot.

With these classes, bot makers will not have to implement the UCI or XBoard interfaces themselves.
"""

from __future__ import annotations
import chess
from chess.engine import PlayResult
import random
from engine_wrapper import MinimalEngine
from typing import Any, Union
import logging
MOVE = Union[chess.engine.PlayResult, list[chess.Move]]


# Use this logger variable to print messages to the console or log files.
# logger.info("message") will always print "message" to the console or log file.
# logger.debug("message") will only print "message" if verbose logging is enabled.
logger = logging.getLogger(__name__)


class ExampleEngine(MinimalEngine):
    """An example engine that all homemade engines inherit."""

    pass


# Strategy names and ideas from tom7's excellent eloWorld video

class RandomMove(ExampleEngine):
    """Get a random move."""

    def search(self, board: chess.Board, *args: Any) -> PlayResult:
        """Choose a random move."""
        return PlayResult(random.choice(list(board.legal_moves)), None)


class Alphabetical(ExampleEngine):
    """Get the first move when sorted by san representation."""

    def search(self, board: chess.Board, *args: Any) -> PlayResult:
        """Choose the first move alphabetically."""
        moves = list(board.legal_moves)
        moves.sort(key=board.san)
        return PlayResult(moves[0], None)


class FirstMove(ExampleEngine):
    """Get the first move when sorted by uci representation."""

    def search(self, board: chess.Board, *args: Any) -> PlayResult:
        """Choose the first move alphabetically in uci representation."""
        moves = list(board.legal_moves)
        moves.sort(key=str)
        return PlayResult(moves[0], None)


class ComboEngine(ExampleEngine):
    """
    Get a move using multiple different methods.

    This engine demonstrates how one can use `time_limit`, `draw_offered`, and `root_moves`.
    """

    def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool,
               root_moves: MOVE) -> chess.engine.PlayResult:
        """
        Choose a move using multiple different methods.

        :param board: The current position.
        :param time_limit: Conditions for how long the engine can search (e.g. we have 10 seconds and search up to depth 10).
        :param ponder: Whether the engine can ponder after playing a move.
        :param draw_offered: Whether the bot was offered a draw.
        :param root_moves: If it is a list, the engine should only play a move that is in `root_moves`.
        :return: The move to play.
        """
        if isinstance(time_limit.time, int):
            my_time = time_limit.time
            my_inc = 0
        elif board.turn == chess.WHITE:
            my_time = time_limit.white_clock if isinstance(time_limit.white_clock, int) else 0
            my_inc = time_limit.white_inc if isinstance(time_limit.white_inc, int) else 0
        else:
            my_time = time_limit.black_clock if isinstance(time_limit.black_clock, int) else 0
            my_inc = time_limit.black_inc if isinstance(time_limit.black_inc, int) else 0

        possible_moves = root_moves if isinstance(root_moves, list) else list(board.legal_moves)

        if my_time / 60 + my_inc > 10:
            # Choose a random move.
            move = random.choice(possible_moves)
        else:
            # Choose the first move alphabetically in uci representation.
            possible_moves.sort(key=str)
            move = possible_moves[0]
        return PlayResult(move, None, draw_offered=draw_offered)

class LLM(ExampleEngine):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        import sys
        import json
        sys.path.append("..")
        from chessllm import ChessLLM
        api_key = open("../OPENAI_API_KEY").read().strip()
        config = json.loads(open("../config.json").read())
        self.my_engine = ChessLLM(api_key, config)

    def search(self, board: chess.Board, *args: Any) -> PlayResult:
        conversation = args[-1]

        new_board = board.copy()
        move = self.my_engine.get_best_move(board, conversation=conversation)
        new_board.push_san(move)

        move = new_board.peek()
        
        return PlayResult(move, None)

        


lichess-bot/strategies.py
carlini/chess-llm
"""pytest won't search `test_bot/` if there is no `__init__.py` file."""

lichess-bot/test_bot/__init__.py
carlini/chess-llm
"""Remove files created when testing lichess-bot."""
import shutil
import os
from typing import Any


def pytest_sessionfinish(session: Any, exitstatus: Any) -> None:
    """Remove files created when testing lichess-bot."""
    shutil.copyfile("correct_lichess.py", "lichess.py")
    os.remove("correct_lichess.py")
    if os.path.exists("TEMP") and not os.getenv("GITHUB_ACTIONS"):
        shutil.rmtree("TEMP")
    if os.path.exists("logs"):
        shutil.rmtree("logs")

lichess-bot/test_bot/conftest.py
carlini/chess-llm
"""Imitate `lichess.py`. Used in tests."""
import time
import chess
import chess.engine
import json
import logging
import traceback
from timer import seconds, to_msec
from typing import Union, Any, Optional, Generator

logger = logging.getLogger(__name__)


def backoff_handler(details: Any) -> None:
    """Log exceptions inside functions with the backoff decorator."""
    logger.debug("Backing off {wait:0.1f} seconds after {tries} tries "
                 "calling function {target} with args {args} and kwargs {kwargs}".format(**details))
    logger.debug(f"Exception: {traceback.format_exc()}")


class GameStream:
    """Imitate lichess.org's GameStream. Used in tests."""

    def __init__(self) -> None:
        """Initialize `self.moves_sent` to an empty string. It stores the moves that we have already sent."""
        self.moves_sent = ""

    def iter_lines(self) -> Generator[bytes, None, None]:
        """Send the game events to lichess-bot."""
        yield json.dumps(
            {"id": "zzzzzzzz",
             "variant": {"key": "standard",
                         "name": "Standard",
                         "short": "Std"},
             "clock": {"initial": 60000,
                       "increment": 2000},
             "speed": "bullet",
             "perf": {"name": "Bullet"},
             "rated": True,
             "createdAt": 1600000000000,
             "white": {"id": "bo",
                       "name": "bo",
                       "title": "BOT",
                       "rating": 3000},
             "black": {"id": "b",
                       "name": "b",
                       "title": "BOT",
                       "rating": 3000,
                       "provisional": True},
             "initialFen": "startpos",
             "type": "gameFull",
             "state": {"type": "gameState",
                       "moves": "",
                       "wtime": 10000,
                       "btime": 10000,
                       "winc": 100,
                       "binc": 100,
                       "status": "started"}}).encode("utf-8")
        time.sleep(1)
        while True:
            time.sleep(0.001)
            with open("./logs/events.txt") as events:
                event = events.read()
            while True:
                try:
                    with open("./logs/states.txt") as states:
                        state = states.read().split("\n")
                    moves = state[0]
                    board = chess.Board()
                    for move in moves.split():
                        board.push_uci(move)
                    wtime, btime = [seconds(float(n)) for n in state[1].split(",")]
                    if len(moves) <= len(self.moves_sent) and not event:
                        time.sleep(0.001)
                        continue
                    self.moves_sent = moves
                    break
                except (IndexError, ValueError):
                    pass
            time.sleep(0.1)
            new_game_state = {"type": "gameState",
                              "moves": moves,
                              "wtime": int(to_msec(wtime)),
                              "btime": int(to_msec(btime)),
                              "winc": 100,
                              "binc": 100}
            if event == "end":
                new_game_state["status"] = "outoftime"
                new_game_state["winner"] = "black"
                yield json.dumps(new_game_state).encode("utf-8")
                break
            if moves:
                new_game_state["status"] = "started"
                yield json.dumps(new_game_state).encode("utf-8")


class EventStream:
    """Imitate lichess.org's EventStream. Used in tests."""

    def __init__(self, sent_game: bool = False) -> None:
        """:param sent_game: If we have already sent the `gameStart` event, so we don't send it again."""
        self.sent_game = sent_game

    def iter_lines(self) -> Generator[bytes, None, None]:
        """Send the events to lichess-bot."""
        if self.sent_game:
            yield b''
            time.sleep(1)
        else:
            yield json.dumps(
                {"type": "gameStart",
                 "game": {"id": "zzzzzzzz",
                          "source": "friend",
                          "compat": {"bot": True,
                                     "board": True}}}).encode("utf-8")


# Docs: https://lichess.org/api.
class Lichess:
    """Imitate communication with lichess.org."""

    def __init__(self, token: str, url: str, version: str) -> None:
        """Has the same parameters as `lichess.Lichess` to be able to be used in its placed without any modification."""
        self.baseUrl = url
        self.game_accepted = False
        self.moves: list[chess.engine.PlayResult] = []
        self.sent_game = False

    def upgrade_to_bot_account(self) -> None:
        """Isn't used in tests."""
        return

    def make_move(self, game_id: str, move: chess.engine.PlayResult) -> None:
        """Write a move to `./logs/states.txt`, to be read by the opponent."""
        self.moves.append(move)
        uci_move = move.move.uci() if move.move else "error"
        with open("./logs/states.txt") as file:
            contents = file.read().split("\n")
        contents[0] += f" {uci_move}"
        with open("./logs/states.txt", "w") as file:
            file.write("\n".join(contents))

    def chat(self, game_id: str, room: str, text: str) -> None:
        """Isn't used in tests."""
        return

    def abort(self, game_id: str) -> None:
        """Isn't used in tests."""
        return

    def get_event_stream(self) -> EventStream:
        """Send the `EventStream`."""
        events = EventStream(self.sent_game)
        self.sent_game = True
        return events

    def get_game_stream(self, game_id: str) -> GameStream:
        """Send the `GameStream`."""
        return GameStream()

    def accept_challenge(self, challenge_id: str) -> None:
        """Set `self.game_accepted` to true."""
        self.game_accepted = True

    def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None:
        """Isn't used in tests."""
        return

    def get_profile(self) -> dict[str, Union[str, bool, dict[str, str]]]:
        """Return a simple profile for the bot that lichess-bot uses when testing."""
        return {"id": "b",
                "username": "b",
                "online": True,
                "title": "BOT",
                "url": "https://lichess.org/@/b",
                "followable": True,
                "following": False,
                "blocking": False,
                "followsYou": False,
                "perfs": {}}

    def get_ongoing_games(self) -> list[str]:
        """Return that the bot isn't playing a game."""
        return []

    def resign(self, game_id: str) -> None:
        """Isn't used in tests."""
        return

    def get_game_pgn(self, game_id: str) -> str:
        """Return a simple PGN."""
        return """
[Event "Test game"]
[Site "pytest"]
[Date "2022.03.11"]
[Round "1"]
[White "Engine"]
[Black "Engine"]
[Result "0-1"]

*
"""

    def get_online_bots(self) -> list[dict[str, Union[str, bool]]]:
        """Return that the only bot online is us."""
        return [{"username": "b", "online": True}]

    def challenge(self, username: str, params: dict[str, str]) -> None:
        """Isn't used in tests."""
        return

    def cancel(self, challenge_id: str) -> None:
        """Isn't used in tests."""
        return

    def online_book_get(self, path: str, params: Optional[dict[str, str]] = None) -> None:
        """Isn't used in tests."""
        return

    def is_online(self, user_id: str) -> bool:
        """Return that a bot is online."""
        return True

lichess-bot/test_bot/lichess.py
carlini/chess-llm
"""Test lichess-bot."""
import pytest
import zipfile
import requests
import time
import yaml
import chess
import chess.engine
import threading
import os
import sys
import stat
import shutil
import importlib
import config
from timer import Timer, to_seconds, seconds
from typing import Any
if __name__ == "__main__":
    sys.exit(f"The script {os.path.basename(__file__)} should only be run by pytest.")
shutil.copyfile("lichess.py", "correct_lichess.py")
shutil.copyfile("test_bot/lichess.py", "lichess.py")
lichess_bot = importlib.import_module("lichess-bot")

platform = sys.platform
file_extension = ".exe" if platform == "win32" else ""
stockfish_path = f"./TEMP/sf{file_extension}"


def download_sf() -> None:
    """Download Stockfish 15."""
    if os.path.exists(stockfish_path):
        return
    windows_or_linux = "win" if platform == "win32" else "linux"
    base_name = f"stockfish_15_{windows_or_linux}_x64"
    exec_name = "stockfish_15_x64"
    zip_link = f"https://files.stockfishchess.org/files/{base_name}.zip"
    response = requests.get(zip_link, allow_redirects=True)
    with open("./TEMP/sf_zip.zip", "wb") as file:
        file.write(response.content)
    with zipfile.ZipFile("./TEMP/sf_zip.zip", "r") as zip_ref:
        zip_ref.extractall("./TEMP/")
    shutil.copyfile(f"./TEMP/{base_name}/{exec_name}{file_extension}", stockfish_path)
    if windows_or_linux == "linux":
        st = os.stat(stockfish_path)
        os.chmod(stockfish_path, st.st_mode | stat.S_IEXEC)


def download_lc0() -> None:
    """Download Leela Chess Zero 0.29.0."""
    if os.path.exists("./TEMP/lc0.exe"):
        return
    response = requests.get("https://github.com/LeelaChessZero/lc0/releases/download/v0.29.0/lc0-v0.29.0-windows-cpu-dnnl.zip",
                            allow_redirects=True)
    with open("./TEMP/lc0_zip.zip", "wb") as file:
        file.write(response.content)
    with zipfile.ZipFile("./TEMP/lc0_zip.zip", "r") as zip_ref:
        zip_ref.extractall("./TEMP/")


def download_sjeng() -> None:
    """Download Sjeng."""
    if os.path.exists("./TEMP/sjeng.exe"):
        return
    response = requests.get("https://sjeng.org/ftp/Sjeng112.zip", allow_redirects=True)
    with open("./TEMP/sjeng_zip.zip", "wb") as file:
        file.write(response.content)
    with zipfile.ZipFile("./TEMP/sjeng_zip.zip", "r") as zip_ref:
        zip_ref.extractall("./TEMP/")
    shutil.copyfile("./TEMP/Release/Sjeng112.exe", "./TEMP/sjeng.exe")


if not os.path.exists("TEMP"):
    os.mkdir("TEMP")
download_sf()
if platform == "win32":
    download_lc0()
    download_sjeng()
logging_level = lichess_bot.logging.DEBUG
lichess_bot.logging_configurer(logging_level, None, None, False)
lichess_bot.logger.info("Downloaded engines")


def thread_for_test() -> None:
    """Play the moves for the opponent of lichess-bot."""
    open("./logs/events.txt", "w").close()
    open("./logs/states.txt", "w").close()
    open("./logs/result.txt", "w").close()

    start_time = seconds(10)
    increment = seconds(0.1)

    board = chess.Board()
    wtime = start_time
    btime = start_time

    with open("./logs/states.txt", "w") as file:
        file.write(f"\n{to_seconds(wtime)},{to_seconds(btime)}")

    engine = chess.engine.SimpleEngine.popen_uci(stockfish_path)
    engine.configure({"Skill Level": 0, "Move Overhead": 1000, "Use NNUE": False})

    while not board.is_game_over():
        if len(board.move_stack) % 2 == 0:
            if not board.move_stack:
                move = engine.play(board,
                                   chess.engine.Limit(time=1),
                                   ponder=False)
            else:
                move_timer = Timer()
                move = engine.play(board,
                                   chess.engine.Limit(white_clock=to_seconds(wtime) - 2,
                                                      white_inc=to_seconds(increment)),
                                   ponder=False)
                wtime -= move_timer.time_since_reset()
                wtime += increment
            engine_move = move.move
            if engine_move is None:
                raise RuntimeError("Engine attempted to make null move.")
            board.push(engine_move)

            uci_move = engine_move.uci()
            with open("./logs/states.txt") as states:
                state_str = states.read()
            state = state_str.split("\n")
            state[0] += f" {uci_move}"
            state_str = "\n".join(state)
            with open("./logs/states.txt", "w") as file:
                file.write(state_str)

        else:  # lichess-bot move.
            move_timer = Timer()
            state2 = state_str
            moves_are_correct = False
            while state2 == state_str or not moves_are_correct:
                with open("./logs/states.txt") as states:
                    state2 = states.read()
                time.sleep(0.001)
                moves = state2.split("\n")[0]
                temp_board = chess.Board()
                moves_are_correct = True
                for move_str in moves.split():
                    try:
                        temp_board.push_uci(move_str)
                    except ValueError:
                        moves_are_correct = False
            with open("./logs/states.txt") as states:
                state2 = states.read()
            if len(board.move_stack) > 1:
                btime -= move_timer.time_since_reset()
                btime += increment
            move_str = state2.split("\n")[0].split(" ")[-1]
            board.push_uci(move_str)

        time.sleep(0.001)
        with open("./logs/states.txt") as states:
            state_str = states.read()
        state = state_str.split("\n")
        state[1] = f"{to_seconds(wtime)},{to_seconds(btime)}"
        state_str = "\n".join(state)
        with open("./logs/states.txt", "w") as file:
            file.write(state_str)

    with open("./logs/events.txt", "w") as file:
        file.write("end")
    engine.quit()
    outcome = board.outcome()
    win = outcome.winner == chess.BLACK if outcome else False
    with open("./logs/result.txt", "w") as file:
        file.write("1" if win else "0")


def run_bot(raw_config: dict[str, Any], logging_level: int) -> str:
    """Start lichess-bot."""
    config.insert_default_values(raw_config)
    CONFIG = config.Configuration(raw_config)
    lichess_bot.logger.info(lichess_bot.intro())
    li = lichess_bot.lichess.Lichess(CONFIG.token, CONFIG.url, lichess_bot.__version__)

    user_profile = li.get_profile()
    username = user_profile["username"]
    if user_profile.get("title") != "BOT":
        return "0"
    lichess_bot.logger.info(f"Welcome {username}!")
    lichess_bot.disable_restart()

    thr = threading.Thread(target=thread_for_test)
    thr.start()
    lichess_bot.start(li, user_profile, CONFIG, logging_level, None, None, one_game=True)
    thr.join()

    with open("./logs/result.txt") as file:
        data = file.read()
    return data


@pytest.mark.timeout(150, method="thread")
def test_sf() -> None:
    """Test lichess-bot with Stockfish (UCI)."""
    if platform != "linux" and platform != "win32":
        assert True
        return
    if os.path.exists("logs"):
        shutil.rmtree("logs")
    os.mkdir("logs")
    with open("./config.yml.default") as file:
        CONFIG = yaml.safe_load(file)
    CONFIG["token"] = ""
    CONFIG["engine"]["dir"] = "./TEMP/"
    CONFIG["engine"]["name"] = f"sf{file_extension}"
    CONFIG["engine"]["uci_options"]["Threads"] = 1
    CONFIG["pgn_directory"] = "TEMP/sf_game_record"
    win = run_bot(CONFIG, logging_level)
    shutil.rmtree("logs")
    lichess_bot.logger.info("Finished Testing SF")
    assert win == "1"
    assert os.path.isfile(os.path.join(CONFIG["pgn_directory"],
                                       "bo vs b - zzzzzzzz.pgn"))


@pytest.mark.timeout(150, method="thread")
def test_lc0() -> None:
    """Test lichess-bot with Leela Chess Zero (UCI)."""
    if platform != "win32":
        assert True
        return
    if os.path.exists("logs"):
        shutil.rmtree("logs")
    os.mkdir("logs")
    with open("./config.yml.default") as file:
        CONFIG = yaml.safe_load(file)
    CONFIG["token"] = ""
    CONFIG["engine"]["dir"] = "./TEMP/"
    CONFIG["engine"]["working_dir"] = "./TEMP/"
    CONFIG["engine"]["name"] = "lc0.exe"
    CONFIG["engine"]["uci_options"]["Threads"] = 1
    CONFIG["engine"]["uci_options"].pop("Hash", None)
    CONFIG["engine"]["uci_options"].pop("Move Overhead", None)
    CONFIG["pgn_directory"] = "TEMP/lc0_game_record"
    win = run_bot(CONFIG, logging_level)
    shutil.rmtree("logs")
    lichess_bot.logger.info("Finished Testing LC0")
    assert win == "1"
    assert os.path.isfile(os.path.join(CONFIG["pgn_directory"],
                                       "bo vs b - zzzzzzzz.pgn"))


@pytest.mark.timeout(150, method="thread")
def test_sjeng() -> None:
    """Test lichess-bot with Sjeng (XBoard)."""
    if platform != "win32":
        assert True
        return
    if os.path.exists("logs"):
        shutil.rmtree("logs")
    os.mkdir("logs")
    with open("./config.yml.default") as file:
        CONFIG = yaml.safe_load(file)
    CONFIG["token"] = ""
    CONFIG["engine"]["dir"] = "./TEMP/"
    CONFIG["engine"]["working_dir"] = "./TEMP/"
    CONFIG["engine"]["protocol"] = "xboard"
    CONFIG["engine"]["name"] = "sjeng.exe"
    CONFIG["engine"]["ponder"] = False
    CONFIG["pgn_directory"] = "TEMP/sjeng_game_record"
    win = run_bot(CONFIG, logging_level)
    shutil.rmtree("logs")
    lichess_bot.logger.info("Finished Testing Sjeng")
    assert win == "1"
    assert os.path.isfile(os.path.join(CONFIG["pgn_directory"],
                                       "bo vs b - zzzzzzzz.pgn"))


@pytest.mark.timeout(150, method="thread")
def test_homemade() -> None:
    """Test lichess-bot with a homemade engine running Stockfish (Homemade)."""
    if platform != "linux" and platform != "win32":
        assert True
        return
    with open("strategies.py") as file:
        original_strategies = file.read()

    with open("strategies.py", "a") as file:
        file.write(f"""
class Stockfish(ExampleEngine):
    def __init__(self, commands, options, stderr, draw_or_resign, **popen_args):
        super().__init__(commands, options, stderr, draw_or_resign, **popen_args)
        import chess
        self.engine = chess.engine.SimpleEngine.popen_uci('{stockfish_path}')

    def search(self, board, time_limit, *args):
        return self.engine.play(board, time_limit)
""")
    if os.path.exists("logs"):
        shutil.rmtree("logs")
    os.mkdir("logs")
    with open("./config.yml.default") as file:
        CONFIG = yaml.safe_load(file)
    CONFIG["token"] = ""
    CONFIG["engine"]["name"] = "Stockfish"
    CONFIG["engine"]["protocol"] = "homemade"
    CONFIG["pgn_directory"] = "TEMP/homemade_game_record"
    win = run_bot(CONFIG, logging_level)
    shutil.rmtree("logs")
    with open("strategies.py", "w") as file:
        file.write(original_strategies)
    lichess_bot.logger.info("Finished Testing Homemade")
    assert win == "1"
    assert os.path.isfile(os.path.join(CONFIG["pgn_directory"],
                                       "bo vs b - zzzzzzzz.pgn"))

lichess-bot/test_bot/test_bot.py
carlini/chess-llm
"""A timer for use in lichess-bot."""
import time
import datetime
from typing import Optional


def msec(time_in_msec: float) -> datetime.timedelta:
    """Create a timedelta duration in milliseconds."""
    return datetime.timedelta(milliseconds=time_in_msec)


def to_msec(duration: datetime.timedelta) -> float:
    """Return a bare number representing the length of the duration in milliseconds."""
    return duration / msec(1)


def msec_str(duration: datetime.timedelta) -> str:
    """Return a string with the duration value in whole number milliseconds."""
    return str(round(to_msec(duration)))


def seconds(time_in_sec: float) -> datetime.timedelta:
    """Create a timedelta duration in seconds."""
    return datetime.timedelta(seconds=time_in_sec)


def to_seconds(duration: datetime.timedelta) -> float:
    """Return a bare number representing the length of the duration in seconds."""
    return duration.total_seconds()


def sec_str(duration: datetime.timedelta) -> str:
    """Return a string with the duration value in whole number seconds."""
    return str(round(to_seconds(duration)))


def minutes(time_in_minutes: float) -> datetime.timedelta:
    """Create a timedelta duration in minutes."""
    return datetime.timedelta(minutes=time_in_minutes)


def hours(time_in_hours: float) -> datetime.timedelta:
    """Create a timedelta duration in hours."""
    return datetime.timedelta(hours=time_in_hours)


def days(time_in_days: float) -> datetime.timedelta:
    """Create a timedelta duration in minutes."""
    return datetime.timedelta(days=time_in_days)


def years(time_in_years: float) -> datetime.timedelta:
    """Create a timedelta duration in median years--i.e., 365 days."""
    return days(365) * time_in_years


class Timer:
    """
    A timer for use in lichess-bot. An instance of timer can be used both as a countdown timer and a stopwatch.

    If the duration argument in the __init__() method is greater than zero, then
    the method is_expired() indicates when the intial duration has passed. The
    method time_until_expiration() gives the amount of time left until the timer
    expires.

    Regardless of the initial duration (even if it's zero), a timer can be used
    as a stopwatch by calling time_since_reset() to get the amount of time since
    the timer was created or since it was last reset.
    """

    def __init__(self, duration: datetime.timedelta = seconds(0),
                 backdated_timestamp: Optional[datetime.datetime] = None) -> None:
        """
        Start the timer.

        :param duration: The duration of time before Timer.is_expired() returns True.
        :param backdated_timestamp: When the timer should have started. Used to keep the timers between sessions.
        """
        self.duration = duration
        self.reset()
        if backdated_timestamp is not None:
            time_already_used = datetime.datetime.now() - backdated_timestamp
            self.starting_time -= to_seconds(time_already_used)

    def is_expired(self) -> bool:
        """Check if a timer is expired."""
        return self.time_since_reset() >= self.duration

    def reset(self) -> None:
        """Reset the timer."""
        self.starting_time = time.perf_counter()

    def time_since_reset(self) -> datetime.timedelta:
        """How much time has passed."""
        return seconds(time.perf_counter() - self.starting_time)

    def time_until_expiration(self) -> datetime.timedelta:
        """How much time is left until it expires."""
        return max(seconds(0), self.duration - self.time_since_reset())

    def starting_timestamp(self, format: str) -> str:
        """When the timer started."""
        return (datetime.datetime.now() - self.time_since_reset()).strftime(format)

lichess-bot/timer.py
carlini/chess-llm
## Copyright (C) 2023, Nicholas Carlini <nicholas@carlini.com>.
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see <http://www.gnu.org/licenses/>.

import chess
import numpy as np
import io
import json
import chessllm
import csv

def convert_pgn_to_game(pgn_moves):
    pgn = io.StringIO(pgn_moves)
    game = chess.pgn.read_game(pgn)
    if len(game.errors) > 0:
        return None
    return game

def solve_puzzle(board, solution):
    solution = solution.split()
    while True:
        guess_next_move = engine.get_best_move(board)
        real_next_move, *solution = solution
        if guess_next_move != real_next_move:
            try:
                board.push_san(guess_next_move)
                if board.is_checkmate():
                    # Lichess puzzles allow multiple mate-in-1 solutions
                    return True
            except:
                pass
            return False
        board.push_san(guess_next_move)
        if len(solution) > 0:
            opponent_move, *solution = solution
            board.push_san(opponent_move)
        else:
            break
    return True

def main():

    ok = [[] for _ in range(30)]
    
    with open("pgn_puzzles.csv", 'r') as f:
        reader = csv.reader(f)
        for puzzleid, rating, pgn, solution in list(reader):
            rating = int(rating)//200
            if len(ok[rating]) >= 20: continue
            
            board = chess.Board()

            # Iterate over the moves and apply them to the board
            for move in convert_pgn_to_game(pgn).mainline_moves():
                board.push(move)

            is_right = solve_puzzle(board, solution)

            ok[rating].append(is_right)
    for i,x in enumerate(ok):
        print('rating',i*200, 'acc',np.mean(x))

    if True:
        import matplotlib.pyplot as plt
        # Remove nan values and get their indices
        ratings = [np.mean(x) for x in ok]
        non_nan_indices = [i for i, val in enumerate(ratings) if not np.isnan(val)]
        non_nan_values = [ratings[i] for i in non_nan_indices]
        
        # Create bucket ranges
        bucket_ranges = [(i*200, (i+1)*200) for i in non_nan_indices]
        bucket_labels = [f"{low}-{high}" for low, high in bucket_ranges]
        
        # Plotting
        plt.figure(figsize=(8, 4))
        plt.bar(bucket_labels, non_nan_values)
        plt.xlabel('Puzzle Rating (Elo)')
        plt.ylabel('Probability correct')
        plt.title('Ratings vs. Buckets')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig("/tmp/a.png", dpi=600)


if __name__ == "__main__":
    api_key = open("OPENAI_API_KEY").read().strip()
    config = json.loads(open("config.json").read())
    engine = chessllm.ChessLLM(api_key, config, num_lookahead_tokens=30)
    main()
    

puzzle_solver.py
carlini/chess-llm
#!/usr/bin/env python

## Copyright (C) 2023, Nicholas Carlini <nicholas@carlini.com>.
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see <http://www.gnu.org/licenses/>.

import chess
from chessllm import ChessLLM
import json

def main():

    config = json.loads(open("config.json").read())
    game = chess.pgn.Game()

    while True:
        line = input()
        log.write(f"Got line: {repr(line)}\n")
        log.flush()
        if line == "uci":
            print(f"id name chess-llm-with-{config['model']}")
            print("id author Nicholas Carlini")
            print("uciok")
        elif line == "isready":
            print("readyok")
        elif line.startswith("position"):
            parts = line.split(" ", 1)
            
            moves_list = []
            if "moves" in parts[1]:
                _, moves_str = parts[1].split("moves")
                moves_list = moves_str.strip().split()

            if parts[1].startswith("startpos"):
                board = chess.Board()
            else:
                raise

            for move_uci in moves_list:
                move = chess.Move.from_uci(move_uci)
                board.push(move)
            log.write(f"now position {repr(board)}\n")
            log.flush()


        elif line.startswith("go"):
            log.write("info string Starting search\n")
            log.flush()

            move = engine.get_best_move(board)
            try:
                log.write("Have move " + move + "\n")
                uci_move = board.push_san(move).uci()
            except:
                log.write(f"info string Invalid move: {repr(move)}\n")
                log.flush()
                
                moves = list(board.legal_moves)
                move = random.choice(moves)
                uci_move = move.uci()
            
            print(f"info pv {uci_move}")
            print(f"bestmove {uci_move}")
        elif line == "quit":
            break

        
if __name__ == "__main__":
    log = open("/tmp/log.txt", "a")
    api_key = open("OPENAI_API_KEY").read().strip()
    config = json.loads(open("config.json").read())
    engine = ChessLLM(api_key, config)
    main()


uci_engine.py
carlini/chess-llm