from __future__ import annotations

import abc

import asyncio

import collections

import concurrent.futures

import contextlib

import copy

import dataclasses

import enum

import inspect

import logging

import math

import shlex

import subprocess

import sys

import threading

import time

import typing

import re

import chess

from chess import Color

from types import TracebackType

from typing import Any, Callable, Coroutine, Deque, Dict, Generator, Generic, Iterable, Iterator, List, Literal, Mapping, MutableMapping, Optional, Tuple, Type, TypedDict, TypeVar, Union

if typing.TYPE_CHECKING:

from typing_extensions import override

else:

F = typing.TypeVar("F", bound=Callable[..., Any])

def override(fn: F, /) -> F:

return fn

if typing.TYPE_CHECKING:

from typing_extensions import Self

WdlModel = Literal["sf", "sf16.1", "sf16", "sf15.1", "sf15", "sf14", "sf12", "lichess"]

T = TypeVar("T")

ProtocolT = TypeVar("ProtocolT", bound="Protocol")

ConfigValue = Union[str, int, bool, None]

ConfigMapping = Mapping[str, ConfigValue]

LOGGER = logging.getLogger(__name__)

MANAGED_OPTIONS = ["uci_chess960", "uci_variant", "multipv", "ponder"]

def run_in_background(coroutine: Callable[[concurrent.futures.Future[T]], Coroutine[Any, Any, None]], *, name: Optional[str] = None, debug: Optional[bool] = None) -> T:

"""

Runs ``coroutine(future)`` in a new event loop on a background thread.

Blocks on *future* and returns the result as soon as it is resolved.

The coroutine and all remaining tasks continue running in the background

until complete.

"""

assert inspect.iscoroutinefunction(coroutine)

future: concurrent.futures.Future[T] = concurrent.futures.Future()

def background() -> None:

try:

asyncio.run(coroutine(future), debug=debug)

future.cancel()

except Exception as exc:

future.set_exception(exc)

threading.Thread(target=background, name=name).start()

return future.result()

class EngineError(RuntimeError):

"""Runtime error caused by a misbehaving engine or incorrect usage."""

class EngineTerminatedError(EngineError):

"""The engine process exited unexpectedly."""

class AnalysisComplete(Exception):

"""

Raised when analysis is complete, all information has been consumed, but

further information was requested.

"""

@dataclasses.dataclass(frozen=True)

class Option:

"""Information about an available engine option."""

name: str

"""The name of the option."""

type: str

"""

The type of the option.

+--------+-----+------+------------------------------------------------+

| type | UCI | CECP | value |

+========+=====+======+================================================+

| check | X | X | ``True`` or ``False`` |

+--------+-----+------+------------------------------------------------+

| spin | X | X | integer, between *min* and *max* |

+--------+-----+------+------------------------------------------------+

| combo | X | X | string, one of *var* |

+--------+-----+------+------------------------------------------------+

| button | X | X | ``None`` |

+--------+-----+------+------------------------------------------------+

| reset | | X | ``None`` |

+--------+-----+------+------------------------------------------------+

| save | | X | ``None`` |

+--------+-----+------+------------------------------------------------+

| string | X | X | string without line breaks |

+--------+-----+------+------------------------------------------------+

| file | | X | string, interpreted as the path to a file |

+--------+-----+------+------------------------------------------------+

| path | | X | string, interpreted as the path to a directory |

+--------+-----+------+------------------------------------------------+

"""

default: ConfigValue

"""The default value of the option."""

min: Optional[int]

"""The minimum integer value of a *spin* option."""

max: Optional[int]

"""The maximum integer value of a *spin* option."""

var: Optional[List[str]]

"""A list of allowed string values for a *combo* option."""

def parse(self, value: ConfigValue) -> ConfigValue:

if self.type == "check":

return value and value != "false"

elif self.type == "spin":

try:

value = int(value) # type: ignore

except ValueError:

raise EngineError(f"expected integer for spin option {self.name!r}, got: {value!r}")

if self.min is not None and value < self.min:

raise EngineError(f"expected value for option {self.name!r} to be at least {self.min}, got: {value}")

if self.max is not None and self.max < value:

raise EngineError(f"expected value for option {self.name!r} to be at most {self.max}, got: {value}")

return value

elif self.type == "combo":

value = str(value)

if value not in (self.var or []):

raise EngineError("invalid value for combo option {!r}, got: {} (available: {})".format(self.name, value, ", ".join(self.var) if self.var else "-"))

