backend.server 

  1import asyncio
  2import json
  3import logging
  4import random
  5from typing import Any, Dict, List, Optional, Tuple, Union
  6
  7from websockets.asyncio.server import ServerConnection, serve
  8
  9logging.basicConfig(level=logging.INFO, format="%(asctime)s - SERVER - %(levelname)s - %(message)s")
 10
 11
 12class BattleshipServer:
 13    """
 14    WebSocket server for the Battleship game.
 15    Manages the game state, connects agents and a single viewer (frontend).
 16    """
 17
 18    def __init__(self) -> None:
 19        """
 20        Initializes the game state and tracking variables.
 21        """
 22        self.frontend_ws: Optional[ServerConnection] = None
 23        self.agent1_ws: Optional[ServerConnection] = None
 24        self.agent2_ws: Optional[ServerConnection] = None
 25
 26        self.size: int = 10
 27        # 0 = Water, Ship Name = Ship
 28        self.p1_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
 29        self.p2_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
 30        # 0 = Unknown, 1 = Miss, 2 = Hit
 31        self.p1_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
 32        self.p2_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
 33
 34        self.ship_fleet: Dict[str, int] = {
 35            "Carrier": 5,
 36            "Battleship": 4,
 37            "Cruiser": 3,
 38            "Submarine": 3,
 39            "Destroyer": 2,
 40        }
 41        self.p1_health: Dict[str, int] = {}
 42        self.p2_health: Dict[str, int] = {}
 43
 44        self.first_player_this_round: int = 1
 45        self.current_turn: int = 1
 46        self.running: bool = False
 47        self.scores: Dict[int, int] = {1: 0, 2: 0}
 48
 49    async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None:
 50        """
 51        Starts the WebSocket server.
 52
 53        Args:
 54            host (str): The host address to bind to.
 55            port (int): The port to listen on.
 56        """
 57        logging.info(f"Battleship Server started on ws://{host}:{port}")
 58        async with serve(self.handle_client, host, port):
 59            await asyncio.Future()
 60
 61    async def handle_client(self, websocket: ServerConnection) -> None:
 62        """
 63        Handles incoming WebSocket connections and assigns them to a role.
 64
 65        Args:
 66            websocket: The connecting WebSocket protocol instance.
 67        """
 68        client_type: str = "Unknown"
 69        try:
 70            init_msg = await websocket.recv()
 71
 72            data: Dict[str, Any] = json.loads(init_msg)
 73            type_val = data.get("client", "Unknown")
 74            client_type = str(type_val) if type_val is not None else "Unknown"
 75
 76            if client_type == "frontend":
 77                logging.info("Frontend connected.")
 78                self.frontend_ws = websocket
 79                await self.update_frontend()
 80                await self.frontend_loop(websocket)
 81            elif client_type == "agent":
 82                if not self.agent1_ws:
 83                    self.agent1_ws = websocket
 84                    logging.info("Player 1 connected.")
 85                    await websocket.send(json.dumps({"type": "setup", "player_id": 1, "size": self.size}))
 86                    await self.check_start_conditions()
 87                    await self.agent_loop(websocket, 1)
 88                elif not self.agent2_ws:
 89                    self.agent2_ws = websocket
 90                    logging.info("Player 2 connected.")
 91                    await websocket.send(json.dumps({"type": "setup", "player_id": 2, "size": self.size}))
 92                    await self.check_start_conditions()
 93                    await self.agent_loop(websocket, 2)
 94                else:
 95                    await websocket.close()
 96        except Exception as e:
 97            logging.error(f"Error handling client ({client_type}): {e}")
 98        finally:
 99            if websocket == self.frontend_ws:
