262 lines
9.3 KiB
Python
262 lines
9.3 KiB
Python
#!/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()
|