agents.base_agent 

 1import json
 2import logging
 3from typing import List, Optional, Any, Tuple
 4
 5import websockets
 6
 7logging.basicConfig(level=logging.INFO, format="%(asctime)s - AGENT - %(message)s")
 8
 9
10class BaseOthelloAgent:
11    r"""
12    Abstract base class for Othello agents.
13
14    Subclasses MUST implement the deliberate() method to define their strategy.
15    The goal is to maximize the number of discs of color $C$ on the board:
16    $\max \sum_{i,j} \mathbb{1}(B_{i,j} = C)$, where $B_{i,j}$ is the board cell.
17    """
18
19    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
20        """
21        Initializes the Othello agent.
22
23        Args:
24            server_uri: The URI of the Othello server. Defaults to "ws://localhost:8765".
25        """
26        self.server_uri: str = server_uri
27        self.player_id: Optional[int] = None
28
29    async def run(self) -> None:
30        """
31        Connects to the server and enters the main event loop.
32
33        Handles setup, game state updates, and game over messages.
34        Utilizes an asynchronous event loop to process WebSocket messages.
35        """
36        try:
37            async with websockets.connect(self.server_uri) as websocket:
38                await websocket.send(json.dumps({"client": "agent"}))
39
40                async for message in websocket:
41                    data: Any = json.loads(message)
42
43                    if data.get("type") == "setup":
44                        self.player_id = data.get("player_id")
45                        color = "Black" if self.player_id == 1 else "White"
46                        logging.info(
47                            f"Connected! Assigned Player {self.player_id} ({color})"
48                        )
49
50                    elif data.get("type") == "state":
51                        current_turn = data.get("current_turn")
52                        valid_actions = data.get("valid_actions")
53                        board = data.get("board")
54
55                        if current_turn == self.player_id:
56                            # It's our turn! Ask the subclass to evaluate the board
57                            action = await self.deliberate(board, valid_actions)
58
59                            if action is not None:
60                                await websocket.send(
61                                    json.dumps(
62                                        {
63                                            "action": "move",
64                                            "x": action[0],
65                                            "y": action[1],
66                                        }
67                                    )
68                                )
69
70                    elif data.get("type") == "game_over":
71                        logging.info(f"Match Over: {data.get('message')}")
72                        logging.info("Waiting for next round to start...")
73
74        except Exception as e:
75            logging.error(f"Connection lost: {e}")
76
77    async def deliberate(
78        self, board: List[List[int]], valid_actions: List[List[int]]
79    ) -> Optional[Tuple[int, int]]:
80        """
81        MUST be implemented by subclasses.
82
83        Args:
84            board: The current $8 \times 8$ game board.
85            valid_actions: A list of valid move coordinates $[x, y]$.
86
87        Returns:
88            The chosen move $(x, y)$ or None to skip.
89        """
90        raise NotImplementedError("Subclasses must implement deliberate()")
class BaseOthelloAgent: 
11class BaseOthelloAgent:
12    r"""
13    Abstract base class for Othello agents.
14
15    Subclasses MUST implement the deliberate() method to define their strategy.
16    The goal is to maximize the number of discs of color $C$ on the board:
17    $\max \sum_{i,j} \mathbb{1}(B_{i,j} = C)$, where $B_{i,j}$ is the board cell.
18    """
19
20    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
21        """
22        Initializes the Othello agent.
23
24        Args:
25            server_uri: The URI of the Othello server. Defaults to "ws://localhost:8765".
26        """
27        self.server_uri: str = server_uri
28        self.player_id: Optional[int] = None
29
30    async def run(self) -> None:
31        """
32        Connects to the server and enters the main event loop.
33
34        Handles setup, game state updates, and game over messages.
35        Utilizes an asynchronous event loop to process WebSocket messages.
36        """
37        try:
38            async with websockets.connect(self.server_uri) as websocket:
39                await websocket.send(json.dumps({"client": "agent"}))
40
41                async for message in websocket:
42                    data: Any = json.loads(message)
43
44                    if data.get("type") == "setup":
45                        self.player_id = data.get("player_id")
46                        color = "Black" if self.player_id == 1 else "White"
47                        logging.info(
48                            f"Connected! Assigned Player {self.player_id} ({color})"
49                        )
50
51                    elif data.get("type") == "state":
52                        current_turn = data.get("current_turn")
53                        valid_actions = data.get("valid_actions")
54                        board = data.get("board")
55
56                        if current_turn == self.player_id:
57                            # It's our turn! Ask the subclass to evaluate the board
58                            action = await self.deliberate(board, valid_actions)
59
60                            if action is not None:
61                                await websocket.send(
62                                    json.dumps(
63                                        {
64                                            "action": "move",
65                                            "x": action[0],
66                                            "y": action[1],
67                                        }
68                                    )
69                                )
70
71                    elif data.get("type") == "game_over":
72                        logging.info(f"Match Over: {data.get('message')}")
73                        logging.info("Waiting for next round to start...")
74
75        except Exception as e:
76            logging.error(f"Connection lost: {e}")
77
78    async def deliberate(
79        self, board: List[List[int]], valid_actions: List[List[int]]
80    ) -> Optional[Tuple[int, int]]:
81        """
82        MUST be implemented by subclasses.
83
84        Args:
85            board: The current $8 \times 8$ game board.
86            valid_actions: A list of valid move coordinates $[x, y]$.
87
88        Returns:
89            The chosen move $(x, y)$ or None to skip.
90        """
91        raise NotImplementedError("Subclasses must implement deliberate()")