return value

elif self.type in ["button", "reset", "save"]:

return None

elif self.type in ["string", "file", "path"]:

value = str(value)

if "\n" in value or "\r" in value:

raise EngineError(f"invalid line-break in string option {self.name!r}: {value!r}")

return value

else:

raise EngineError(f"unknown option type: {self.type!r}")

def is_managed(self) -> bool:

"""

Some options are managed automatically: ``UCI_Chess960``,

``UCI_Variant``, ``MultiPV``, ``Ponder``.

"""

return self.name.lower() in MANAGED_OPTIONS

@dataclasses.dataclass

class Limit:

"""Search-termination condition."""

time: Optional[float] = None

"""Search exactly *time* seconds."""

depth: Optional[int] = None

"""Search *depth* ply only."""

nodes: Optional[int] = None

"""Search only a limited number of *nodes*."""

mate: Optional[int] = None

"""Search for a mate in *mate* moves."""

white_clock: Optional[float] = None

"""Time in seconds remaining for White."""

black_clock: Optional[float] = None

"""Time in seconds remaining for Black."""

white_inc: Optional[float] = None

"""Fisher increment for White, in seconds."""

black_inc: Optional[float] = None

"""Fisher increment for Black, in seconds."""

remaining_moves: Optional[int] = None

"""

Number of moves to the next time control. If this is not set, but

*white_clock* and *black_clock* are, then it is sudden death.

"""

clock_id: object = None

"""

An identifier to use with XBoard engines to signal that the time

control has changed. When this field changes, Xboard engines are

sent level or st commands as appropriate. Otherwise, only time

and otim commands are sent to update the engine's clock.

"""

def __repr__(self) -> str:

# Like default __repr__, but without None values.

return "{}({})".format(

type(self).__name__,

", ".join("{}={!r}".format(attr, getattr(self, attr))

for attr in ["time", "depth", "nodes", "mate", "white_clock", "black_clock", "white_inc", "black_inc", "remaining_moves"]

if getattr(self, attr) is not None))

class InfoDict(TypedDict, total=False):

"""

Dictionary of aggregated information sent by the engine.

Commonly used keys are: ``score`` (a :class:`~chess.engine.PovScore`),

``pv`` (a list of :class:`~chess.Move` objects), ``depth``,

``seldepth``, ``time`` (in seconds), ``nodes``, ``nps``, ``multipv``

(``1`` for the mainline).

Others: ``tbhits``, ``currmove``, ``currmovenumber``, ``hashfull``,

``cpuload``, ``refutation``, ``currline``, ``ebf`` (effective branching factor),

``wdl`` (a :class:`~chess.engine.PovWdl`), and ``string``.

"""

score: PovScore

pv: List[chess.Move]

depth: int

seldepth: int

time: float

nodes: int

nps: int

tbhits: int

multipv: int

currmove: chess.Move

currmovenumber: int

hashfull: int

cpuload: int

refutation: Dict[chess.Move, List[chess.Move]]

currline: Dict[int, List[chess.Move]]

ebf: float

wdl: PovWdl

string: str

class PlayResult:

"""Returned by :func:`chess.engine.Protocol.play()`."""

move: Optional[chess.Move]

"""The best move according to the engine, or ``None``."""

ponder: Optional[chess.Move]

"""The response that the engine expects after *move*, or ``None``."""

info: InfoDict

"""

A dictionary of extra :class:`information <chess.engine.InfoDict>`

sent by the engine, if selected with the *info* argument of

:func:`~chess.engine.Protocol.play()`.

"""

draw_offered: bool

"""Whether the engine offered a draw before moving."""

resigned: bool

"""Whether the engine resigned."""

def __init__(self,

move: Optional[chess.Move],

ponder: Optional[chess.Move],

info: Optional[InfoDict] = None,

*,

draw_offered: bool = False,

resigned: bool = False) -> None:

self.move = move

self.ponder = ponder

self.info = info or {}

self.draw_offered = draw_offered

self.resigned = resigned

def __repr__(self) -> str:

return "<{} at {:#x} (move={}, ponder={}, info={}, draw_offered={}, resigned={})>".format(

type(self).__name__, id(self), self.move, self.ponder, self.info,

self.draw_offered, self.resigned)

class Info(enum.IntFlag):

"""Used to filter information sent by the chess engine."""

NONE = 0

BASIC = 1

