Skip to content

Instantly share code, notes, and snippets.

@AndyGrant AndyGrant/texel.py
Last active Jan 13, 2019

Embed
What would you like to do?
# Ethereal is a UCI chess playing engine authored by Andrew Grant.
# <https://github.com/AndyGrant/Ethereal> <andrew@grantnet.us>
#
# Ethereal is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ethereal is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import chess.pgn
import os, time, random, multiprocessing
## General Configuration
THREADS = 30
INPUT_NAME = "games.pgn"
OUTPUT_NAME = "games.fens"
## Draw Adjucation Configuration
ADJUDICATE_DRAW_SCORE = 5
ADJUDICATE_DRAW_PLY = 16
ADJUDICATE_DRAW_START = 40
## Win Adjucation Configuration
ADJUDICATE_WIN_SCORE = 250
ADJUDICATE_WIN_PLY = 6
ADJUDICATE_WIN_START = 0
## Selectivity Configuration
SHORTEST_GAME = 40
FENS_PER_GAME = 4
class Visitor(chess.pgn.BaseVisitor):
def begin_game(self):
self.fens = []; self.evals = []
def visit_comment(self, comment):
evalstr = comment.split("/")[0]
if "M" in evalstr: self.evals.append(32000)
else: self.evals.append(int(100 * float(evalstr)))
def visit_move(self, board, move):
self.fens.append(board.fen())
def splitPGN():
# Split games into THREADS-lists
games = parseGamesFromPGN()
pieces = [[] for f in range(THREADS)]
for ii, game in enumerate(games):
pieces[ii % THREADS].append(game)
# Output to SPLIT_OUT_N for each piece
for ii, piece in enumerate(pieces):
with open("SPLIT_OUT_{0}.pgn".format(ii), "w") as fout:
for game in piece:
for line in game:
fout.write(line)
time.sleep(5) # Give time to handle files
def parseGamesFromPGN():
# Parse each game from the PGN
games = []; tokens = []; count = 0
with open(INPUT_NAME, "r") as pgn:
# Parsing into individual games
for line in pgn.readlines():
# python-chess expects half/full move counter
if line.startswith("[FEN"):
if len(line.split(" ")) <= 6:
line = line.replace("\"]", " 0 1\"]")
tokens.append(line)
# Count empty lines to check for new games
if line.strip() == "":
count += 1
# Second empty line denotes a new game
if count == 2:
games.append(tokens[::])
tokens = []
count = 0
return games
def parseFENSFromPGNS(id):
accepted = rejected = 0; outputs = []
# Parse only games from our designated PGN split
with open("SPLIT_OUT_{0}.pgn".format(id), "r") as pgn:
# Parse all games from the PGN
while True:
# Grab the next game in the PGN
game = chess.pgn.read_game(pgn)
if game == None: break
# Skip PGNs with strange end results (crashes, timelosses, disconnects, ...)
if "Termination" in game.headers and game.headers["Termination"] != "adjudication":
rejected += 1
continue
# Collect each position and evaluation
visitor = Visitor(); game.accept(visitor)
# Save off each FEN until we hit adjudication
fens = []; draw = 0; win = 0; ply = 0
for fen, eval in zip(visitor.fens, visitor.evals):
ply += 1 # Update ply count to know game length
# Update consecutive plys which would fall under a draw
if ply >= ADJUDICATE_DRAW_START and abs(eval) <= ADJUDICATE_DRAW_SCORE: draw += 1
else: draw = 0
# Update consecutive plys which would fall under a win
if ply >= ADJUDICATE_WIN_START and abs(eval) >= ADJUDICATE_WIN_SCORE : win += 1
else: win = 0
# Check for either form of adjudication
if draw >= ADJUDICATE_DRAW_PLY or win >= ADJUDICATE_WIN_PLY: break
# Ensure long enough and big enough sample
if ply < SHORTEST_GAME or ply < FENS_PER_GAME:
rejected += 1
continue;
# Sample FENS_PER_GAME times and save the position
for fen in random.sample(visitor.fens[:ply], FENS_PER_GAME):
outputs.append("{0} {1}\n".format(fen, game.headers["Result"]))
# No criteria met to skip this game
accepted += 1;
# Output FENS to a thread specific file
with open("SPLIT_PARSE_{0}.fen".format(id), "w") as fout:
for fen in outputs:
fout.write(fen)
# Final stat reporting
print "Thread # {0:<2} Accepted {1} Rejected {2}".format(id, accepted, rejected)
def buildTexelBook():
splitPGN() # Split main file into THREADS-pieces
processes = [] # Process for each PGN parser
# Launch all of the procsses
for ii in range(THREADS):
processes.append(
multiprocessing.Process(
target=parseFENSFromPGNS, args=(ii,)))
# Wait for each parser to finish
for p in processes: p.start()
for p in processes: p.join()
# Build final FEN file from process outputs
os.system("rm {0}".format(OUTPUT_NAME))
os.system("touch {0}".format(OUTPUT_NAME))
for ii in range(THREADS):
os.system("cat SPLIT_PARSE_{0}.fen >> {1}".format(ii, OUTPUT_NAME))
os.system("rm SPLIT_OUT_{0}.pgn".format(ii))
os.system("rm SPLIT_PARSE_{0}.fen".format(ii))
if __name__ == "__main__":
buildTexelBook()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.