-
Notifications
You must be signed in to change notification settings - Fork 8
/
lishogibot.py
222 lines (189 loc) · 7.98 KB
/
lishogibot.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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
import argparse
import shogi
import engine_wrapper
import model
import json
import lishogi
import logging
import multiprocessing
from multiprocessing import Process
import signal
import backoff
from config import load_config
from conversation import Conversation, ChatLine
from functools import partial
from requests.exceptions import HTTPError, ReadTimeout
from util import switchusiuci,makeusi,makesfenfromfen
logger = logging.getLogger(__name__)
try:
from http.client import RemoteDisconnected
# New in version 3.5: Previously, BadStatusLine('') was raised.
except ImportError:
from http.client import BadStatusLine as RemoteDisconnected
__version__ = "0.6.0"
terminated = False
def signal_handler(signal, frame):
global terminated
logger.debug("Recieved SIGINT. Terminating client.")
terminated = True
signal.signal(signal.SIGINT, signal_handler)
def is_final(exception):
return isinstance(exception, HTTPError) and exception.response.status_code < 500
def upgrade_account(li):
if li.upgrade_to_bot_account() is None:
return False
logger.info("Succesfully upgraded to Bot Account!")
return True
def watch_control_stream(control_queue, li):
logger.info("start")
while not terminated:
try:
response = li.get_event_stream()
lines = response.iter_lines()
for line in lines:
if line:
event = json.loads(line.decode('utf-8'))
control_queue.put_nowait(event)
logger.info(event)
except:
logger.info("except")
pass
def start(li, user_profile, engine_factory, config):
challenge_config = config["challenge"]
logger.info("You're now connected to {} and awaiting challenges.".format(config["url"]))
control_queue=multiprocessing.Manager().Queue()
control_stream = Process(target=watch_control_stream, args=[control_queue,li])
control_stream.start()
while not terminated:
event=control_queue.get()
if event["type"] == "terminated":
break
elif event["type"] == "challenge":
logger.info("chlng detected")
chlng = model.Challenge(event["challenge"])
if chlng.is_supported(challenge_config):
logger.info("chlng supported")
try:
logger.info(" Accept {}".format(chlng))
response = li.accept_challenge(chlng.id)
logger.info(chlng.id)
except (HTTPError, ReadTimeout) as exception:
if isinstance(exception, HTTPError) and exception.response.status_code == 404: # ignore missing challenge
logger.info(" Skip missing {}".format(chlng))
else:
try:
li.decline_challenge(chlng.id)
logger.info(" Decline {}".format(chlng))
except:
pass
elif event["type"] == "gameStart":
logger.info("game detected")
game_id = event["game"]["id"]
play_game(li, game_id, engine_factory, user_profile, config)
logger.info("Terminated")
control_stream.terminate()
control_stream.join()
ponder_results = {}
@backoff.on_exception(backoff.expo, BaseException, max_time=600, giveup=is_final)
def play_game(li, game_id, engine_factory, user_profile, config):
response = li.get_game_stream(game_id)
lines = response.iter_lines()
#Initial response of stream will be the full game info. Store it
initial_state = json.loads(next(lines).decode('utf-8'))
game = model.Game(initial_state, user_profile["username"], li.baseUrl, config.get("abort_time", 20))
board = setup_board(game)
engineeng = engine_factory(board)
logger.info("+++ {}".format(game))
if is_engine_move(game,board.move_stack) and not is_game_over(game):
move=engineeng.search(board,5000,5000,1,1)
finalmove=switchusiuci(move)
board.push(shogi.Move.from_usi(move))
li.make_move(game.id, finalmove)
while not terminated:
try:
binary_chunk = next(lines)
except(StopIteration):
break
upd = json.loads(binary_chunk.decode('utf-8')) if binary_chunk else None
u_type = upd["type"] if upd else "ping"
if not board.is_game_over():
if u_type == "gameState":
game.state=upd
moves = upd["moves"].split()
board = update_board(board, moves[-1])
if not is_game_over(game) and is_engine_move(game, moves):
move=engineeng.search(board,upd['wtime'],upd['btime'],upd['winc'],upd['binc'])
finalmove=switchusiuci(move)
board.push(shogi.Move.from_usi(move))
li.make_move(game.id, finalmove)
if board.turn == shogi.BLACK:
game.ping(config.get("abort_time", 20), (upd["btime"] + upd["binc"]) / 1000 + 60)
else:
game.ping(config.get("abort_time", 20), (upd["wtime"] + upd["winc"]) / 1000 + 60)
elif u_type == "ping":
if game.should_abort_now():
logger.info(" Aborting {} by lack of activity".format(game.url()))
li.abort(game.id)
break
elif game.should_terminate_now():
logger.info(" Terminating {} by lack of activity".format(game.url()))
if game.is_abortable():
li.abort(game.id)
break
else:
logger.info("game over")
engineeng.quit()
break
def is_white_to_move(game, moves):
return len(moves) % 2 == (0 if game.white_starts else 1)
def setup_board(game):
if game.variant_name == "From Position":
board = shogi.Board(makesfenfromfen(game.initial_fen))
else:
board = shogi.Board() # Standard
moves = game.state["moves"].split()
for move in moves:
board = update_board(board, move)
return board
def is_engine_move(game, moves):
return game.is_white == is_white_to_move(game, moves)
def is_game_over(game):
return game.state["status"] != "started"
def update_board(board, move):
usi_move = shogi.Move.from_usi(makeusi(move))
if board.is_legal(usi_move):
board.push(usi_move)
else:
logger.debug('Ignoring illegal move {} on board {}'.format(makeusi(move), board.sfen()))
return board
def intro():
return r"""
. _/\_
. //o o\\
. || || lishogi-bot %s
. || ||
. ||____|| Play on Lishogi with a bot
""" % __version__
if __name__=="__main__":
parser = argparse.ArgumentParser(description='Play on Lishogi with a bot')
parser.add_argument('-u', action='store_true', help='Add this flag to upgrade your account to a bot account.')
parser.add_argument('-v', action='store_true', help='Verbose output. Changes log level from INFO to DEBUG.')
parser.add_argument('--config', help='Specify a configuration file (defaults to ./config.yml)')
parser.add_argument('-l', '--logfile', help="Log file to append logs to.", default=None)
args = parser.parse_args()
logging.basicConfig(level=logging.DEBUG if args.v else logging.INFO, filename=args.logfile,
format="%(asctime)-15s: %(message)s")
logger.info(intro())
CONFIG = load_config(args.config or "./config.yml")
li = lishogi.Lishogi(CONFIG["token"], CONFIG["url"], __version__)
user_profile = li.get_profile()
username = user_profile["username"]
is_bot = user_profile.get("title") == "BOT"
logger.info("Welcome {}!".format(username))
if is_bot is False:
is_bot = upgrade_account(li)
if is_bot:
engine_factory = partial(engine_wrapper.create_engine, CONFIG)
start(li, user_profile, engine_factory, CONFIG)
else:
logger.error("{} is not a bot account. Please upgrade it to a bot account!".format(user_profile["username"]))