dockered

django/games/objects/pong/Ball.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from ... import config
+
+from .Position import Position
+from .Point import Point
+
+import time
+import math
+class Ball:
+    
+    def __init__(self) -> None:
+        self.size: float
+        self.position: Position
+        self.angle: float
+        self.speed: float
+        
+        self.reset()
+        self.speed = 0
+    
+    def reset(self) -> None:
+        self.size = config.BALL_SIZE
+        self.position = Position(Point(config.BALL_SPAWN_POS_X + self.size / 2, config.BALL_SPAWN_POS_Y + self.size / 2), time.time())
+        self.angle = math.pi * 1
+        self.speed = config.BALL_SPEED_START
+    
+    def to_dict(self) -> dict:
+        
+        data: dict = {
+            "size": self.size,
+            "speed": self.speed,
+            "position": self.position.to_dict(),
+            "angle": self.angle,
+        }
+        
+        return data
+    
+    def __str__(self) -> str:
+        return f"Ball(size: {self.size}, speed: {self.speed}, angle: {self.angle}, position: {self.position})"

django/games/objects/pong/Point.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from math import dist
+class Point:
+    
+    def __init__(self, x: float, y: float) -> None:
+        self.x = x
+        self.y = y
+    
+    def __str__(self) -> str:
+        return f"Point(x: {self.x}, y: {self.y})"
+    
+    def __repr__(self) -> str:
+        return f"Point(x: {self.x}, y: {self.x})"
+    
+    def __eq__(self, __value: object) -> bool:
+        return (self.x == __value.x and self.y == __value.y)
+    
+    def distance(self, point: Point):
+        return dist((point.x, point.y), (self.x, self.y))
+    
+    def copy(self):
+        return Point(self.x, self.y)
+    
+    def to_dict(self) -> dict:
+        
+        data: dict[str: float] = {
+            "x": self.x,
+            "y": self.y,
+        }
+        
+        return data

django/games/objects/pong/PongGame.py