SCORE = 2

PV = 4

REFUTATION = 8

CURRLINE = 16

ALL = BASIC | SCORE | PV | REFUTATION | CURRLINE

INFO_NONE = Info.NONE

INFO_BASIC = Info.BASIC

INFO_SCORE = Info.SCORE

INFO_PV = Info.PV

INFO_REFUTATION = Info.REFUTATION

INFO_CURRLINE = Info.CURRLINE

INFO_ALL = Info.ALL

@dataclasses.dataclass

class Opponent:

"""Used to store information about an engine's opponent."""

name: Optional[str]

"""The name of the opponent."""

title: Optional[str]

"""The opponent's title--for example, GM, IM, or BOT."""

rating: Optional[int]

"""The opponent's ELO rating."""

is_engine: Optional[bool]

"""Whether the opponent is a chess engine/computer program."""

class PovScore:

"""A relative :class:`~chess.engine.Score` and the point of view."""

relative: Score

"""The relative :class:`~chess.engine.Score`."""

turn: Color

"""The point of view (``chess.WHITE`` or ``chess.BLACK``)."""

def __init__(self, relative: Score, turn: Color) -> None:

self.relative = relative

self.turn = turn

def white(self) -> Score:

"""Gets the score from White's point of view."""

return self.pov(chess.WHITE)

def black(self) -> Score:

"""Gets the score from Black's point of view."""

return self.pov(chess.BLACK)

def pov(self, color: Color) -> Score:

"""Gets the score from the point of view of the given *color*."""

return self.relative if self.turn == color else -self.relative

def is_mate(self) -> bool:

"""Tests if this is a mate score."""

return self.relative.is_mate()

def wdl(self, *, model: WdlModel = "sf", ply: int = 30) -> PovWdl:

"""See :func:`~chess.engine.Score.wdl()`."""

return PovWdl(self.relative.wdl(model=model, ply=ply), self.turn)

def __repr__(self) -> str:

return "PovScore({!r}, {})".format(self.relative, "WHITE" if self.turn else "BLACK")

def __eq__(self, other: object) -> bool:

if isinstance(other, PovScore):

return self.white() == other.white()

else:

return NotImplemented

class Score(abc.ABC):

"""

Evaluation of a position.

The score can be :class:`~chess.engine.Cp` (centi-pawns),

:class:`~chess.engine.Mate` or :py:data:`~chess.engine.MateGiven`.

A positive value indicates an advantage.

There is a total order defined on centi-pawn and mate scores.

>>> from chess.engine import Cp, Mate, MateGiven

>>>

>>> Mate(-0) < Mate(-1) < Cp(-50) < Cp(200) < Mate(4) < Mate(1) < MateGiven

True

Scores can be negated to change the point of view:

>>> -Cp(20)

Cp(-20)

>>> -Mate(-4)

Mate(+4)

>>> -Mate(0)

MateGiven

"""

@typing.overload

@abc.abstractmethod

def score(self, *, mate_score: int) -> int: ...

@typing.overload

@abc.abstractmethod

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]: ...

@abc.abstractmethod

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]:

"""

Returns the centi-pawn score as an integer or ``None``.

You can optionally pass a large value to convert mate scores to

centi-pawn scores.

>>> Cp(-300).score()

-300

>>> Mate(5).score() is None

True

>>> Mate(5).score(mate_score=100000)

99995

"""

@abc.abstractmethod

def mate(self) -> Optional[int]:

"""

Returns the number of plies to mate, negative if we are getting

mated, or ``None``.

.. warning::

This conflates ``Mate(0)`` (we lost) and ``MateGiven``

(we won) to ``0``.

"""

def is_mate(self) -> bool:

"""Tests if this is a mate score."""

return self.mate() is not None

@abc.abstractmethod

def wdl(self, *, model: WdlModel = "sf", ply: int = 30) -> Wdl:

