from __future__ import annotations from channels.generic.websocket import WebsocketConsumer from django.contrib.auth.models import User import json import sys from .objects.GameManager import GameManager from .objects.tictactoe.TicTacToePlayer import TicTacToePlayer import time from .objects.pong.PongPlayer import PongPlayer from typing import TYPE_CHECKING if TYPE_CHECKING: from .objects.pong.PongSpectator import PongSpectator from .objects.pong.PongGame import PongGame from .objects.tictactoe.TicTacToeGame import TicTacToeGame game_manager: GameManager = GameManager() class TicTacToeWebSocket(WebsocketConsumer): def connect(self): self.user: User = self.scope["user"] if (self.user.pk is None): self.user.pk = 0 self.accept() self.game_id = int(self.scope['url_route']['kwargs']['game_id']) self.game: TicTacToeGame = game_manager.get(self.game_id, "tictactoe") if (self.game is None): return self.member = self.game.join(self.user.pk, self) if (isinstance(self.member, TicTacToePlayer)): # Send player sign (x or o) to the player self.member.send(self.member.sign) if (self.game._everbody_is_here() and self.game.model.started == True and self.game.model.finished == False): # if an user disconnected and joined back self.member.send("catchup", {"morpion" : self.game._map, "turn" : self.game.turn, "currentMorpion": self.member.currentMorpion, "timer" : 20 - (time.time() - self.game.time), "finished" : self.game.model.finished}) return if (self.game._everbody_is_here() and self.game.model.started == False): # Launch the game if the two players are connected self.game.broadcast("game_start") self.game.time = time.time() self.game.model.start() def receive(self, text_data=None, bytes_data=None): data = json.loads(text_data) if (data.get("targetMorpion") is not None and data.get("targetCase") is not None): # A move has been played if (self.game.add(data, self.member) == False): # If the move is invalid return self.game.broadcast("game_move", data, [self.member]) if (data.get("timerIsDue") is not None and self.game.time + 20 < time.time()): # Frontend asking if the timer is due self.winner = "x" if self.game.turn == "o" else "o" self.game.get_members() self.game.model.finish(self.game.get_players_by_sign(self.winner).user) self.game.broadcast("game_end", {"winning_sign": self.winner}) if (self.game.checkWin() != False): # Check if after a move, the game is finished self.winner = self.game.checkWin() self.game.model.finish(self.user) self.game.broadcast("game_end", {"winning_sign": self.winner}) def disconnect(self, event): try: self.member.socket = None if (len(self.game.get_players_connected()) == 0): self.game.model.finish(None) except: pass class PongWebSocket(WebsocketConsumer): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.member = None def connect(self): self.user: User = self.scope["user"] if (self.user.pk is None): self.user.pk = 0 self.accept() self.game_id = int(self.scope['url_route']['kwargs']['game_id']) self.game: PongGame = game_manager.get(self.game_id, "pong") if (self.game is None): self.send(text_data=json.dumps({"detail": "Game not found"})) self.disconnect(1404) return self.member: PongPlayer | PongSpectator = self.game.join(self.user, self) def disconnect(self, code): if (self.member is not None): self.member.disconnect() super().disconnect(code) def receive(self, text_data: str = None, bytes_data: bytes = None): if (text_data is None): return data: dict = json.loads(text_data) if (isinstance(self.member, PongPlayer)): self.member.receive(data)