Merge pull request #998 from MarkZH/send-result

Send game result to engines

chess/engine.py

@@ -1266,6 +1266,31 @@ async def analysis(self, board: chess.Board, limit: Optional[Limit] = None, *, m
         and stopping the analysis at any time.
         """
 
+    @abc.abstractmethod
+    async def send_game_result(self, board: chess.Board, winner: Optional[Color] = None, game_ending: Optional[str] = None, game_complete: bool = True) -> None:
+        """
+        Sends the engine the result of the game.
+
+        XBoard engines receive the final moves and a line of the form
+        "result <winner> {<ending>}". The <winner> field is one of "1-0",
+        "0-1", "1/2-1/2", or "*" to indicate white won, black won, draw,
+        or adjournment, respectively. The <ending> field is a description
+        of the specific reason for the end of the game: "White mates",
+        "Time forfeiture", "Stalemate", etc.
+
+        UCI engines do not expect end-of-game information and so are not
+        sent anything.
+
+        :param board: The final state of the board.
+        :param winner: Optional. Specify the winner of the game. This is useful
+            if the result of the game is not evident from the board--e.g., time
+            forfeiture or draw by agreement.
+        :param game_ending: Optional. Text describing the reason for the game
+            ending. Similarly to the winner paramter, this overrides any game
+            result derivable from the board.
+        :param game_complete: Optional. Whether the game reached completion.
+        """
+
     @abc.abstractmethod
     async def quit(self) -> None:
         """Asks the engine to shut down."""
@@ -1817,6 +1842,9 @@ def engine_terminated(self, engine: UciProtocol, exc: Exception) -> None:
 
         return await self.communicate(UciAnalysisCommand)
 
+    async def send_game_result(self, board: chess.Board, winner: Optional[Color] = None, game_ending: Optional[str] = None, game_complete: bool = True) -> None:
+        pass
+
     async def quit(self) -> None:
         self.send_line("quit")
         await asyncio.shield(self.returncode)
@@ -2502,6 +2530,41 @@ def _opponent_configuration(self, *, opponent: Optional[Opponent] = None, engine
     async def send_opponent_information(self, *, opponent: Optional[Opponent] = None, engine_rating: Optional[int] = None) -> None:
         return await self.configure(self._opponent_configuration(opponent=opponent, engine_rating=engine_rating))
 
+    async def send_game_result(self, board: chess.Board, winner: Optional[Color] = None, game_ending: Optional[str] = None, game_complete: bool = True) -> None:
+        class XBoardGameResultCommand(BaseCommand[XBoardProtocol, None]):
+            def start(self, engine: XBoardProtocol) -> None:
+                if game_ending and any(c in game_ending for c in "{}\n\r"):
+                    raise EngineError(f"invalid line break or curly braces in game ending message: {game_ending!r}")
+
+                engine._new(board, engine.game, {})  # Send final moves to engine.
+
+                outcome = board.outcome(claim_draw=True)
+
+                if not game_complete:
+                    result = "*"
+                    ending = game_ending or ""
+                elif winner is not None or game_ending:
+                    result = "1-0" if winner == chess.WHITE else "0-1" if winner == chess.BLACK else "1/2-1/2"
+                    ending = game_ending or ""
+                elif outcome is not None and outcome.winner is not None:
+                    result = outcome.result()
+                    winning_color = "White" if outcome.winner == chess.WHITE else "Black"
+                    is_checkmate = outcome.termination == chess.Termination.CHECKMATE
+                    ending = f"{winning_color} {'mates' if is_checkmate else 'variant win'}"
+                elif outcome is not None:
+                    result = outcome.result()
+                    ending = outcome.termination.name.capitalize().replace("_", " ")
+                else:
+                    result = "*"
+                    ending = ""
+
+                ending_text = f"{{{ending}}}" if ending else ""
+                engine.send_line(f"result {result} {ending_text}".strip())
+                self.result.set_result(None)
+                self.set_finished()
+
+        return await self.communicate(XBoardGameResultCommand)
+
     async def quit(self) -> None:
         self.send_line("quit")
         await asyncio.shield(self.returncode)
@@ -2929,6 +2992,12 @@ def analysis(self, board: chess.Board, limit: Optional[Limit] = None, *, multipv
             future = asyncio.run_coroutine_threadsafe(coro, self.protocol.loop)
         return SimpleAnalysisResult(self, future.result())
 
+    def send_game_result(self, board: chess.Board, winner: Optional[Color] = None, game_ending: Optional[str] = None, game_complete: bool = True) -> None:
+        with self._not_shut_down():
+            coro = asyncio.wait_for(self.protocol.send_game_result(board, winner, game_ending, game_complete), self.timeout)
+            future = asyncio.run_coroutine_threadsafe(coro, self.protocol.loop)
+        return future.result()
+
     def quit(self) -> None:
         with self._not_shut_down():
             coro = asyncio.wait_for(self.protocol.quit(), self.timeout)

test.py

@@ -3466,6 +3466,32 @@ def test_uci_info(self):
         info = chess.engine._parse_uci_info("depth 1 seldepth 2 time 16 nodes 1 score cp 72 wdl 249 747 4 hashfull 0 nps 400 tbhits 0 multipv 1", board)
         self.assertEqual(info["wdl"], (249, 747, 4))
 
+    def test_uci_result(self):
+        async def main():
+            protocol = chess.engine.UciProtocol()
+            mock = chess.engine.MockTransport(protocol)
+
+            mock.expect("uci", ["uciok"])
+            await protocol.initialize()
+            mock.assert_done()
+
+            limit = chess.engine.Limit(time=5)
+            checkmate_board = chess.Board("k7/7R/6R1/8/8/8/8/K7 w - - 0 1")
+
+            mock.expect("ucinewgame")
+            mock.expect("isready", ["readyok"])
+            mock.expect("position fen k7/7R/6R1/8/8/8/8/K7 w - - 0 1")
+            mock.expect("go movetime 5000", ["bestmove g6g8"])
+            result = await protocol.play(checkmate_board, limit, game="checkmate")
+            self.assertEqual(result.move, checkmate_board.parse_uci("g6g8"))
+            checkmate_board.push(result.move)
+            self.assertTrue(checkmate_board.is_checkmate())
+            await protocol.send_game_result(checkmate_board)
+            mock.assert_done()
+
+        asyncio.set_event_loop_policy(chess.engine.EventLoopPolicy())
+        asyncio.run(main())
+
     def test_hiarcs_bestmove(self):
         async def main():
             protocol = chess.engine.UciProtocol()
@@ -3665,6 +3691,96 @@ async def main():
         asyncio.set_event_loop_policy(chess.engine.EventLoopPolicy())
         asyncio.run(main())
 
+    def test_xboard_result(self):
+        async def main():
+            protocol = chess.engine.XBoardProtocol()
+            mock = chess.engine.MockTransport(protocol)
+
+            mock.expect("xboard")
+            mock.expect("protover 2", ["feature ping=1 setboard=1 done=1"])
+            await protocol.initialize()
+            mock.assert_done()
+
+            limit = chess.engine.Limit(time=5)
+            checkmate_board = chess.Board("k7/7R/6R1/8/8/8/8/K7 w - - 0 1")
+
+            mock.expect("new")
+            mock.expect("force")
+            mock.expect("setboard k7/7R/6R1/8/8/8/8/K7 w - - 0 1")
+            mock.expect("st 5")
+            mock.expect("nopost")
+            mock.expect("easy")
+            mock.expect("go", ["move g6g8"])
+            mock.expect_ping()
+            mock.expect("force")
+            mock.expect("result 1-0 {White mates}")
+            result = await protocol.play(checkmate_board, limit, game="checkmate")
+            self.assertEqual(result.move, checkmate_board.parse_uci("g6g8"))
+            checkmate_board.push(result.move)
+            self.assertTrue(checkmate_board.is_checkmate())
+            await protocol.send_game_result(checkmate_board)
+            mock.assert_done()
+
+            unfinished_board = chess.Board()
+            mock.expect("new")
+            mock.expect("force")
+            mock.expect("st 5")
+            mock.expect("nopost")
+            mock.expect("easy")
+            mock.expect("go", ["move e2e4"])
+            mock.expect_ping()
+            mock.expect("force")
+            mock.expect("result *")
+            result = await protocol.play(unfinished_board, limit, game="unfinished")
+            self.assertEqual(result.move, unfinished_board.parse_uci("e2e4"))
+            unfinished_board.push(result.move)
+            await protocol.send_game_result(unfinished_board, game_complete=False)
+            mock.assert_done()
+
+            timeout_board = chess.Board()
+            mock.expect("new")
+            mock.expect("force")
+            mock.expect("st 5")
+            mock.expect("nopost")
+            mock.expect("easy")
+            mock.expect("go", ["move e2e4"])
+            mock.expect_ping()
+            mock.expect("force")
+            mock.expect("result 0-1 {Time forfeiture}")
+            result = await protocol.play(timeout_board, limit, game="timeout")
+            self.assertEqual(result.move, timeout_board.parse_uci("e2e4"))
+            timeout_board.push(result.move)
+            await protocol.send_game_result(timeout_board, chess.BLACK, "Time forfeiture")
+            mock.assert_done()
+
+            error_board = chess.Board()
+            mock.expect("new")
+            mock.expect("force")
+            mock.expect("st 5")
+            mock.expect("nopost")
+            mock.expect("easy")
+            mock.expect("go", ["move e2e4"])
+            mock.expect_ping()
+            result = await protocol.play(error_board, limit, game="error")
+            self.assertEqual(result.move, error_board.parse_uci("e2e4"))
+            error_board.push(result.move)
+            for c in "\n\r{}":
+                with self.assertRaises(chess.engine.EngineError):
+                    await protocol.send_game_result(error_board, chess.BLACK, f"Time{c}forfeiture")
+            mock.assert_done()
+
+            material_board = chess.Board("k7/8/8/8/8/8/8/K7 b - - 0 1")
+            self.assertTrue(material_board.is_insufficient_material())
+            mock.expect("new")
+            mock.expect("force")
+            mock.expect("setboard k7/8/8/8/8/8/8/K7 b - - 0 1")
+            mock.expect("result 1/2-1/2 {Insufficient material}")
+            await protocol.send_game_result(material_board)
+            mock.assert_done()
+
+        asyncio.set_event_loop_policy(chess.engine.EventLoopPolicy())
+        asyncio.run(main())
+
     def test_xboard_analyse(self):
         async def main():
             protocol = chess.engine.XBoardProtocol()