backend.server
1import asyncio 2import json 3import logging 4from typing import Any, Dict, List, Optional 5 6import websockets 7from websockets.server import WebSocketServerProtocol 8 9logging.basicConfig( 10 level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" 11) 12 13 14class UTTTServer: 15 """ 16 Ultimate Tic-Tac-Toe (UTTT) Server. 17 18 Handles map loading, agent movement, and state broadcasting. 19 The game is played on a $9 \times 9$ micro-board, divided into $3 \times 3$ macro-boards. 20 """ 21 22 def __init__(self) -> None: 23 """ 24 Initializes the UTTTServer. 25 """ 26 self.frontend_ws: Optional[WebSocketServerProtocol] = None 27 self.agent1_ws: Optional[WebSocketServerProtocol] = None 28 self.agent2_ws: Optional[WebSocketServerProtocol] = None 29 30 # 9x9 Micro Board (0=Empty, 1=P1, 2=P2) 31 self.board: List[List[int]] = [[0] * 9 for _ in range(9)] 32 # 3x3 Macro Board (0=Ongoing, 1=P1 Win, 2=P2 Win, 3=Draw) 33 self.macro_board: List[List[int]] = [[0] * 3 for _ in range(3)] 34 35 # (my, mx) indicating which macro-board the current player MUST play in. 36 # None means the player can play in ANY available macro-board. 37 self.active_macro: Optional[List[int]] = None 38 39 self.first_player_this_round: int = 1 40 self.current_turn: int = 1 41 self.running: bool = False 42 self.match_scores: Dict[int, int] = {1: 0, 2: 0} 43 44 async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None: 45 """ 46 Starts the UTTT server. 47 48 Args: 49 host (str): The host address to bind to. 50 port (int): The port to listen on. 51 """ 52 logging.info(f"UTTT Server started on ws://{host}:{port}") 53 async with websockets.serve(self.handle_client, host, port): 54 await asyncio.Future() 55 56 async def handle_client(self, websocket: WebSocketServerProtocol) -> None: 57 """ 58 Handles incoming WebSocket connections. 59 60 Args: 61 websocket (WebSocketServerProtocol): The connected WebSocket client. 62 """ 63 client_type = "Unknown" 64 try: 65 init_msg = await websocket.recv() 66 if isinstance(init_msg, bytes): 67 init_msg = init_msg.decode("utf-8") 68 data: Dict[str, Any] = json.loads(init_msg) 69 client_type = data.get("client", "Unknown") 70 71 if client_type == "frontend": 72 logging.info("Frontend connected.") 73 self.frontend_ws = websocket 74 await self.update_frontend() 75 await self.frontend_loop(websocket) 76 elif client_type == "agent": 77 if not self.agent1_ws: 78 self.agent1_ws = websocket 79 logging.info("Player 1 (X) connected.") 80 await websocket.send(json.dumps({"type": "setup", "player_id": 1})) 81 # Start the agent loop and check conditions in parallel 82 await asyncio.gather( 83 self.agent_loop(websocket, 1), 84 self.check_start_conditions() 85 ) 86 elif not self.agent2_ws: 87 self.agent2_ws = websocket 88 logging.info("Player 2 (O) connected.") 89 await websocket.send(json.dumps({"type": "setup", "player_id": 2})) 90 # Start the agent loop and check conditions in parallel 91 await asyncio.gather( 92 self.agent_loop(websocket, 2), 93 self.check_start_conditions() 94 ) 95 else: 96 await websocket.close() 97 except Exception as e: 98 logging.error(f"Error: {e}") 99 finally: 100 if websocket == self.frontend_ws: 101 self.frontend_ws = None 102 elif websocket == self.agent1_ws: 103 self.agent1_ws = None 104 self.running = False 105 await self.update_frontend() 106 elif websocket == self.agent2_ws: 107 self.agent2_ws = None 108 self.running = False 109 await self.update_frontend() 110 111 async def frontend_loop(self, websocket: WebSocketServerProtocol) -> None: 112 """ 113 Main loop for handling frontend communication. 114 115 Args: 116 websocket (WebSocketServerProtocol): The connected frontend client. 117 """ 118 async for _ in websocket: 119 pass 120 121 async def agent_loop(self, websocket: WebSocketServerProtocol, player_id: int) -> None: 122 """ 123 Main loop for handling agent communication. 124 125 Args: 126 websocket (WebSocketServerProtocol): The connected agent client. 127 player_id (int): The ID of the player (1 or 2). 128 """ 129 async for message in websocket: 130 if not self.running or self.current_turn != player_id: 131 continue 132 try: 133 if isinstance(message, bytes): 134 message = message.decode("utf-8") 135 data: Dict[str, Any] = json.loads(message) 136 if data.get("action") == "move": 137 x, y = data.get("x"), data.get("y") 138 if x is not None and y is not None and self.process_move(player_id, x, y): 139 await self.check_game_over() 140 if self.running: 141 self.current_turn = 3 - self.current_turn 142 await self.broadcast_state() 143 await self.update_frontend() 144 except Exception as e: 145 logging.error(f"Error processing move: {e}") 146 147 async def check_start_conditions(self) -> None: 148 """ 149 Checks if both agents are connected and starts the game if so. 150 """ 151 if self.agent1_ws and self.agent2_ws and not self.running: 152 self.running = True 153 self.board = [[0] * 9 for _ in range(9)] 154 self.macro_board = [[0] * 3 for _ in range(3)] 155 self.active_macro = None 156 self.current_turn = self.first_player_this_round 157 await self.update_frontend() 158 # Give enough time for the loops to start and agents to be ready 159 await asyncio.sleep(1.0) 160 await self.broadcast_state() 161 162 def get_valid_actions(self) -> List[List[int]]: 163 """ 164 Returns a list of all currently valid actions for the current player. 165 166 Returns: 167 List[List[int]]: A list of [x, y] coordinates representing valid moves. 168 """ 169 actions: List[List[int]] = [] 170 for y in range(9): 171 for x in range(9): 172 my, mx = y // 3, x // 3 173 # Cannot play in a resolved macro-board 174 if self.macro_board[my][mx] != 0: 175 continue 176 # Cannot play in an occupied cell 177 if self.board[y][x] != 0: 178 continue 179 # Must play in the active macro-board, unless free move is granted 180 if self.active_macro is not None and self.active_macro != [my, mx]: 181 continue 182 183 actions.append([x, y]) 184 return actions 185 186 def process_move(self, player_id: int, x: int, y: int) -> bool: 187 """ 188 Processes a move from a player. 189 190 Args: 191 player_id (int): The ID of the player making the move. 192 x (int): The x-coordinate of the move. 193 y (int): The y-coordinate of the move. 194 195 Returns: 196 bool: True if the move was processed successfully, False otherwise. 197 """ 198 if [x, y] not in self.get_valid_actions(): 199 return False 200 201 self.board[y][x] = player_id 202 my, mx = y // 3, x // 3 203 micro_y, micro_x = y % 3, x % 3 204 205 # Check if this move won the local macro-board 206 local_winner = self.check_3x3_win(self.board, mx * 3, my * 3) 207 if local_winner: 208 self.macro_board[my][mx] = local_winner 209 elif self.is_3x3_full(self.board, mx * 3, my * 3): 210 self.macro_board[my][mx] = 3 # Draw 211 212 # Set next active macro-board 213 next_my, next_mx = micro_y, micro_x 214 if self.macro_board[next_my][next_mx] != 0: 215 self.active_macro = None # Free move! 216 else: 217 self.active_macro = [next_my, next_mx] 218 219 return True 220 221 def check_3x3_win(self, grid: List[List[int]], start_x: int, start_y: int) -> int: 222 """ 223 Checks for a win in a specific 3x3 subset of a grid. 224 225 Args: 226 grid (List[List[int]]): The grid to check (9x9 or 3x3). 227 start_x (int): The starting x-coordinate of the 3x3 subset. 228 start_y (int): The starting y-coordinate of the 3x3 subset. 229 230 Returns: 231 int: The ID of the winning player (1 or 2), 0 if no winner, 3 if draw. 232 """ 233 for i in range(3): 234 # Rows 235 if ( 236 grid[start_y + i][start_x] != 0 237 and grid[start_y + i][start_x] 238 == grid[start_y + i][start_x + 1] 239 == grid[start_y + i][start_x + 2] 240 ): 241 return grid[start_y + i][start_x] 242 # Cols 243 if ( 244 grid[start_y][start_x + i] != 0 245 and grid[start_y][start_x + i] 246 == grid[start_y + 1][start_x + i] 247 == grid[start_y + 2][start_x + i] 248 ): 249 return grid[start_y][start_x + i] 250 # Diagonals 251 if ( 252 grid[start_y][start_x] != 0 253 and grid[start_y][start_x] 254 == grid[start_y + 1][start_x + 1] 255 == grid[start_y + 2][start_x + 2] 256 ): 257 return grid[start_y][start_x] 258 if ( 259 grid[start_y + 2][start_x] != 0 260 and grid[start_y + 2][start_x] 261 == grid[start_y + 1][start_x + 1] 262 == grid[start_y][start_x + 2] 263 ): 264 return grid[start_y + 2][start_x] 265 return 0 266 267 def is_3x3_full(self, grid: List[List[int]], start_x: int, start_y: int) -> bool: 268 """ 269 Checks if a 3x3 subset of a grid is full. 270 271 Args: 272 grid (List[List[int]]): The grid to check. 273 start_x (int): The starting x-coordinate of the 3x3 subset. 274 start_y (int): The starting y-coordinate of the 3x3 subset. 275 276 Returns: 277 bool: True if the 3x3 subset is full, False otherwise. 278 """ 279 for y in range(3): 280 for x in range(3): 281 if grid[start_y + y][start_x + x] == 0: 282 return False 283 return True 284 285 async def check_game_over(self) -> None: 286 """ 287 Checks if the game is over and handles the results. 288 """ 289 # Treat the macro_board as a standard Tic-Tac-Toe board 290 winner = self.check_3x3_win(self.macro_board, 0, 0) 291 is_draw = self.is_3x3_full(self.macro_board, 0, 0) 292 293 if winner in [1, 2]: 294 self.match_scores[winner] += 1 295 await self.end_round(f"Player {winner} Wins!") 296 elif is_draw or winner == 3: 297 await self.end_round("Global Draw!") 298 299 async def end_round(self, message: str) -> None: 300 """ 301 Ends the current round. 302 303 Args: 304 message (str): The message to display at the end of the round. 305 """ 306 self.running = False 307 payload = {"type": "game_over", "message": message} 308 if self.agent1_ws: 309 await self.agent1_ws.send(json.dumps(payload)) 310 if self.agent2_ws: 311 await self.agent2_ws.send(json.dumps(payload)) 312 await self.update_frontend() 313 314 await asyncio.sleep(3.0) 315 self.first_player_this_round = 3 - self.first_player_this_round 316 await self.check_start_conditions() 317 318 async def broadcast_state(self) -> None: 319 """ 320 Broadcasts the current game state to both agents. 321 """ 322 payload = { 323 "type": "state", 324 "current_turn": self.current_turn, 325 "board": self.board, 326 "macro_board": self.macro_board, 327 "active_macro": self.active_macro, 328 "valid_actions": self.get_valid_actions(), 329 } 330 msg = json.dumps(payload) 331 if self.agent1_ws: 332 await self.agent1_ws.send(msg) 333 if self.agent2_ws: 334 await self.agent2_ws.send(msg) 335 336 async def update_frontend(self) -> None: 337 """ 338 Sends an update to the frontend. 339 """ 340 if self.frontend_ws: 341 await self.frontend_ws.send( 342 json.dumps( 343 { 344 "type": "update", 345 "current_turn": self.current_turn, 346 "board": self.board, 347 "macro_board": self.macro_board, 348 "active_macro": self.active_macro, 349 "match_scores": self.match_scores, 350 "p1_connected": self.agent1_ws is not None, 351 "p2_connected": self.agent2_ws is not None, 352 } 353 ) 354 ) 355 356 357if __name__ == "__main__": 358 server = UTTTServer() 359 asyncio.run(server.start()) 360 361 362if __name__ == "__main__": 363 server = UTTTServer() 364 asyncio.run(server.start())
class
UTTTServer:
15class UTTTServer: 16 """ 17 Ultimate Tic-Tac-Toe (UTTT) Server. 18 19 Handles map loading, agent movement, and state broadcasting. 20 The game is played on a $9 \times 9$ micro-board, divided into $3 \times 3$ macro-boards. 21 """ 22 23 def __init__(self) -> None: 24 """ 25 Initializes the UTTTServer. 26 """ 27 self.frontend_ws: Optional[WebSocketServerProtocol] = None 28 self.agent1_ws: Optional[WebSocketServerProtocol] = None 29 self.agent2_ws: Optional[WebSocketServerProtocol] = None 30 31 # 9x9 Micro Board (0=Empty, 1=P1, 2=P2) 32 self.board: List[List[int]] = [[0] * 9 for _ in range(9)] 33 # 3x3 Macro Board (0=Ongoing, 1=P1 Win, 2=P2 Win, 3=Draw) 34 self.macro_board: List[List[int]] = [[0] * 3 for _ in range(3)] 35 36 # (my, mx) indicating which macro-board the current player MUST play in. 37 # None means the player can play in ANY available macro-board. 38 self.active_macro: Optional[List[int]] = None 39 40 self.first_player_this_round: int = 1 41 self.current_turn: int = 1 42 self.running: bool = False 43 self.match_scores: Dict[int, int] = {1: 0, 2: 0} 44 45 async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None: 46 """ 47 Starts the UTTT server. 48 49 Args: 50 host (str): The host address to bind to. 51 port (int): The port to listen on. 52 """ 53 logging.info(f"UTTT Server started on ws://{host}:{port}") 54 async with websockets.serve(self.handle_client, host, port): 55 await asyncio.Future() 56 57 async def handle_client(self, websocket: WebSocketServerProtocol) -> None: 58 """ 59 Handles incoming WebSocket connections. 60 61 Args: 62 websocket (WebSocketServerProtocol): The connected WebSocket client. 63 """ 64 client_type = "Unknown" 65 try: 66 init_msg = await websocket.recv() 67 if isinstance(init_msg, bytes): 68 init_msg = init_msg.decode("utf-8") 69 data: Dict[str, Any] = json.loads(init_msg) 70 client_type = data.get("client", "Unknown") 71 72 if client_type == "frontend": 73 logging.info("Frontend connected.") 74 self.frontend_ws = websocket 75 await self.update_frontend() 76 await self.frontend_loop(websocket) 77 elif client_type == "agent": 78 if not self.agent1_ws: 79 self.agent1_ws = websocket 80 logging.info("Player 1 (X) connected.") 81 await websocket.send(json.dumps({"type": "setup", "player_id": 1})) 82 # Start the agent loop and check conditions in parallel 83 await asyncio.gather( 84 self.agent_loop(websocket, 1), 85 self.check_start_conditions() 86 ) 87 elif not self.agent2_ws: 88 self.agent2_ws = websocket 89 logging.info("Player 2 (O) connected.") 90 await websocket.send(json.dumps({"type": "setup", "player_id": 2})) 91 # Start the agent loop and check conditions in parallel 92 await asyncio.gather( 93 self.agent_loop(websocket, 2), 94 self.check_start_conditions() 95 ) 96 else: 97 await websocket.close() 98 except Exception as e: 99 logging.error(f"Error: {e}") 100 finally: 101 if websocket == self.frontend_ws: 102 self.frontend_ws = None 103 elif websocket == self.agent1_ws: 104 self.agent1_ws = None 105 self.running = False 106 await self.update_frontend() 107 elif websocket == self.agent2_ws: 108 self.agent2_ws = None 109 self.running = False 110 await self.update_frontend() 111 112 async def frontend_loop(self, websocket: WebSocketServerProtocol) -> None: 113 """ 114 Main loop for handling frontend communication. 115 116 Args: 117 websocket (WebSocketServerProtocol): The connected frontend client. 118 """ 119 async for _ in websocket: 120 pass 121 122 async def agent_loop(self, websocket: WebSocketServerProtocol, player_id: int) -> None: 123 """ 124 Main loop for handling agent communication. 125 126 Args: 127 websocket (WebSocketServerProtocol): The connected agent client. 128 player_id (int): The ID of the player (1 or 2). 129 """ 130 async for message in websocket: 131 if not self.running or self.current_turn != player_id: 132 continue 133 try: 134 if isinstance(message, bytes): 135 message = message.decode("utf-8") 136 data: Dict[str, Any] = json.loads(message) 137 if data.get("action") == "move": 138 x, y = data.get("x"), data.get("y") 139 if x is not None and y is not None and self.process_move(player_id, x, y): 140 await self.check_game_over() 141 if self.running: 142 self.current_turn = 3 - self.current_turn 143 await self.broadcast_state() 144 await self.update_frontend() 145 except Exception as e: 146 logging.error(f"Error processing move: {e}") 147 148 async def check_start_conditions(self) -> None: 149 """ 150 Checks if both agents are connected and starts the game if so. 151 """ 152 if self.agent1_ws and self.agent2_ws and not self.running: 153 self.running = True 154 self.board = [[0] * 9 for _ in range(9)] 155 self.macro_board = [[0] * 3 for _ in range(3)] 156 self.active_macro = None 157 self.current_turn = self.first_player_this_round 158 await self.update_frontend() 159 # Give enough time for the loops to start and agents to be ready 160 await asyncio.sleep(1.0) 161 await self.broadcast_state() 162 163 def get_valid_actions(self) -> List[List[int]]: 164 """ 165 Returns a list of all currently valid actions for the current player. 166 167 Returns: 168 List[List[int]]: A list of [x, y] coordinates representing valid moves. 169 """ 170 actions: List[List[int]] = [] 171 for y in range(9): 172 for x in range(9): 173 my, mx = y // 3, x // 3 174 # Cannot play in a resolved macro-board 175 if self.macro_board[my][mx] != 0: 176 continue 177 # Cannot play in an occupied cell 178 if self.board[y][x] != 0: 179 continue 180 # Must play in the active macro-board, unless free move is granted 181 if self.active_macro is not None and self.active_macro != [my, mx]: 182 continue 183 184 actions.append([x, y]) 185 return actions 186 187 def process_move(self, player_id: int, x: int, y: int) -> bool: 188 """ 189 Processes a move from a player. 190 191 Args: 192 player_id (int): The ID of the player making the move. 193 x (int): The x-coordinate of the move. 194 y (int): The y-coordinate of the move. 195 196 Returns: 197 bool: True if the move was processed successfully, False otherwise. 198 """ 199 if [x, y] not in self.get_valid_actions(): 200 return False 201 202 self.board[y][x] = player_id 203 my, mx = y // 3, x // 3 204 micro_y, micro_x = y % 3, x % 3 205 206 # Check if this move won the local macro-board 207 local_winner = self.check_3x3_win(self.board, mx * 3, my * 3) 208 if local_winner: 209 self.macro_board[my][mx] = local_winner 210 elif self.is_3x3_full(self.board, mx * 3, my * 3): 211 self.macro_board[my][mx] = 3 # Draw 212 213 # Set next active macro-board 214 next_my, next_mx = micro_y, micro_x 215 if self.macro_board[next_my][next_mx] != 0: 216 self.active_macro = None # Free move! 217 else: 218 self.active_macro = [next_my, next_mx] 219 220 return True 221 222 def check_3x3_win(self, grid: List[List[int]], start_x: int, start_y: int) -> int: 223 """ 224 Checks for a win in a specific 3x3 subset of a grid. 225 226 Args: 227 grid (List[List[int]]): The grid to check (9x9 or 3x3). 228 start_x (int): The starting x-coordinate of the 3x3 subset. 229 start_y (int): The starting y-coordinate of the 3x3 subset. 230 231 Returns: 232 int: The ID of the winning player (1 or 2), 0 if no winner, 3 if draw. 233 """ 234 for i in range(3): 235 # Rows 236 if ( 237 grid[start_y + i][start_x] != 0 238 and grid[start_y + i][start_x] 239 == grid[start_y + i][start_x + 1] 240 == grid[start_y + i][start_x + 2] 241 ): 242 return grid[start_y + i][start_x] 243 # Cols 244 if ( 245 grid[start_y][start_x + i] != 0 246 and grid[start_y][start_x + i] 247 == grid[start_y + 1][start_x + i] 248 == grid[start_y + 2][start_x + i] 249 ): 250 return grid[start_y][start_x + i] 251 # Diagonals 252 if ( 253 grid[start_y][start_x] != 0 254 and grid[start_y][start_x] 255 == grid[start_y + 1][start_x + 1] 256 == grid[start_y + 2][start_x + 2] 257 ): 258 return grid[start_y][start_x] 259 if ( 260 grid[start_y + 2][start_x] != 0 261 and grid[start_y + 2][start_x] 262 == grid[start_y + 1][start_x + 1] 263 == grid[start_y][start_x + 2] 264 ): 265 return grid[start_y + 2][start_x] 266 return 0 267 268 def is_3x3_full(self, grid: List[List[int]], start_x: int, start_y: int) -> bool: 269 """ 270 Checks if a 3x3 subset of a grid is full. 271 272 Args: 273 grid (List[List[int]]): The grid to check. 274 start_x (int): The starting x-coordinate of the 3x3 subset. 275 start_y (int): The starting y-coordinate of the 3x3 subset. 276 277 Returns: 278 bool: True if the 3x3 subset is full, False otherwise. 279 """ 280 for y in range(3): 281 for x in range(3): 282 if grid[start_y + y][start_x + x] == 0: 283 return False 284 return True 285 286 async def check_game_over(self) -> None: 287 """ 288 Checks if the game is over and handles the results. 289 """ 290 # Treat the macro_board as a standard Tic-Tac-Toe board 291 winner = self.check_3x3_win(self.macro_board, 0, 0) 292 is_draw = self.is_3x3_full(self.macro_board, 0, 0) 293 294 if winner in [1, 2]: 295 self.match_scores[winner] += 1 296 await self.end_round(f"Player {winner} Wins!") 297 elif is_draw or winner == 3: 298 await self.end_round("Global Draw!") 299 300 async def end_round(self, message: str) -> None: 301 """ 302 Ends the current round. 303 304 Args: 305 message (str): The message to display at the end of the round. 306 """ 307 self.running = False 308 payload = {"type": "game_over", "message": message} 309 if self.agent1_ws: 310 await self.agent1_ws.send(json.dumps(payload)) 311 if self.agent2_ws: 312 await self.agent2_ws.send(json.dumps(payload)) 313 await self.update_frontend() 314 315 await asyncio.sleep(3.0) 316 self.first_player_this_round = 3 - self.first_player_this_round 317 await self.check_start_conditions() 318 319 async def broadcast_state(self) -> None: 320 """ 321 Broadcasts the current game state to both agents. 322 """ 323 payload = { 324 "type": "state", 325 "current_turn": self.current_turn, 326 "board": self.board, 327 "macro_board": self.macro_board, 328 "active_macro": self.active_macro, 329 "valid_actions": self.get_valid_actions(), 330 } 331 msg = json.dumps(payload) 332 if self.agent1_ws: 333 await self.agent1_ws.send(msg) 334 if self.agent2_ws: 335 await self.agent2_ws.send(msg) 336 337 async def update_frontend(self) -> None: 338 """ 339 Sends an update to the frontend. 340 """ 341 if self.frontend_ws: 342 await self.frontend_ws.send( 343 json.dumps( 344 { 345 "type": "update", 346 "current_turn": self.current_turn, 347 "board": self.board, 348 "macro_board": self.macro_board, 349 "active_macro": self.active_macro, 350 "match_scores": self.match_scores, 351 "p1_connected": self.agent1_ws is not None, 352 "p2_connected": self.agent2_ws is not None, 353 } 354 ) 355 )
Ultimate Tic-Tac-Toe (UTTT) Server.
Handles map loading, agent movement, and state broadcasting. The game is played on a $9 imes 9$ micro-board, divided into $3 imes 3$ macro-boards.
UTTTServer()
23 def __init__(self) -> None: 24 """ 25 Initializes the UTTTServer. 26 """ 27 self.frontend_ws: Optional[WebSocketServerProtocol] = None 28 self.agent1_ws: Optional[WebSocketServerProtocol] = None 29 self.agent2_ws: Optional[WebSocketServerProtocol] = None 30 31 # 9x9 Micro Board (0=Empty, 1=P1, 2=P2) 32 self.board: List[List[int]] = [[0] * 9 for _ in range(9)] 33 # 3x3 Macro Board (0=Ongoing, 1=P1 Win, 2=P2 Win, 3=Draw) 34 self.macro_board: List[List[int]] = [[0] * 3 for _ in range(3)] 35 36 # (my, mx) indicating which macro-board the current player MUST play in. 37 # None means the player can play in ANY available macro-board. 38 self.active_macro: Optional[List[int]] = None 39 40 self.first_player_this_round: int = 1 41 self.current_turn: int = 1 42 self.running: bool = False 43 self.match_scores: Dict[int, int] = {1: 0, 2: 0}
Initializes the UTTTServer.
async def
start(self, host: str = '0.0.0.0', port: int = 8765) -> None:
45 async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None: 46 """ 47 Starts the UTTT server. 48 49 Args: 50 host (str): The host address to bind to. 51 port (int): The port to listen on. 52 """ 53 logging.info(f"UTTT Server started on ws://{host}:{port}") 54 async with websockets.serve(self.handle_client, host, port): 55 await asyncio.Future()
Starts the UTTT server.
Args: host (str): The host address to bind to. port (int): The port to listen on.
async def
handle_client( self, websocket: websockets.legacy.server.WebSocketServerProtocol) -> None:
57 async def handle_client(self, websocket: WebSocketServerProtocol) -> None: 58 """ 59 Handles incoming WebSocket connections. 60 61 Args: 62 websocket (WebSocketServerProtocol): The connected WebSocket client. 63 """ 64 client_type = "Unknown" 65 try: 66 init_msg = await websocket.recv() 67 if isinstance(init_msg, bytes): 68 init_msg = init_msg.decode("utf-8") 69 data: Dict[str, Any] = json.loads(init_msg) 70 client_type = data.get("client", "Unknown") 71 72 if client_type == "frontend": 73 logging.info("Frontend connected.") 74 self.frontend_ws = websocket 75 await self.update_frontend() 76 await self.frontend_loop(websocket) 77 elif client_type == "agent": 78 if not self.agent1_ws: 79 self.agent1_ws = websocket 80 logging.info("Player 1 (X) connected.") 81 await websocket.send(json.dumps({"type": "setup", "player_id": 1})) 82 # Start the agent loop and check conditions in parallel 83 await asyncio.gather( 84 self.agent_loop(websocket, 1), 85 self.check_start_conditions() 86 ) 87 elif not self.agent2_ws: 88 self.agent2_ws = websocket 89 logging.info("Player 2 (O) connected.") 90 await websocket.send(json.dumps({"type": "setup", "player_id": 2})) 91 # Start the agent loop and check conditions in parallel 92 await asyncio.gather( 93 self.agent_loop(websocket, 2), 94 self.check_start_conditions() 95 ) 96 else: 97 await websocket.close() 98 except Exception as e: 99 logging.error(f"Error: {e}") 100 finally: 101 if websocket == self.frontend_ws: 102 self.frontend_ws = None 103 elif websocket == self.agent1_ws: 104 self.agent1_ws = None 105 self.running = False 106 await self.update_frontend() 107 elif websocket == self.agent2_ws: 108 self.agent2_ws = None 109 self.running = False 110 await self.update_frontend()
Handles incoming WebSocket connections.
Args: websocket (WebSocketServerProtocol): The connected WebSocket client.
async def
frontend_loop( self, websocket: websockets.legacy.server.WebSocketServerProtocol) -> None:
112 async def frontend_loop(self, websocket: WebSocketServerProtocol) -> None: 113 """ 114 Main loop for handling frontend communication. 115 116 Args: 117 websocket (WebSocketServerProtocol): The connected frontend client. 118 """ 119 async for _ in websocket: 120 pass
Main loop for handling frontend communication.
Args: websocket (WebSocketServerProtocol): The connected frontend client.
async def
agent_loop( self, websocket: websockets.legacy.server.WebSocketServerProtocol, player_id: int) -> None:
122 async def agent_loop(self, websocket: WebSocketServerProtocol, player_id: int) -> None: 123 """ 124 Main loop for handling agent communication. 125 126 Args: 127 websocket (WebSocketServerProtocol): The connected agent client. 128 player_id (int): The ID of the player (1 or 2). 129 """ 130 async for message in websocket: 131 if not self.running or self.current_turn != player_id: 132 continue 133 try: 134 if isinstance(message, bytes): 135 message = message.decode("utf-8") 136 data: Dict[str, Any] = json.loads(message) 137 if data.get("action") == "move": 138 x, y = data.get("x"), data.get("y") 139 if x is not None and y is not None and self.process_move(player_id, x, y): 140 await self.check_game_over() 141 if self.running: 142 self.current_turn = 3 - self.current_turn 143 await self.broadcast_state() 144 await self.update_frontend() 145 except Exception as e: 146 logging.error(f"Error processing move: {e}")
Main loop for handling agent communication.
Args: websocket (WebSocketServerProtocol): The connected agent client. player_id (int): The ID of the player (1 or 2).
async def
check_start_conditions(self) -> None:
148 async def check_start_conditions(self) -> None: 149 """ 150 Checks if both agents are connected and starts the game if so. 151 """ 152 if self.agent1_ws and self.agent2_ws and not self.running: 153 self.running = True 154 self.board = [[0] * 9 for _ in range(9)] 155 self.macro_board = [[0] * 3 for _ in range(3)] 156 self.active_macro = None 157 self.current_turn = self.first_player_this_round 158 await self.update_frontend() 159 # Give enough time for the loops to start and agents to be ready 160 await asyncio.sleep(1.0) 161 await self.broadcast_state()
Checks if both agents are connected and starts the game if so.
def
get_valid_actions(self) -> List[List[int]]:
163 def get_valid_actions(self) -> List[List[int]]: 164 """ 165 Returns a list of all currently valid actions for the current player. 166 167 Returns: 168 List[List[int]]: A list of [x, y] coordinates representing valid moves. 169 """ 170 actions: List[List[int]] = [] 171 for y in range(9): 172 for x in range(9): 173 my, mx = y // 3, x // 3 174 # Cannot play in a resolved macro-board 175 if self.macro_board[my][mx] != 0: 176 continue 177 # Cannot play in an occupied cell 178 if self.board[y][x] != 0: 179 continue 180 # Must play in the active macro-board, unless free move is granted 181 if self.active_macro is not None and self.active_macro != [my, mx]: 182 continue 183 184 actions.append([x, y]) 185 return actions
Returns a list of all currently valid actions for the current player.
Returns: List[List[int]]: A list of [x, y] coordinates representing valid moves.
def
process_move(self, player_id: int, x: int, y: int) -> bool:
187 def process_move(self, player_id: int, x: int, y: int) -> bool: 188 """ 189 Processes a move from a player. 190 191 Args: 192 player_id (int): The ID of the player making the move. 193 x (int): The x-coordinate of the move. 194 y (int): The y-coordinate of the move. 195 196 Returns: 197 bool: True if the move was processed successfully, False otherwise. 198 """ 199 if [x, y] not in self.get_valid_actions(): 200 return False 201 202 self.board[y][x] = player_id 203 my, mx = y // 3, x // 3 204 micro_y, micro_x = y % 3, x % 3 205 206 # Check if this move won the local macro-board 207 local_winner = self.check_3x3_win(self.board, mx * 3, my * 3) 208 if local_winner: 209 self.macro_board[my][mx] = local_winner 210 elif self.is_3x3_full(self.board, mx * 3, my * 3): 211 self.macro_board[my][mx] = 3 # Draw 212 213 # Set next active macro-board 214 next_my, next_mx = micro_y, micro_x 215 if self.macro_board[next_my][next_mx] != 0: 216 self.active_macro = None # Free move! 217 else: 218 self.active_macro = [next_my, next_mx] 219 220 return True
Processes a move from a player.
Args: player_id (int): The ID of the player making the move. x (int): The x-coordinate of the move. y (int): The y-coordinate of the move.
Returns: bool: True if the move was processed successfully, False otherwise.
def
check_3x3_win(self, grid: List[List[int]], start_x: int, start_y: int) -> int:
222 def check_3x3_win(self, grid: List[List[int]], start_x: int, start_y: int) -> int: 223 """ 224 Checks for a win in a specific 3x3 subset of a grid. 225 226 Args: 227 grid (List[List[int]]): The grid to check (9x9 or 3x3). 228 start_x (int): The starting x-coordinate of the 3x3 subset. 229 start_y (int): The starting y-coordinate of the 3x3 subset. 230 231 Returns: 232 int: The ID of the winning player (1 or 2), 0 if no winner, 3 if draw. 233 """ 234 for i in range(3): 235 # Rows 236 if ( 237 grid[start_y + i][start_x] != 0 238 and grid[start_y + i][start_x] 239 == grid[start_y + i][start_x + 1] 240 == grid[start_y + i][start_x + 2] 241 ): 242 return grid[start_y + i][start_x] 243 # Cols 244 if ( 245 grid[start_y][start_x + i] != 0 246 and grid[start_y][start_x + i] 247 == grid[start_y + 1][start_x + i] 248 == grid[start_y + 2][start_x + i] 249 ): 250 return grid[start_y][start_x + i] 251 # Diagonals 252 if ( 253 grid[start_y][start_x] != 0 254 and grid[start_y][start_x] 255 == grid[start_y + 1][start_x + 1] 256 == grid[start_y + 2][start_x + 2] 257 ): 258 return grid[start_y][start_x] 259 if ( 260 grid[start_y + 2][start_x] != 0 261 and grid[start_y + 2][start_x] 262 == grid[start_y + 1][start_x + 1] 263 == grid[start_y][start_x + 2] 264 ): 265 return grid[start_y + 2][start_x] 266 return 0
Checks for a win in a specific 3x3 subset of a grid.
Args: grid (List[List[int]]): The grid to check (9x9 or 3x3). start_x (int): The starting x-coordinate of the 3x3 subset. start_y (int): The starting y-coordinate of the 3x3 subset.
Returns: int: The ID of the winning player (1 or 2), 0 if no winner, 3 if draw.
def
is_3x3_full(self, grid: List[List[int]], start_x: int, start_y: int) -> bool:
268 def is_3x3_full(self, grid: List[List[int]], start_x: int, start_y: int) -> bool: 269 """ 270 Checks if a 3x3 subset of a grid is full. 271 272 Args: 273 grid (List[List[int]]): The grid to check. 274 start_x (int): The starting x-coordinate of the 3x3 subset. 275 start_y (int): The starting y-coordinate of the 3x3 subset. 276 277 Returns: 278 bool: True if the 3x3 subset is full, False otherwise. 279 """ 280 for y in range(3): 281 for x in range(3): 282 if grid[start_y + y][start_x + x] == 0: 283 return False 284 return True
Checks if a 3x3 subset of a grid is full.
Args: grid (List[List[int]]): The grid to check. start_x (int): The starting x-coordinate of the 3x3 subset. start_y (int): The starting y-coordinate of the 3x3 subset.
Returns: bool: True if the 3x3 subset is full, False otherwise.
async def
check_game_over(self) -> None:
286 async def check_game_over(self) -> None: 287 """ 288 Checks if the game is over and handles the results. 289 """ 290 # Treat the macro_board as a standard Tic-Tac-Toe board 291 winner = self.check_3x3_win(self.macro_board, 0, 0) 292 is_draw = self.is_3x3_full(self.macro_board, 0, 0) 293 294 if winner in [1, 2]: 295 self.match_scores[winner] += 1 296 await self.end_round(f"Player {winner} Wins!") 297 elif is_draw or winner == 3: 298 await self.end_round("Global Draw!")
Checks if the game is over and handles the results.
async def
end_round(self, message: str) -> None:
300 async def end_round(self, message: str) -> None: 301 """ 302 Ends the current round. 303 304 Args: 305 message (str): The message to display at the end of the round. 306 """ 307 self.running = False 308 payload = {"type": "game_over", "message": message} 309 if self.agent1_ws: 310 await self.agent1_ws.send(json.dumps(payload)) 311 if self.agent2_ws: 312 await self.agent2_ws.send(json.dumps(payload)) 313 await self.update_frontend() 314 315 await asyncio.sleep(3.0) 316 self.first_player_this_round = 3 - self.first_player_this_round 317 await self.check_start_conditions()
Ends the current round.
Args: message (str): The message to display at the end of the round.
async def
broadcast_state(self) -> None:
319 async def broadcast_state(self) -> None: 320 """ 321 Broadcasts the current game state to both agents. 322 """ 323 payload = { 324 "type": "state", 325 "current_turn": self.current_turn, 326 "board": self.board, 327 "macro_board": self.macro_board, 328 "active_macro": self.active_macro, 329 "valid_actions": self.get_valid_actions(), 330 } 331 msg = json.dumps(payload) 332 if self.agent1_ws: 333 await self.agent1_ws.send(msg) 334 if self.agent2_ws: 335 await self.agent2_ws.send(msg)
Broadcasts the current game state to both agents.
async def
update_frontend(self) -> None:
337 async def update_frontend(self) -> None: 338 """ 339 Sends an update to the frontend. 340 """ 341 if self.frontend_ws: 342 await self.frontend_ws.send( 343 json.dumps( 344 { 345 "type": "update", 346 "current_turn": self.current_turn, 347 "board": self.board, 348 "macro_board": self.macro_board, 349 "active_macro": self.active_macro, 350 "match_scores": self.match_scores, 351 "p1_connected": self.agent1_ws is not None, 352 "p2_connected": self.agent2_ws is not None, 353 } 354 ) 355 )
Sends an update to the frontend.