chessai.core.gameparser

  1import enum
  2import gzip
  3import logging
  4import os
  5import random
  6import re
  7import typing
  8
  9import edq.util.serial
 10
 11import chessai.core.action
 12import chessai.core.board
 13import chessai.core.castling
 14import chessai.core.coordinate
 15import chessai.core.gamestate
 16import chessai.core.piece
 17import chessai.core.types
 18
 19THIS_DIR: str = os.path.join(os.path.dirname(os.path.realpath(__file__)))
 20GAMES_DIR: str = os.path.join(THIS_DIR, '..', 'resources', 'games')
 21
 22PGN_FILE_EXTENSION: str = '.pgn'
 23GZIP_FILE_EXTENSION: str = '.gz'
 24
 25DEFAULT_ENCODING: str = 'utf-8'
 26
 27# File parsing patterns.
 28SEPARATOR_PATTERN: re.Pattern = re.compile(r'^\s*-{3,}\s*$')
 29
 30# PGN parsing patterns.
 31TAG_PATTERN: re.Pattern= re.compile(r"^\[([A-Za-z0-9][A-Za-z0-9_+#=:-]*)\s+\"([^\r]*)\"\]\s*$")
 32TAG_NAME_PATTERN: re.Pattern = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_+#=:-]*\Z")
 33MOVETEXT_PATTERN: re.Pattern = re.compile(r"""
 34    (
 35        [NBKRQ]?[a-h]?[1-8]?[\-x]?[a-h][1-8](?:=?[nbrqkNBRQK])?
 36        |[PNBRQK]?@[a-h][1-8]
 37        |--
 38        |Z0
 39        |0000
 40        |@@@@
 41        |O-O(?:-O)?
 42        |0-0(?:-0)?
 43    )
 44    |(\{.*?})
 45    |(\{.*)
 46    |(;.*)
 47    |(\$[0-9]+)
 48    |(\()
 49    |(\))
 50    |(\*|1-0|0-1|1/2-1/2)
 51    |([\?!]{1,2})
 52    """, re.DOTALL | re.VERBOSE)
 53
 54SKIP_MOVETEXT_PATTERN: re.Pattern = re.compile(r""";|\{|\}""")
 55
 56FEN_HEADER_KEY: str = 'FEN'
 57SET_UP_HEADER_KEY: str = 'SetUp'
 58
 59MAX_PGN_LINE_LENGTH: int = 80
 60
 61PGN_HEADERS: list[str] = [
 62    "event",
 63    "site",
 64    "date",
 65    "round",
 66    "white",
 67    "black",
 68    "result",
 69]
 70
 71class StandardHeaders(edq.util.serial.DictConverter):
 72    """
 73    These seven standard tags will always be present in the headers:
 74     - Event: the name of the tournament or match event
 75     - Site: the location of the event
 76     - Date: the starting date of the game
 77     - Round: the playing round ordinal of the game
 78     - White: the player of the white pieces
 79     - Black: the player of the black pieces
 80     - Result: the result of the game
 81
 82    To see the full description of the possible headers and their format,
 83    see https://www.saremba.de/chessgml/standards/pgn/pgn-complete.htm#c8.1 .
 84    """
 85
 86    def __init__(self,
 87            event: str = "?",
 88            site: str = "?",
 89            date: str = "????.??.??",
 90            game_round: str = "?",
 91            white: str = "?",
 92            black: str = "?",
 93            result: str = "*",
 94            ) -> None:
 95        self.event: str = event
 96        """ The name of the tournament or match event. """
 97
 98        self.site: str = site
 99        """ The location of the event. """
