File size: 6,201 Bytes
51c8e36
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Move Processor: Utilities for parsing and analyzing moves.



Handles:

- Move parsing from LLM output

- Piece extraction from board

- Board state serialization

- Move direction computation

"""

import re
from typing import Optional, Tuple, List
from dataclasses import dataclass


@dataclass
class MoveDetails:
    """Parsed move information for logging."""
    src_pos: str = ""
    dst_pos: str = ""
    piece_type: str = ""
    target_piece: str = ""  # Piece at destination (if any)
    move_direction: str = ""  # N/S/E/W
    board_state: str = ""  # Serialized board state
    available_moves: str = ""  # List of valid moves


def parse_move(action: str) -> Tuple[str, str]:
    """

    Extract source and destination positions from move string.

    

    Args:

        action: Move string like "[D4 E4]"

        

    Returns:

        Tuple of (src_pos, dst_pos) or ("", "") if parsing fails

    """
    move_pattern = r'\[([A-J]\d+)\s+([A-J]\d+)\]'
    match = re.search(move_pattern, action)
    if match:
        return match.group(1), match.group(2)
    return "", ""


def get_piece_at_position(board: List[List], position: str) -> str:
    """

    Get piece type at a board position.

    

    Args:

        board: 10x10 game board

        position: Position string like "D4"

        

    Returns:

        piece_type or "" if not found

    """
    if not position:
        return ""
    
    try:
        row = ord(position[0]) - ord('A')
        col = int(position[1:])
        piece = board[row][col]
        
        if piece and isinstance(piece, dict) and 'rank' in piece:
            return piece['rank']
    except (IndexError, ValueError, TypeError):
        pass
    
    return ""


def compute_move_direction(src_pos: str, dst_pos: str) -> str:
    """

    Compute move direction from source to destination.

    

    Args:

        src_pos: Source position like "D4"

        dst_pos: Destination position like "E4"

        

    Returns:

        Direction: "N", "S", "E", "W", or "" if invalid

    """
    if not src_pos or not dst_pos:
        return ""
    try:
        src_row = ord(src_pos[0]) - ord('A')
        dst_row = ord(dst_pos[0]) - ord('A')
        src_col = int(src_pos[1:])
        dst_col = int(dst_pos[1:])
        
        if dst_row < src_row:
            return "N"  # Moving up (toward A)
        elif dst_row > src_row:
            return "S"  # Moving down (toward J)
        elif dst_col > src_col:
            return "E"  # Moving right
        elif dst_col < src_col:
            return "W"  # Moving left
    except (IndexError, ValueError):
        pass
    return ""


def serialize_board(board: List[List], player_id: int = 0) -> str:
    """

    Serialize board state to a compact string for training.

    Format: Each cell as "RC:PIECE" where R=row, C=col, PIECE=short rank or ?/~/.

    

    Args:

        board: 10x10 game board

        player_id: Current player (to show their pieces, hide opponent's)

        

    Returns:

        Compact board string representation

    """
    # Short forms matching the game board display (to store it in csv and datasets)
    RANK_SHORT = {
        "Flag": "FL",
        "Spy": "SP",
        "Scout": "SC",
        "Miner": "MN",
        "Sergeant": "SG",
        "Lieutenant": "LT",
        "Captain": "CP",
        "Major": "MJ",
        "Colonel": "CL",
        "General": "GN",
        "Marshal": "MS",
        "Bomb": "BM",
    }
    
    if not board:
        return ""
    
    cells = []
    row_labels = "ABCDEFGHIJ"
    
    for r, row in enumerate(board):
        for c, cell in enumerate(row):
            pos = f"{row_labels[r]}{c}"
            if cell is None:
                cells.append(f"{pos}:.")
            elif cell == "~":
                cells.append(f"{pos}:~")
            elif isinstance(cell, dict):
                rank = cell.get('rank', '?')
                owner = cell.get('player', -1)
                if owner == player_id:
                    # Show own piece rank (use short form)
                    short_rank = RANK_SHORT.get(rank, rank)
                    cells.append(f"{pos}:{short_rank}")
                else:
                    # Hide opponent's piece
                    cells.append(f"{pos}:?")
            else:
                cells.append(f"{pos}:{cell}")
    
    return "|".join(cells)


def extract_available_moves(observation: str) -> str:
    """

    Extract available moves from observation string.

    

    Args:

        observation: Full observation text from environment

        

    Returns:

        Comma-separated list of available moves

    """
    # Pattern: "Available Moves: [D1 E1], [D5 E5], ..."
    match = re.search(r'Available Moves:\s*(.+?)(?:\n|$)', observation)
    if match:
        moves_str = match.group(1).strip()
        # Extract just the moves like "[D1 E1]"
        moves = re.findall(r'\[[A-J]\d+\s+[A-J]\d+\]', moves_str)
        return ",".join(moves)
    return ""


def process_move(action: str, board: List[List], observation: str = "", player_id: int = 0) -> MoveDetails:
    """

    Process a move and extract all relevant details for logging.

    

    Args:

        action: Move string from LLM

        board: Current game board

        observation: Full observation text (for available moves)

        player_id: Current player ID

        

    Returns:

        MoveDetails dataclass with all extracted info

    """
    src_pos, dst_pos = parse_move(action)
    piece_type = get_piece_at_position(board, src_pos)
    target_piece = get_piece_at_position(board, dst_pos)
    move_direction = compute_move_direction(src_pos, dst_pos)
    board_state = serialize_board(board, player_id)
    available_moves = extract_available_moves(observation)
    
    return MoveDetails(
        src_pos=src_pos,
        dst_pos=dst_pos,
        piece_type=piece_type,
        target_piece=target_piece,
        move_direction=move_direction,
        board_state=board_state,
        available_moves=available_moves
    )