backend.server

  1import asyncio
  2import json
  3import logging
  4from typing import Dict, List, Optional, Tuple, Any
  5
  6from websockets.server import serve
  7
  8logging.basicConfig(
  9    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
 10)
 11
 12
 13class OthelloServer:
 14    r"""
 15    Othello game server implementation using WebSockets.
 16
 17    Manages the game state, connects agents and frontend, and broadcasts updates.
 18    The game board is an $N \times N$ grid, where $N = 8$.
 19    The objective is to maximize the number of discs of the player's color $P$:
 20    $\max \sum_{i=1}^{N} \sum_{j=1}^{N} [B_{i,j} = P]$, where $B_{i,j}$ is the board state at $(i,j)$.
 21    """
 22
 23    def __init__(self) -> None:
 24        """Initializes the OthelloServer with default values."""
 25        self.frontend_ws: Optional[Any] = None
 26        self.agent1_ws: Optional[Any] = None
 27        self.agent2_ws: Optional[Any] = None
 28
 29        self.size: int = 8
 30        self.board: List[List[int]] = [[0] * self.size for _ in range(self.size)]
 31        self.directions: List[Tuple[int, int]] = [
 32            (-1, -1),
 33            (-1, 0),
 34            (-1, 1),
 35            (0, -1),
 36            (0, 1),
 37            (1, -1),
 38            (1, 0),
 39            (1, 1),
 40        ]
 41
 42        self.first_player_this_round: int = 1
 43        self.current_turn: int = 1
 44        self.running: bool = False
 45        self.match_scores: Dict[int, int] = {1: 0, 2: 0}  # Tracks rounds won
 46        self.broadcast_lock = asyncio.Lock()
 47
 48    async def start(self, host: str = "0.0.0.0", port: int = 8765) -> None:
 49        """
 50        Starts the Othello server.
 51
 52        Args:
 53            host (str): The host to bind to. Defaults to "0.0.0.0".
 54            port (int): The port to listen on. Defaults to 8765.
 55        """
 56        logging.info(f"Othello Server started on ws://{host}:{port}")
 57        async with serve(self.handle_client, host, port):
 58            await asyncio.Future()
 59
 60    async def handle_client(self, websocket: Any) -> None:
 61        """
 62        Handles incoming WebSocket connections.
 63
 64        Args:
 65            websocket: The incoming WebSocket connection.
 66        """
 67        logging.info(f"New connection attempt from {websocket.remote_address}")
 68        client_type: str = "Unknown"
 69        try:
 70            init_msg = await websocket.recv()
 71            logging.info(f"Received initial message: {init_msg}")
 72            data: Dict[str, Any] = json.loads(init_msg)
 73            client_type = data.get("client", "Unknown")
 74
 75            if client_type == "frontend":
 76                logging.info("Frontend connected.")
 77                self.frontend_ws = websocket
 78                await self.update_frontend()
 79                await self.frontend_loop(websocket)
 80            elif client_type == "agent":
 81                if not self.agent1_ws:
 82                    self.agent1_ws = websocket
 83                    logging.info("Player 1 (Black) connected.")
 84                    await websocket.send(json.dumps({"type": "setup", "player_id": 1}))
 85                    await self.check_start_conditions()
 86                    await self.agent_loop(websocket, 1)
 87                elif not self.agent2_ws:
 88                    self.agent2_ws = websocket
 89                    logging.info("Player 2 (White) connected.")
 90                    await websocket.send(json.dumps({"type": "setup", "player_id": 2}))
 91                    await self.check_start_conditions()
 92                    await self.agent_loop(websocket, 2)
 93                else:
 94                    logging.warning("Extra agent attempted to connect. Closing connection.")
 95                    await websocket.close()
 96            else:
 97                logging.warning(f"Unknown client type: {client_type}")
 98                await websocket.close()
 99        except Exception as e:
