agents.base_agent

  1import json
  2import logging
  3from typing import Any, Dict, List, Optional, Tuple, Union, cast
  4
  5import websockets
  6
  7logging.basicConfig(level=logging.INFO, format="%(asctime)s - AGENT - %(levelname)s - %(message)s")
  8
  9
 10class BaseBSAgent:
 11    """
 12    Abstract base class for Battleship agents.
 13
 14    Subclasses MUST implement the deliberate() method to define their
 15    decision-making logic.
 16    """
 17
 18    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
 19        """
 20        Initializes the agent with the server URI.
 21
 22        Args:
 23            server_uri (str): The WebSocket URI of the Battleship server.
 24        """
 25        self.server_uri: str = server_uri
 26        self.player_id: Optional[int] = None
 27        self.board_size: Optional[int] = None
 28
 29    async def run(self) -> None:
 30        """
 31        Connects to the server and enters the main message loop.
 32        """
 33        try:
 34            async with websockets.connect(self.server_uri) as websocket:
 35                await websocket.send(json.dumps({"client": "agent"}))
 36
 37                async for message in websocket:
 38                    if not isinstance(message, str):
 39                        continue
 40
 41                    data = cast(Dict[str, Any], json.loads(message))
 42
 43                    msg_type = str(data.get("type", ""))
 44                    if msg_type == "setup":
 45                        self.player_id = cast(Optional[int], data.get("player_id"))
 46                        self.board_size = cast(Optional[int], data.get("size"))
 47                        logging.info(f"Connected! Assigned Player {self.player_id}")
 48
 49                    elif msg_type == "state":
 50                        current_turn = data.get("current_turn", -1)
 51                        my_ships_val = data.get("my_ships", [])
 52                        my_shots_val = data.get("my_shots", [])
 53                        valid_actions_val = data.get("valid_actions", [])
 54
 55                        if (
 56                            isinstance(current_turn, int)
 57                            and isinstance(my_ships_val, list)
 58                            and isinstance(my_shots_val, list)
 59                            and isinstance(valid_actions_val, list)
 60                            and self.player_id is not None
 61                            and current_turn == self.player_id
 62                        ):
 63                            # Type narrowing for my_ships, my_shots, valid_actions
 64                            my_ships = cast(List[List[Union[int, str]]], my_ships_val)
 65                            my_shots = cast(List[List[int]], my_shots_val)
 66                            valid_actions = cast(List[List[int]], valid_actions_val)
 67
 68                            # Pass the partial state to the subclass logic
 69                            target_coord = await self.deliberate(my_ships, my_shots, valid_actions)
 70
 71                            if target_coord is not None:
 72                                await websocket.send(
 73                                    json.dumps(
 74                                        {
 75                                            "action": "fire",
 76                                            "x": target_coord[0],
 77                                            "y": target_coord[1],
 78                                        }
 79                                    )
 80                                )
 81
 82                    elif msg_type == "game_over":
 83                        logging.info(f"Round Over: {data.get('message', 'Unknown result')}")
 84                        logging.info("Waiting for next round to start...")
 85
 86        except Exception as e:
 87            logging.error(f"Connection lost: {e}")
 88
 89    async def deliberate(
 90        self,
 91        my_ships: List[List[Union[int, str]]],
 92        my_shots: List[List[int]],
 93        valid_actions: List[List[int]],
 94    ) -> Optional[Union[List[int], Tuple[int, int]]]:
 95        r"""
 96        Determines the next target to fire upon.
 97
 98        MUST be implemented by subclasses.
 99
100        Args:
101            my_ships: 10x10 grid with your ship names or 0 for water.
102                $M_{ships} \in \{0, \text{'Carrier'}, \dots\}^{10 \times 10}$
103            my_shots: 10x10 grid with your shot history (0=unknown, 1=miss, 2=hit).
104                $M_{shots} \in \{0, 1, 2\}^{10 \times 10}$
105            valid_actions: List of [x, y] coordinates you haven't shot at yet.
106                $A = \{(x, y) \mid M_{shots}[y][x] = 0\}$
107
108        Returns:
109            A list or tuple of [x, y] representing the target coordinate.
110        """
111        raise NotImplementedError("Subclasses must implement deliberate()")