""" Gestionnaire WebSocket de synchronisation temps réel pour les Outils Tactiques. Permet de synchroniser instantanément minuteurs et compteurs (points/respawns) entre plusieurs smartphones sur le terrain, sans historique lourd ni base de données. """ import time import random import string import math import asyncio from typing import Dict, Set, Any, Optional from fastapi import WebSocket class ToolsRoom: """Représente l'état en direct d'un salon d'outils synchronisés.""" def __init__(self, code: str): self.code = code self.created_at = time.time() self.last_activity = time.time() # Minuteur synchronisé self.timer_initial_seconds = 20 * 60 # 20 minutes par défaut self.timer_seconds_left = 20 * 60 self.timer_running = False self.timer_last_start_time: Optional[float] = None # Compteur / Incrémenteur synchronisé self.counter_mode = "teams" # "single" ou "teams" self.counter_single = 0 self.counter_label = "Points" self.counter_teams_count = 2 # 2 par défaut, jusqu'à 4 équipes self.counter_alpha = 0 # Bleu self.counter_bravo = 0 # Rouge self.counter_charlie = 0 # Jaune self.counter_delta = 0 # Vert # Onglet actif suggéré self.active_tab = "timer" def get_current_timer_seconds(self) -> int: """Calcule le temps restant réel si le chrono tourne.""" if not self.timer_running or self.timer_last_start_time is None: return max(0, int(self.timer_seconds_left)) elapsed = time.time() - self.timer_last_start_time remaining = max(0, int(self.timer_seconds_left - elapsed)) return remaining def to_dict(self, connected_count: int = 1) -> Dict[str, Any]: """Sérialise l'état du salon pour diffusion JSON aux clients.""" current_time = self.get_current_timer_seconds() return { "type": "sync_state", "room_code": self.code, "connected_count": connected_count, "active_tab": self.active_tab, "timer": { "running": self.timer_running, "seconds_left": current_time, "initial_seconds": self.timer_initial_seconds, }, "counter": { "mode": self.counter_mode, "single": self.counter_single, "label": self.counter_label, "teams_count": self.counter_teams_count, "alpha": self.counter_alpha, "bravo": self.counter_bravo, "charlie": self.counter_charlie, "delta": self.counter_delta, } } class ToolsSyncManager: """Gère l'ensemble des salons d'outils et leurs connexions WebSocket.""" def __init__(self): # Code de salon -> ensemble des connexions WebSocket actives self.active_connections: Dict[str, Set[WebSocket]] = {} # Code de salon -> objet ToolsRoom self.rooms: Dict[str, ToolsRoom] = {} def get_or_create_room(self, code: str) -> ToolsRoom: """Récupère une room existante ou en instancie une nouvelle.""" clean_code = code.upper().strip() if clean_code not in self.rooms: self.rooms[clean_code] = ToolsRoom(clean_code) return self.rooms[clean_code] def generate_room_code(self) -> str: """Génère un code de salon court et lisible (ex: SAT78 ou V78A).""" chars = string.ascii_uppercase + string.digits for _ in range(100): suffix = "".join(random.choices(chars, k=3)) code = f"T{suffix}" if code not in self.rooms: return code return f"T{int(time.time()) % 10000:04d}" async def connect(self, room_code: str, websocket: WebSocket): """Accepte et enregistre une connexion WebSocket dans un salon.""" await websocket.accept() clean_code = room_code.upper().strip() if clean_code not in self.active_connections: self.active_connections[clean_code] = set() self.active_connections[clean_code].add(websocket) self.get_or_create_room(clean_code) def disconnect(self, room_code: str, websocket: WebSocket): """Retire une connexion WebSocket d'un salon. Si plus personne n'est connecté, supprime définitivement la room de la mémoire RAM.""" clean_code = room_code.upper().strip() if clean_code in self.active_connections: self.active_connections[clean_code].discard(websocket) if not self.active_connections[clean_code]: del self.active_connections[clean_code] if clean_code in self.rooms: del self.rooms[clean_code] def get_connected_count(self, room_code: str) -> int: """Retourne le nombre de téléphones connectés au salon.""" clean_code = room_code.upper().strip() return len(self.active_connections.get(clean_code, set())) async def broadcast(self, room_code: str, data: Dict[str, Any]): """Diffuse un message JSON à tous les clients du salon.""" clean_code = room_code.upper().strip() if clean_code not in self.active_connections: return dead_connections = set() for connection in list(self.active_connections[clean_code]): try: await connection.send_json(data) except Exception: dead_connections.add(connection) for dead in dead_connections: self.disconnect(clean_code, dead) async def broadcast_room_state(self, room_code: str): """Diffuse l'état complet du salon à tous les appareils connectés.""" clean_code = room_code.upper().strip() if clean_code not in self.active_connections: return room = self.get_or_create_room(clean_code) count = self.get_connected_count(clean_code) await self.broadcast(clean_code, room.to_dict(count)) # Instance singleton globale pour les outils tools_sync_manager = ToolsSyncManager() # ============================================================================== # GESTIONNAIRE TEMPS RÉEL DU MODE : LARGUÉ EN INCONNU (RÉSEAU LOCAL & ATTRIBUTION 50/50) # ============================================================================== class LargagePlayer: """Représente un opérateur connecté dans le salon de Largage en Inconnu.""" def __init__(self, player_id: str, name: str, is_host: bool = False): self.player_id = player_id self.name = name self.is_host = is_host self.team: Optional[str] = None # "bleue" ou "rouge" self.connected: bool = True self.joined_at: float = time.time() def to_dict(self) -> Dict[str, Any]: return { "id": self.player_id, "name": self.name, "is_host": self.is_host, "team": self.team, "connected": self.connected, "joined_at": self.joined_at, } class LargageRoom: """Salon de coordination tactique pour le mode de jeu 'Largué en Inconnu'.""" def __init__(self, code: str): self.code = code self.created_at = time.time() self.last_activity = time.time() self.status = "lobby" # "lobby", "countdown", "started", "finished" self.host_player_id: Optional[str] = None self.round_number = 1 self.countdown_seconds = 5 self.countdown_end_time: Optional[float] = None self.started_at: Optional[float] = None self.players: Dict[str, LargagePlayer] = {} def get_or_create_player(self, player_id: str, name: str) -> LargagePlayer: self.last_activity = time.time() if not self.host_player_id: self.host_player_id = player_id if player_id in self.players: p = self.players[player_id] if name and name.strip(): p.name = name.strip() p.connected = True p.is_host = (player_id == self.host_player_id) return p is_host = (self.host_player_id == player_id) clean_name = name.strip() if name and name.strip() else f"Opérateur #{len(self.players) + 1}" new_player = LargagePlayer(player_id, clean_name, is_host=is_host) self.players[player_id] = new_player # Si la partie a déjà commencé et qu'un joueur rejoint en retard, # équilibre automatique 50/50 pour maintenir la balance : if self.status == "started": blues = sum(1 for p in self.players.values() if p.team == "bleue") reds = sum(1 for p in self.players.values() if p.team == "rouge") new_player.team = "bleue" if blues <= reds else "rouge" return new_player def start_game(self, countdown: int = 5): """Distribue équitablement 50/50 les joueurs entre Équipe Rouge et Équipe Bleue.""" self.last_activity = time.time() # Récupère tous les joueurs connectés (ou tous si testing solo) connected_ids = [pid for pid, p in self.players.items() if p.connected] if not connected_ids: connected_ids = list(self.players.keys()) # Tirage aléatoire équitable 50/50 random.shuffle(connected_ids) total = len(connected_ids) half = total // 2 extra_blue = 1 if (total % 2 != 0 and random.choice([True, False])) else 0 blue_limit = half + extra_blue for idx, pid in enumerate(connected_ids): self.players[pid].team = "bleue" if idx < blue_limit else "rouge" # Les joueurs hors-ligne sans équipe reçoivent aussi une assignation équilibrée for pid, p in self.players.items(): if pid not in connected_ids and not p.team: blues = sum(1 for pl in self.players.values() if pl.team == "bleue") reds = sum(1 for pl in self.players.values() if pl.team == "rouge") p.team = "bleue" if blues <= reds else "rouge" self.started_at = time.time() self.countdown_seconds = max(0, countdown) if self.countdown_seconds > 0: self.status = "countdown" self.countdown_end_time = time.time() + self.countdown_seconds else: self.status = "started" self.countdown_end_time = None def trigger_reveal(self): """Active l'état démarré après la fin du compte à rebours.""" self.status = "started" self.countdown_end_time = None def reset_round(self): """Réinitialise les équipes et relance le lobby pour une nouvelle manche.""" self.last_activity = time.time() self.status = "lobby" self.countdown_end_time = None self.started_at = None self.round_number += 1 for p in self.players.values(): p.team = None def to_dict_for(self, player_id: Optional[str] = None) -> Dict[str, Any]: """Formate l'état temps réel adapté à l'opérateur concerné.""" connected_count = sum(1 for p in self.players.values() if p.connected) blue_count = sum(1 for p in self.players.values() if p.team == "bleue") red_count = sum(1 for p in self.players.values() if p.team == "rouge") my_team = None is_host = False my_name = "Opérateur" if player_id and player_id in self.players: p = self.players[player_id] my_team = p.team is_host = p.is_host my_name = p.name # Liste des joueurs : les rôles ne sont dévoilés qu'au créateur du salon (hôte) players_summary = [ { "id": p.player_id, "name": p.name, "is_host": p.is_host, "connected": p.connected, "team": p.team if is_host else None, } for p in self.players.values() ] countdown_left = 0 if self.status == "countdown" and self.countdown_end_time: countdown_left = max(0, int(round(self.countdown_end_time - time.time()))) elapsed = 0 if self.status == "started" and self.started_at: elapsed = max(0, int(time.time() - self.started_at)) return { "type": "largage_state", "room_code": self.code, "status": self.status, "round_number": self.round_number, "connected_count": connected_count, "total_players": len(self.players), "is_host": is_host, "my_player_id": player_id, "my_name": my_name, "my_team": my_team, "team_counts": { "bleue": blue_count, "rouge": red_count, }, "countdown_seconds": countdown_left, "elapsed_seconds": elapsed, "players": players_summary, } class LargageSyncManager: """Gestionnaire temps réel des salons de dispersion 'Largué en Inconnu'.""" def __init__(self): # room_code -> Dict[WebSocket, str (player_id)] self.active_connections: Dict[str, Dict[WebSocket, str]] = {} # room_code -> LargageRoom self.rooms: Dict[str, LargageRoom] = {} def get_or_create_room(self, code: str) -> LargageRoom: clean_code = code.upper().strip() if clean_code not in self.rooms: self.rooms[clean_code] = LargageRoom(clean_code) return self.rooms[clean_code] def generate_room_code(self) -> str: chars = string.ascii_uppercase + string.digits for _ in range(100): suffix = "".join(random.choices(chars, k=3)) code = f"L{suffix}" if code not in self.rooms: return code return f"L{int(time.time()) % 10000:04d}" async def connect(self, room_code: str, websocket: WebSocket, player_id: str, name: str): await websocket.accept() clean_code = room_code.upper().strip() if clean_code not in self.active_connections: self.active_connections[clean_code] = {} self.active_connections[clean_code][websocket] = player_id room = self.get_or_create_room(clean_code) room.get_or_create_player(player_id, name) await self.broadcast_room_state(clean_code) def disconnect(self, room_code: str, websocket: WebSocket): clean_code = room_code.upper().strip() if clean_code in self.active_connections: player_id = self.active_connections[clean_code].pop(websocket, None) if clean_code in self.rooms and player_id: if player_id in self.rooms[clean_code].players: self.rooms[clean_code].players[player_id].connected = False if not self.active_connections[clean_code]: del self.active_connections[clean_code] if clean_code in self.rooms: # Si aucun joueur n'est connecté et statut lobby, libérer la mémoire if all(not p.connected for p in self.rooms[clean_code].players.values()): del self.rooms[clean_code] def get_connected_count(self, room_code: str) -> int: clean_code = room_code.upper().strip() return len(self.active_connections.get(clean_code, {})) async def broadcast_room_state(self, room_code: str): clean_code = room_code.upper().strip() if clean_code not in self.active_connections or clean_code not in self.rooms: return room = self.rooms[clean_code] dead_connections = [] for ws, pid in list(self.active_connections[clean_code].items()): try: state_data = room.to_dict_for(pid) await ws.send_json(state_data) except Exception: dead_connections.append(ws) for dead in dead_connections: self.disconnect(clean_code, dead) largage_sync_manager = LargageSyncManager() # ============================================================================== # GESTIONNAIRE TEMPS RÉEL DU MODE : CAPTURE DE ZONES (ARBITRAGE GPS & TICKS) # ============================================================================== def calculate_haversine_distance(lat1: float, lng1: float, lat2: float, lng2: float) -> float: """Calcule la distance géodésique en mètres entre deux coordonnées GPS.""" r = 6371000.0 # Rayon moyen de la Terre en mètres phi1 = math.radians(lat1) phi2 = math.radians(lat2) delta_phi = math.radians(lat2 - lat1) delta_lambda = math.radians(lng2 - lng1) a = (math.sin(delta_phi / 2.0) ** 2 + math.cos(phi1) * math.cos(phi2) * (math.sin(delta_lambda / 2.0) ** 2)) c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(max(0.0, 1.0 - a))) return r * c class ZoneCapturePlayer: """Représente un joueur connecté dans le salon de Capture de zones.""" def __init__(self, player_id: str, name: str, is_host: bool = False, team: str = "bleue"): self.player_id = player_id self.name = name self.is_host = is_host self.team = team # "bleue" ou "rouge" self.connected = True self.lat: Optional[float] = None self.lng: Optional[float] = None self.accuracy: Optional[float] = None self.last_update: float = time.time() self.joined_at: float = time.time() def to_dict(self) -> Dict[str, Any]: return { "id": self.player_id, "name": self.name, "is_host": self.is_host, "team": self.team, "connected": self.connected, "lat": self.lat, "lng": self.lng, "accuracy": self.accuracy, "last_update": self.last_update, "joined_at": self.joined_at, } class ZoneObjective: """Représente l'une des 4 zones stratégiques sur le terrain de Satory.""" def __init__(self, zone_id: str, name: str, lat: float, lng: float, radius: float = 20.0): self.id = zone_id self.name = name self.lat = lat self.lng = lng self.radius = radius # Rayon en mètres self.state = "neutre" # "neutre", "en_cours_capture_bleu", "bleu", "en_cours_capture_rouge", "rouge" self.owner: Optional[str] = None # None, "bleue", "rouge" self.capturing_team: Optional[str] = None self.capture_progress: float = 0.0 # 0 à 10 secondes self.capture_duration: float = 10.0 self.is_contested: bool = False self.blues_in_zone: int = 0 self.reds_in_zone: int = 0 def to_dict(self) -> Dict[str, Any]: return { "id": self.id, "name": self.name, "lat": self.lat, "lng": self.lng, "radius": self.radius, "state": self.state, "owner": self.owner, "capturing_team": self.capturing_team, "capture_progress": round(self.capture_progress, 1), "capture_duration": self.capture_duration, "is_contested": self.is_contested, "blues_in_zone": self.blues_in_zone, "reds_in_zone": self.reds_in_zone, } class ZoneCaptureRoom: """Salon de jeu et moteur d'arbitrage temps réel pour la Capture de zones.""" def __init__(self, code: str): self.code = code self.created_at = time.time() self.last_activity = time.time() self.status = "lobby" # "lobby", "started", "paused", "finished" self.host_player_id: Optional[str] = None # Minuteur synchronisé self.initial_timer_seconds = 30 * 60 # 30 minutes par défaut self.timer_seconds_left = 30 * 60 # Scores en direct self.score_bleu = 0 self.score_rouge = 0 self.winner: Optional[str] = None # "bleue", "rouge", "draw" self.victory_reason: Optional[str] = None # "total_control", "time_expired" # Configuration de capture self.capture_duration_seconds = 10.0 self.capture_radius_meters = 20.0 # Joueurs enregistrés self.players: Dict[str, ZoneCapturePlayer] = {} # 4 zones officielles configurées à Satory (issues du scénario) self.zones: Dict[str, ZoneObjective] = { "zone-charly": ZoneObjective( zone_id="zone-charly", name="Zone Charly", lat=48.79320, lng=2.10150, radius=self.capture_radius_meters ), "zone-delta": ZoneObjective( zone_id="zone-delta", name="Zone Delta", lat=48.79360, lng=2.10330, radius=self.capture_radius_meters ), "zone-echo": ZoneObjective( zone_id="zone-echo", name="Zone Echo", lat=48.79280, lng=2.10250, radius=self.capture_radius_meters ), "zone-foxtrot": ZoneObjective( zone_id="zone-foxtrot", name="Zone Fox-trot", lat=48.79220, lng=2.10180, radius=self.capture_radius_meters ), } def get_or_create_player(self, player_id: str, name: str) -> ZoneCapturePlayer: self.last_activity = time.time() if not self.host_player_id: self.host_player_id = player_id if player_id in self.players: p = self.players[player_id] if name and name.strip(): p.name = name.strip() p.connected = True p.is_host = (player_id == self.host_player_id) return p is_host = (self.host_player_id == player_id) clean_name = name.strip() if name and name.strip() else f"Opérateur #{len(self.players) + 1}" # Équilibrage initial automatique à l'entrée dans le lobby (Bleu vs Rouge) blues = sum(1 for pl in self.players.values() if pl.team == "bleue") reds = sum(1 for pl in self.players.values() if pl.team == "rouge") initial_team = "bleue" if blues <= reds else "rouge" new_player = ZoneCapturePlayer(player_id, clean_name, is_host=is_host, team=initial_team) self.players[player_id] = new_player return new_player def move_player_team(self, player_id: str, new_team: str): """Permet à l'hôte de déplacer un joueur via le drag & drop.""" if player_id in self.players and new_team in ["bleue", "rouge"]: self.players[player_id].team = new_team self.last_activity = time.time() if self.status == "started": self.evaluate_presences() def set_timer_duration(self, seconds: int): """Définit la durée du chronomètre de la partie.""" self.initial_timer_seconds = max(10, int(seconds)) if self.status == "lobby": self.timer_seconds_left = self.initial_timer_seconds self.last_activity = time.time() def start_game(self): """Lance la mission de capture.""" self.last_activity = time.time() self.status = "started" self.winner = None self.victory_reason = None def pause_game(self): self.last_activity = time.time() if self.status == "started": self.status = "paused" def resume_game(self): self.last_activity = time.time() if self.status == "paused": self.status = "started" def reset_game(self): """Réinitialise les scores et remet le salon en mode lobby.""" self.last_activity = time.time() self.status = "lobby" self.score_bleu = 0 self.score_rouge = 0 self.winner = None self.victory_reason = None self.timer_seconds_left = self.initial_timer_seconds for z in self.zones.values(): z.state = "neutre" z.owner = None z.capturing_team = None z.capture_progress = 0.0 z.is_contested = False z.blues_in_zone = 0 z.reds_in_zone = 0 def update_player_position(self, player_id: str, lat: float, lng: float, accuracy: Optional[float] = None): """Met à jour les coordonnées GPS d'un opérateur.""" if player_id in self.players: p = self.players[player_id] p.lat = float(lat) p.lng = float(lng) if accuracy is not None: p.accuracy = float(accuracy) p.last_update = time.time() self.last_activity = time.time() if self.status == "started": self.evaluate_presences() def evaluate_presences(self): """Calcule la présence des joueurs dans chaque zone selon leur position GPS.""" for z in self.zones.values(): blues = 0 reds = 0 for p in self.players.values(): if not p.connected: continue if p.lat is None or p.lng is None: continue dist = calculate_haversine_distance(p.lat, p.lng, z.lat, z.lng) if dist <= z.radius: if p.team == "bleue": blues += 1 elif p.team == "rouge": reds += 1 z.blues_in_zone = blues z.reds_in_zone = reds z.is_contested = (blues >= 1 and reds >= 1) def evaluate_capture_rules(self, delta_seconds: float = 1.0): """ Applique les règles d'arbitrage de capture : - Capture active : au moins 2 joueurs de la même équipe. - Contestation : dès qu'au moins 1 adversaire est présent, toute capture en cours est stoppée et redescend vers 'neutre'. - Durée de capture : 10 secondes pleines consécutives sans contestation. """ for z in self.zones.values(): b = z.blues_in_zone r = z.reds_in_zone # 1. Équipe Bleue en condition de capture (>= 2 Bleus et 0 Rouge) if b >= 2 and r == 0: if z.state == "bleu": z.capture_progress = z.capture_duration z.owner = "bleue" z.capturing_team = None elif z.state == "en_cours_capture_bleu": z.capture_progress = min(z.capture_duration, z.capture_progress + delta_seconds) if z.capture_progress >= z.capture_duration: z.state = "bleu" z.owner = "bleue" z.capturing_team = None else: # Était neutre, rouge ou en_cours_capture_rouge -> amorce capture Bleue z.state = "en_cours_capture_bleu" z.owner = None z.capturing_team = "bleue" z.capture_progress = min(z.capture_duration, delta_seconds) if z.capture_progress >= z.capture_duration: z.state = "bleu" z.owner = "bleue" z.capturing_team = None # 2. Équipe Rouge en condition de capture (>= 2 Rouges et 0 Bleu) elif r >= 2 and b == 0: if z.state == "rouge": z.capture_progress = z.capture_duration z.owner = "rouge" z.capturing_team = None elif z.state == "en_cours_capture_rouge": z.capture_progress = min(z.capture_duration, z.capture_progress + delta_seconds) if z.capture_progress >= z.capture_duration: z.state = "rouge" z.owner = "rouge" z.capturing_team = None else: # Était neutre, bleu ou en_cours_capture_bleu -> amorce capture Rouge z.state = "en_cours_capture_rouge" z.owner = None z.capturing_team = "rouge" z.capture_progress = min(z.capture_duration, delta_seconds) if z.capture_progress >= z.capture_duration: z.state = "rouge" z.owner = "rouge" z.capturing_team = None # 3. Contestation ou abandon de capture en cours : elif z.state in ["en_cours_capture_bleu", "en_cours_capture_rouge"]: # Dès qu'au moins 1 adversaire arrive, ou si l'équipe n'a plus 2 joueurs dans la zone, # la capture s'annule immédiatement et la zone redevient Neutre. z.state = "neutre" z.owner = None z.capturing_team = None z.capture_progress = 0.0 # 4. Maintien de possession existante (sans capture adverse active) elif z.state == "bleu": z.owner = "bleue" z.capturing_team = None z.capture_progress = z.capture_duration elif z.state == "rouge": z.owner = "rouge" z.capturing_team = None z.capture_progress = z.capture_duration else: z.state = "neutre" z.owner = None z.capturing_team = None z.capture_progress = 0.0 def tick(self, delta_seconds: float = 1.0): """Boucle de calcul exécutée chaque seconde : arbitrage, points et chrono.""" if self.status != "started": return self.last_activity = time.time() self.evaluate_presences() self.evaluate_capture_rules(delta_seconds) # Décompte des points en temps réel : # +1 point par seconde pour chaque zone activement possédée (validée Bleu ou Rouge) blue_zones = sum(1 for z in self.zones.values() if z.state == "bleu") red_zones = sum(1 for z in self.zones.values() if z.state == "rouge") self.score_bleu += int(round(blue_zones * delta_seconds)) self.score_rouge += int(round(red_zones * delta_seconds)) # Condition de Victoire 1 : Contrôle total simultané des 4 zones (Victoire immédiate) if blue_zones == 4: self.status = "finished" self.winner = "bleue" self.victory_reason = "total_control" return elif red_zones == 4: self.status = "finished" self.winner = "rouge" self.victory_reason = "total_control" return # Condition de Victoire 2 : Expiration du chronomètre self.timer_seconds_left = max(0, self.timer_seconds_left - int(round(delta_seconds))) if self.timer_seconds_left <= 0: self.status = "finished" self.victory_reason = "time_expired" if self.score_bleu > self.score_rouge: self.winner = "bleue" elif self.score_rouge > self.score_bleu: self.winner = "rouge" else: self.winner = "draw" def to_dict_for(self, player_id: Optional[str] = None) -> Dict[str, Any]: """Sérialise l'état complet du salon pour diffusion JSON.""" connected_count = sum(1 for p in self.players.values() if p.connected) blue_players_count = sum(1 for p in self.players.values() if p.team == "bleue") red_players_count = sum(1 for p in self.players.values() if p.team == "rouge") my_team = None is_host = False my_name = "Opérateur" if player_id and player_id in self.players: p = self.players[player_id] my_team = p.team is_host = p.is_host my_name = p.name blue_zones = sum(1 for z in self.zones.values() if z.state == "bleu") red_zones = sum(1 for z in self.zones.values() if z.state == "rouge") return { "type": "zone_capture_state", "room_code": self.code, "status": self.status, "is_host": is_host, "my_player_id": player_id, "my_name": my_name, "my_team": my_team, "connected_count": connected_count, "total_players": len(self.players), "team_counts": { "bleue": blue_players_count, "rouge": red_players_count, }, "timer": { "initial_seconds": self.initial_timer_seconds, "seconds_left": self.timer_seconds_left, }, "scores": { "bleue": self.score_bleu, "rouge": self.score_rouge, }, "controlled_zones": { "bleue": blue_zones, "rouge": red_zones, "total": len(self.zones), }, "winner": self.winner, "victory_reason": self.victory_reason, "players": [p.to_dict() for p in self.players.values()], "zones": [z.to_dict() for z in self.zones.values()], } class ZoneCaptureSyncManager: """Gestionnaire temps réel des salons de jeu du mode 'Capture de zones'.""" def __init__(self): # room_code -> Dict[WebSocket, str (player_id)] self.active_connections: Dict[str, Dict[WebSocket, str]] = {} # room_code -> ZoneCaptureRoom self.rooms: Dict[str, ZoneCaptureRoom] = {} # room_code -> asyncio.Task self.room_tasks: Dict[str, asyncio.Task] = {} def get_or_create_room(self, code: str) -> ZoneCaptureRoom: clean_code = code.upper().strip() if clean_code not in self.rooms: self.rooms[clean_code] = ZoneCaptureRoom(clean_code) return self.rooms[clean_code] def generate_room_code(self) -> str: """Génère un code de salle court sous la forme Z-XXXX.""" chars = string.ascii_uppercase + string.digits for _ in range(100): suffix = "".join(random.choices(chars, k=4)) code = f"Z-{suffix}" if code not in self.rooms: return code return f"Z-{int(time.time()) % 10000:04d}" async def connect(self, room_code: str, websocket: WebSocket, player_id: str, name: str): await websocket.accept() clean_code = room_code.upper().strip() if clean_code not in self.active_connections: self.active_connections[clean_code] = {} self.active_connections[clean_code][websocket] = player_id room = self.get_or_create_room(clean_code) room.get_or_create_player(player_id, name) await self.broadcast_room_state(clean_code) def disconnect(self, room_code: str, websocket: WebSocket): clean_code = room_code.upper().strip() if clean_code in self.active_connections: player_id = self.active_connections[clean_code].pop(websocket, None) if clean_code in self.rooms and player_id: if player_id in self.rooms[clean_code].players: self.rooms[clean_code].players[player_id].connected = False if not self.active_connections[clean_code]: del self.active_connections[clean_code] if clean_code in self.rooms: if all(not p.connected for p in self.rooms[clean_code].players.values()): self.stop_room_loop(clean_code) del self.rooms[clean_code] def get_connected_count(self, room_code: str) -> int: clean_code = room_code.upper().strip() return len(self.active_connections.get(clean_code, {})) async def broadcast_room_state(self, room_code: str): clean_code = room_code.upper().strip() if clean_code not in self.active_connections or clean_code not in self.rooms: return room = self.rooms[clean_code] dead_connections = [] for ws, pid in list(self.active_connections[clean_code].items()): try: state_data = room.to_dict_for(pid) await ws.send_json(state_data) except Exception: dead_connections.append(ws) for dead in dead_connections: self.disconnect(clean_code, dead) def ensure_room_loop(self, room_code: str): """S'assure que la boucle 1-seconde tick loop est active pour ce salon.""" clean_code = room_code.upper().strip() task = self.room_tasks.get(clean_code) if task is None or task.done(): self.room_tasks[clean_code] = asyncio.create_task(self._run_room_loop(clean_code)) def stop_room_loop(self, room_code: str): clean_code = room_code.upper().strip() task = self.room_tasks.pop(clean_code, None) if task and not task.done(): task.cancel() async def _run_room_loop(self, room_code: str): clean_code = room_code.upper().strip() try: while True: await asyncio.sleep(1) room = self.rooms.get(clean_code) if not room: break if room.status == "started": room.tick(1.0) await self.broadcast_room_state(clean_code) if room.status == "finished": # Partie terminée, conserver l'affichage final break except asyncio.CancelledError: pass except Exception: pass zone_capture_sync_manager = ZoneCaptureSyncManager()