agent.py (9138B)
1 #!/usr/bin/env python3 2 3 from __future__ import annotations 4 5 import enum 6 import random 7 import socket 8 import struct 9 import sys 10 11 12 class PlayerType(enum.Enum): 13 PLAYER_BLACK = 0 14 PLAYER_WHITE = 1 15 16 def __str__(self) -> str: 17 match self: 18 case self.PLAYER_BLACK: return 'black' 19 case self.PLAYER_WHITE: return 'white' 20 21 22 class MsgType(enum.Enum): 23 MSG_START = 0 24 MSG_MOVE = 1 25 MSG_SWAP = 2 26 MSG_END = 3 27 28 29 class MsgStartData: 30 def __init__(self, player: int, board_size: int, game_secs: int, thread_limit: int, mem_limit_mib: int): 31 self.player = PlayerType(player) 32 self.board_size = board_size 33 self.game_secs = game_secs 34 self.thread_limit = thread_limit 35 self.mem_limit_mib = mem_limit_mib 36 37 def as_tuple(self) -> tuple[PlayerType, int, int, int, int]: 38 return self.player, self.board_size, self.game_secs, self.thread_limit, self.mem_limit_mib 39 40 41 class MsgMoveData: 42 def __init__(self, board_x: int, board_y: int): 43 self.board_x = board_x 44 self.board_y = board_y 45 46 def __repr__(self) -> str: 47 return f'move: ({self.board_x}, {self.board_y})' 48 49 def as_tuple(self) -> tuple[int, int]: 50 return self.board_x, self.board_y 51 52 53 class MsgSwapData: 54 def __init__(self): 55 pass 56 57 def __repr__(self) -> str: 58 return f'swap' 59 60 61 class MsgEndData: 62 def __init__(self, winner: int): 63 self.winner = PlayerType(winner) 64 65 def __repr__(self) -> str: 66 return f'end: {self.winner}' 67 68 def as_tuple(self) -> tuple[PlayerType]: 69 return self.winner, 70 71 72 MsgData = MsgStartData | MsgMoveData | MsgSwapData | MsgEndData 73 74 75 class Msg: 76 def __init__(self, typ: MsgType, dat: MsgData): 77 self.typ = typ 78 self.dat = dat 79 80 def __repr__(self) -> str: 81 return f'msg: {self.typ}, {self.dat}' 82 83 @classmethod 84 def size(cls) -> int: 85 return 32 86 87 def serialise_into(self, buffer: memoryview) -> int: 88 assert Msg.size() <= len(buffer) 89 90 match self.typ: 91 case MsgType.MSG_START: # this message type is never sent by the client 92 pass 93 94 case MsgType.MSG_MOVE: 95 struct.pack_into('!III', buffer, 0, 96 self.typ.value, 97 self.dat.board_x, 98 self.dat.board_y) 99 100 case MsgType.MSG_SWAP: 101 struct.pack_into('!I', buffer, 0, self.typ.value) 102 103 case MsgType.MSG_END: # this message type is never sent by the client 104 pass 105 106 return Msg.size() 107 108 @classmethod 109 def deserialise_from(cls, buffer: memoryview) -> Msg: 110 assert cls.size() <= len(buffer) 111 112 raw_typ, = struct.unpack_from('!I', buffer, 0) 113 typ = MsgType(raw_typ) 114 115 match typ: 116 case MsgType.MSG_START: dat = MsgStartData(*struct.unpack_from('!IIIII', buffer, 4)) 117 case MsgType.MSG_MOVE: dat = MsgMoveData(*struct.unpack_from('!II', buffer, 4)) 118 case MsgType.MSG_SWAP: dat = MsgSwapData() 119 case MsgType.MSG_END: dat = MsgEndData(*struct.unpack_from('!I', buffer, 4)) 120 121 return Msg(typ, dat) 122 123 124 def recv_msg(sock: socket.socket, *, expected_msg_types: list[MsgType]) -> Msg: 125 buffer = bytearray(Msg.size()) 126 127 def recv_all_bytes(sock: socket.socket, buf: memoryview, sz: int) -> int: 128 total = 0 129 while total < sz: 130 curr = sock.recv_into(buf[total:], sz - total) 131 if curr == 0: return total 132 total += curr 133 134 return total 135 136 recv_all_bytes(sock, memoryview(buffer), len(buffer)) 137 138 return Msg.deserialise_from(buffer) 139 140 141 def send_msg(sock: socket.socket, msg: Msg) -> None: 142 buffer = bytearray(Msg.size()) 143 144 def send_all_bytes(sock: socket.socket, buf: memoryview, sz: int) -> int: 145 total = 0 146 while total < sz: 147 curr = sock.send(buf[total:], sz - total) 148 if curr == 0: return total 149 total += curr 150 151 return total 152 153 msg.serialise_into(buffer) 154 155 send_all_bytes(sock, memoryview(buffer), len(buffer)) 156 157 158 class Board: 159 class Cell(enum.Enum): 160 BLACK = PlayerType.PLAYER_BLACK.value 161 WHITE = PlayerType.PLAYER_WHITE.value 162 EMPTY = enum.auto() 163 164 def __init__(self, board_size: int): 165 self.board_size = board_size 166 self.board = [self.Cell.EMPTY for _ in range(board_size * board_size)] 167 self.remaining_moves = [(i, j) for i in range(board_size) for j in range(board_size)] 168 random.shuffle(self.remaining_moves) 169 170 def swap(self) -> None: 171 self.remaining_moves = [] 172 for j in range(self.board_size): 173 for i in range(self.board_size): 174 match self.board[j * self.board_size + i]: 175 case self.Cell.BLACK: 176 self.board[j * self.board_size + i] = self.Cell.WHITE 177 178 case self.Cell.WHITE: 179 self.board[j * self.board_size + i] = self.Cell.BLACK 180 181 case self.Cell.EMPTY: 182 self.remaining_moves.append((i, j)) 183 184 random.shuffle(self.remaining_moves) 185 186 def play(self, player: PlayerType, px: int, py: int) -> bool: 187 old = self.board[py * self.board_size + px] 188 189 if old != self.Cell.EMPTY: 190 return False 191 192 new = None 193 match player: 194 case PlayerType.PLAYER_BLACK: new = self.Cell.BLACK 195 case PlayerType.PLAYER_WHITE: new = self.Cell.WHITE 196 197 self.board[py * self.board_size + px] = new 198 199 for idx, t in enumerate(self.remaining_moves): 200 if t[0] == px and t[1] == py: 201 self.remaining_moves.pop(idx) 202 break 203 204 return True 205 206 def get_next_move(self) -> tuple[int, int]: 207 return self.remaining_moves.pop(0) 208 209 210 class GameState(enum.Enum): 211 START = enum.auto() 212 RECV = enum.auto() 213 SEND = enum.auto() 214 END = enum.auto() 215 216 217 def main() -> None: 218 if len(sys.argv) < 3: 219 print(f'Usage: {sys.argv[0]} <host> <port>', file=sys.stderr) 220 quit(1) 221 222 host, port, *args = sys.argv[1:] 223 with socket.create_connection((host, port)) as sock: 224 state = GameState.START 225 226 player = None 227 game_secs = None # unused 228 thread_limit = None # unused 229 mem_limit_mib = None # unused 230 231 board = None 232 other_player = None 233 winner = None 234 235 first_round = True 236 game_is_over = False 237 while not game_is_over: 238 match state: 239 case GameState.START: 240 msg = recv_msg(sock, expected_msg_types=[MsgType.MSG_START]) 241 player, board_size, game_secs, thread_limit, mem_limit_mib = msg.dat.as_tuple() 242 243 board = Board(board_size) 244 245 print(f'[{player}] Started game: {board_size}x{board_size}, {game_secs} secs, {thread_limit} threads, {mem_limit_mib} MiB') 246 247 if player == PlayerType.PLAYER_BLACK: 248 other_player = PlayerType.PLAYER_WHITE 249 state = GameState.SEND 250 251 elif player == PlayerType.PLAYER_WHITE: 252 other_player = PlayerType.PLAYER_BLACK 253 state = GameState.RECV 254 255 case GameState.RECV: 256 msg = recv_msg(sock, expected_msg_types=[MsgType.MSG_MOVE, MsgType.MSG_SWAP, MsgType.MSG_END]) 257 258 if msg.typ == MsgType.MSG_MOVE: 259 board_x, board_y = msg.dat.as_tuple() 260 board.play(other_player, board_x, board_y) 261 262 if first_round and random.choice([True, False]): 263 board.swap() 264 265 msg = Msg(MsgType.MSG_SWAP, MsgSwapData()) 266 send_msg(sock, msg) 267 268 state = GameState.RECV 269 270 else: 271 state = GameState.SEND 272 273 elif msg.typ == MsgType.MSG_SWAP: 274 board.swap() 275 276 state = GameState.SEND 277 278 elif msg.typ == MsgType.MSG_END: 279 winner, = msg.dat.as_tuple() 280 281 state = GameState.END 282 283 first_round = False 284 285 case GameState.SEND: 286 board_x, board_y = board.get_next_move() 287 board.play(player, board_x, board_y) 288 289 msg = Msg(MsgType.MSG_MOVE, MsgMoveData(board_x, board_y)) 290 291 send_msg(sock, msg) 292 293 state = GameState.RECV 294 295 first_round = False 296 297 case GameState.END: 298 print(f'[{player}] Player {winner} has won the game') 299 break 300 301 case _: 302 print(f'[{player}] Unknown state encountered: {state}') 303 break 304 305 306 if __name__ == '__main__': 307 main() 308