@@ -0,0 +1,195 @@
+from ..AGame import AGame
+from .PongPlayer import PongPlayer
+from .PongSpectator import PongSpectator
+from .Wall import Wall
+from .Segment import Segment
+from .Point import Point
+from .Ball import Ball
+
+from ... import config
+
+from channels.generic.websocket import WebsocketConsumer
+
+from django.contrib.auth.models import User
+
+from ...routine import routine
+
+import threading
+
+class PongGame(AGame):
+    
+    def __init__(self, game_id: int, game_manager):
+        super().__init__("pong", game_id, game_manager)
+        
+        self.players: list[PongPlayer]
+        self.walls: list[Wall]
+        
+        players: list[User] = self.model.get_players()
+        
+        nb_players: int = len(players)
+        if (nb_players == 2):
+            corners = [Point(50, config.MAP_CENTER_Y - config.MAP_SIZE_Y / 4),
+                       Point(config.MAP_SIZE_X - 50, config.MAP_CENTER_Y - config.MAP_SIZE_Y / 4),
+                       Point(config.MAP_SIZE_X - 50, config.MAP_CENTER_Y + config.MAP_SIZE_Y / 4),
+                       Point(50, config.MAP_CENTER_Y + config.MAP_SIZE_Y / 4)
+                      ]
+            
+            self.players = [PongPlayer(self, players[0], None, Segment(corners[1].copy(), corners[2].copy())),
+                            PongPlayer(self, players[1], None, Segment(corners[0].copy(), corners[3].copy()))]
+            
+            self.walls = [Wall(corners[0], corners[1]),
+                          Wall(corners[2], corners[3])]
+        else:
+            corners: list[Point] = [Point(50, 50),
+                                    Point(config.MAP_SIZE_X - 50, 50),
+                                    Point(config.MAP_SIZE_X - 50, config.MAP_CENTER_Y - 50),
+                                    Point(50, config.MAP_SIZE_Y - 50)]
+            
+            self.players = []
+            self.walls = []
+            
+            for i in range(4):
+                if i < nb_players:
+                    self.players.append(PongPlayer(self, players[i], None, Segment(corners[i], corners[(i + 1) % 4])))
+                else:
+                    self.walls.append(Segment(corners[i], corners[(i + 1) % 4]))
+    
+        self.ball: Ball = Ball()
+    
+    def goal(self, goal_defenser: PongPlayer) -> None:
+        
+        timestamp: int = goal_defenser.add_goal()
+        
+        self.broadcast("goal", {"player_id": goal_defenser.user.pk,
+                                "timestamp": timestamp})
+        
+        if len(goal_defenser.score) >= config.GAME_MAX_SCORE:
+            
+            self.eliminate(goal_defenser)
+            
+            player_list = self.get_valid_players()
+            
+            if len(player_list) == 1:
+                self.finish(player_list[0])
+                return
+        
+        self.ball.reset()
+        self.broadcast("update_ball", self.ball.to_dict())
+
+    def get_valid_players(self) -> list[PongPlayer]:
+        return [player for player in self.players if player.is_connected and not player.is_eliminated] 
+            
+    def finish(self, winner: PongPlayer) -> bool:
+        self.broadcast("finish", {'winner_id': winner.user.pk})
+        self.model.finish(winner.user)
+        self.stopped = True
+    
+    def start(self):
+        
+        # Set to true to stop the thread routine
+        self.stopped: bool = False
+        
+        self.model.start()
+        
+        self.broadcast("start")
+
+        self.ball.reset()
+        
+        self.broadcast("update_ball", self.ball.to_dict())
+        
+        self.thread = threading.Thread(target=routine, args=(self,))
+        
+        self.thread.start()
+    
+    def eliminate(self, eliminated: PongPlayer):
+        
+        self.broadcast("eliminated", {"eliminated_id": eliminated.user.pk})
+        
+        eliminated.eliminate()
+        
+    def _player_join(self, user: User, socket: WebsocketConsumer) -> PongPlayer | None:
+        
+        if (self.model.started):
+            return None
+        
+        player = self.get_player_by_user_id(user.pk)
+        if (player is None):
+            return None
+
+        # check if player is already connected
+        if (player.is_connected()):
+            player.disconnect(1001)
+            
+        player.socket = socket
+        
+        self.update_player(player)
+        
+        if len(self.players) == len(self.get_players_connected()):
+            self.start()
+            
+        return player
+
+    def _spectator_join(self, user: User, socket: WebsocketConsumer) -> PongSpectator:
+        
+        spectator: PongSpectator = PongSpectator(user, socket, self)
+        
+        self.spectators.append(spectator)
+        
+        return spectator
+    
+    def join(self, user: User, socket: WebsocketConsumer) -> PongSpectator | PongPlayer:
+        
+        member: PongPlayer | PongSpectator
+        
+        member = self._player_join(user, socket)
+        
+        if member is None:
+            member = self._spectator_join(user, socket)
+        
+        self._send_game_data(member)
+        
+        return member
+        
+    def _player_leave(self, player: PongPlayer):
+        
+        if self.model.started:
+            
+            self.eliminate(player)
+            
+            players: list[PongPlayer] = self.get_valid_players()
+            
+            if len(players) == 1:
+                self.finish(players[0])
+    
+    def _spectator_leave(self, spectator: PongSpectator):
+        self.spectators.remove(spectator)
+        
+    def leave(self, member: PongSpectator | PongPlayer):
+        
+        if (isinstance(member, PongPlayer)):
+            self._player_leave(member)
+        elif (isinstance(member, PongSpectator)):
+            self._spectator_leave(member)
+        
+        if self.model.started:
+            if len(self.get_players_connected()) + len(self.spectators) == 0:
+                self.stopped = True
+                if hasattr(self, 'thread'):
+                    self.thread.join(10)
+                self.game_manager.remove(self)
+    
+    def _send_game_data(self, member: PongSpectator | PongPlayer):
+    
+        member.send("init_game", self.to_dict())
+        
+    def update_player(self, player: PongPlayer):
+        self.broadcast("update_player", player.to_dict(), [player])
+
+    def to_dict(self): 
+        
+        data: dict = {"ball": self.ball.to_dict(),
+                      "players": [player.to_dict() for player in self.players],
+                      "walls": [wall.to_dict() for wall in self.walls],
+                      }
+        
+        return data

