Public/Chess_Probabilistic_Programming
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added lichess bot

Browse Source
This commit is contained in:
Lukas Kuess
2024-01-25 23:11:25 +01:00
parent 35d3f456e9
commit 62410d239f
45 changed files with 6990 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
lichess_bot/lib/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""This lib folder contains the library code necessary for running lichess_bot."""

382
lichess_bot/lib/config.py Normal file
View File

@@ -0,0 +1,382 @@
"""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", key="max_depth", default=math.inf, 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="rating_preference", default="none")
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
def has_valid_list(name: str) -> bool:
entries = matchmaking.get(name)
return isinstance(entries, list) and entries[0] is not None
matchmaking_has_values = (has_valid_list("challenge_initial_time")
and has_valid_list("challenge_increment")
or has_valid_list("challenge_days"))
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 = matchmaking.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)}.")
config_assert(matchmaking.get("rating_preference") in ["none", "high", "low"],
f"{matchmaking.get('rating_preference')} is not a valid `matchmaking:rating_preference` option. "
f"Valid options are 'none', 'high', or 'low'.")
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)

102
lichess_bot/lib/conversation.py Normal file
View File

@@ -0,0 +1,102 @@
"""Allows lichess_bot to send messages to the chat."""
from __future__ import annotations
import logging
from lib import model
from lib.engine_wrapper import EngineWrapper
from lib.lichess import Lichess
from collections.abc import Sequence
from lib.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."""

1337
lichess_bot/lib/engine_wrapper.py Normal file
View File

File diff suppressed because it is too large Load Diff

385
lichess_bot/lib/lichess.py Normal file
View File

@@ -0,0 +1,385 @@
"""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 lib.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 is not None 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)

373
lichess_bot/lib/matchmaking.py Normal file
View File

@@ -0,0 +1,373 @@
"""Challenge other bots."""
import random
import logging
from lib import model
from lib.timer import Timer, seconds, minutes, days
from collections import defaultdict
from collections.abc import Sequence
from lib import lichess
import datetime
from lib.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 get_weights(self, online_bots: list[USER_PROFILE_TYPE], rating_preference: str, min_rating: int, max_rating: int,
game_type: str) -> list[int]:
"""Get the weight for each bot. A higher weights means the bot is more likely to get challenged."""
def rating(bot: USER_PROFILE_TYPE) -> int:
return int(bot.get("perfs", {}).get(game_type, {}).get("rating", 0))
if rating_preference == "high":
# A bot with max_rating rating will be twice as likely to get picked than a bot with min_rating rating.
reduce_ratings_by = min(min_rating - (max_rating - min_rating), min_rating - 1)
weights = [rating(bot) - reduce_ratings_by for bot in online_bots]
elif rating_preference == "low":
# A bot with min_rating rating will be twice as likely to get picked than a bot with max_rating rating.
reduce_ratings_by = max(max_rating - (min_rating - max_rating), max_rating + 1)
weights = [reduce_ratings_by - rating(bot) for bot in online_bots]
else:
weights = [1] * len(online_bots)
return weights
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"])
rating_preference = match_config.rating_preference
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
weights = self.get_weights(online_bots, rating_preference, min_rating, max_rating, game_type)
try:
bot = random.choices(online_bots, weights=weights)[0]
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"

284
lichess_bot/lib/model.py Normal file
View File

@@ -0,0 +1,284 @@
"""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 lib.timer import Timer, msec, seconds, sec_str, to_msec, to_seconds, years
from lib.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__()

95
lichess_bot/lib/strategies.py Normal file
View File

@@ -0,0 +1,95 @@
"""
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, Limit
import random
from lib.engine_wrapper import MinimalEngine, MOVE
from typing import Any
import logging
# 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: Limit, ponder: bool, draw_offered: bool, root_moves: MOVE) -> 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)

103
lichess_bot/lib/timer.py Normal file
View File

@@ -0,0 +1,103 @@
"""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)

4
lichess_bot/lib/versioning.yml Normal file
View File

@@ -0,0 +1,4 @@
lichess_bot_version: 2024.1.21.1
minimum_python_version: '3.9'
deprecated_python_version: '3.8'
deprecation_date: 2023-05-01