121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from channels.generic.websocket import WebsocketConsumer
|
|
|
|
from django.contrib.auth.models import User
|
|
|
|
import json
|
|
|
|
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})
|
|
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.model.finish(self.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
|
|
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) |