django/games/objects/pong/PongPlayer.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from ... import config
+
+from channels.generic.websocket import WebsocketConsumer
+
+from django.contrib.auth.models import User
+
+from .Position import Position
+from ..APlayer import APlayer
+from .Segment import Segment
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .PongGame import PongGame
+
+class PongPlayer(APlayer):
+    
+    def __init__(self, game: PongGame, user: User, socket: WebsocketConsumer, rail: Segment) -> None:
+        
+        super().__init__(user, socket, game)
+        
+        self.position: Position = Position(0.5, 0)
+        
+        self.score: list[int] = []
+        
+        self.rail: Segment = rail
+
+        self.is_eliminated: bool = False
+
+        self.game: PongGame
+        
+    def eliminate(self):
+        
+        self.is_eliminated = True
+
+    def receive(self, data: dict):
+        
+        detail: str = data.get("detail")
+
+        if (detail is None):
+            return
+        
+        if (detail == "update_my_paddle_pos"):
+            self.update_position(data)
+    
+    def update_position(self, data: dict):
+        
+        new_position: Position = Position(None)
+        
+        position_dict = data.get("position")
+        if (position_dict is None):
+            self.send_error("missing position")
+            return
+        
+        new_position.location = position_dict.get("location")
+        if (new_position.location is None):
+            self.send_error("missing location")
+            return
+        
+        new_position.time = data.get("time")
+        
+        if (new_position.time is None):
+            self.send_error("missing time")
+            return
+        
+        if (self.position.time > new_position.time):
+            self.send_error("time error")
+            return
+        
+        distance: float = abs(self.position.location - new_position.location) 
+        
+        sign: int = 1 if self.position.location >= new_position.location else -1
+        
+        time_difference: float = (new_position.time - self.position.time) / 1000
+        
+        max_distance: float = config.PADDLE_SPEED_PER_SECOND_MAX * (time_difference) * config.PADDLE_SPEED_PER_SECOND_TOLERANCE
+        
+        new_position_verified: Position = new_position.copy()
+
+        if (distance > max_distance):
+            new_position_verified.location = self.position.location + max_distance * sign
+        
+        if (not config.PADDLE_POSITION_MIN <= new_position_verified.location <= config.PADDLE_POSITION_MAX):
+            
+            new_position_verified.location = max(new_position_verified.location, config.PADDLE_POSITION_MIN)
+            new_position_verified.location = min(new_position_verified.location, config.PADDLE_POSITION_MAX)
+
+        invalid_pos: bool = new_position.location != new_position_verified.location
+        
+        if (new_position.location != self.position.location):
+            self.game.update_player(self)
+
+        self.position.location = new_position.location
+        
+        if (invalid_pos):
+            self.send("update_player", self.to_dict())
+    
+    def connect(self, socket: WebsocketConsumer):
+        self.socket = socket
+        self.accept()
+        self.game.update_player(self)
+    
+    def disconnect(self, code: int = 1000):
+        self.socket = None
+        self.game.leave(self)
+    
+    def add_goal(self):
+        
+        timestamp = self.game.model.add_goal(self.user)
+        
+        self.score.append(timestamp)
+        
+        return timestamp
+    
+    def to_dict(self) -> dict:
+        
+        data = {
+            "username": self.user.username,
+            "id": self.user.pk,
+            "position": self.position.to_dict(),
+            "score": self.score,
+            
+            "isEliminated": self.is_eliminated,
+            
+            "rail": self.rail.to_dict(),
+            
+            "isConnected": self.is_connected(),
+        }
+        
+        return data

