agents.base_agent

 1import json
 2import logging
 3from typing import List, Optional, Tuple, Union
 4
 5import websockets
 6
 7logging.basicConfig(level=logging.INFO, format="%(asctime)s - AGENT - %(message)s")
 8
 9
10class BaseUTTTAgent:
11    """
12    Abstract base class for all Ultimate Tic-Tac-Toe agents.
13    The game is played on a $9 \times 9$ micro-board, divided into $3 \times 3$ macro-boards.
14    """
15
16    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
17        """
18        Initializes the BaseUTTTAgent.
19
20        Args:
21            server_uri (str): The URI of the UTTT server.
22        """
23        self.server_uri: str = server_uri
24        self.player_id: Optional[int] = None
25
26    async def run(self) -> None:
27        """
28        Connects to the server and enters the main communication loop.
29        """
30        try:
31            async with websockets.connect(self.server_uri) as websocket:
32                await websocket.send(json.dumps({"client": "agent"}))
33
34                async for message in websocket:
35                    if isinstance(message, bytes):
36                        message = message.decode("utf-8")
37                    data = json.loads(message)
38
39                    if data.get("type") == "setup":
40                        self.player_id = data.get("player_id")
41                        logging.info(f"Connected! Assigned Player {self.player_id}")
42
43                    elif data.get("type") == "state":
44                        current_turn = data.get("current_turn")
45                        board = data.get("board")
46                        macro_board = data.get("macro_board")
47                        active_macro = data.get("active_macro")
48                        valid_actions = data.get("valid_actions")
49
50                        if current_turn == self.player_id:
51                            action = await self.deliberate(
52                                board, macro_board, active_macro, valid_actions
53                            )
54
55                            if action is not None:
56                                await websocket.send(
57                                    json.dumps(
58                                        {
59                                            "action": "move",
60                                            "x": action[0],
61                                            "y": action[1],
62                                        }
63                                    )
64                                )
65
66                    elif data.get("type") == "game_over":
67                        logging.info(f"Match Over: {data.get('message')}")
68
69        except Exception as e:
70            logging.error(f"Connection lost: {e}")
71
72    async def deliberate(
73        self,
74        board: List[List[int]],
75        macro_board: List[List[int]],
76        active_macro: Optional[List[int]],
77        valid_actions: List[List[int]],
78    ) -> Optional[Union[List[int], Tuple[int, int]]]:
79        """
80        Deliberates on the next move. MUST be implemented by subclasses.
81
82        Args:
83            board (List[List[int]]): The current 9x9 board state.
84            macro_board (List[List[int]]): The current 3x3 macro board state.
85            active_macro (Optional[List[int]]): The active macro board coordinates [my, mx].
86            valid_actions (List[List[int]]): A list of valid moves [x, y].
87
88        Returns:
89            Optional[Union[List[int], Tuple[int, int]]]: The chosen move [x, y] or None.
90        """
91        raise NotImplementedError("Subclasses must implement deliberate()")