100
101        self.date: str = date
102        """ The starting date of the game. """
103
104        self.round: str = game_round
105        """ The playing round ordinal of the game. """
106
107        self.white: str = white
108        """ The player of the white pieces. """
109
110        self.black: str = black
111        """ The player of the black pieces. """
112
113        self.result: str = result
114        """ The result of the game. """
115
116    def _set(self, attribute: str, value: str) -> None:
117        """ Update the header attribute to the given value. """
118
119        match attribute.lower():
120            case "event":
121                self.event = value
122            case "site":
123                self.site = value
124            case "date":
125                self.date = value
126            case "round":
127                self.round = value
128            case "white":
129                self.white = value
130            case "black":
131                self.black = value
132            case "result":
133                self.result = value
134            case _:
135                return
136
137    def to_pgn(self) -> list[str]:
138        """ Convert the headers to a list of PGN tags. """
139
140        tags: list[str] = []
141
142        tags.append(_to_tag("Event", self.event))
143        tags.append(_to_tag("Site", self.site))
144        tags.append(_to_tag("Date", self.date))
145        tags.append(_to_tag("Round", self.round))
146        tags.append(_to_tag("White", self.white))
147        tags.append(_to_tag("Black", self.black))
148        tags.append(_to_tag("Result", self.result))
149
150        return tags
151
152    def __eq__(self, other: object) -> bool:
153        if (not isinstance(other, StandardHeaders)):
154            return False
155
156        return (self.event == other.event
157            and self.site == other.site
158            and self.date == other.date
159            and self.round == other.round
160            and self.white == other.white
161            and self.black == other.black
162            and self.result == other.result)
163
164class PGNResult(enum.StrEnum):
165    """ The possible game endings denoted in a PGN. """
166
167    WHITE_WIN = '1-0'
168    """ White won the game. """
169
170    BLACK_WIN = '0-1'
171    """ Black won the game. """
172
173    TIE = '1/2-1/2'
174    """ The game was a tie. """
175
176    IN_PROGRESS = '*'
177    """ The game is still in progress. """
178
179    UNKNOWN = 'Unknown'
180    """ The ending of the game is unknown. """
181
182class ParsedPGN(edq.util.serial.DictConverter):
183    """
184    The parsed result of a single PGN game.
185    """
186
187    def __init__(self,
188                 headers: StandardHeaders | None = None,
189                 optional_headers: dict[str, typing.Any] | None = None,
190                 starting_fen: str | None = None,
191                 initial_actions: list[chessai.core.action.Action] | None = None,
192                 comments: list[str] | None = None,
193                 result: PGNResult = PGNResult.UNKNOWN) -> None:
194
195        if (headers is None):
196            headers = StandardHeaders()
197
198        self.headers: StandardHeaders = headers
199        """ The standard headers from a single PGN game. """
200
201        if (optional_headers is None):
202            optional_headers = {}
203
204        self.optional_headers: dict[str, typing.Any] = optional_headers
205        """ Any additional headers parsed from the PGN. """
206
207        self.starting_fen: str | None = starting_fen
208        """ The starting FEN for the game, which is often omitted to denote the default FEN. """
209
210        if (initial_actions is None):
211            initial_actions = []
212
213        self.initial_actions: list[chessai.core.action.Action] = initial_actions
214        """ The mainline moves in the game. """
215
216        if (comments is None):
217            comments = []
218
219        self.comments: list[str] = comments
220        """ The comments found in the game. """
221
222        self.result: PGNResult = result
223        """ The result of the PGN or if the game is still in progress. """
224
225    def get_starting_fen(self) -> str | None:
226        """ Get the optional FEN header from the PGN. """
227
228        return self.optional_headers.get(FEN_HEADER_KEY, None)
229
230def parse_pgn(pgn: str, state_class: typing.Type[chessai.core.gamestate.GameState]) -> ParsedPGN | None:
231    """
232    Parse a single PGN string into a ParsedPGN.
233
234    Raises ValueError if the PGN is malformed.
235    See https://en.wikipedia.org/wiki/Portable_Game_Notation .
236    """
237
238    lines = pgn.splitlines()
239
240    # If it is only one line, try loading it from a file.
241    if (len(lines) == 1):
242        pgn = load_pgn_from_file(pgn)
243        lines = pgn.splitlines()
244
245    index = 0
246
247    headers: StandardHeaders = StandardHeaders()
248
249    optional_headers: dict[str, typing.Any] = {}
250
251    # Parse the header fields.
252    for i, line in enumerate(lines):
253        # Track where we are in the lines.
254        index = i
255
256        clean_line = line.strip()
257
258        # Skip empty lines.
259        if (len(line) == 0):
260            continue
261
262        # Skip comments.
263        if (_is_pgn_comment_line(clean_line)):
264            continue
265
266        # Headers must begin with an open bracket.
267        if (not clean_line.startswith("[")):
268            break
269
270        tag_match = TAG_PATTERN.match(clean_line)
271
272        # Ignore malformed tags.
273        if (not tag_match):
274            continue
275
276        tag_header_key = tag_match.group(1)
277        tag_header_value = tag_match.group(2)
278
279        # Determine if this is a standard header key or not.
280        if (tag_header_key.lower() in PGN_HEADERS):
281            headers._set(tag_header_key, tag_header_value)
282        else:
283            optional_headers[tag_header_key] = tag_header_value
284
285    # No game found.
286    if (headers == StandardHeaders()):
287        return None
288
289    # Start scanning for moves from the end of the headers.
290    lines = lines[index:]
291
292    # Set up a gamestate to follow along and parse SANs.
293    fen = optional_headers.get(FEN_HEADER_KEY, None)
294    state = state_class.from_fen(fen = fen)
295    rng = random.Random(1)
296
297    initial_actions: list[chessai.core.action.Action] = []
298    comments: list[str] = []
299    result: PGNResult = PGNResult.UNKNOWN
300
301    in_comment = False
302    comment_buffer: list[str] = []
303
304    skip_variation_depth = 0
305
306    for line in lines:
307        line = line.strip()
308
309        if (not line):
310            break
311
312        # Skip whole-line comments.
313        if _is_pgn_comment_line(line):
314            continue
315
316        i = 0
317        while (i < len(line)):
318            if in_comment:
319                end_index = line.find("}", i)
320
321                # The rest of the line is part of the comment.
322                if (end_index == -1):
323                    comment_buffer.append(line[i:])
324                    break
325
326                # Add the remainder of the comment and continue processing the current line.
327                comment_buffer.append(line[i:end_index])
328                comments.append("\n".join(comment_buffer).strip())
329
330                comment_buffer = []
331                in_comment = False
332
333                i = end_index + 1
334
335                continue
336
337            match = MOVETEXT_PATTERN.match(line, i)
338
339            if (not match):
340                i += 1
341                continue
342
343            token = match.group(0)
344            i = match.end()
345
346            # Handle block comments {...}.
347            if token.startswith("{"):
348                content = token[1:]
349
350                end_index = content.find("}")
351
352                if (end_index == -1):
353                    # Track part of a multi-line comment.
354                    in_comment = True
355                    comment_buffer.append(content)
356                else:
357                    # Add the in-line comment and continue processing this line.
358                    comments.append(content[:end_index].strip())
359
360                continue
361
362            # Skip variations (...) entirely.
363            if (token == "("):
364                skip_variation_depth += 1
365                continue
366
367            if (token == ")"):
368                if (skip_variation_depth > 0):
369                    skip_variation_depth -= 1
370
371                continue
372
373            if (skip_variation_depth > 0):
374                continue
375
376            # Skip line comments.
377            if token.startswith(";"):
378                break
379
380            # Skip move numbers like "1." or "23.".
381            if token.endswith("."):
382                continue
383
384            # Skip results.
385            if (token in ["1-0", "0-1", "1/2-1/2", "*"]):
386                result = PGNResult(token)
387                break
388
389            # Skip NAGs / annotations.
390            if (token.startswith("$") or (token in ["!", "?", "!!", "??", "!?", "?!"])):
391                continue
392
393            # Otherwise, this must be a SAN move.
394            action = state.get_action_from_san(token)
395            if (action == chessai.core.action.NoneAction()):
396                raise ValueError(f"Unable to find a legal action for the SAN: '{token}'.")
397
398            state.process_turn_full(action, rng)
399            initial_actions.append(action)
400
401    # Check if the gamestate agrees with the expected result.
402    if ((result in [PGNResult.WHITE_WIN, PGNResult.BLACK_WIN]) and (not state.is_game_over())):
403        # The game is not over when the PGN believes it should be, so add a forfeit action.
404        initial_actions.append(chessai.core.action.ForfeitAction())
405
406    if ((result == PGNResult.TIE) and (not state.is_game_over())):
407        # The game must be a proposed draw, so add the draw handshake.
408        initial_actions.append(chessai.core.action.ProposeDrawAction())
409        initial_actions.append(chessai.core.action.AcceptDrawAction())
410
411    return ParsedPGN(
412        headers          = headers,
413        optional_headers = optional_headers,
414        initial_actions  = initial_actions,
415        comments         = comments,
416        result           = result,
417    )
418
419def load_pgn_from_file(path: str) -> str:
420    """
421    Load the PGN string from a file.
422
423    If the given path does not exist,
424    try to prefix the path with the standard game directory and suffix with the standard extension.
425    """
426
427    raw_path = path
428
429    # If the path does not exist, try the boards directory.
430    if (not os.path.exists(path)):
431        path = os.path.join(GAMES_DIR, path)
432
433        # If this path does not have a good extension, add one.
434        if ((os.path.splitext(path)[-1] != PGN_FILE_EXTENSION) and (os.path.splitext(path)[-1] != GZIP_FILE_EXTENSION)):
435            path = path + PGN_FILE_EXTENSION
436
437    # If the path still does not exist, try adding the GZip extension.
438    if (not os.path.exists(path)):
439        if (os.path.splitext(path)[-1] != GZIP_FILE_EXTENSION):
440            path = path + GZIP_FILE_EXTENSION
441
442    if (not os.path.exists(path)):
443        raise ValueError(f"Could not find PGN file, path does not exist: '{raw_path}'.")
444
445    if (os.path.splitext(path)[-1] == GZIP_FILE_EXTENSION):
446        return load_pgn_from_gzip(path)
447
448    contents = edq.util.dirent.read_file(path, strip = False)
449    return contents
450
451def load_pgn_from_gzip(path: str) -> str:
452    """
453    Load the PGN string from a file.
454
455    If the given path does not exist,
456    try to prefix the path with the standard game directory and suffix with the standard extension.
457    """
458
459    raw_path = path
460
461    # If the path does not exist, try the boards directory.
462    if (not os.path.exists(path)):
463        path = os.path.join(GAMES_DIR, path)
464
465        # If this path does not have a good extension, add one.
466        if (os.path.splitext(path)[-1] != GZIP_FILE_EXTENSION):
467            path = path + GZIP_FILE_EXTENSION
468
469    if (not os.path.exists(path)):
470        raise ValueError(f"Could not find PGN file, path does not exist: '{raw_path}'.")
471
472    with gzip.open(path, 'rb') as file:
473        uncompressed_contents = file.read()
474        decompressed_contents = uncompressed_contents.decode(DEFAULT_ENCODING)
475
476    return decompressed_contents
477
478def to_pgn(headers: StandardHeaders, optional_headers: dict[str, typing.Any],
479           state_class: typing.Type[chessai.core.gamestate.GameState], start_fen: str,
480           actions: list[chessai.core.action.Action]) -> str:
481    """
482    Convert a game into a PGN.
483
484    Raises an error if any of the seven required headers are missing.
485    If a move cannot be generated from an action and gamestate pair, an error is raised.
486    """
487
488    lines: list[str] = headers.to_pgn()
489
490    # Non-default starting FENs require additional PGN information.
491    if (start_fen != chessai.core.gamestate.DEFAULT_FEN):
492        lines.append(_to_tag(SET_UP_HEADER_KEY, '1'))
493        lines.append(_to_tag(FEN_HEADER_KEY, start_fen))
494
495    # Add optional headers.
496    for (key, value) in optional_headers.items():
497        try:
498            str_value = str(value)
499        except Exception:
500            logging.warning("Could not convert optional header value for key '%s' to a string.", key)
501            continue
502
503        lines.append(_to_tag(str(key), str_value))
504
505    # Add an extra blank line after the header section.
506    lines.append('')
507
508    # Build the move SAN tokens while replaying the game.
509    state = state_class.from_fen(fen = start_fen)
510    pgn_tokens: list[str] = []
511    current_fullmove_number = 0
512
513    for action in actions:
514        # Ignore meta actions as they do not have a valid SAN representation.
515        if (not isinstance(action, chessai.core.action.MoveAction)):
516            continue
517
518        if (state.fullmove_number != current_fullmove_number):
519            pgn_tokens.append(f"{state.fullmove_number}.")
520            current_fullmove_number = state.fullmove_number
521
522        san = _action_to_san(action, state)
523        pgn_tokens.append(san)
524
525        state.push(action)
526
527    # Add the result token.
528    pgn_tokens.append(headers.result)
529
530    movetext_lines: list[str] = []
531    current_line = ''
532
533    for token in pgn_tokens:
534        if (len(current_line) == 0):
535            candidate_line = token
536        else:
537            candidate_line = f"{current_line} {token}"
538
539        # Keep building a line until it surpasses the max line length.
540        if (len(candidate_line) <= MAX_PGN_LINE_LENGTH):
541            current_line = candidate_line
542        else:
543            movetext_lines.append(current_line)
544            current_line = token
545
546    if (len(current_line) > 0):
547        movetext_lines.append(current_line)
548
549    lines.extend(movetext_lines)
550
551    # Add a trailing newline after the movetext.
552    lines.append('')
553
554    return '\n'.join(lines)
555
556def _escape_tag_value(value: str) -> str:
557    """ Escape the tag value to avoid parsing errors. """
558
559    escaped = value.replace('\\', '\\\\').replace('"', '\\"')
560    return escaped
561
562def _to_tag(header: str, value: str) -> str:
563    """ Write a header and its value to a PGN tag. """
564
565    escaped_value = _escape_tag_value(value)
566    return f'[{header} "{escaped_value}"]'
567
568def _action_to_san(action: chessai.core.action.MoveAction, state: chessai.core.gamestate.GameState) -> str:
569    """
570    Convert an action to a SAN using the state before the action is applied evaluate SAN move ambiguity.
571
572    This method does not understand meta actions (i.e., forfeits or draw actions).
573    """
574
575    piece = state.get(action.start_coordinate)
576    if (piece is None):
577        raise ValueError(f"The start coordinate for action '{action.uci()}' does not have a piece.")
578
579    piece_symbol_upper = piece.symbol().upper()
580
581    move_san: str | None = None
582
583    # Detect castling by checking for a king moving two squares horizontally.
584    if (piece_symbol_upper == 'K'):
585        file_delta = action.end_coordinate.file - action.start_coordinate.file
586
587        if (file_delta == 2):
588            # Kingside castle.
589            move_san = 'O-O'
590        elif (file_delta == -2):
591            # Queenside castle.
592            move_san = 'O-O-O'
593
594    is_pawn = (piece_symbol_upper == 'P')
595
596    if (move_san is None):
597        is_capture = state.is_capture(action)
598
599        # Detect en-passant captures.
600        if (is_pawn and (action.end_coordinate == state.en_passant_coordinate)):
601            is_capture = True
602
603        # Check if the move is ambiguous.
604        ambiguous_actions: list[chessai.core.action.MoveAction] = []
605        for alternate_action in state.get_legal_actions():
606            if (not isinstance(alternate_action, chessai.core.action.MoveAction)):
607                continue
608
609            # An ambiguous move must have the same action type.
610            if (type(alternate_action) != type(action)):
611                continue
612
613            # An ambiguous move must end at the same coordinate.
614            if (alternate_action.end_coordinate != action.end_coordinate):
615                continue
616
617            # An ambiguous move cannot start at the same coordinate because that would be the same piece moving.
618            if (alternate_action.start_coordinate == action.start_coordinate):
619                continue
620
621            # An ambiguous action must have the same promotion.
622            if (isinstance(action, chessai.core.action.PromotionAction)
623                    and isinstance(alternate_action, chessai.core.action.PromotionAction)
624                    and (alternate_action.promotion != action.promotion)):
625                continue
626
627            # An ambiguous action must have a different piece type.
628            alternate_action_piece = state.get(alternate_action.start_coordinate)
629            if (alternate_action_piece is None):
630                continue
631
632            if (alternate_action_piece.symbol().upper() != piece_symbol_upper):
633                continue
634
635            ambiguous_actions.append(alternate_action)
636
637        disambiguation_str = ''
638
639        # Pawn actions cannot be ambiguous.
640        if ((len(ambiguous_actions) > 0) and (not is_pawn)):
641            # Check if there is an ambiguous action on the same file.
642            same_file = False
643            for ambiguous_action in ambiguous_actions:
644                if (ambiguous_action.start_coordinate.file != action.start_coordinate.file):
645                    continue
646
647                same_file = True
648
649            # Check if there is an ambiguous action on the same rank.
650            same_rank = False
651            for ambiguous_action in ambiguous_actions:
652                if (ambiguous_action.start_coordinate.rank != action.start_coordinate.rank):
653                    continue
654
655                same_rank = True
656
657            if (not same_file):
658                disambiguation_str = action.start_coordinate.uci(only_file = True)
659            elif (not same_rank):
660                disambiguation_str = action.start_coordinate.uci(only_rank = True)
661            else:
662                # Fallback to the entire coordinate.
663                disambiguation_str = action.start_coordinate.uci()
664
665        # Build the final SAN with the above information.
666        destination_str = action.end_coordinate.uci()
667
668        if (is_pawn):
669            if is_capture:
670                # Pawn captures require the starting file to disambiguate.
671                start_file = action.start_coordinate.uci(only_file = True)
672                move_san = f"{start_file}x{destination_str}"
673            else:
674                move_san = destination_str
675        else:
676            if is_capture:
677                capture_str = 'x'
678            else:
679                capture_str = ''
680
681            move_san = f"{piece_symbol_upper}{disambiguation_str}{capture_str}{destination_str}"
682
683        # Add the piece promotion suffix.
684        if isinstance(action, chessai.core.action.PromotionAction):
685            move_san += f"={action.promotion.symbol().upper()}"
686
687    # Apply the action and check for check and checkmate.
688    successor = state.generate_successor(action)
689
690    if (successor.is_checkmate()):
691        move_san += '#'
692    elif (successor.is_check(successor.turn)):
693        move_san += '+'
694
695    return move_san
696
697def _is_pgn_comment_line(line: str) -> bool:
698    """
699    Returns if the line is a comment.
700
701    A line is a comment in a PGN if it  starts with a '%' or ';'.
702    """
703
704    line = line.strip()
705    if (len(line) == 0):
706        return True
707
708    if (line[0] == '%'):
709        return True
710
711    if (line[0] == ';'):
712        return True
713
714    return False