from __future__ import annotations from channels.generic.websocket import WebsocketConsumer from django.contrib.auth.models import User from django.db.models import QuerySet from django.utils.translation import gettext as _ from profiles.models import ProfileModel from profiles.serializers import ProfileSerializer from .models import TournamentModel from .models import TournamentGameModel from .serializers import TournamentGameSerializer import json class TournamentMember: def __init__(self, socket: TournamentWebConsumer, room: TournamentRoom, is_participating: bool = False) -> None: self._socket: TournamentWebConsumer = socket self._room: TournamentRoom = room self.is_participating: bool = is_participating def send(self, detail: str, data: dict = {}): data_to_send: dict = {"detail": detail} data_to_send.update(data) self._socket.send(json.dumps(data_to_send)) def send_error(self, error_message: str, data: dict = {}): data_to_send: dict = {"error_message": error_message} data_to_send.update(data) self.send("error", data_to_send) def send_goto(self, game: TournamentGameModel): self.send("go_to", {"game_id": game.pk}) def _receive_participating(self, data: dict) -> None: is_participating: bool | None = data.get("is_participating") if (is_participating is None): self.send_error(_("Missing is_participating statement.")) return self._room.set_participation(self, is_participating) def receive(self, data: dict): detail: str | None = data.get("detail") if (detail is None): return match(detail): case "update_participating": self._receive_participating(data) case _: print("bozo_send") class TournamentRoomManager: def __init__(self): self._room_list: list[TournamentRoom] = [] def get(self, tournament: TournamentModel) -> TournamentRoom: for room in self._room_list: if room._model is tournament: return room room: TournamentRoom = TournamentRoom(self, tournament) self._room_list.append(room) return room def remove(self, room: TournamentRoom) -> None: self._room_list.remove(room) class TournamentRoom: def __init__(self, room_manager: TournamentRoomManager, tournament: TournamentModel): self._room_manager: TournamentRoomManager = room_manager self._member_list: set[TournamentMember] = set() self._model: TournamentModel = tournament self._game_in_progress_list: set[TournamentGameModel] = set() self._current_round = 0 def join(self, socket: TournamentWebConsumer) -> TournamentMember: member: TournamentMember = TournamentMember(socket, self) self._member_list.add(member) return member def set_game_as_finished(self, game: TournamentGameModel): self._game_in_progress_list.remove(game) self.broadcast("game_update", TournamentGameSerializer(game).data) if len(self._game_in_progress_list) == 0: self._round_finished() def set_game_as_started(self, game: TournamentGameModel): self._game_in_progress_list.add(game) self.broadcast("game_update", TournamentGameSerializer(game).data) def _finish(self, winner: User): self._model.finish(winner) def _round_finished(self): if self._current_round == self._model.round: last_game: TournamentGameSerializer = self._model.get_games_by_round(self._current_round)[0] self._finish(last_game.winner) return self._start_round() def _start_round(self): self._current_round += 1 participant_list: set[User] = self._model.get_participants_by_round(self._current_round) game_list = self._model.create_round(participant_list, self._current_round) for game in game_list: for player in game.get_players(): participant: TournamentMember = self.get_participant_by_profile(player) participant.send_goto(game) def get_participants_profiles(self) -> list[ProfileModel]: return [participant._socket.user.profilemodel for participant in self.get_participants()] def start(self) -> None: self._model.start() self.broadcast("start") self._start_round() def get_participant_by_profile(self, profile: ProfileModel): for participant in self.get_participants(): if (participant._socket.user.profilemodel == profile): return participant def leave(self, member: TournamentMember) -> None: # Delete room if nobody connected, no cringe memory leak if (len(self._member_list) == 1): self._room_manager.remove(self) return self._member_list.remove(member) self.set_participation(member, False) def everybody_is_here(self): return len(self.get_participants()) == self._model.nb_participants def broadcast(self, detail: str, data: dict, excludes: set[TournamentMember] = set()) -> None: member_list: list[TournamentMember] = self._member_list - excludes for member in member_list: member.send(detail, data) def get_participants(self) -> list[TournamentMember]: return [member for member in self._member_list if member.is_participating] def set_participation(self, member: TournamentMember, is_participating: bool) -> None: if (self._model.started): return if (member.is_participating == is_participating): return if (is_participating == True): self.broadcast("add_participant", {"participant": ProfileSerializer(member._socket.user.profilemodel).data}) else: self.broadcast("del_participant", {"participant": ProfileSerializer(member._socket.user.profilemodel).data}) member.is_participating = is_participating if self.everybody_is_here(): self.start() tournament_manager: TournamentRoomManager = TournamentRoomManager() class TournamentWebConsumer(WebsocketConsumer): def connect(self): self.user: User = self.scope["user"] if (not self.user.is_authenticated): return tournament_id: int = int(self.scope['url_route']['kwargs']['tournament_id']) query: QuerySet = TournamentModel.objects.filter(pk=tournament_id) if (not query.exists()): return tournament: TournamentModel = query[0] self.room = tournament_manager.get(tournament) self.member: TournamentMember = self.room.join(self) self.accept() def receive(self, text_data: str = None, bytes_data: bytes = None): user: User = self.scope["user"] if (user != self.user): return data: dict = json.loads(text_data) self.member.receive(data) def disconnect(self, close_code): self.room.leave(self.member) super().disconnect(close_code) # proutman à encore frappé