More complex challenge filtering and game-dependent option management (…

…#924)

* Allow more complex challenge filtering

* Change engine options based on game parameters

* Less misleading comment

* Handle missing initialFen field in Challenges

* Rename Game.opponent field

* Create default extra_game_handles.py file

* Update documentation

* Use actual player color if randomly chosen

README.md

@@ -25,6 +25,7 @@ Supports:
 
 ## Advanced options
 - [Create a homemade engine](https://github.com/lichess-bot-devs/lichess-bot/wiki/Create-a-homemade-engine)
+- [Add extra customizations](https://github.com/lichess-bot-devs/lichess-bot/wiki/Extra-customizations)
 
 <br />
 

extra_game_handlers.py

@@ -0,0 +1,21 @@
+"""Functions for the user to implement when the config file is not adequate to express bot requirements."""
+from lib import model
+from typing import Any
+
+
+def game_specific_options(game: model.Game) -> dict[str, Any]:
+    """
+    Return a dictionary of engine options based on game aspects.
+
+    By default, an empty dict is returned so that the options in the configuration file are used.
+    """
+    return {}
+
+
+def is_supported_extra(challenge: model.Challenge) -> bool:
+    """
+    Determine whether to accept a challenge.
+
+    By default, True is always returned so that there are no extra restrictions beyond those in the config file.
+    """
+    return True

lib/engine_wrapper.py

@@ -18,6 +18,7 @@
 from lib import config, model, lichess
 from lib.config import Configuration
 from lib.timer import Timer, msec, seconds, msec_str, sec_str, to_seconds
+from extra_game_handlers import game_specific_options
 from typing import Any, Optional, Union, Literal, Type
 from types import TracebackType
 OPTIONS_TYPE = dict[str, Any]
@@ -28,12 +29,13 @@
 MOVE = Union[chess.engine.PlayResult, list[chess.Move]]
 LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess]
 
+
 logger = logging.getLogger(__name__)
 
 out_of_online_opening_book_moves: Counter[str] = Counter()
 
 
-def create_engine(engine_config: config.Configuration) -> EngineWrapper:
+def create_engine(engine_config: config.Configuration, game: Optional[model.Game] = None) -> EngineWrapper:
     """
     Create the engine.
 
@@ -64,7 +66,7 @@ def create_engine(engine_config: config.Configuration) -> EngineWrapper:
             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}")
-    return Engine(commands, options, stderr, cfg.draw_or_resign, cwd=cfg.working_dir)
+    return Engine(commands, options, stderr, cfg.draw_or_resign, game, cwd=cfg.working_dir)
 
 
 def remove_managed_options(config: config.Configuration) -> OPTIONS_TYPE:
@@ -95,7 +97,7 @@ def __init__(self, options: OPTIONS_TYPE, draw_or_resign: config.Configuration)
         self.move_commentary: list[MOVE_INFO_TYPE] = []
         self.comment_start_index = -1
 
-    def configure(self, options: OPTIONS_TYPE) -> None:
+    def configure(self, options: OPTIONS_TYPE, game: Optional[model.Game]) -> None:
         """
         Send configurations to the engine.
 
@@ -104,7 +106,8 @@ def configure(self, options: OPTIONS_TYPE) -> None:
         Raises chess.engine.EngineError if an option is sent that the engine does not support.
         """
         try:
-            self.engine.configure(options)
+            extra_options = {} if game is None else game_specific_options(game)
+            self.engine.configure(options | extra_options)
         except Exception:
             self.engine.close()
             raise
@@ -468,34 +471,36 @@ 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:
+                 draw_or_resign: config.Configuration, game: Optional[model.Game], **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 game: The first Game message from the game stream.
         :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=True, stderr=stderr,
                                                           **popen_args)
-        self.configure(options)
+        self.configure(options, game)
 
 
 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:
+                 draw_or_resign: config.Configuration, game: Optional[model.Game], **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 game: The first Game message from the game stream.
         :param popen_args: The cwd of the engine.
         """
         super().__init__(options, draw_or_resign)
@@ -512,7 +517,7 @@ def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optio
                     options[f"egtpath {egt_type}"] = egt_paths[egt_type]
                 else:
                     logger.debug(f"No paths found for egt type: {egt_type}.")
-        self.configure(options)
+        self.configure(options, game)
 
 
 class MinimalEngine(EngineWrapper):
@@ -528,7 +533,8 @@ class MinimalEngine(EngineWrapper):
     """
 
     def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optional[int],
-                 draw_or_resign: Configuration, name: Optional[str] = None, **popen_args: str) -> None:
+                 draw_or_resign: Configuration, game: Optional[model.Game] = None, name: Optional[str] = None,
+                 **popen_args: str) -> None:
         """
         Initialize the values of the engine that all homemade engines inherit.
 

lib/matchmaking.py

@@ -336,7 +336,7 @@ def declined_challenge(self, event: EVENT_TYPE) -> None:
         Depends on whether `FilterType` is `NONE`, `COARSE`, or `FINE`.
         """
         challenge = model.Challenge(event["challenge"], self.user_profile)
-        opponent = challenge.opponent
+        opponent = challenge.challenge_target
         reason = event["challenge"]["declineReason"]
         logger.info(f"{opponent} declined {challenge}: {reason}")
         self.discard_challenge(challenge.id)

lib/model.py

@@ -26,8 +26,12 @@ def __init__(self, challenge_info: dict[str, Any], user_profile: dict[str, Any])
         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.challenge_target = Player(challenge_info.get("destUser") or {})
         self.from_self = self.challenger.name == user_profile["username"]
+        self.initial_fen = challenge_info.get("initialFen", "startpos")
+        color = challenge_info["color"]
+        self.color = color if color != "random" else challenge_info["finalColor"]
+        self.time_control = challenge_info["timeControl"]
 
     def is_supported_variant(self, challenge_cfg: Configuration) -> bool:
         """Check whether the variant is supported."""
@@ -94,6 +98,8 @@ def is_supported(self, config: Configuration,
             if self.from_self:
                 return True, ""
 
+            from extra_game_handlers import is_supported_extra
+
             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")
@@ -102,7 +108,8 @@ def is_supported(self, config: Configuration,
                               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"))
+                              or self.decline_due_to(self.is_supported_recent(config, recent_bot_challenges), "later")
+                              or self.decline_due_to(is_supported_extra(self), "generic"))
 
             return not decline_reason, decline_reason
 

lichess-bot.py

@@ -606,7 +606,7 @@ def play_game(li: LICHESS_TYPE,
     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:
+    with engine_wrapper.create_engine(config, game) 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)

test_bot/homemade.py

@@ -4,6 +4,7 @@
 import chess.engine
 import sys
 from lib.config import Configuration
+from lib import model
 from typing import Any, Optional, Union
 OPTIONS_TYPE = dict[str, Any]
 COMMANDS_TYPE = list[str]
@@ -17,9 +18,9 @@ class Stockfish(ExampleEngine):
     """A homemade engine that uses Stockfish."""
 
     def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_TYPE, stderr: Optional[int],
-                 draw_or_resign: Configuration, **popen_args: str):
+                 draw_or_resign: Configuration, game: Optional[model.Game], **popen_args: str):
         """Start Stockfish."""
-        super().__init__(commands, options, stderr, draw_or_resign, **popen_args)
+        super().__init__(commands, options, stderr, draw_or_resign, game, **popen_args)
         self.engine = chess.engine.SimpleEngine.popen_uci(f"./TEMP/sf{file_extension}")
 
     def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool,

wiki/Extra-customizations.md

@@ -0,0 +1,46 @@
+## Extra customizations
+
+If your bot has more complex requirements than can be expressed in the configuration file, edit the file named `extra_game_handlers.py` in the main lichess-bot directory.
+Within this file, write whatever code is needed.
+
+Each section below describes a customization.
+Only one function is needed to make that customization work.
+However, if writing other functions makes implementing the customization easier, do so.
+Only the named function will be used in lichess-bot.
+
+
+### Filtering challenges
+
+The function `is_supported_extra()` allows for finer control over which challenges from other players are accepted.
+It should use the data in the `Challenge` argument (see `lib/model.py`) and return `True` to accept the challenge or `False` to reject it.
+As an example, here's a version that will only only accept games where the bot plays black:
+``` python
+def is_supported_extra(challenge):
+    return challenge.color == "white"
+```
+For another example, this function will reject any board that contains queens:
+``` python
+def is_supported_extra(challenge):
+    # https://en.wikipedia.org/wiki/Forsyth%E2%80%93Edwards_Notation
+    starting_position = challenge.initial_fen
+    return starting_position != "startpos" and "Q" not in starting_position.upper()
+```
+The body of the function can be as complex as needed and combine any conditions to suit the needs of the bot.
+Information within the `Challenge` instance is detailed in the [Lichess API documentation](https://lichess.org/api#tag/Bot/operation/apiStreamEvent) (click on 200 under Responses, then select `ChallengeEvent` under Response Schema and expand the `challenge` heading).
+
+### Tailoring engine options
+
+The function `game_specific_options()` can modify the engine options for UCI and XBoard engines based on aspects of the game about to be played.
+It use the data in the `Game` argument (see `lib/model.py`) and return a dictionary of `str` to values.
+This dictionary will add or replace values in the `uci_options` or `xboard_options` section of the bot's configuration file.
+For example, this version of the function will changes the move overhead value for longer games:
+``` python
+from datetime import timedelta
+
+def game_specific_options(game):
+    if game.clock_initial >= timedelta(minutes=5):
+        return {"Move Overhead": 5000}
+    else:
+        return {}
+```
+Returning an empty dictionary leaves the engine options unchanged.