100                self.frontend_ws = None
101            elif websocket == self.agent1_ws:
102                self.agent1_ws = None
103                self.running = False
104            elif websocket == self.agent2_ws:
105                self.agent2_ws = None
106                self.running = False
107
108    async def frontend_loop(self, websocket: ServerConnection) -> None:
109        """
110        Main loop for the frontend viewer connection.
111
112        Args:
113            websocket: The frontend WebSocket protocol instance.
114        """
115        async for _ in websocket:
116            pass
117
118    async def agent_loop(self, websocket: ServerConnection, player_id: int) -> None:
119        """
120        Main loop for an agent connection.
121
122        Args:
123            websocket: The agent's WebSocket protocol instance.
124            player_id: The ID of the player (1 or 2).
125        """
126        async for message in websocket:
127            if not self.running or self.current_turn != player_id:
128                continue
129            try:
130                data: Dict[str, Any] = json.loads(message)
131                if data.get("action") == "fire":
132                    x, y = data.get("x"), data.get("y")
133                    if isinstance(x, int) and isinstance(y, int):
134                        valid, hit = self.process_shot(player_id, x, y)
135                        if valid:
136                            await self.update_frontend()
137                            await self.check_game_over()
138                            if self.running:
139                                if not hit:
140                                    self.current_turn = 3 - self.current_turn
141                                await self.broadcast_state()
142            except Exception as e:
143                logging.error(f"Error processing move from Player {player_id}: {e}")
144
145    async def check_start_conditions(self) -> None:
146        """
147        Checks if both agents are connected and starts a new round if needed.
148        """
149        if self.agent1_ws and self.agent2_ws and not self.running:
150            self.running = True
151            self.current_turn = self.first_player_this_round
152
153            # Initialize empty boards
154            self.p1_ships = [[0] * self.size for _ in range(self.size)]
155            self.p2_ships = [[0] * self.size for _ in range(self.size)]
156            self.p1_shots = [[0] * self.size for _ in range(self.size)]
157            self.p2_shots = [[0] * self.size for _ in range(self.size)]
158
159            # Place ships and track health
160            self.p1_health = self.place_fleet(self.p1_ships)
161            self.p2_health = self.place_fleet(self.p2_ships)
162
163            await self.update_frontend()
164            await self.broadcast_state()
165
166    def place_fleet(self, board: List[List[Union[int, str]]]) -> Dict[str, int]:
167        r"""
168        Randomly places ships on the provided board.
169
170        Args:
171            board: The 10x10 matrix to place ships on.
172
173        Returns:
174            A dictionary tracking the health of each placed ship.
175        """
176        health_tracker: Dict[str, int] = {}
177        for ship_name, length in self.ship_fleet.items():
178            health_tracker[ship_name] = length
179            placed = False
180            while not placed:
181                x, y = (
182                    random.randint(0, self.size - 1),
183                    random.randint(0, self.size - 1),
184                )
185                horizontal = random.choice([True, False])
186
187                # Check bounds
188                if horizontal and x + length > self.size:
189                    continue
190                if not horizontal and y + length > self.size:
191                    continue
192
193                # Check overlap
194                overlap = False
195                for i in range(length):
196                    if horizontal:
197                        if board[y][x + i] != 0:
198                            overlap = True
199                            break
200                    else:
201                        if board[y + i][x] != 0:
202                            overlap = True
203                            break
204
205                if not overlap:
206                    for i in range(length):
207                        if horizontal:
208                            board[y][x + i] = ship_name
209                        else:
210                            board[y + i][x] = ship_name
211                    placed = True
212        return health_tracker
213
214    def get_valid_actions(self, player_id: int) -> List[List[int]]:
215        """
216        Calculates all valid target coordinates for a player.
217
218        Args:
219            player_id: The ID of the player (1 or 2).
220
221        Returns:
222            A list of [x, y] coordinates that have not been targeted yet.
223        """
224        shots = self.p1_shots if player_id == 1 else self.p2_shots
225        actions: List[List[int]] = []
226        for y in range(self.size):
227            for x in range(self.size):
228                if shots[y][x] == 0:
229                    actions.append([x, y])
230        return actions
231
232    def process_shot(self, player_id: int, x: int, y: int) -> Tuple[bool, bool]:
233        r"""
234        Processes an agent's firing action.
235
236        Given a coordinate $(x, y)$, where $x, y \in [0, 9]$, this method
237        updates the shot history $M_{shots}$ and checks for a hit on the
238        opponent's fleet $F$.
239
240        Args:
241            player_id: The ID of the player taking the shot.
242            x: The x-coordinate target.
243            y: The y-coordinate target.
244
245        Returns:
246            A tuple (valid_shot, was_hit).
247        """
248        shots = self.p1_shots if player_id == 1 else self.p2_shots
249        enemy_ships = self.p2_ships if player_id == 1 else self.p1_ships
250        enemy_health = self.p2_health if player_id == 1 else self.p1_health
251
252        if not (0 <= x < self.size and 0 <= y < self.size) or shots[y][x] != 0:
253            return False, False  # Invalid or already shot
254
255        target = enemy_ships[y][x]
256        is_hit = False
257        if isinstance(target, str):
258            shots[y][x] = 2  # Hit
259            is_hit = True
260            enemy_health[target] -= 1
261            if enemy_health[target] == 0:
262                logging.info(f"Player {player_id} SUNK the {target}!")
263        else:
264            shots[y][x] = 1  # Miss
265
266        return True, is_hit
267
268    async def check_game_over(self) -> None:
269        """
270        Checks if the current game session has concluded.
271        """
272        p1_dead = all(h == 0 for h in self.p1_health.values())
273        p2_dead = all(h == 0 for h in self.p2_health.values())
274
275        winner = None
276        if p1_dead:
277            winner = 2
278        elif p2_dead:
279            winner = 1
280
281        if winner:
282            self.scores[winner] += 1
283            await self.end_round(f"Player {winner} Wins!")
284
285    async def end_round(self, message: str) -> None:
286        """
287        Finishes a game round and schedules a restart.
288
289        Args:
290            message: The result message to send to clients.
291        """
292        self.running = False
293        payload = {"type": "game_over", "message": message}
294        if self.agent1_ws:
295            await self.agent1_ws.send(json.dumps(payload))
296        if self.agent2_ws:
297            await self.agent2_ws.send(json.dumps(payload))
298        await self.update_frontend()
299
300        await asyncio.sleep(3.0)
301        self.first_player_this_round = 3 - self.first_player_this_round
302        await self.check_start_conditions()
303
304    async def broadcast_state(self) -> None:
305        """Sends partial state information to each connected agent."""
306        if self.agent1_ws:
307            await self.agent1_ws.send(
308                json.dumps(
309                    {
310                        "type": "state",
311                        "current_turn": self.current_turn,
312                        "my_ships": self.p1_ships,
313                        "my_shots": self.p1_shots,
314                        "valid_actions": self.get_valid_actions(1),
315                    }
316                )
317            )
318        if self.agent2_ws:
319            await self.agent2_ws.send(
320                json.dumps(
321                    {
322                        "type": "state",
323                        "current_turn": self.current_turn,
324                        "my_ships": self.p2_ships,
325                        "my_shots": self.p2_shots,
326                        "valid_actions": self.get_valid_actions(2),
327                    }
328                )
329            )
330
331    async def update_frontend(self) -> None:
332        """Sends the full game state to the frontend viewer."""
333        if self.frontend_ws:
334            await self.frontend_ws.send(
335                json.dumps(
336                    {
337                        "type": "update",
338                        "current_turn": self.current_turn,
339                        "p1_ships": self.p1_ships,
340                        "p2_ships": self.p2_ships,
341                        "p1_shots": self.p1_shots,
342                        "p2_shots": self.p2_shots,
343                        "scores": self.scores,
344                        "p1_connected": self.agent1_ws is not None,
345                        "p2_connected": self.agent2_ws is not None,
346                    }
347                )
348            )
349
350
351if __name__ == "__main__":
352    server = BattleshipServer()
353    asyncio.run(server.start())
class BattleshipServer: 
 13class BattleshipServer:
 14    """
 15    WebSocket server for the Battleship game.
 16    Manages the game state, connects agents and a single viewer (frontend).
 17    """
 18
 19    def __init__(self) -> None:
 20        """
 21        Initializes the game state and tracking variables.
 22        """
 23        self.frontend_ws: Optional[ServerConnection] = None
 24        self.agent1_ws: Optional[ServerConnection] = None
 25        self.agent2_ws: Optional[ServerConnection] = None
 26
 27        self.size: int = 10
 28        # 0 = Water, Ship Name = Ship
 29        self.p1_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
 30        self.p2_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
 31        # 0 = Unknown, 1 = Miss, 2 = Hit
 32        self.p1_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
 33        self.p2_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
 34
 35        self.ship_fleet: Dict[str, int] = {
 36            "Carrier": 5,
 37            "Battleship": 4,
 38            "Cruiser": 3,
 39            "Submarine": 3,
 40            "Destroyer": 2,
 41        }
 42        self.p1_health: Dict[str, int] = {}
 43        self.p2_health: Dict[str, int] = {}
 44
 45        self.first_player_this_round: int = 1
 46        self.current_turn: int = 1
 47        self.running: bool = False
 48        self.scores: Dict[int, int] = {1: 0, 2: 0}
 49
 50    async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None:
 51        """
 52        Starts the WebSocket server.
 53
 54        Args:
 55            host (str): The host address to bind to.
 56            port (int): The port to listen on.
 57        """
 58        logging.info(f"Battleship Server started on ws://{host}:{port}")
 59        async with serve(self.handle_client, host, port):
 60            await asyncio.Future()
 61
 62    async def handle_client(self, websocket: ServerConnection) -> None:
 63        """
 64        Handles incoming WebSocket connections and assigns them to a role.
 65
 66        Args:
 67            websocket: The connecting WebSocket protocol instance.
 68        """
 69        client_type: str = "Unknown"
 70        try:
 71            init_msg = await websocket.recv()
 72
 73            data: Dict[str, Any] = json.loads(init_msg)
 74            type_val = data.get("client", "Unknown")
 75            client_type = str(type_val) if type_val is not None else "Unknown"
 76
 77            if client_type == "frontend":
 78                logging.info("Frontend connected.")
 79                self.frontend_ws = websocket
 80                await self.update_frontend()
 81                await self.frontend_loop(websocket)
 82            elif client_type == "agent":
 83                if not self.agent1_ws:
 84                    self.agent1_ws = websocket
 85                    logging.info("Player 1 connected.")
 86                    await websocket.send(json.dumps({"type": "setup", "player_id": 1, "size": self.size}))
 87                    await self.check_start_conditions()
 88                    await self.agent_loop(websocket, 1)
 89                elif not self.agent2_ws:
 90                    self.agent2_ws = websocket
 91                    logging.info("Player 2 connected.")
 92                    await websocket.send(json.dumps({"type": "setup", "player_id": 2, "size": self.size}))
 93                    await self.check_start_conditions()
 94                    await self.agent_loop(websocket, 2)
 95                else:
 96                    await websocket.close()
 97        except Exception as e:
 98            logging.error(f"Error handling client ({client_type}): {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            elif websocket == self.agent2_ws:
106                self.agent2_ws = None
107                self.running = False
108
109    async def frontend_loop(self, websocket: ServerConnection) -> None:
110        """
111        Main loop for the frontend viewer connection.
112
113        Args:
114            websocket: The frontend WebSocket protocol instance.
115        """
116        async for _ in websocket:
117            pass
118
119    async def agent_loop(self, websocket: ServerConnection, player_id: int) -> None:
120        """
121        Main loop for an agent connection.
122
123        Args:
124            websocket: The agent's WebSocket protocol instance.
125            player_id: The ID of the player (1 or 2).
126        """
127        async for message in websocket:
128            if not self.running or self.current_turn != player_id:
129                continue
130            try:
131                data: Dict[str, Any] = json.loads(message)
132                if data.get("action") == "fire":
133                    x, y = data.get("x"), data.get("y")
134                    if isinstance(x, int) and isinstance(y, int):
135                        valid, hit = self.process_shot(player_id, x, y)
136                        if valid:
137                            await self.update_frontend()
138                            await self.check_game_over()
139                            if self.running:
140                                if not hit:
141                                    self.current_turn = 3 - self.current_turn
142                                await self.broadcast_state()
143            except Exception as e:
144                logging.error(f"Error processing move from Player {player_id}: {e}")
145
146    async def check_start_conditions(self) -> None:
147        """
148        Checks if both agents are connected and starts a new round if needed.
149        """
150        if self.agent1_ws and self.agent2_ws and not self.running:
151            self.running = True
152            self.current_turn = self.first_player_this_round
153
154            # Initialize empty boards
155            self.p1_ships = [[0] * self.size for _ in range(self.size)]
156            self.p2_ships = [[0] * self.size for _ in range(self.size)]
157            self.p1_shots = [[0] * self.size for _ in range(self.size)]
158            self.p2_shots = [[0] * self.size for _ in range(self.size)]
159
160            # Place ships and track health
161            self.p1_health = self.place_fleet(self.p1_ships)
162            self.p2_health = self.place_fleet(self.p2_ships)
163
164            await self.update_frontend()
165            await self.broadcast_state()
166
167    def place_fleet(self, board: List[List[Union[int, str]]]) -> Dict[str, int]:
168        r"""
169        Randomly places ships on the provided board.
170
171        Args:
172            board: The 10x10 matrix to place ships on.
173
174        Returns:
175            A dictionary tracking the health of each placed ship.
176        """
177        health_tracker: Dict[str, int] = {}
178        for ship_name, length in self.ship_fleet.items():
179            health_tracker[ship_name] = length
180            placed = False
181            while not placed:
182                x, y = (
183                    random.randint(0, self.size - 1),
184                    random.randint(0, self.size - 1),
185                )
186                horizontal = random.choice([True, False])
187
188                # Check bounds
189                if horizontal and x + length > self.size:
190                    continue
191                if not horizontal and y + length > self.size:
192                    continue
193
194                # Check overlap
195                overlap = False
196                for i in range(length):
197                    if horizontal:
198                        if board[y][x + i] != 0:
199                            overlap = True
200                            break
201                    else:
202                        if board[y + i][x] != 0:
203                            overlap = True
204                            break
205
206                if not overlap:
207                    for i in range(length):
208                        if horizontal:
209                            board[y][x + i] = ship_name
210                        else:
211                            board[y + i][x] = ship_name
212                    placed = True
213        return health_tracker
214
215    def get_valid_actions(self, player_id: int) -> List[List[int]]:
216        """
217        Calculates all valid target coordinates for a player.
218
219        Args:
220            player_id: The ID of the player (1 or 2).
221
222        Returns:
223            A list of [x, y] coordinates that have not been targeted yet.
224        """
225        shots = self.p1_shots if player_id == 1 else self.p2_shots
226        actions: List[List[int]] = []
227        for y in range(self.size):
228            for x in range(self.size):
229                if shots[y][x] == 0:
230                    actions.append([x, y])
231        return actions
232
233    def process_shot(self, player_id: int, x: int, y: int) -> Tuple[bool, bool]:
234        r"""
235        Processes an agent's firing action.
236
237        Given a coordinate $(x, y)$, where $x, y \in [0, 9]$, this method
238        updates the shot history $M_{shots}$ and checks for a hit on the
239        opponent's fleet $F$.
240
241        Args:
242            player_id: The ID of the player taking the shot.
243            x: The x-coordinate target.
244            y: The y-coordinate target.
245
246        Returns:
247            A tuple (valid_shot, was_hit).
248        """
249        shots = self.p1_shots if player_id == 1 else self.p2_shots
250        enemy_ships = self.p2_ships if player_id == 1 else self.p1_ships
251        enemy_health = self.p2_health if player_id == 1 else self.p1_health
252
253        if not (0 <= x < self.size and 0 <= y < self.size) or shots[y][x] != 0:
254            return False, False  # Invalid or already shot
255
256        target = enemy_ships[y][x]
257        is_hit = False
258        if isinstance(target, str):
259            shots[y][x] = 2  # Hit
260            is_hit = True
261            enemy_health[target] -= 1
262            if enemy_health[target] == 0:
263                logging.info(f"Player {player_id} SUNK the {target}!")
264        else:
265            shots[y][x] = 1  # Miss
266
267        return True, is_hit
268
269    async def check_game_over(self) -> None:
270        """
271        Checks if the current game session has concluded.
272        """
273        p1_dead = all(h == 0 for h in self.p1_health.values())
274        p2_dead = all(h == 0 for h in self.p2_health.values())
275
276        winner = None
277        if p1_dead:
278            winner = 2
279        elif p2_dead:
280            winner = 1
281
282        if winner:
283            self.scores[winner] += 1
284            await self.end_round(f"Player {winner} Wins!")
285
286    async def end_round(self, message: str) -> None:
287        """
288        Finishes a game round and schedules a restart.
289
290        Args:
291            message: The result message to send to clients.
292        """
293        self.running = False
294        payload = {"type": "game_over", "message": message}
295        if self.agent1_ws:
296            await self.agent1_ws.send(json.dumps(payload))
297        if self.agent2_ws:
298            await self.agent2_ws.send(json.dumps(payload))
299        await self.update_frontend()
300
301        await asyncio.sleep(3.0)
302        self.first_player_this_round = 3 - self.first_player_this_round
303        await self.check_start_conditions()
304
305    async def broadcast_state(self) -> None:
306        """Sends partial state information to each connected agent."""
307        if self.agent1_ws:
308            await self.agent1_ws.send(
309                json.dumps(
310                    {
311                        "type": "state",
312                        "current_turn": self.current_turn,
313                        "my_ships": self.p1_ships,
314                        "my_shots": self.p1_shots,
315                        "valid_actions": self.get_valid_actions(1),
316                    }
317                )
318            )
319        if self.agent2_ws:
320            await self.agent2_ws.send(
321                json.dumps(
322                    {
323                        "type": "state",
324                        "current_turn": self.current_turn,
325                        "my_ships": self.p2_ships,
326                        "my_shots": self.p2_shots,
327                        "valid_actions": self.get_valid_actions(2),
328                    }
329                )
330            )
331
332    async def update_frontend(self) -> None:
333        """Sends the full game state to the frontend viewer."""
334        if self.frontend_ws:
335            await self.frontend_ws.send(
336                json.dumps(
337                    {
338                        "type": "update",
339                        "current_turn": self.current_turn,
340                        "p1_ships": self.p1_ships,
341                        "p2_ships": self.p2_ships,
342                        "p1_shots": self.p1_shots,
343                        "p2_shots": self.p2_shots,
344                        "scores": self.scores,
345                        "p1_connected": self.agent1_ws is not None,
346                        "p2_connected": self.agent2_ws is not None,
347                    }
348                )
349            )

WebSocket server for the Battleship game. Manages the game state, connects agents and a single viewer (frontend).

BattleshipServer() 
19    def __init__(self) -> None:
20        """
21        Initializes the game state and tracking variables.
22        """
23        self.frontend_ws: Optional[ServerConnection] = None
24        self.agent1_ws: Optional[ServerConnection] = None
25        self.agent2_ws: Optional[ServerConnection] = None
26
27        self.size: int = 10
28        # 0 = Water, Ship Name = Ship
29        self.p1_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
30        self.p2_ships: List[List[Union[int, str]]] = [[0] * self.size for _ in range(self.size)]
31        # 0 = Unknown, 1 = Miss, 2 = Hit
32        self.p1_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
33        self.p2_shots: List[List[int]] = [[0] * self.size for _ in range(self.size)]
34
35        self.ship_fleet: Dict[str, int] = {
36            "Carrier": 5,
37            "Battleship": 4,
38            "Cruiser": 3,
39            "Submarine": 3,
40            "Destroyer": 2,
41        }
42        self.p1_health: Dict[str, int] = {}
43        self.p2_health: Dict[str, int] = {}
44
45        self.first_player_this_round: int = 1
46        self.current_turn: int = 1
47        self.running: bool = False
48        self.scores: Dict[int, int] = {1: 0, 2: 0}

Initializes the game state and tracking variables.

frontend_ws: Optional[websockets.asyncio.server.ServerConnection]
agent1_ws: Optional[websockets.asyncio.server.ServerConnection]
agent2_ws: Optional[websockets.asyncio.server.ServerConnection]
size: int
p1_ships: List[List[Union[str, int]]]
p2_ships: List[List[Union[str, int]]]
p1_shots: List[List[int]]
p2_shots: List[List[int]]
ship_fleet: Dict[str, int]
p1_health: Dict[str, int]
p2_health: Dict[str, int]
first_player_this_round: int
current_turn: int
running: bool
scores: Dict[int, int]
async def start(self, host: str = '0.0.0.0', port: int = 8765) -> None: 
50    async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None:
51        """
52        Starts the WebSocket server.
53
54        Args:
55            host (str): The host address to bind to.
56            port (int): The port to listen on.
57        """
58        logging.info(f"Battleship Server started on ws://{host}:{port}")
59        async with serve(self.handle_client, host, port):
60            await asyncio.Future()

Starts the WebSocket server.

Args: host (str): The host address to bind to. port (int): The port to listen on.

async def handle_client(self, websocket: websockets.asyncio.server.ServerConnection) -> None: 
 62    async def handle_client(self, websocket: ServerConnection) -> None:
 63        """
 64        Handles incoming WebSocket connections and assigns them to a role.
 65
 66        Args:
 67            websocket: The connecting WebSocket protocol instance.
 68        """
 69        client_type: str = "Unknown"
 70        try:
 71            init_msg = await websocket.recv()
 72
 73            data: Dict[str, Any] = json.loads(init_msg)
 74            type_val = data.get("client", "Unknown")
 75            client_type = str(type_val) if type_val is not None else "Unknown"
 76
 77            if client_type == "frontend":
 78                logging.info("Frontend connected.")
 79                self.frontend_ws = websocket
 80                await self.update_frontend()
 81                await self.frontend_loop(websocket)
 82            elif client_type == "agent":
 83                if not self.agent1_ws:
 84                    self.agent1_ws = websocket
 85                    logging.info("Player 1 connected.")
 86                    await websocket.send(json.dumps({"type": "setup", "player_id": 1, "size": self.size}))
 87                    await self.check_start_conditions()
 88                    await self.agent_loop(websocket, 1)
 89                elif not self.agent2_ws:
 90                    self.agent2_ws = websocket
 91                    logging.info("Player 2 connected.")
 92                    await websocket.send(json.dumps({"type": "setup", "player_id": 2, "size": self.size}))
 93                    await self.check_start_conditions()
 94                    await self.agent_loop(websocket, 2)
 95                else:
 96                    await websocket.close()
 97        except Exception as e:
 98            logging.error(f"Error handling client ({client_type}): {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            elif websocket == self.agent2_ws:
106                self.agent2_ws = None
107                self.running = False

Handles incoming WebSocket connections and assigns them to a role.

Args: websocket: The connecting WebSocket protocol instance.

async def frontend_loop(self, websocket: websockets.asyncio.server.ServerConnection) -> None: 
109    async def frontend_loop(self, websocket: ServerConnection) -> None:
110        """
111        Main loop for the frontend viewer connection.
112
113        Args:
114            websocket: The frontend WebSocket protocol instance.
115        """
116        async for _ in websocket:
117            pass

Main loop for the frontend viewer connection.

Args: websocket: The frontend WebSocket protocol instance.

async def agent_loop( self, websocket: websockets.asyncio.server.ServerConnection, player_id: int) -> None: 
119    async def agent_loop(self, websocket: ServerConnection, player_id: int) -> None:
120        """
121        Main loop for an agent connection.
122
123        Args:
124            websocket: The agent's WebSocket protocol instance.
125            player_id: The ID of the player (1 or 2).
126        """
127        async for message in websocket:
128            if not self.running or self.current_turn != player_id:
129                continue
130            try:
131                data: Dict[str, Any] = json.loads(message)
132                if data.get("action") == "fire":
133                    x, y = data.get("x"), data.get("y")
134                    if isinstance(x, int) and isinstance(y, int):
135                        valid, hit = self.process_shot(player_id, x, y)
136                        if valid:
137                            await self.update_frontend()
138                            await self.check_game_over()
139                            if self.running:
140                                if not hit:
141                                    self.current_turn = 3 - self.current_turn
142                                await self.broadcast_state()
143            except Exception as e:
144                logging.error(f"Error processing move from Player {player_id}: {e}")

Main loop for an agent connection.

Args: websocket: The agent's WebSocket protocol instance. player_id: The ID of the player (1 or 2).

async def check_start_conditions(self) -> None: 
146    async def check_start_conditions(self) -> None:
147        """
148        Checks if both agents are connected and starts a new round if needed.
149        """
150        if self.agent1_ws and self.agent2_ws and not self.running:
151            self.running = True
152            self.current_turn = self.first_player_this_round
153
154            # Initialize empty boards
155            self.p1_ships = [[0] * self.size for _ in range(self.size)]
156            self.p2_ships = [[0] * self.size for _ in range(self.size)]
157            self.p1_shots = [[0] * self.size for _ in range(self.size)]
158            self.p2_shots = [[0] * self.size for _ in range(self.size)]
159
160            # Place ships and track health
161            self.p1_health = self.place_fleet(self.p1_ships)
162            self.p2_health = self.place_fleet(self.p2_ships)
163
164            await self.update_frontend()
165            await self.broadcast_state()

Checks if both agents are connected and starts a new round if needed.

def place_fleet(self, board: List[List[Union[str, int]]]) -> Dict[str, int]: 
167    def place_fleet(self, board: List[List[Union[int, str]]]) -> Dict[str, int]:
168        r"""
169        Randomly places ships on the provided board.
170
171        Args:
172            board: The 10x10 matrix to place ships on.
173
174        Returns:
175            A dictionary tracking the health of each placed ship.
176        """
177        health_tracker: Dict[str, int] = {}
178        for ship_name, length in self.ship_fleet.items():
179            health_tracker[ship_name] = length
180            placed = False
181            while not placed:
182                x, y = (
183                    random.randint(0, self.size - 1),
184                    random.randint(0, self.size - 1),
185                )
186                horizontal = random.choice([True, False])
187
188                # Check bounds
189                if horizontal and x + length > self.size:
190                    continue
191                if not horizontal and y + length > self.size:
192                    continue
193
194                # Check overlap
195                overlap = False
196                for i in range(length):
197                    if horizontal:
198                        if board[y][x + i] != 0:
199                            overlap = True
200                            break
201                    else:
202                        if board[y + i][x] != 0:
203                            overlap = True
204                            break
205
206                if not overlap:
207                    for i in range(length):
208                        if horizontal:
209                            board[y][x + i] = ship_name
210                        else:
211                            board[y + i][x] = ship_name
212                    placed = True
213        return health_tracker

Randomly places ships on the provided board.

Args: board: The 10x10 matrix to place ships on.

Returns: A dictionary tracking the health of each placed ship.

def get_valid_actions(self, player_id: int) -> List[List[int]]: 
215    def get_valid_actions(self, player_id: int) -> List[List[int]]:
216        """
217        Calculates all valid target coordinates for a player.
218
219        Args:
220            player_id: The ID of the player (1 or 2).
221
222        Returns:
223            A list of [x, y] coordinates that have not been targeted yet.
224        """
225        shots = self.p1_shots if player_id == 1 else self.p2_shots
226        actions: List[List[int]] = []
227        for y in range(self.size):
228            for x in range(self.size):
229                if shots[y][x] == 0:
230                    actions.append([x, y])
231        return actions

Calculates all valid target coordinates for a player.

Args: player_id: The ID of the player (1 or 2).

Returns: A list of [x, y] coordinates that have not been targeted yet.

def process_shot(self, player_id: int, x: int, y: int) -> Tuple[bool, bool]: 
233    def process_shot(self, player_id: int, x: int, y: int) -> Tuple[bool, bool]:
234        r"""
235        Processes an agent's firing action.
236
237        Given a coordinate $(x, y)$, where $x, y \in [0, 9]$, this method
238        updates the shot history $M_{shots}$ and checks for a hit on the
239        opponent's fleet $F$.
240
241        Args:
242            player_id: The ID of the player taking the shot.
243            x: The x-coordinate target.
244            y: The y-coordinate target.
245
246        Returns:
247            A tuple (valid_shot, was_hit).
248        """
249        shots = self.p1_shots if player_id == 1 else self.p2_shots
250        enemy_ships = self.p2_ships if player_id == 1 else self.p1_ships
251        enemy_health = self.p2_health if player_id == 1 else self.p1_health
252
253        if not (0 <= x < self.size and 0 <= y < self.size) or shots[y][x] != 0:
254            return False, False  # Invalid or already shot
255
256        target = enemy_ships[y][x]
257        is_hit = False
258        if isinstance(target, str):
259            shots[y][x] = 2  # Hit
260            is_hit = True
261            enemy_health[target] -= 1
262            if enemy_health[target] == 0:
263                logging.info(f"Player {player_id} SUNK the {target}!")
264        else:
265            shots[y][x] = 1  # Miss
266
267        return True, is_hit

Processes an agent's firing action.

Given a coordinate $(x, y)$, where $x, y \in [0, 9]$, this method updates the shot history $M_{shots}$ and checks for a hit on the opponent's fleet $F$.

Args: player_id: The ID of the player taking the shot. x: The x-coordinate target. y: The y-coordinate target.

Returns: A tuple (valid_shot, was_hit).

async def check_game_over(self) -> None: 
269    async def check_game_over(self) -> None:
270        """
271        Checks if the current game session has concluded.
272        """
273        p1_dead = all(h == 0 for h in self.p1_health.values())
274        p2_dead = all(h == 0 for h in self.p2_health.values())
275
276        winner = None
277        if p1_dead:
278            winner = 2
279        elif p2_dead:
280            winner = 1
281
282        if winner:
283            self.scores[winner] += 1
284            await self.end_round(f"Player {winner} Wins!")

Checks if the current game session has concluded.

async def end_round(self, message: str) -> None: 
286    async def end_round(self, message: str) -> None:
287        """
288        Finishes a game round and schedules a restart.
289
290        Args:
291            message: The result message to send to clients.
292        """
293        self.running = False
294        payload = {"type": "game_over", "message": message}
295        if self.agent1_ws:
296            await self.agent1_ws.send(json.dumps(payload))
297        if self.agent2_ws:
298            await self.agent2_ws.send(json.dumps(payload))
299        await self.update_frontend()
300
301        await asyncio.sleep(3.0)
302        self.first_player_this_round = 3 - self.first_player_this_round
303        await self.check_start_conditions()

Finishes a game round and schedules a restart.

Args: message: The result message to send to clients.

async def broadcast_state(self) -> None: 
305    async def broadcast_state(self) -> None:
306        """Sends partial state information to each connected agent."""
307        if self.agent1_ws:
308            await self.agent1_ws.send(
309                json.dumps(
310                    {
311                        "type": "state",
312                        "current_turn": self.current_turn,
313                        "my_ships": self.p1_ships,
314                        "my_shots": self.p1_shots,
315                        "valid_actions": self.get_valid_actions(1),
316                    }
317                )
318            )
319        if self.agent2_ws:
320            await self.agent2_ws.send(
321                json.dumps(
322                    {
323                        "type": "state",
324                        "current_turn": self.current_turn,
325                        "my_ships": self.p2_ships,
326                        "my_shots": self.p2_shots,
327                        "valid_actions": self.get_valid_actions(2),
328                    }
329                )
330            )

Sends partial state information to each connected agent.

async def update_frontend(self) -> None: 
332    async def update_frontend(self) -> None:
333        """Sends the full game state to the frontend viewer."""
334        if self.frontend_ws:
335            await self.frontend_ws.send(
336                json.dumps(
337                    {
338                        "type": "update",
339                        "current_turn": self.current_turn,
340                        "p1_ships": self.p1_ships,
341                        "p2_ships": self.p2_ships,
342                        "p1_shots": self.p1_shots,
343                        "p2_shots": self.p2_shots,
344                        "scores": self.scores,
345                        "p1_connected": self.agent1_ws is not None,
346                        "p2_connected": self.agent2_ws is not None,
347                    }
348                )
349            )

Sends the full game state to the frontend viewer.