Abstract base class for Othello agents.

Subclasses MUST implement the deliberate() method to define their strategy. The goal is to maximize the number of discs of color $C$ on the board: $\max \sum_{i,j} \mathbb{1}(B_{i,j} = C)$, where $B_{i,j}$ is the board cell.

BaseOthelloAgent(server_uri: str = 'ws://localhost:8765') 
20    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
21        """
22        Initializes the Othello agent.
23
24        Args:
25            server_uri: The URI of the Othello server. Defaults to "ws://localhost:8765".
26        """
27        self.server_uri: str = server_uri
28        self.player_id: Optional[int] = None

Initializes the Othello agent.

Args: server_uri: The URI of the Othello server. Defaults to "ws://localhost:8765".

server_uri: str
player_id: Optional[int]
async def run(self) -> None: 
30    async def run(self) -> None:
31        """
32        Connects to the server and enters the main event loop.
33
34        Handles setup, game state updates, and game over messages.
35        Utilizes an asynchronous event loop to process WebSocket messages.
36        """
37        try:
38            async with websockets.connect(self.server_uri) as websocket:
39                await websocket.send(json.dumps({"client": "agent"}))
40
41                async for message in websocket:
42                    data: Any = json.loads(message)
43
44                    if data.get("type") == "setup":
45                        self.player_id = data.get("player_id")
46                        color = "Black" if self.player_id == 1 else "White"
47                        logging.info(
48                            f"Connected! Assigned Player {self.player_id} ({color})"
49                        )
50
51                    elif data.get("type") == "state":
52                        current_turn = data.get("current_turn")
53                        valid_actions = data.get("valid_actions")
54                        board = data.get("board")
55
56                        if current_turn == self.player_id:
57                            # It's our turn! Ask the subclass to evaluate the board
58                            action = await self.deliberate(board, valid_actions)
59
60                            if action is not None:
61                                await websocket.send(
62                                    json.dumps(
63                                        {
64                                            "action": "move",
65                                            "x": action[0],
66                                            "y": action[1],
67                                        }
68                                    )
69                                )
70
71                    elif data.get("type") == "game_over":
72                        logging.info(f"Match Over: {data.get('message')}")
73                        logging.info("Waiting for next round to start...")
74
75        except Exception as e:
76            logging.error(f"Connection lost: {e}")

Connects to the server and enters the main event loop.

Handles setup, game state updates, and game over messages. Utilizes an asynchronous event loop to process WebSocket messages.

async def deliberate( self, board: List[List[int]], valid_actions: List[List[int]]) -> Optional[Tuple[int, int]]: 
78    async def deliberate(
79        self, board: List[List[int]], valid_actions: List[List[int]]
80    ) -> Optional[Tuple[int, int]]:
81        """
82        MUST be implemented by subclasses.
83
84        Args:
85            board: The current $8 \times 8$ game board.
86            valid_actions: A list of valid move coordinates $[x, y]$.
87
88        Returns:
89            The chosen move $(x, y)$ or None to skip.
90        """
91        raise NotImplementedError("Subclasses must implement deliberate()")

MUST be implemented by subclasses.

Args: board: The current $8 imes 8$ game board. valid_actions: A list of valid move coordinates $[x, y]$.

Returns: The chosen move $(x, y)$ or None to skip.