"""

Returns statistics for the expected outcome of this game, based on

a *model*, given that this score is reached at *ply*.

Scores have a total order, but it makes little sense to compute

the difference between two scores. For example, going from

``Cp(-100)`` to ``Cp(+100)`` is much more significant than going

from ``Cp(+300)`` to ``Cp(+500)``. It is better to compute differences

of the expectation values for the outcome of the game (based on winning

chances and drawing chances).

>>> Cp(100).wdl().expectation() - Cp(-100).wdl().expectation() # doctest: +ELLIPSIS

0.379...

>>> Cp(500).wdl().expectation() - Cp(300).wdl().expectation() # doctest: +ELLIPSIS

0.015...

:param model:

* ``sf``, the WDL model used by the latest Stockfish

(currently ``sf16``).

* ``sf16``, the WDL model used by Stockfish 16.

* ``sf15.1``, the WDL model used by Stockfish 15.1.

* ``sf15``, the WDL model used by Stockfish 15.

* ``sf14``, the WDL model used by Stockfish 14.

* ``sf12``, the WDL model used by Stockfish 12.

* ``lichess``, the win rate model used by Lichess.

Does not use *ply*, and does not consider drawing chances.

:param ply: The number of half-moves played since the starting

position. Models may scale scores slightly differently based on

this. Defaults to middle game.

"""

@abc.abstractmethod

def __neg__(self) -> Score:

...

@abc.abstractmethod

def __pos__(self) -> Score:

...

@abc.abstractmethod

def __abs__(self) -> Score:

...

def _score_tuple(self) -> Tuple[bool, bool, bool, int, Optional[int]]:

mate = self.mate()

return (

isinstance(self, MateGivenType),

mate is not None and mate > 0,

mate is None,

-(mate or 0),

self.score(),

)

def __eq__(self, other: object) -> bool:

if isinstance(other, Score):

return self._score_tuple() == other._score_tuple()

else:

return NotImplemented

def __lt__(self, other: object) -> bool:

if isinstance(other, Score):

return self._score_tuple() < other._score_tuple()

else:

return NotImplemented

def __le__(self, other: object) -> bool:

if isinstance(other, Score):

return self._score_tuple() <= other._score_tuple()

else:

return NotImplemented

def __gt__(self, other: object) -> bool:

if isinstance(other, Score):

return self._score_tuple() > other._score_tuple()

else:

return NotImplemented

def __ge__(self, other: object) -> bool:

if isinstance(other, Score):

return self._score_tuple() >= other._score_tuple()

else:

return NotImplemented

def _sf16_1_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_16.1/src/uci.cpp#L48

NormalizeToPawnValue = 356

# https://github.com/official-stockfish/Stockfish/blob/sf_16.1/src/uci.cpp#L383-L384

m = min(120, max(8, ply / 2 + 1)) / 32

a = (((-1.06249702 * m + 7.42016937) * m + 0.89425629) * m) + 348.60356174

b = (((-5.33122190 * m + 39.57831533) * m + -90.84473771) * m) + 123.40620748

