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 .Spectator import Spectator from .Point import Point from .Segment import Segment from typing import TYPE_CHECKING if TYPE_CHECKING: from ...transcendence.abstract.AbstractRoomMember import AbstractRoomMember from .Game import Game class Player(Spectator): def __init__(self, game: Game, user_id: int, socket: WebsocketConsumer, rail: Segment) -> None: super().__init__(user_id, socket, game) self.position: Position = Position(0.5, 0) self.score: list[int] = [] self.rail: Segment = rail self.username: str = User.objects.get(pk = self.user_id).username def eliminate(self): self.disconnect(1000) self.game.update_player(self) 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 send_error(self, error_message: str, error_data = {}): data: dict = { "error_message": error_message } data.update(error_data) self.send("error", data) def update_position(self, data: dict): new_position: Position = Position() new_position.position: float = data.get("position") if (new_position.position is None): self.send_error("missing new_position") return new_position.time: float = data.get("time") if (new_position.time is None): self.game_member.send_error("missing time") return if (self.position.time > new_position.time): self.game_member.send_error("time error") return distance: float = abs(self.position.position - new_position.position) sign: int = 1 if self.position.position >= new_position.position 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.position = self.position.position + max_distance * sign if (not config.PADDLE_POSITION_MIN <= new_position_verified.position <= config.PADDLE_POSITION_MAX): new_position_verified.position = max(new_position_verified.position, config.PADDLE_POSITION_MIN) new_position_verified.position = min(new_position_verified.position, config.PADDLE_POSITION_MAX) invalid_pos: bool = new_position.position != new_position_verified.position if (new_position != self.position): self.game.update_player(self) self.position = new_position 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 is_connected(self): return self.socket != None 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_id) self.score.append(timestamp) return timestamp def to_dict(self) -> dict: data = { "username": self.username, "id": self.user_id, "position": self.position.to_dict(), "score": [*self.score], "rail": self.rail.to_dict(), "isConnected": self.is_connected(), } return data