django/games/objects/pong/PongSpectator.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from channels.generic.websocket import WebsocketConsumer
+
+from django.contrib.auth.models import User
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .PongPlayer import PongPlayer
+    from .PongGame import PongGame
+
+from ..ASpectator import ASpectator
+from .Ball import Ball
+
+class PongSpectator(ASpectator):
+
+    def __init__(self, user: User, socket: WebsocketConsumer, game: PongGame):
+        super().__init__(user, socket, game)
+        self.game: PongGame = game
+
+    def send_paddle(self, player: PongPlayer):
+        self.send("update_player", player.to_dict())    
+
+    def send_ball(self, ball: Ball):
+        self.send("update_ball", ball.to_dict())
+    
+    def disconnect(self, code: int = 1000):
+        self.game.leave(self)

django/games/objects/pong/Position.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from .Point import Point
+class Position:
+    
+    def __init__(self, location: int | Point = 0, time: int = 0) -> None:
+        self.time: float = time
+        self.location: float | Point = location
+    
+    def copy(self):
+        try:
+            return Position(self.location.copy(), self.time)
+        except:
+            return Position(self.location, self.time)
+    
+    def to_dict(self):
+        
+        data: dict = {
+            "time": self.time,
+        }
+        
+        try:
+            data.update({"location": self.location.to_dict()})
+        except:
+            data.update({"location": self.location})
+                
+        return data
+
+    def __eq__(self, __value: Position) -> bool:
+        return (self.location == __value.location)

django/games/objects/pong/Segment.py

@@ -0,0 +1,38 @@
+
+from .Point import Point
+from .Vector import Vector
+
+import math
+
+class Segment:
+    
+    def __init__(self, start: Point, stop: Point) -> None:
+        self.start: Point = start
+        self.stop: Point = stop
+
+    def angle(self) -> float:
+        return math.atan2((self.start.y - self.stop.y), (self.start.x - self.stop.x))
+        
+    def length(self):
+        return self.start.distance(self.stop)
+    
+    def is_on(self, C: Point):
+        return (self.start.distance(C) + self.stop.distance(C) == self.length())
+    
+    def __repr__(self) -> str:
+        return f"Segment(start: {self.start}, stop: {self.stop})"
+    
+    def __str__(self) -> str:
+        return f"Segment(start: {self.start}, stop: {self.stop})"
+    
+    def copy(self):
+        return Segment(self.start.copy(), self.stop.copy())
+    
+    def to_dict(self) -> dict:
+        
+        data: dict[str: dict] = {
+            "start": self.start.to_dict(),
+            "stop": self.stop.to_dict(),
+        }
+        
+        return data

django/games/objects/pong/Vector.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import math
+from .Point import Point
+
+class Vector:
+
+    def __init__(self, x: float, y: float) -> None:
+        self.norm: float = math.dist((0, 0), (x, y))
+        self.x: float = x
+        self.y: float = y
+
+    def __truediv__(self, denominator: float):
+        return Vector(self.x / denominator, self.y / denominator)
+    
+    def angle(self, vector: Vector):
+        scalar_product: float = self.scalar(vector)
+        if (scalar_product is None):
+            return None
+        cos: float = scalar_product / (vector.norm * self.norm)
+        
+        angle: float = math.acos(cos)
+        
+        return angle
+
+    def scalar(self, vector: Vector):
+        return self.x * vector.x + vector.y * self.y
+    
+    def __str__(self) -> str:
+        return f"Vector(x: {self.x}, y: {self.y}, norme: {self.norm})"
+    
+    def __eq__(self, __value: Vector) -> bool:
+        return (self.x == __value.x and
+                self.x == __value.x and 
+                self.norm == __value.norm)

django/games/objects/pong/Wall.py

@@ -0,0 +1,5 @@
+
+
+from .Segment import Segment
+class Wall(Segment):
+    pass