Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Various small changes #974

Merged
merged 3 commits into from
Jun 9, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion lib/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
from __future__ import annotations
import yaml
import os
import os.path
import logging
import math
from abc import ABCMeta
Expand Down
8 changes: 3 additions & 5 deletions lib/conversation.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,17 +69,15 @@ def command(self, line: ChatLine, cmd: str) -> None:
from_self = line.username == self.game.username
is_eval = cmd.startswith("eval")
if cmd == "commands" or cmd == "help":
self.send_reply(line, (
"Supported commands: !wait (wait a minute for my first move), !name, !howto, "
"!eval (or any text starting with !eval), !queue"))
self.send_reply(line,
"Supported commands: !wait (wait a minute for my first move), !name, "
"!eval (or any text starting with !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 is_eval and (from_self or line.room == "spectator"):
stats = self.engine.get_stats(for_chat=True)
self.send_reply(line, ", ".join(stats))
Expand Down
44 changes: 22 additions & 22 deletions lib/engine_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@
import test_bot.lichess
from collections import Counter
from collections.abc import Callable
from lib import config, model, lichess
from lib.config import Configuration
from lib import model, lichess
from lib.config import Configuration, change_value_to_list
from lib.timer import Timer, msec, seconds, msec_str, sec_str, to_seconds
from lib.types import (ReadableType, ChessDBMoveType, LichessEGTBMoveType, OPTIONS_GO_EGTB_TYPE, OPTIONS_TYPE,
COMMANDS_TYPE, MOVE, InfoStrDict, InfoDictKeys, InfoDictValue, GO_COMMANDS_TYPE, EGTPATH_TYPE,
Expand All @@ -32,7 +32,7 @@
out_of_online_opening_book_moves: Counter[str] = Counter()


def create_engine(engine_config: config.Configuration, game: Optional[model.Game] = None) -> EngineWrapper:
def create_engine(engine_config: Configuration, game: Optional[model.Game] = None) -> EngineWrapper:
"""
Create the engine.

Expand Down Expand Up @@ -61,12 +61,12 @@ def create_engine(engine_config: config.Configuration, game: Optional[model.Game
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({}))
options = remove_managed_options(cfg.lookup(f"{engine_type}_options") or Configuration({}))
logger.debug(f"Starting engine: {commands}")
return Engine(commands, options, stderr, cfg.draw_or_resign, game, cwd=cfg.working_dir)


def remove_managed_options(config: config.Configuration) -> OPTIONS_GO_EGTB_TYPE:
def remove_managed_options(config: Configuration) -> OPTIONS_GO_EGTB_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()
Expand All @@ -80,7 +80,7 @@ def is_managed(key: str) -> bool:
class EngineWrapper:
"""A wrapper used by all engines (UCI, XBoard, Homemade)."""

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

Expand All @@ -90,7 +90,7 @@ def __init__(self, options: OPTIONS_GO_EGTB_TYPE, draw_or_resign: config.Configu
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(cast(GO_COMMANDS_TYPE, options.pop("go_commands", {})) or {})
self.go_commands = Configuration(cast(GO_COMMANDS_TYPE, options.pop("go_commands", {})) or {})
self.move_commentary: list[InfoStrDict] = []
self.comment_start_index = -1

Expand Down Expand Up @@ -132,7 +132,7 @@ def play_move(self,
can_ponder: bool,
is_correspondence: bool,
correspondence_move_time: datetime.timedelta,
engine_cfg: config.Configuration,
engine_cfg: Configuration,
min_time: datetime.timedelta) -> None:
"""
Play a move.
Expand Down Expand Up @@ -478,7 +478,7 @@ class UCIEngine(EngineWrapper):
"""The class used to communicate with UCI engines."""

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

Expand All @@ -499,7 +499,7 @@ class XBoardEngine(EngineWrapper):
"""The class used to communicate with XBoard engines."""

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

Expand Down Expand Up @@ -738,7 +738,7 @@ def check_for_draw_offer(game: model.Game) -> bool:


def get_book_move(board: chess.Board, game: model.Game,
polyglot_cfg: config.Configuration) -> chess.engine.PlayResult:
polyglot_cfg: Configuration) -> chess.engine.PlayResult:
"""Get a move from an opening book."""
no_book_move = chess.engine.PlayResult(None, None)
use_book = polyglot_cfg.enabled
Expand All @@ -751,7 +751,7 @@ def get_book_move(board: chess.Board, game: model.Game,
else:
variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant)

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

for book in books:
Expand All @@ -776,8 +776,8 @@ def get_book_move(board: chess.Board, game: model.Game,
return no_book_move


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

Expand Down Expand Up @@ -829,7 +829,7 @@ def get_online_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, onli


def get_chessdb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
chessdb_cfg: config.Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
chessdb_cfg: Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
"""Get a move from chessdb.cn's opening book."""
use_chessdb = chessdb_cfg.enabled
time_left = msec(game.state[wbtime(board)])
Expand Down Expand Up @@ -868,7 +868,7 @@ def get_chessdb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,


def get_lichess_cloud_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
lichess_cloud_cfg: config.Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
lichess_cloud_cfg: Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
"""Get a move from the lichess's cloud analysis."""
side = wbtime(board)
time_left = msec(game.state[side])
Expand Down Expand Up @@ -922,7 +922,7 @@ def get_lichess_cloud_move(li: LICHESS_TYPE, board: chess.Board, game: model.Gam


def get_opening_explorer_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
opening_explorer_cfg: config.Configuration
opening_explorer_cfg: Configuration
) -> tuple[Optional[str], chess.engine.InfoDict]:
"""Get a move from lichess's opening explorer."""
side = wbtime(board)
Expand Down Expand Up @@ -974,7 +974,7 @@ def get_opening_explorer_move(li: LICHESS_TYPE, board: chess.Board, game: model.
return move, comment


def get_online_egtb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, online_egtb_cfg: config.Configuration
def get_online_egtb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, online_egtb_cfg: Configuration
) -> tuple[Union[str, list[str], None], int, chess.engine.InfoDict]:
"""
Get a move from an online egtb (either by lichess or chessdb).
Expand Down Expand Up @@ -1011,8 +1011,8 @@ def get_online_egtb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
return None, -3, {}


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

Expand Down Expand Up @@ -1148,7 +1148,7 @@ def good_enough(move: ChessDBMoveType) -> bool:


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

Expand Down Expand Up @@ -1220,7 +1220,7 @@ def dtz_to_wdl(dtz: Union[int, float]) -> int:


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

Expand Down
6 changes: 3 additions & 3 deletions lichess-bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import chess
import chess.pgn
from chess.variant import find_variant
import lib.config
from lib import engine_wrapper, model, lichess, matchmaking
import json
import logging
Expand All @@ -22,6 +21,7 @@
import traceback
import itertools
import glob
import platform
import test_bot.lichess
from lib.config import load_config, Configuration
from lib.conversation import Conversation, ChatLine
Expand All @@ -48,7 +48,7 @@ class PlayGameArgsType(TypedDict, total=False):
li: LICHESS_TYPE
control_queue: CONTROL_QUEUE_TYPE
user_profile: UserProfileType
config: lib.config.Configuration
config: Configuration
challenge_queue: MULTIPROCESSING_LIST_TYPE
correspondence_queue: CORRESPONDENCE_QUEUE_TYPE
logging_queue: LOGGING_QUEUE_TYPE
Expand Down Expand Up @@ -1101,7 +1101,7 @@ def intro() -> str:
return fr"""
. _/|
. // o\
. || ._) lichess-bot {__version__}
. || ._) lichess-bot {__version__} on {platform.system()} {platform.release()}
. //__\
. )___( Play on Lichess with a bot
"""
Expand Down
Loading