""" 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 # Chaque téléphone possède sa propre base. Les scores restent communs au salon. self.ecapture_bases: Dict[str, Dict[str, Any]] = {} self.ecapture_hold_duration: int = 10 self.ecapture_pts_per_second: int = 10 self.ecapture_teams_count: int = 2 self.ecapture_score_alpha: int = 0 self.ecapture_score_bravo: int = 0 self.ecapture_score_charlie: int = 0 self.ecapture_score_delta: int = 0 self.ecapture_target_score: Optional[int] = None # Onglet actif suggéré self.active_tab = "timer" def get_ecapture_base(self, client_id: str) -> Dict[str, Any]: """Récupère l'état de capture propre à un téléphone.""" return self.ecapture_bases.setdefault(client_id, { "owner": None, "running": False, "control_seconds": 0, "holding_team": None, }) 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, client_id: str = "default") -> Dict[str, Any]: """Sérialise l'état du salon pour diffusion JSON aux clients.""" current_time = self.get_current_timer_seconds() base = self.get_ecapture_base(client_id) 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, }, "ecapture": { "owner": base["owner"], "running": base["running"], "hold_duration": self.ecapture_hold_duration, "pts_per_second": self.ecapture_pts_per_second, "teams_count": self.ecapture_teams_count, "score_alpha": self.ecapture_score_alpha, "score_bravo": self.ecapture_score_bravo, "score_charlie": self.ecapture_score_charlie, "score_delta": self.ecapture_score_delta, "control_seconds": base["control_seconds"], "holding_team": base["holding_team"], "target_score": self.ecapture_target_score, } } 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]] = {} self.connection_client_ids: Dict[WebSocket, str] = {} # Code de salon -> objet ToolsRoom self.rooms: Dict[str, ToolsRoom] = {} # Code de salon -> tâche asyncio de boucle d'incrémentation temps réel self.room_tasks: Dict[str, asyncio.Task] = {} 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, client_id: str = "default"): """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.connection_client_ids[websocket] = client_id 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) self.connection_client_ids.pop(websocket, None) if not self.active_connections[clean_code]: del self.active_connections[clean_code] self.stop_room_loop(clean_code) if clean_code in self.rooms: del self.rooms[clean_code] def ensure_room_loop(self, room_code: str): """Démarre la boucle 1s pour l'e-capture si elle n'est pas déjà active.""" 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): """Arrête la tâche de fond associée au salon.""" 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): """Boucle d'incrémentation automatique des points de capture (+10 pts/sec).""" clean_code = room_code.upper().strip() try: while True: await asyncio.sleep(1) room = self.rooms.get(clean_code) if not room: break needs_broadcast = False for base in room.ecapture_bases.values(): if not base["running"] or not base["owner"]: continue pts = room.ecapture_pts_per_second score_attr = f"ecapture_score_{base['owner']}" setattr(room, score_attr, getattr(room, score_attr) + pts) base["control_seconds"] += 1 needs_broadcast = True if room.timer_running and room.get_current_timer_seconds() <= 0: room.timer_running = False room.timer_seconds_left = 0 needs_broadcast = True if needs_broadcast: await self.broadcast_room_state(clean_code) if not any(base["running"] and base["owner"] for base in room.ecapture_bases.values()) and not room.timer_running: break except asyncio.CancelledError: pass 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) for connection in list(self.active_connections[clean_code]): try: client_id = self.connection_client_ids.get(connection, "default") await connection.send_json(room.to_dict(count, client_id)) except Exception: self.disconnect(clean_code, connection) # Instance singleton globale pour les outils tools_sync_manager = ToolsSyncManager() # ============================================================================== # OUTIL DE RÉPARTITION ALÉATOIRE DES ÉQUIPES (SALONS TEMPS RÉEL) # ============================================================================== TEAM_DRAW_TEAMS = [ {"id": "alpha", "name": "Équipe Alpha", "color": "bleue", "badge": "BLEU"}, {"id": "bravo", "name": "Équipe Bravo", "color": "rouge", "badge": "ROUGE"}, {"id": "charlie", "name": "Équipe Charlie", "color": "jaune", "badge": "JAUNE"}, {"id": "delta", "name": "Équipe Delta", "color": "verte", "badge": "VERT"}, ] class TeamDrawPlayer: """Joueur participant au tirage aléatoire d'équipes.""" 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 self.connected = True self.joined_at = time.time() def to_dict(self, reveal_team: bool = False) -> Dict[str, Any]: return { "id": self.player_id, "name": self.name, "is_host": self.is_host, "team": self.team if reveal_team else None, "connected": self.connected, "joined_at": self.joined_at, } class TeamDrawRoom: """Salon temporaire de répartition équilibrée en deux à quatre équipes.""" def __init__(self, code: str): self.code = code self.created_at = time.time() self.last_activity = time.time() self.status = "lobby" self.host_player_id: Optional[str] = None self.round_number = 1 self.teams_count = 2 self.countdown_end_time: Optional[float] = None self.started_at: Optional[float] = None self.players: Dict[str, TeamDrawPlayer] = {} def get_or_create_player(self, player_id: str, name: str) -> TeamDrawPlayer: self.last_activity = time.time() if not self.host_player_id: self.host_player_id = player_id if player_id in self.players: player = self.players[player_id] if name and name.strip(): player.name = name.strip() player.connected = True player.is_host = player_id == self.host_player_id return player clean_name = name.strip() if name and name.strip() else f"Opérateur #{len(self.players) + 1}" player = TeamDrawPlayer(player_id, clean_name, is_host=player_id == self.host_player_id) self.players[player_id] = player if self.status in ("countdown", "started"): self._assign_late_player(player) return player def _active_teams(self) -> list: return TEAM_DRAW_TEAMS[:self.teams_count] def _assign_late_player(self, player: TeamDrawPlayer) -> None: counts = {team["id"]: 0 for team in self._active_teams()} for existing in self.players.values(): if existing is not player and existing.team in counts: counts[existing.team] += 1 least_populated = min(counts.values()) choices = [team_id for team_id, count in counts.items() if count == least_populated] player.team = random.choice(choices) def set_teams_count(self, count: int) -> bool: if self.status != "lobby" or count not in (2, 3, 4): return False self.teams_count = count self.last_activity = time.time() return True def start_game(self) -> None: self.last_activity = time.time() player_ids = list(self.players.keys()) random.shuffle(player_ids) team_ids = [team["id"] for team in self._active_teams()] random.shuffle(team_ids) for index, player_id in enumerate(player_ids): self.players[player_id].team = team_ids[index % self.teams_count] self.started_at = time.time() self.status = "countdown" self.countdown_end_time = time.time() + 5 def trigger_reveal(self) -> None: self.status = "started" self.countdown_end_time = None self.last_activity = time.time() def reset_round(self) -> None: self.last_activity = time.time() self.status = "lobby" self.countdown_end_time = None self.started_at = None self.round_number += 1 for player in self.players.values(): player.team = None def to_dict_for(self, player_id: Optional[str] = None) -> Dict[str, Any]: player = self.players.get(player_id) if player_id else None is_host = bool(player and player.is_host) counts = {team["id"]: 0 for team in self._active_teams()} for entry in self.players.values(): if entry.team in counts: counts[entry.team] += 1 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 = max(0, int(time.time() - self.started_at)) if self.status == "started" and self.started_at else 0 return { "type": "team_draw_state", "room_code": self.code, "status": self.status, "round_number": self.round_number, "teams_count": self.teams_count, "teams": self._active_teams(), "connected_count": sum(1 for entry in self.players.values() if entry.connected), "total_players": len(self.players), "is_host": is_host, "my_player_id": player_id, "my_name": player.name if player else "Opérateur", "my_team": player.team if player and self.status == "started" else None, "team_counts": counts, "countdown_seconds": countdown_left, "elapsed_seconds": elapsed, "players": [entry.to_dict(reveal_team=is_host) for entry in self.players.values()], } class TeamDrawSyncManager: """Synchronise les salons de tirage d'équipes entre téléphones.""" def __init__(self): self.active_connections: Dict[str, Dict[WebSocket, str]] = {} self.rooms: Dict[str, TeamDrawRoom] = {} def get_or_create_room(self, code: str) -> TeamDrawRoom: clean_code = code.upper().strip() if clean_code not in self.rooms: self.rooms[clean_code] = TeamDrawRoom(clean_code) return self.rooms[clean_code] def generate_room_code(self) -> str: chars = string.ascii_uppercase + string.digits for _ in range(100): code = "E-" + "".join(random.choices(chars, k=4)) if code not in self.rooms: return code return f"E-{int(time.time()) % 10000:04d}" def disconnect(self, room_code: str, websocket: WebSocket) -> None: clean_code = room_code.upper().strip() connections = self.active_connections.get(clean_code) if not connections: return player_id = connections.pop(websocket, None) if player_id and clean_code in self.rooms and player_id in self.rooms[clean_code].players: self.rooms[clean_code].players[player_id].connected = False if not connections: self.active_connections.pop(clean_code, None) room = self.rooms.get(clean_code) if room and all(not player.connected for player in room.players.values()): self.rooms.pop(clean_code, None) async def broadcast_room_state(self, room_code: str) -> None: 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 = [] for websocket, player_id in list(self.active_connections[clean_code].items()): try: await websocket.send_json(room.to_dict_for(player_id)) except Exception: dead.append(websocket) for websocket in dead: self.disconnect(clean_code, websocket) random_teams_sync_manager = TeamDrawSyncManager()