forked from Torom/BotLi
-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathgame.py
executable file
·180 lines (143 loc) · 6.74 KB
/
game.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
from datetime import datetime, timedelta
from queue import Queue
from threading import Event, Thread
from api import API
from botli_dataclasses import Game_Information
from chatter import Chatter
from enums import Game_Status
from lichess_game import Lichess_Game
class Game(Thread):
def __init__(self, config: dict, api: API, game_id: str, game_finished_event: Event) -> None:
Thread.__init__(self)
self.config = config
self.api = api
self.game_id = game_id
self.game_finished_event = game_finished_event
self.lichess_game: Lichess_Game | None = None
self.chatter: Chatter | None = None
self.game_info: Game_Information | None = None
def start(self):
Thread.start(self)
def run(self) -> None:
game_queue = Queue()
game_queue_thread = Thread(target=self.api.get_game_stream, args=(self.game_id, game_queue), daemon=True)
game_queue_thread.start()
self.game_info = Game_Information.from_gameFull_event(game_queue.get(), self.api.username)
self._print_game_information()
self.lichess_game = Lichess_Game(self.api, self.game_info, self.config)
self.chatter = Chatter(self.api, self.config, self.game_info, self.lichess_game)
self.chatter.send_greetings()
if self._finish_game(self.game_info.state.get('winner')):
self.lichess_game.end_game()
return
if self.lichess_game.is_our_turn:
self._make_move()
else:
self.lichess_game.start_pondering()
abortion_seconds = 30.0 if self.game_info.opponent_is_bot else 60.0
abortion_time = datetime.now() + timedelta(seconds=abortion_seconds)
while True:
event = game_queue.get()
if event['type'] not in ['gameFull', 'gameState']:
if self.lichess_game.is_abortable and datetime.now() >= abortion_time:
print('Aborting game ...')
self.api.abort_game(self.game_id)
self.chatter.send_abortion_message()
if event['type'] == 'gameFull':
self.lichess_game.update(event['state'])
if self._finish_game(event['state'].get('winner')):
break
if self.lichess_game.is_our_turn:
self._make_move()
else:
self.lichess_game.start_pondering()
elif event['type'] == 'gameState':
updated = self.lichess_game.update(event)
if self._finish_game(event.get('winner')):
break
if self.lichess_game.is_game_over:
continue
if self.lichess_game.is_our_turn and updated:
self._make_move()
elif event['type'] == 'chatLine':
self.chatter.handle_chat_message(event)
elif event['type'] == 'opponentGone':
continue
elif event['type'] == 'ping':
continue
else:
print(event)
self.lichess_game.end_game()
self.game_finished_event.set()
def _make_move(self) -> None:
assert self.lichess_game
assert self.chatter
uci_move, offer_draw, resign = self.lichess_game.make_move()
if resign:
self.api.resign_game(self.game_id)
else:
self.api.send_move(self.game_id, uci_move, offer_draw)
self.chatter.print_eval()
def _finish_game(self, winner: str | None) -> bool:
assert self.lichess_game
assert self.chatter
if self.lichess_game.is_finished:
self._print_result_message(winner)
self.chatter.send_goodbyes()
return True
return False
def _print_game_information(self) -> None:
assert self.game_info
opponents_str = f'{self.game_info.white_str} - {self.game_info.black_str}'
delimiter = 5 * ' '
print()
print(delimiter.join([self.game_info.id_str, opponents_str, self.game_info.tc_str,
self.game_info.rated_str, self.game_info.variant_str]))
print(128 * '‾')
def _print_result_message(self, winner: str | None) -> None:
assert self.lichess_game
assert self.game_info
winning_name = self.game_info.white_name if winner == 'white' else self.game_info.black_name
winning_title = self.game_info.white_title if winner == 'white' else self.game_info.black_title
losing_name = self.game_info.white_name if winner == 'black' else self.game_info.black_name
losing_title = self.game_info.white_title if winner == 'black' else self.game_info.black_title
if winner:
if winner == 'white':
white_result = '1'
black_result = '0'
else:
white_result = '0'
black_result = '1'
message = f'{winning_title}{" " if winning_title else ""}{winning_name} won'
if self.lichess_game.status == Game_Status.MATE:
message += ' by checkmate!'
elif self.lichess_game.status == Game_Status.OUT_OF_TIME:
message += f'! {losing_title}{" " if losing_title else ""}{losing_name} ran out of time.'
elif self.lichess_game.status == Game_Status.RESIGN:
message += f'! {losing_title}{" " if losing_title else ""}{losing_name} resigned.'
elif self.lichess_game.status == Game_Status.VARIANT_END:
message += ' by variant rules!'
else:
white_result = '½'
black_result = '½'
if self.lichess_game.status == Game_Status.DRAW:
if self.lichess_game.board.is_fifty_moves():
message = 'Game drawn by 50-move rule.'
elif self.lichess_game.board.is_repetition():
message = 'Game drawn by threefold repetition.'
elif self.lichess_game.board.is_insufficient_material():
message = 'Game drawn due to insufficient material.'
elif self.lichess_game.board.is_variant_draw():
message = 'Game drawn by variant rules.'
else:
message = 'Game drawn by agreement.'
elif self.lichess_game.status == Game_Status.STALEMATE:
message = 'Game drawn by stalemate.'
else:
message = 'Game aborted.'
white_result = 'X'
black_result = 'X'
opponents_str = f'{self.game_info.white_str} {white_result} - {black_result} {self.game_info.black_str}'
delimiter = 5 * ' '
print(delimiter.join([self.game_info.id_str, opponents_str, message]))
print(128 * '‾')