chess0.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. import time
  2. import chess
  3. import chess.svg
  4. import chess.engine
  5. class GameOver(Exception):
  6. pass
  7. class IllegalMove(Exception):
  8. pass
  9. class ChessSession:
  10. def __init__(self, engine):
  11. self.engine = engine
  12. self.board = chess.Board()
  13. self.ts = time.time()
  14. self.move_ts = time.time()
  15. def check_game_over(self):
  16. if self.board.is_game_over():
  17. outcome = self.board.outcome()
  18. raise GameOver(f"{outcome.result()} {outcome.termination.name}")
  19. elif self.board.status() & chess.STATUS_NO_WHITE_KING:
  20. raise GameOver(f"0-1 CHECKMATE")
  21. elif self.board.status() & chess.STATUS_NO_BLACK_KING:
  22. raise GameOver(f"1-0 CHECKMATE")
  23. elif not self.board.is_valid():
  24. raise GameOver(f"Некорректное состояние доски.")
  25. async def move(self, move=None):
  26. if move is not None:
  27. try:
  28. move = chess.Move.from_uci(move)
  29. except chess.InvalidMoveError:
  30. raise IllegalMove(move)
  31. if move not in self.board.legal_moves:
  32. raise IllegalMove(move)
  33. else:
  34. move = await self.engine.play(self.board, chess.engine.Limit(nodes=1))
  35. move = move.move
  36. self.board.push(move)
  37. self.move_ts = time.time()
  38. self.check_game_over()
  39. def skip(self):
  40. self.board.push(chess.Move.null())
  41. self.move_ts = time.time()
  42. class ChessManager:
  43. def __init__(self, engine_path):
  44. self.engine_path = engine_path
  45. self.engine = None
  46. self.sessions = {}
  47. async def cleanup(self):
  48. for id in list(self.sessions.keys()):
  49. session = self.sessions[id]
  50. if (
  51. time.time() - session.move_ts >= 60 * 60 * 12
  52. or time.time() - session.ts >= 60 * 60 * 24 * 7
  53. ):
  54. await self.end(id)
  55. async def begin(self, id):
  56. if id in self.sessions:
  57. await self.end(id)
  58. if not self.engine:
  59. _, self.engine = await chess.engine.popen_uci(self.engine_path)
  60. self.sessions[id] = ChessSession(self.engine)
  61. async def end(self, id):
  62. session = self.sessions.get(id)
  63. if not session:
  64. return False
  65. del self.sessions[id]
  66. return True
  67. async def move(self, id, move=None):
  68. session = self.sessions.get(id)
  69. if not session:
  70. raise KeyError(id)
  71. await session.move(move)
  72. def undo(self, id):
  73. session = self.sessions.get(id)
  74. if not session:
  75. raise KeyError(id)
  76. session.board.pop()
  77. def is_check(self, id):
  78. session = self.sessions.get(id)
  79. if not session:
  80. raise KeyError(id)
  81. return session.board.is_check()
  82. async def skip(self, id):
  83. session = self.sessions.get(id)
  84. if not session:
  85. raise KeyError(id)
  86. session.skip()
  87. def svg(self, id, size=256):
  88. session = self.sessions.get(id)
  89. if not session:
  90. raise KeyError(id)
  91. arrows = []
  92. if session.board.move_stack:
  93. arrows.append((session.board.move_stack[-1].from_square, session.board.move_stack[-1].to_square))
  94. if len(session.board.move_stack) > 1 and session.board.move_stack[-2] != chess.Move.null():
  95. arrows.append((session.board.move_stack[-2].from_square, session.board.move_stack[-2].to_square))
  96. move = None
  97. if session.board.attackers_mask(chess.WHITE, session.board.king(chess.BLACK)) != 0:
  98. move = session.board.king(chess.BLACK)
  99. if session.board.is_check():
  100. return chess.svg.board(session.board, size=size, lastmove=move, arrows=arrows, fill={checker: "#cc0000cc" for checker in session.board.checkers()}, check=session.board.king(session.board.turn))
  101. return chess.svg.board(session.board, size=size, lastmove=move, arrows=arrows)
  102. def ascii(self, id):
  103. session = self.sessions.get(id)
  104. if not session:
  105. raise KeyError(id)
  106. return str(session.board)
  107. def has_moves(self, id):
  108. session = self.sessions.get(id)
  109. if not session:
  110. raise KeyError(id)
  111. return bool(session.board.move_stack)
  112. def moves_count(self, id):
  113. session = self.sessions.get(id)
  114. if not session:
  115. raise KeyError(id)
  116. return len(session.board.move_stack)
  117. def moves(self, id, offset=None):
  118. session = self.sessions.get(id)
  119. if not session:
  120. raise KeyError(id)
  121. if offset is not None:
  122. moves = session.board.move_stack[-offset:]
  123. else:
  124. moves = session.board.move_stack
  125. return " ".join(map(lambda m: "пропуск" if m == chess.Move.null() else str(m), moves))