engine_wrapper.py

"""Provides communication with the engine."""
from __future__ import annotations
import os
import chess.engine
import chess.polyglot
import chess.syzygy
import chess.gaviota
import subprocess
import logging
import time
import random
from collections import Counter
from contextlib import contextmanager
import config
import model
import lichess
from config import Configuration
from typing import Dict, Any, List, Optional, Union, Tuple, Generator, Callable, Type
OPTIONS_TYPE = Dict[str, Any]
MOVE_INFO_TYPE = Dict[str, Any]
COMMANDS_TYPE = List[str]
LICHESS_EGTB_MOVE = Dict[str, Any]
CHESSDB_EGTB_MOVE = Dict[str, Any]
MOVE = Union[chess.engine.PlayResult, List[chess.Move]]

logger = logging.getLogger(__name__)

out_of_online_opening_book_moves: Counter[str] = Counter()


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

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

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

    stderr = None if cfg.silence_stderr else subprocess.DEVNULL

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


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

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


def translate_termination(game: model.Game, board: chess.Board) -> str:
    """Get a human-readable string with the result of the game."""
    winner_color = game.state.get("winner", "")
    termination: Optional[str] = game.state.get("status")

    if termination == model.Termination.MATE:
        return f"{winner_color.title()} mates"
    elif termination == model.Termination.TIMEOUT:
        return "Time forfeiture" if winner_color else "Timeout with insufficient material"
    elif termination == model.Termination.RESIGN:
        resigner = "black" if winner_color == "white" else "white"
        return f"{resigner.title()} resigns"
    elif termination == model.Termination.ABORT:
        return "Game aborted"
    elif termination == model.Termination.DRAW:
        if board.is_fifty_moves():
            return "50-move rule"
        elif board.is_repetition():
            return "Threefold repetition"
        elif board.is_insufficient_material():
            return "Insufficient material"
        else:
            return "Draw by agreement"
    elif termination:
        return termination
    else:
        return ""


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


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

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

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

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

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

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

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

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

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

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

            best_move = self.search(board, time_limit, can_ponder, draw_offered, best_move)
        else:
            self.stop()

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        def to_readable_value(stat: str, info: MOVE_INFO_TYPE) -> str:
            readable: Dict[str, Callable[[Any], str]] = {"score": self.readable_score, "wdl": self.readable_wdl,
                                                         "hashfull": lambda x: f"{round(x / 10, 1)}%",
                                                         "nodes": self.readable_number,
                                                         "nps": lambda x: f"{self.readable_number(x)}nps",
                                                         "tbhits": self.readable_number,
                                                         "cpuload": lambda x: f"{round(x / 10, 1)}%"}

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

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

        def to_readable_key(stat: str) -> str:
            readable = {"wdl": "winrate", "ponderpv": "PV", "nps": "speed", "score": "evaluation"}
            stat = readable.get(stat, stat)
            return stat.title()

        stats = ["score", "wdl", "depth", "nodes", "nps", "ponderpv"]
        if for_chat and "ponderpv" in info:
            bot_stats = [f"{to_readable_key(stat)}: {to_readable_value(stat, info)}"
                         for stat in stats if stat in info and stat != "ponderpv"]
            len_bot_stats = len(", ".join(bot_stats)) + PONDERPV_CHARACTERS
            ponder_pv = info["ponderpv"].split()
            try:
                while len(" ".join(ponder_pv)) + len_bot_stats > lichess.MAX_CHAT_MESSAGE_LEN:
                    ponder_pv.pop()
                if ponder_pv[-1].endswith("."):
                    ponder_pv.pop()
                info["ponderpv"] = " ".join(ponder_pv)
            except IndexError:
                pass
            if not info["ponderpv"]:
                info.pop("ponderpv")
        return [f"{to_readable_key(stat)}: {to_readable_value(stat, info)}" for stat in stats if stat in info]

    def get_opponent_info(self, game: model.Game) -> None:
        """Get the opponent's information and sends it to the engine. Depends on the protocol."""
        pass

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

    def report_game_result(self, game: model.Game, board: chess.Board) -> None:
        """Report the game result to the engine. Depends on the protocol."""
        pass

    def stop(self) -> None:
        """Stop the engine. Depends on the protocol."""
        pass

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

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

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


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

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

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

    def stop(self) -> None:
        """Tell the engine to stop searching."""
        self.engine.protocol.send_line("stop")

    def get_opponent_info(self, game: model.Game) -> None:
        """Get the opponent's info and send it to the engine."""
        name = game.opponent.name
        if (name and isinstance(self.engine.protocol, chess.engine.UciProtocol)
                and "UCI_Opponent" in self.engine.protocol.config):
            rating = game.opponent.rating or "none"
            title = game.opponent.title or "none"
            player_type = "computer" if title == "BOT" else "human"
            self.engine.configure({"UCI_Opponent": f"{title} {rating} {player_type} {name}"})

    def report_game_result(self, game: model.Game, board: chess.Board) -> None:
        """Send the game result to the engine."""
        if isinstance(self.engine.protocol, chess.engine.UciProtocol):
            self.engine.protocol._position(board)


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

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

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

    def report_game_result(self, game: model.Game, board: chess.Board) -> None:
        """Send the game result to the engine."""
        # Send final moves, if any, to engine.
        if isinstance(self.engine.protocol, chess.engine.XBoardProtocol):
            self.engine.protocol._new(board, None, {})

        endgame_message = translate_termination(game, board)
        if endgame_message:
            endgame_message = " {" + endgame_message + "}"

        self.engine.protocol.send_line(f"result {game.result()}{endgame_message}")

    def stop(self) -> None:
        """Tell the engine to stop searching."""
        self.engine.protocol.send_line("?")

    def get_opponent_info(self, game: model.Game) -> None:
        """Get the opponent's info and send it to the engine."""
        if (game.opponent.name and isinstance(self.engine.protocol, chess.engine.XBoardProtocol)
                and self.engine.protocol.features.get("name", True)):
            title = f"{game.opponent.title} " if game.opponent.title else ""
            self.engine.protocol.send_line(f"name {title}{game.opponent.name}")
        if game.me.rating and game.opponent.rating:
            self.engine.protocol.send_line(f"rating {game.me.rating} {game.opponent.rating}")
        if game.opponent.title == "BOT":
            self.engine.protocol.send_line("computer")


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

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

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

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

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

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

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

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

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

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

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

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

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

        Nothing happens unless the main engine does something.

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


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

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

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

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

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

        return method


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

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


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

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


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

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


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

    :param board: The current positions.
    :param game: The game that the bot is playing.
    :param start_time: The time we have left.
    :param move_overhead: The time it takes to communicate between the engine and lichess-bot.
    :return: The time to play a move.
    """
    pre_move_time = int((time.perf_counter_ns() - start_time) / 1e6)
    overhead = pre_move_time + move_overhead
    wb = "w" if board.turn == chess.WHITE else "b"
    game.state[f"{wb}time"] = max(0, game.state[f"{wb}time"] - overhead)
    logger.info("Searching for wtime {wtime} btime {btime}".format_map(game.state) + f" for game {game.id}")
    return chess.engine.Limit(white_clock=game.state["wtime"] / 1000,
                              black_clock=game.state["btime"] / 1000,
                              white_inc=game.state["winc"] / 1000,
                              black_inc=game.state["binc"] / 1000)


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


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

    variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant)
    config.change_value_to_list(polyglot_cfg.config, "book", key=variant)
    books = polyglot_cfg.book.lookup(variant)

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

        if move is not None:
            logger.info(f"Got move {move} from book {book} for game {game.id}")
            return chess.engine.PlayResult(move, None)

    return no_book_move


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

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

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

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

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

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


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

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

        if chessdb_cfg.contribute:
            params["action"] = "queue"
            li.online_book_get(site, params=params)
    except Exception:
        pass

    return move, comment


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

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

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

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

    return move, comment


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

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

        return None, -3

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

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

    return None, -3


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

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

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


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

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

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

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

            def good_enough(possible_move: LICHESS_EGTB_MOVE) -> bool:
                return name_to_wld[possible_move["category"]] == best_wdl
            possible_moves = list(filter(good_enough, data["moves"]))
            random_move = random.choice(possible_moves)
            move = random_move["uci"]
            wdl = name_to_wld[random_move["category"]] * -1
            dtz = random_move["dtz"] * -1
            dtm = random_move["dtm"]
            if dtm:
                dtm *= -1
            logger.info(f"Got move {move} from tablebase.lichess.ovh (wdl: {wdl}, dtz: {dtz}, dtm: {dtm}) for game {game.id}")

        return move, wdl
    return None, -3


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

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

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

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

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

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

            def good_enough(move: CHESSDB_EGTB_MOVE) -> bool:
                return score_to_wdl(move["score"]) == best_wdl
            possible_moves = list(filter(good_enough, data["moves"]))
            random_move = random.choice(possible_moves)
            score = random_move["score"]
            move = random_move["uci"]
            wdl = score_to_wdl(score)
            dtz = score_to_dtz(score)
            logger.info(f"Got move {move} from chessdb.cn (wdl: {wdl}, dtz: {dtz}) for game {game.id}")

        return move, wdl
    return None, -3


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

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

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

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


def dtz_scorer(tablebase: chess.syzygy.Tablebase, board: chess.Board) -> int:
    """Score a position based on a syzygy DTZ egtb."""
    dtz = -tablebase.probe_dtz(board)
    return dtz + (1 if dtz > 0 else -1) * board.halfmove_clock * (0 if dtz == 0 else 1)


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


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

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

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

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

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


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


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


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


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

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


def piecewise_function(range_definitions: List[Tuple[int, int]], last_value: int, position: int) -> int:
    """
    Return a value according to a position argument.

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

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

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

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


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


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