207 lines
7.4 KiB
Python
207 lines
7.4 KiB
Python
# lobby.py : gestion des lobbies (creation, classe, ready, demarrage)
|
|
# les methodes retournent (conn_id, message), c'est main.py qui envoie
|
|
|
|
import logging
|
|
import random
|
|
from dataclasses import dataclass, field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# pas de 0/O/I/1 pour eviter les confusions
|
|
_LOBBY_CODE_CHARS = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
|
|
_LOBBY_CODE_LENGTH = 6
|
|
MAX_PLAYERS = 3
|
|
VALID_CLASSES = {"kael", "seris", "aldric"}
|
|
|
|
|
|
@dataclass
|
|
class LobbyPlayer:
|
|
conn_id: str
|
|
username: str
|
|
player_class: str | None = None
|
|
is_ready: bool = False
|
|
is_host: bool = False
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"id": self.conn_id,
|
|
"username": self.username,
|
|
"class": self.player_class,
|
|
"ready": self.is_ready,
|
|
"host": self.is_host,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Lobby:
|
|
code: str
|
|
players: list[LobbyPlayer] = field(default_factory=list)
|
|
started: bool = False
|
|
|
|
def get_conn_ids(self) -> list[str]:
|
|
return [p.conn_id for p in self.players]
|
|
|
|
def get_player(self, conn_id):
|
|
return next((p for p in self.players if p.conn_id == conn_id), None)
|
|
|
|
def is_class_taken(self, class_name):
|
|
return any(p.player_class == class_name for p in self.players)
|
|
|
|
def all_ready(self):
|
|
return (
|
|
len(self.players) == MAX_PLAYERS
|
|
and all(p.is_ready and p.player_class for p in self.players)
|
|
)
|
|
|
|
|
|
class LobbyManager:
|
|
def __init__(self):
|
|
self.lobbies: dict[str, Lobby] = {}
|
|
self.conn_to_lobby: dict[str, str] = {}
|
|
|
|
def _generate_code(self):
|
|
while True:
|
|
code = "".join(random.choices(_LOBBY_CODE_CHARS, k=_LOBBY_CODE_LENGTH))
|
|
if code not in self.lobbies:
|
|
return code
|
|
|
|
def _error(self, conn_id, code, message):
|
|
return [(conn_id, {"type": "error", "code": code, "message": message})]
|
|
|
|
def create_lobby(self, conn_id, username):
|
|
# si on est deja dans un lobby on le libere d'abord, evite les blocages au retour menu
|
|
leave_msgs = []
|
|
if conn_id in self.conn_to_lobby:
|
|
leave_msgs = self.disconnect(conn_id)
|
|
|
|
code = self._generate_code()
|
|
player = LobbyPlayer(conn_id=conn_id, username=username, is_host=True)
|
|
lobby = Lobby(code=code, players=[player])
|
|
|
|
self.lobbies[code] = lobby
|
|
self.conn_to_lobby[conn_id] = code
|
|
|
|
logger.info("Lobby %s créé par %s", code, username)
|
|
return leave_msgs + [(conn_id, {"type": "lobby_created", "code": code})]
|
|
|
|
def join_lobby(self, conn_id, username, code):
|
|
code = code.upper()
|
|
|
|
leave_msgs = []
|
|
if conn_id in self.conn_to_lobby:
|
|
leave_msgs = self.disconnect(conn_id)
|
|
|
|
lobby = self.lobbies.get(code)
|
|
if not lobby:
|
|
return self._error(conn_id, "lobby_not_found", f"Lobby {code} introuvable.")
|
|
if lobby.started:
|
|
return self._error(conn_id, "game_in_progress", "La partie a déjà commencé.")
|
|
if len(lobby.players) >= MAX_PLAYERS:
|
|
return self._error(conn_id, "lobby_full", "Le lobby est plein (3/3).")
|
|
|
|
player = LobbyPlayer(conn_id=conn_id, username=username)
|
|
lobby.players.append(player)
|
|
self.conn_to_lobby[conn_id] = code
|
|
|
|
messages = [
|
|
(conn_id, {"type": "lobby_joined", "code": code, "players": [p.to_dict() for p in lobby.players]}),
|
|
]
|
|
# notifier les autres
|
|
for p in lobby.players:
|
|
if p.conn_id != conn_id:
|
|
messages.append((p.conn_id, {"type": "player_joined", "player": player.to_dict()}))
|
|
|
|
logger.info("%s a rejoint le lobby %s (%d/3)", username, code, len(lobby.players))
|
|
return leave_msgs + messages
|
|
|
|
def select_class(self, conn_id, class_name):
|
|
code = self.conn_to_lobby.get(conn_id)
|
|
if not code:
|
|
return self._error(conn_id, "not_in_lobby", "Tu n'es pas dans un lobby.")
|
|
|
|
lobby = self.lobbies[code]
|
|
player = lobby.get_player(conn_id)
|
|
if not player:
|
|
return self._error(conn_id, "not_in_lobby", "Joueur introuvable dans le lobby.")
|
|
if class_name not in VALID_CLASSES:
|
|
return self._error(conn_id, "invalid_class", f"Classe invalide : {class_name}.")
|
|
# un joueur peut reselectionner sa propre classe sans erreur
|
|
if lobby.is_class_taken(class_name) and player.player_class != class_name:
|
|
return self._error(conn_id, "class_taken", f"La classe {class_name} est déjà prise.")
|
|
|
|
player.player_class = class_name
|
|
player.is_ready = False # reset du ready si on change de classe
|
|
|
|
logger.info("%s a choisi la classe %s", player.username, class_name)
|
|
return [(p.conn_id, {"type": "class_selected", "player_id": conn_id, "class": class_name}) for p in lobby.players]
|
|
|
|
def set_ready(self, conn_id):
|
|
code = self.conn_to_lobby.get(conn_id)
|
|
if not code:
|
|
return self._error(conn_id, "not_in_lobby", "Tu n'es pas dans un lobby.")
|
|
|
|
lobby = self.lobbies[code]
|
|
player = lobby.get_player(conn_id)
|
|
if not player:
|
|
return self._error(conn_id, "not_in_lobby", "Joueur introuvable.")
|
|
if not player.player_class:
|
|
return self._error(conn_id, "no_class_selected", "Choisis une classe avant de te mettre prêt.")
|
|
|
|
player.is_ready = True
|
|
logger.info("%s est prêt", player.username)
|
|
return [(p.conn_id, {"type": "player_ready", "player_id": conn_id}) for p in lobby.players]
|
|
|
|
def start_game(self, conn_id):
|
|
code = self.conn_to_lobby.get(conn_id)
|
|
if not code:
|
|
return self._error(conn_id, "not_in_lobby", "Tu n'es pas dans un lobby.")
|
|
|
|
lobby = self.lobbies[code]
|
|
player = lobby.get_player(conn_id)
|
|
if not player or not player.is_host:
|
|
return self._error(conn_id, "not_host", "Seul le chef de partie peut démarrer.")
|
|
if not lobby.all_ready():
|
|
return self._error(
|
|
conn_id, "not_all_ready",
|
|
f"Tous les joueurs doivent être prêts avec une classe ({len(lobby.players)}/3).",
|
|
)
|
|
|
|
lobby.started = True
|
|
logger.info("Partie démarrée dans le lobby %s", code)
|
|
return [(p.conn_id, {"type": "game_starting", "countdown": 3}) for p in lobby.players]
|
|
|
|
def leave_lobby(self, conn_id):
|
|
# bouton Back : on quitte volontairement, meme logique que disconnect + confirmation
|
|
if conn_id not in self.conn_to_lobby:
|
|
return []
|
|
notifications = self.disconnect(conn_id)
|
|
notifications.insert(0, (conn_id, {"type": "lobby_left"}))
|
|
return notifications
|
|
|
|
def disconnect(self, conn_id):
|
|
code = self.conn_to_lobby.pop(conn_id, None)
|
|
if not code:
|
|
return []
|
|
|
|
lobby = self.lobbies.get(code)
|
|
if not lobby:
|
|
return []
|
|
|
|
player = lobby.get_player(conn_id)
|
|
if not player:
|
|
return []
|
|
|
|
lobby.players = [p for p in lobby.players if p.conn_id != conn_id]
|
|
logger.info("%s a quitté le lobby %s", player.username, code)
|
|
|
|
if not lobby.players:
|
|
del self.lobbies[code]
|
|
logger.info("Lobby %s supprimé (vide)", code)
|
|
return []
|
|
|
|
# transferer l'host au premier restant
|
|
if player.is_host:
|
|
lobby.players[0].is_host = True
|
|
|
|
return [(p.conn_id, {"type": "player_left", "player_id": conn_id}) for p in lobby.players]
|