100            logging.error(f"Error in handle_client ({client_type}): {e}")
101        finally:
102            logging.info(f"Connection closed for {client_type}")
103            if websocket == self.frontend_ws:
104                self.frontend_ws = None
105            elif websocket == self.agent1_ws:
106                self.agent1_ws = None
107                self.running = False
108                logging.info("Player 1 disconnected. Game stopped.")
109                await self.update_frontend()
110            elif websocket == self.agent2_ws:
111                self.agent2_ws = None
112                self.running = False
113                logging.info("Player 2 disconnected. Game stopped.")
114                await self.update_frontend()
115
116    async def frontend_loop(self, websocket: Any) -> None:
117        """
118        Main loop for frontend WebSocket connection.
119
120        Args:
121            websocket: The frontend WebSocket connection.
122        """
123        async for _ in websocket:
124            pass
125
126    async def agent_loop(self, websocket: Any, player_id: int) -> None:
127        """
128        Main loop for agent WebSocket connection.
129
130        Args:
131            websocket: The agent WebSocket connection.
132            player_id (int): The ID of the player associated with this agent.
133        """
134        async for message in websocket:
135            logging.debug(f"Received message from Player {player_id}: {message}")
136            if not self.running or self.current_turn != player_id:
137                continue
138            try:
139                data: Dict[str, Any] = json.loads(message)
140                if data.get("action") == "move":
141                    x, y = data.get("x"), data.get("y")
142                    logging.info(f"Player {player_id} moves at ({x}, {y})")
143                    if x is not None and y is not None:
144                        if self.process_move(player_id, x, y):
145                            await self.advance_turn()
146                        else:
147                            logging.warning(f"Invalid move from Player {player_id}: ({x}, {y})")
148            except Exception as e:
149                logging.error(f"Error processing move: {e}")
150
151    async def check_start_conditions(self) -> None:
152        """Checks if all conditions are met to start a new game round."""
153        if self.agent1_ws and self.agent2_ws and not self.running:
154            logging.info("Both agents connected. Starting game.")
155            self.running = True
156            self.board = [[0] * self.size for _ in range(self.size)]
157            # Othello starting position
158            self.board[3][3], self.board[4][4] = 2, 2  # White
159            self.board[3][4], self.board[4][3] = 1, 1  # Black
160
161            self.current_turn = self.first_player_this_round
162            await self.update_frontend()
163            await self.broadcast_state()
164
165    def get_flips(self, player_id: int, x: int, y: int) -> List[Tuple[int, int]]:
166        """
167        Calculates which pieces would be flipped if player_id moves at (x,y).
168
169        Args:
170            player_id (int): The ID of the player making the move.
171            x (int): The x-coordinate of the move.
172            y (int): The y-coordinate of the move.
173
174        Returns:
175            List[Tuple[int, int]]: A list of coordinates to be flipped.
176        """
177        if x < 0 or x >= self.size or y < 0 or y >= self.size:
178            return []
179
180        if self.board[y][x] != 0:
181            return []
182
183        opponent: int = 3 - player_id
184        flips: List[Tuple[int, int]] = []
185
186        for dx, dy in self.directions:
187            nx, ny = x + dx, y + dy
188            temp_flips: List[Tuple[int, int]] = []
189
190            while (
191                0 <= nx < self.size
192                and 0 <= ny < self.size
193                and self.board[ny][nx] == opponent
194            ):
195                temp_flips.append((nx, ny))
196                nx += dx
197                ny += dy
198
199            if (
200                0 <= nx < self.size
201                and 0 <= ny < self.size
202                and self.board[ny][nx] == player_id
203            ):
204                flips.extend(temp_flips)
205
206        return flips
207
208    def get_valid_actions(self, player_id: int) -> List[List[int]]:
209        """
210        Returns a list of valid moves for a given player.
211
212        Args:
213            player_id (int): The ID of the player.
214
215        Returns:
216            List[List[int]]: A list of valid move coordinates [x, y].
217        """
218        actions: List[List[int]] = []
219        for y in range(self.size):
220            for x in range(self.size):
221                if len(self.get_flips(player_id, x, y)) > 0:
222                    actions.append([x, y])
223        return actions
224
225    def process_move(self, player_id: int, x: int, y: int) -> bool:
226        """
227        Processes a move for a player.
228
229        Args:
230            player_id (int): The ID of the player making the move.
231            x (int): The x-coordinate of the move.
232            y (int): The y-coordinate of the move.
233
234        Returns:
235            bool: True if the move was successful, False otherwise.
236        """
237        flips: List[Tuple[int, int]] = self.get_flips(player_id, x, y)
238        if not flips:
239            return False
240
241        self.board[y][x] = player_id
242        for fx, fy in flips:
243            self.board[fy][fx] = player_id
244        return True
245
246    async def advance_turn(self) -> None:
247        """Advances the turn, handling Othello's skip rules."""
248        await self.update_frontend()
249
250        next_player: int = 3 - self.current_turn
251        valid_next: List[List[int]] = self.get_valid_actions(next_player)
252
253        if valid_next:
254            self.current_turn = next_player
255            logging.info(f"Turn advanced to Player {self.current_turn}")
256        else:
257            # Next player has no moves. Do we have moves?
258            valid_current: List[List[int]] = self.get_valid_actions(self.current_turn)
259            if not valid_current:
260                logging.info("No more valid moves for either player. Game over.")
261                await self.check_game_over()
262                return
263            else:
264                logging.info(f"Player {next_player} has no moves. Turn remains with Player {self.current_turn}")
265
266        if self.running:
267            await self.broadcast_state()
268
269    def count_discs(self) -> Tuple[int, int]:
270        """
271        Counts the number of discs for each player.
272
273        Returns:
274            Tuple[int, int]: A tuple containing (player1_count, player2_count).
275        """
276        p1, p2 = 0, 0
277        for row in self.board:
278            for cell in row:
279                if cell == 1:
280                    p1 += 1
281                elif cell == 2:
282                    p2 += 1
283        return p1, p2
284
285    async def check_game_over(self) -> None:
286        """Checks if the game is over and determines the winner."""
287        p1, p2 = self.count_discs()
288        winner: int = 1 if p1 > p2 else (2 if p2 > p1 else 0)
289
290        if winner:
291            self.match_scores[winner] += 1
292
293        msg: str = (
294            f"Player {winner} Wins ({p1}-{p2})!" if winner else f"Draw ({p1}-{p2})!"
295        )
296        logging.info(f"Game Over: {msg}")
297        await self.end_round(msg)
298
299    async def end_round(self, message: str) -> None:
300        """
301        Ends the current round and prepares for the next.
302
303        Args:
304            message (str): The message to broadcast (winner/draw).
305        """
306        self.running = False
307        payload: Dict[str, Any] = {"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        logging.info("Round ended. Preparing for next round in 3 seconds...")
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        """Broadcasts the current game state to all agents."""
321        async with self.broadcast_lock:
322            valid_actions: List[List[int]] = self.get_valid_actions(self.current_turn)
323            logging.info(f"Broadcasting state for Player {self.current_turn}. Valid moves: {valid_actions}")
324            payload: Dict[str, Any] = {
325                "type": "state",
326                "board": self.board,
327                "current_turn": self.current_turn,
328                "valid_actions": valid_actions,
329            }
330            msg: str = 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        """Sends an update message to the frontend."""
338        if self.frontend_ws:
339            p1, p2 = self.count_discs()
340            try:
341                await self.frontend_ws.send(
342                    json.dumps(
343                        {
344                            "type": "update",
345                            "board": self.board,
346                            "current_turn": self.current_turn,
347                            "valid_actions": self.get_valid_actions(self.current_turn)
348                            if self.running
349                            else [],
350                            "disc_counts": {1: p1, 2: p2},
351                            "match_scores": self.match_scores,
352                            "p1_connected": self.agent1_ws is not None,
353                            "p2_connected": self.agent2_ws is not None,
354                        }
355                    )
356                )
357            except Exception as e:
358                logging.error(f"Error updating frontend: {e}")
359                self.frontend_ws = None
360
361
362if __name__ == "__main__":
363    server = OthelloServer()
364    asyncio.run(server.start())