TP03

battleship_server.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Bataille Navale - Serveur LAN (arbitre)
+Lance le serveur, attend 2 clients, échange messages JSON ligne par ligne.
+"""
+
+import socket
+import threading
+import json
+import time
+
+HOST = "0.0.0.0"
+PORT = 5000
+
+# Board constants
+ROWS = 10
+COLS = 10
+SHIPS_DEF = {
+    "Carrier": 5,
+    "Battleship": 4,
+    "Cruiser": 3,
+    "Submarine": 3,
+    "Destroyer": 2
+}
+
+# Helper: send JSON line
+def send_line(conn, data):
+    try:
+        conn.sendall((json.dumps(data) + "\n").encode())
+    except Exception as e:
+        print("Send error:", e)
+
+# Helper: read JSON line from file-like socket
+def recv_line(f):
+    line = f.readline()
+    if not line:
+        return None
+    return json.loads(line.strip())
+
+class Player:
+    def __init__(self, conn, addr, idx):
+        self.conn = conn
+        self.f = conn.makefile(mode="r", encoding="utf-8")
+        self.addr = addr
+        self.idx = idx
+        self.ready = False
+        self.board_ships = None  # dict of ship_id -> list of (r,c)
+        self.hits_received = set()  # positions hit by opponent
+
+    def close(self):
+        try:
+            self.f.close()
+        except:
+            pass
+        try:
+            self.conn.close()
+        except:
+            pass
+
+class GameServer:
+    def __init__(self, host, port):
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.sock.bind((host, port))
+        self.sock.listen(4)
+        print(f"[SERVER] Listening on {host}:{port}")
+        self.players = []
+        self.lock = threading.Lock()
+        self.turn = 0  # player index whose turn it is
+
+    def accept_players(self):
+        while len(self.players) < 2:
+            conn, addr = self.sock.accept()
+            with self.lock:
+                idx = len(self.players)
+                player = Player(conn, addr, idx)
+                self.players.append(player)
+            print(f"[SERVER] Player {idx+1} connected from {addr}")
+            threading.Thread(target=self.handle_player, args=(player,), daemon=True).start()
+
+        # when two players connected start handshake
+        self.broadcast({"type": "info", "msg": "2 players connected. Place your ships."})
+        # ask for placements
+        self.broadcast({"type": "start_place", "ships": SHIPS_DEF})
+
+    def broadcast(self, data):
+        with self.lock:
+            for p in list(self.players):
+                try:
+                    send_line(p.conn, data)
+                except:
+                    pass
+
+    def handle_player(self, player):
+        try:
+            send_line(player.conn, {"type": "welcome", "player_idx": player.idx, "msg": f"You are player {player.idx+1}"})
+            while True:
+                data = recv_line(player.f)
+                if data is None:
+                    print(f"[SERVER] Player {player.idx+1} disconnected")
+                    break
+                self.process_message(player, data)
+        except Exception as e:
+            print("Player handler error:", e)
+        finally:
+            player.close()
+
+    def process_message(self, player, data):
+        t = data.get("type")
+        if t == "place":
+            # expects 'ships': {name: [[r,c],...], ...}
+            ships = data.get("ships")
+            if not self.validate_placement(ships):
+                send_line(player.conn, {"type": "error", "msg": "Invalid placement"})
+                return
+            player.board_ships = ships
+            player.ready = True
+            send_line(player.conn, {"type": "placed_ok"})
+            print(f"[SERVER] Player {player.idx+1} placed ships")
+            # If both ready, start game
+            if all(p.ready for p in self.players) and len(self.players) >= 2:
+                time.sleep(0.2)
+                self.start_game()
+        elif t == "shot":
+            # only process if both ready
+            if not all(p.ready for p in self.players):
+                send_line(player.conn, {"type": "error", "msg": "Both players not ready"})
+                return
+            if player.idx != self.turn:
+                send_line(player.conn, {"type": "error", "msg": "Not your turn"})
+                return
+            r = data.get("r")
+            c = data.get("c")
+            self.handle_shot(player.idx, r, c)
+        elif t == "forfeit":
+            # opponent wins
+            winner = 1 - player.idx
+            self.end_game(winner, reason=f"Player {player.idx+1} forfeited")
+        else:
+            send_line(player.conn, {"type": "error", "msg": "Unknown message type"})
+
+    def validate_placement(self, ships):
+        # ships is a dict name->list of [r,c]
+        if not isinstance(ships, dict):
+            return False
+        occupied = set()
+        for name, coords in ships.items():
+            if name not in SHIPS_DEF:
+                return False
+            if not isinstance(coords, list) or len(coords) != SHIPS_DEF[name]:
+                return False
+            # each coord valid and not overlapping, must form straight contiguous line
+            rows = [p[0] for p in coords]
+            cols = [p[1] for p in coords]
+            if any(not (0 <= rr < ROWS and 0 <= cc < COLS) for rr, cc in coords):
+                return False
+            # contiguous line check: either same row or same col
+            if not (len(set(rows)) == 1 or len(set(cols)) == 1):
+                return False
+            # ensure consecutive
+            if len(set(rows)) == 1:
+                sorted_cols = sorted(cols)
+                if sorted_cols != list(range(min(cols), max(cols)+1)):
+                    return False
+            else:
+                sorted_rows = sorted(rows)
+                if sorted_rows != list(range(min(rows), max(rows)+1)):
+                    return False
+            for p in coords:
+                if tuple(p) in occupied:
+                    return False
+                occupied.add(tuple(p))
+        return True
+
+    def start_game(self):
+        # tell players game starts and whose turn
+        self.turn = 0
+        for p in self.players:
+            send_line(p.conn, {"type": "start_game", "your_idx": p.idx})
+        self.notify_turn()
+
+    def notify_turn(self):
+        # notify both players whose turn it is
+        for p in self.players:
+            send_line(p.conn, {"type": "turn", "current": self.turn})
+
+    def handle_shot(self, shooter_idx, r, c):
+        shooter = self.players[shooter_idx]
+        target = self.players[1 - shooter_idx]
+        pos = (r, c)
+        # check bounds
+        if not (0 <= r < ROWS and 0 <= c < COLS):
+            send_line(shooter.conn, {"type": "error", "msg": "Shot out of bounds"})
+            return
+        # if already shot there? check by looking at hits_received of target
+        if pos in target.hits_received:
+            send_line(shooter.conn, {"type": "error", "msg": "Position already shot"})
+            return
+        # mark hit
+        target.hits_received.add(pos)
+        # determine result
+        hit_ship = None
+        for name, coords in target.board_ships.items():
+            coords_t = [tuple(p) for p in coords]
+            if pos in coords_t:
+                hit_ship = name
+                break
+        if hit_ship is None:
+            # miss
+            send_line(shooter.conn, {"type": "result", "res": "miss", "r": r, "c": c})
+            send_line(target.conn, {"type": "update", "opponent_shot": {"r": r, "c": c, "res": "miss"}})
+            # switch turn
+            self.turn = 1 - self.turn
+            self.notify_turn()
+        else:
+            # check if ship sunk
+            coords_t = [tuple(p) for p in target.board_ships[hit_ship]]
+            sunk = all(p in target.hits_received for p in coords_t)
+            if sunk:
+                send_line(shooter.conn, {"type": "result", "res": "sunk", "ship": hit_ship, "r": r, "c": c})
+                send_line(target.conn, {"type": "update", "opponent_shot": {"r": r, "c": c, "res": "sunk", "ship": hit_ship}})
+                # check win: all ships coords in hits_received?
+                all_coords = []
+                for coords in target.board_ships.values():
+                    all_coords.extend([tuple(p) for p in coords])
+                if set(all_coords).issubset(target.hits_received):
+                    # shooter wins
+                    self.end_game(shooter_idx, reason="All ships sunk")
+                    return
+                else:
+                    # shooter keeps the turn after hit (classic rule variations exist; we keep shooter on turn)
+                    self.notify_turn()
+            else:
+                send_line(shooter.conn, {"type": "result", "res": "hit", "r": r, "c": c})
+                send_line(target.conn, {"type": "update", "opponent_shot": {"r": r, "c": c, "res": "hit"}})
+                # shooter keeps turn
+                self.notify_turn()
+
+    def end_game(self, winner_idx, reason=""):
+        print(f"[SERVER] Game ended. Winner: Player {winner_idx+1}. Reason: {reason}")
+        for p in self.players:
+            send_line(p.conn, {"type": "end", "winner": winner_idx, "reason": reason})
+        # close sockets after a short pause
+        time.sleep(1.0)
+        for p in self.players:
+            try:
+                p.close()
+            except:
+                pass
+        # reset server for new game? here we exit
+        print("[SERVER] Shutting down.")
+        try:
+            self.sock.close()
+        except:
+            pass
+        exit(0)
+
+
+if __name__ == "__main__":
+    server = GameServer(HOST, PORT)
+    server.accept_players()