x = min(4000, max(cp * NormalizeToPawnValue / 100, -4000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _sf16_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_16/src/uci.h#L38

NormalizeToPawnValue = 328

# https://github.com/official-stockfish/Stockfish/blob/sf_16/src/uci.cpp#L200-L224

m = min(240, max(ply, 0)) / 64

a = (((0.38036525 * m + -2.82015070) * m + 23.17882135) * m) + 307.36768407

b = (((-2.29434733 * m + 13.27689788) * m + -14.26828904) * m) + 63.45318330

x = min(4000, max(cp * NormalizeToPawnValue / 100, -4000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _sf15_1_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_15.1/src/uci.h#L38

NormalizeToPawnValue = 361

# https://github.com/official-stockfish/Stockfish/blob/sf_15.1/src/uci.cpp#L200-L224

m = min(240, max(ply, 0)) / 64

a = (((-0.58270499 * m + 2.68512549) * m + 15.24638015) * m) + 344.49745382

b = (((-2.65734562 * m + 15.96509799) * m + -20.69040836) * m) + 73.61029937

x = min(4000, max(cp * NormalizeToPawnValue / 100, -4000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _sf15_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_15/src/uci.cpp#L200-L220

m = min(240, max(ply, 0)) / 64

a = (((-1.17202460e-1 * m + 5.94729104e-1) * m + 1.12065546e+1) * m) + 1.22606222e+2

b = (((-1.79066759 * m + 11.30759193) * m + -17.43677612) * m) + 36.47147479

x = min(2000, max(cp, -2000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _sf14_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_14/src/uci.cpp#L200-L220

m = min(240, max(ply, 0)) / 64

a = (((-3.68389304 * m + 30.07065921) * m + -60.52878723) * m) + 149.53378557

b = (((-2.01818570 * m + 15.85685038) * m + -29.83452023) * m) + 47.59078827

x = min(2000, max(cp, -2000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _sf12_wins(cp: int, *, ply: int) -> int:

# https://github.com/official-stockfish/Stockfish/blob/sf_12/src/uci.cpp#L198-L218

m = min(240, max(ply, 0)) / 64

a = (((-8.24404295 * m + 64.23892342) * m + -95.73056462) * m) + 153.86478679

b = (((-3.37154371 * m + 28.44489198) * m + -56.67657741) * m) + 72.05858751

x = min(1000, max(cp, -1000))

return int(0.5 + 1000 / (1 + math.exp((a - x) / b)))

def _lichess_raw_wins(cp: int) -> int:

# https://github.com/lichess-org/lila/pull/11148

# https://github.com/lichess-org/lila/blob/2242b0a08faa06e7be5508d338ede7bb09049777/modules/analyse/src/main/WinPercent.scala#L26-L30

return round(1000 / (1 + math.exp(-0.00368208 * cp)))

class Cp(Score):

"""Centi-pawn score."""

def __init__(self, cp: int) -> None:

self.cp = cp

def mate(self) -> None:

return None

def score(self, *, mate_score: Optional[int] = None) -> int:

return self.cp

def wdl(self, *, model: WdlModel = "sf", ply: int = 30) -> Wdl:

if model == "lichess":

wins = _lichess_raw_wins(max(-1000, min(self.cp, 1000)))

losses = 1000 - wins

elif model == "sf12":

wins = _sf12_wins(self.cp, ply=ply)

losses = _sf12_wins(-self.cp, ply=ply)

elif model == "sf14":

wins = _sf14_wins(self.cp, ply=ply)

losses = _sf14_wins(-self.cp, ply=ply)

elif model == "sf15":

wins = _sf15_wins(self.cp, ply=ply)

losses = _sf15_wins(-self.cp, ply=ply)

elif model == "sf15.1":

wins = _sf15_1_wins(self.cp, ply=ply)

losses = _sf15_1_wins(-self.cp, ply=ply)

elif model == "sf16":

wins = _sf16_wins(self.cp, ply=ply)

losses = _sf16_wins(-self.cp, ply=ply)

else:

wins = _sf16_1_wins(self.cp, ply=ply)

losses = _sf16_1_wins(-self.cp, ply=ply)

draws = 1000 - wins - losses

return Wdl(wins, draws, losses)

def __str__(self) -> str:

return f"+{self.cp:d}" if self.cp > 0 else str(self.cp)

def __repr__(self) -> str:

return f"Cp({self})"

def __neg__(self) -> Cp:

return Cp(-self.cp)

def __pos__(self) -> Cp:

return Cp(self.cp)

def __abs__(self) -> Cp:

return Cp(abs(self.cp))

class Mate(Score):

"""Mate score."""

def __init__(self, moves: int) -> None:

self.moves = moves

def mate(self) -> int:

return self.moves

@typing.overload

def score(self, *, mate_score: int) -> int: ...

@typing.overload

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]: ...

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]:

if mate_score is None:

return None

elif self.moves > 0:

return mate_score - self.moves

else:

return -mate_score - self.moves

def wdl(self, *, model: WdlModel = "sf", ply: int = 30) -> Wdl:

if model == "lichess":

cp = (21 - min(10, abs(self.moves))) * 100

wins = _lichess_raw_wins(cp)

return Wdl(wins, 0, 1000 - wins) if self.moves > 0 else Wdl(1000 - wins, 0, wins)

else:

return Wdl(1000, 0, 0) if self.moves > 0 else Wdl(0, 0, 1000)

def __str__(self) -> str:

return f"#+{self.moves}" if self.moves > 0 else f"#-{abs(self.moves)}"

def __repr__(self) -> str:

return "Mate({})".format(str(self).lstrip("#"))

def __neg__(self) -> Union[MateGivenType, Mate]:

return MateGiven if not self.moves else Mate(-self.moves)

def __pos__(self) -> Mate:

return Mate(self.moves)

def __abs__(self) -> Union[MateGivenType, Mate]:

return MateGiven if not self.moves else Mate(abs(self.moves))

class MateGivenType(Score):

"""Winning mate score, equivalent to ``-Mate(0)``."""

def mate(self) -> int:

return 0

@typing.overload

def score(self, *, mate_score: int) -> int: ...

@typing.overload

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]: ...

def score(self, *, mate_score: Optional[int] = None) -> Optional[int]:

return mate_score

def wdl(self, *, model: WdlModel = "sf", ply: int = 30) -> Wdl:

return Wdl(1000, 0, 0)

def __neg__(self) -> Mate:

return Mate(0)

def __pos__(self) -> MateGivenType:

return self

def __abs__(self) -> MateGivenType:

return self

def __repr__(self) -> str:

return "MateGiven"

def __str__(self) -> str:

return "#+0"

MateGiven = MateGivenType()

@dataclasses.dataclass

class PovWdl:

"""

Relative :class:`win/draw/loss statistics <chess.engine.Wdl>` and the point

of view.

"""

relative: Wdl

"""The relative :class:`~chess.engine.Wdl`."""

turn: Color

"""The point of view (``chess.WHITE`` or ``chess.BLACK``)."""

def white(self) -> Wdl:

"""Gets the :class:`~chess.engine.Wdl` from White's point of view."""

return self.pov(chess.WHITE)

def black(self) -> Wdl:

"""Gets the :class:`~chess.engine.Wdl` from Black's point of view."""

return self.pov(chess.BLACK)

def pov(self, color: Color) -> Wdl:

"""

Gets the :class:`~chess.engine.Wdl` from the point of view of the given

*color*.

"""

return self.relative if self.turn == color else -self.relative

def __bool__(self) -> bool:

return bool(self.relative)

def __repr__(self) -> str:

return "PovWdl({!r}, {})".format(self.relative, "WHITE" if self.turn else "BLACK")

@dataclasses.dataclass

class Wdl:

"""Win/draw/loss statistics."""

wins: int

"""The number of wins."""

draws: int

"""The number of draws."""

losses: int

"""The number of losses."""

def total(self) -> int:

"""

Returns the total number of games. Usually, ``wdl`` reported by engines

is scaled to 1000 games.

"""

return self.wins + self.draws + self.losses

def winning_chance(self) -> float:

"""Returns the relative frequency of wins."""

return self.wins / self.total()

def drawing_chance(self) -> float:

"""Returns the relative frequency of draws."""

return self.draws / self.total()

def losing_chance(self) -> float:

"""Returns the relative frequency of losses."""

return self.losses / self.total()

def expectation(self) -> float:

"""

Returns the expectation value, where a win is valued 1, a draw is

valued 0.5, and a loss is valued 0.

"""

return (self.wins + 0.5 * self.draws) / self.total()

def __bool__(self) -> bool:

return bool(self.total())

def __pos__(self) -> Wdl:

return self

def __neg__(self) -> Wdl:

return Wdl(self.losses, self.draws, self.wins)

class MockTransport(asyncio.SubprocessTransport, asyncio.WriteTransport):

def __init__(self, protocol: Protocol) -> None:

super().__init__()

self.protocol = protocol

self.expectations: Deque[Tuple[str, List[str]]] = collections.deque()

self.expected_pings = 0

self.stdin_buffer = bytearray()

self.protocol.connection_made(self)

def expect(self, expectation: str, responses: List[str] = []) -> None:

self.expectations.append((expectation, responses))

def expect_ping(self) -> None:

self.expected_pings += 1

def assert_done(self) -> None:

assert not self.expectations, f"pending expectations: {self.expectations}"

def get_pipe_transport(self, fd: int) -> Optional[asyncio.BaseTransport]:

assert fd == 0, f"expected 0 for stdin, got {fd}"

return self

def write(self, data: bytes | bytearray | memoryview) -> None:

self.stdin_buffer.extend(data)

while b"\n" in self.stdin_buffer:

line_bytes, self.stdin_buffer = self.stdin_buffer.split(b"\n", 1)

line = line_bytes.decode("utf-8")

if line.startswith("ping ") and self.expected_pings:

self.expected_pings -= 1

self.protocol.pipe_data_received(1, (line.replace("ping ", "pong ") + "\n").encode("utf-8"))

else:

assert self.expectations, f"unexpected: {line!r}"

expectation, responses = self.expectations.popleft()

assert expectation == line, f"expected {expectation}, got: {line}"

if responses:

self.protocol.loop.call_soon(self.protocol.pipe_data_received, 1, "\n".join(responses + [""]).encode("utf-8"))

def get_pid(self) -> int:

return id(self)

def get_returncode(self) -> Optional[int]:

return None if self.expectations else 0

class Protocol(asyncio.SubprocessProtocol, metaclass=abc.ABCMeta):

"""Protocol for communicating with a chess engine process."""

id: Dict[str, str]

"""

Dictionary of information about the engine. Common keys are ``name``

and ``author``.

"""

returncode: asyncio.Future[int]

"""Future: Exit code of the process."""

def __init__(self) -> None:

self.loop = asyncio.get_running_loop()

self.transport: Optional[asyncio.SubprocessTransport] = None

self.buffer = {

1: bytearray(), # stdout

2: bytearray(), # stderr

}

self.command: Optional[BaseCommand[Any]] = None

self.next_command: Optional[BaseCommand[Any]] = None

self.initialized = False

self.returncode: asyncio.Future[int] = asyncio.Future()

def connection_made(self, transport: asyncio.BaseTransport) -> None:

# SubprocessTransport expected, but not checked to allow duck typing.

self.transport = transport # type: ignore

LOGGER.debug("%s: Connection made", self)

def connection_lost(self, exc: Optional[Exception]) -> None:

assert self.transport is not None

code = self.transport.get_returncode()

assert code is not None, "connect lost, but got no returncode"

LOGGER.debug("%s: Connection lost (exit code: %d, error: %s)", self, code, exc)

# Terminate commands.

command, self.command = self.command, None

next_command, self.next_command = self.next_command, None

if command:

command._engine_terminated(code)

if next_command:

next_command._engine_terminated(code)

self.returncode.set_result(code)

def process_exited(self) -> None:

LOGGER.debug("%s: Process exited", self)

def send_line(self, line: str) -> None:

LOGGER.debug("%s: << %s", self, line)

assert self.transport is not None, "cannot send line before connection is made"

stdin = self.transport.get_pipe_transport(0)

# WriteTransport expected, but not checked to allow duck typing.

stdin.write((line + "\n").encode("utf-8")) # type: ignore

def pipe_data_received(self, fd: int, data: Union[bytes, str]) -> None:

self.buffer[fd].extend(data) # type: ignore

while b"\n" in self.buffer[fd]:

line_bytes, self.buffer[fd] = self.buffer[fd].split(b"\n", 1)

if line_bytes.endswith(b"\r"):

line_bytes = line_bytes[:-1]

try:

line = line_bytes.decode("utf-8")

except UnicodeDecodeError as err:

LOGGER.warning("%s: >> %r (%s)", self, bytes(line_bytes), err)

else:

if fd == 1:

self._line_received(line)

else:

self.error_line_received(line)

def error_line_received(self, line: str) -> None:

LOGGER.warning("%s: stderr >> %s", self, line)

def _line_received(self, line: str) -> None:

LOGGER.debug("%s: >> %s", self, line)

self.line_received(line)

if self.command:

self.command._line_received(line)

def line_received(self, line: str) -> None:

pass

async def communicate(self, command_factory: Callable[[Self], BaseCommand[T]]) -> T:

command = command_factory(self)

if self.returncode.done():

raise EngineTerminatedError(f"engine process dead (exit code: {self.returncode.result()})")

assert command.state == CommandState.NEW, command.state

if self.next_command is not None:

self.next_command.result.cancel()

self.next_command.finished.cancel()

self.next_command.set_finished()

self.next_command = command

def previous_command_finished() -> None:

self.command, self.next_command = self.next_command, None

if self.command is not None:

cmd = self.command

def cancel_if_cancelled(result: asyncio.Future[T]) -> None:

if result.cancelled():

cmd._cancel()

cmd.result.add_done_callback(cancel_if_cancelled)

cmd._start()

cmd.add_finished_callback(previous_command_finished)

if self.command is None:

previous_command_finished()

elif not self.command.result.done():

self.command.result.cancel()

elif not self.command.result.cancelled():

self.command._cancel()

return await command.result

def __repr__(self) -> str:

pid = self.transport.get_pid() if self.transport is not None else "?"

return f"<{type(self).__name__} (pid={pid})>"

@property

@abc.abstractmethod

def options(self) -> MutableMapping[str, Option]:

"""Dictionary of available options."""

@abc.abstractmethod

async def initialize(self) -> None:

"""Initializes the engine."""

@abc.abstractmethod

async def ping(self) -> None:

"""

Pings the engine and waits for a response. Used to ensure the engine

is still alive and idle.

"""

@abc.abstractmethod

async def configure(self, options: ConfigMapping) -> None:

"""

Configures global engine options.

:param options: A dictionary of engine options where the keys are

names of :data:`~chess.engine.Protocol.options`. Do not set options