Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def score(self, board):
try:
info = self.engine.analyse(board, chess.engine.Limit(time=0.500))
cp = chess.engine.PovScore(info['score'], chess.WHITE).pov(chess.WHITE).relative.score()
return cp
except chess.engine.EngineTerminatedError:
return None
if (key == 'bm' or key == 'am'):
san_moves = []
for move in position.ops[key]:
try:
san_moves.append(position.board.san(move))
except:
print(key + " solution move " + str(i) + " could not be parsed",file=sys.stderr, flush=True)
i = i + 1
position.ops[key] = san_moves
results.solution_time = 0
results.solution_nodes = 0
results.bestmove = None
results.infos = {}
with engine.analysis(board=position.board,
limit=chess.engine.Limit(time=options.search_time),multipv=options.multi_pv_value) as analysis:
for info in analysis:
process_info(info,results)
# print last group of results
for i in range(1,options.multi_pv_value+1):
group = 0
infos = results.infos[i]
for key in ["depth","seldepth","multipv","score","nodes",
"nps","hashfull","tbhits","time","pv"]:
if key in infos:
if (group != 0):
print(" ",end="")
print(key + " ",end="")
if (key == "pv"):
fen = position.board.fen()
for m in infos[key]:
san = position.board.san(m)
def _readyok(self, engine: UciProtocol) -> None:
self.sent_isready = False
engine._position(board)
if limit:
engine._go(limit, root_moves=root_moves)
else:
engine._go(Limit(), root_moves=root_moves, infinite=True)
self.result.set_result(self.analysis)
# If we use "go infinite" we stop the search by time and depth
if not is_time_check and \
time.time() - start_thinking_time >= self.max_time:
logging.info('Max time limit is reached.')
is_time_check = True
break
if 'depth' in info:
if int(info['depth']) >= self.max_depth:
logging.info('Max depth limit is reached.')
break
except:
logging.info('Error in parsing engine search info')
else:
start_thinking_time = time.time()
result = self.engine.play(self.board, chess.engine.Limit(time=self.max_time,
depth=self.max_depth))
self.bm = result.move
# Apply engine move delay
while True:
if time.time() - start_thinking_time >= self.move_delay_sec:
break
logging.info('Delay sending of best move')
time.sleep(0.25)
self.eng_queue.put('bestmove {}' .format(self.bm))
logging.info('bestmove {}'.format(self.bm))
def isThereMate(board, moves, engine):
info = engine.analyse(board.board, chess.engine.Limit(time=0.005))
if str(info["score"])[0] == "#":
#print("mate found")
move = str(info["pv"][0])
# find where this move is in the possible actions
for i in range(len(moves)):
if moves[i] == move:
#print("found index")
return i
#print("found mate but not index")
return None
async def match_task(self):
self.match_message = await self.ctx.embed_send("Loading..")
await self.update_match_embed()
while True:
player = [self.black_player, self.white_player][int(self.turn)]
embed = self.match_message.embeds[0]
if player == self.bot.user:
await self.match_message.edit(embed = embed.set_footer(text = "I'm thinking.."))
result = await self.chess_engine.play(self, chess.engine.Limit(time = 2))
self.push(result.move)
await self.update_match_embed(footer_text = f"I moved {result.move}")
else:
message = await self.bot.wait_for("message",
check = lambda msg: msg.author == player and
msg.channel == self.ctx.channel and
self.valid_move(msg.content))
await self.match_message.edit(embed = embed.set_footer(text = "Processing move.."))
self.make_move(message.content)
if self.is_game_over():
footer_text = discord.Embed.Empty
else:
footer_text = f"It is {['black', 'white'][int(self.turn)]}'s ({[self.black_player, self.white_player][int(self.turn)]}'s) turn to move"
await self.update_match_embed(footer_text = footer_text)
await self.bot.attempt_delete_message(message)
def _analyze(board, depth, **kwargs) -> Union[List[InfoDict], InfoDict]:
try:
info = AnalysisEngine.instance().analyse(board, Limit(depth=depth), **kwargs)
except EngineTerminatedError:
log(Color.RED, "Analysis engine crashed... restarting")
AnalysisEngine.quit()
info = AnalysisEngine._analyze(board, depth, **kwargs)
return info