292 lines
8.7 KiB
Python
292 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from .objects.Spectator import Spectator
|
|
from .objects.Player import Player
|
|
from .objects.Game import Game
|
|
from .objects.Ball import Ball
|
|
|
|
from .objects.Point import Point
|
|
from .objects.Vector import Point
|
|
from .objects.Segment import Segment
|
|
from .objects.Vector import Vector
|
|
|
|
from . import config
|
|
|
|
from . import config
|
|
|
|
import math
|
|
import asyncio
|
|
|
|
from asgiref.sync import SyncToAsync
|
|
|
|
from time import sleep
|
|
|
|
VERTICALLY = 1
|
|
NORMAL = 2
|
|
|
|
def get_sign(num: float) -> int:
|
|
if (num == 0):
|
|
return 0
|
|
if (num > 0):
|
|
return 1
|
|
if (num < 0):
|
|
return -1
|
|
|
|
def get_derive(segment: Segment) -> float:
|
|
|
|
if (segment.start.x == segment.stop.x):
|
|
return None
|
|
|
|
return (segment.stop.y - segment.start.y) / (segment.stop.x - segment.start.x)
|
|
|
|
def get_intercept(derive: float, point: Point) -> float:
|
|
|
|
if (derive is None):
|
|
return None
|
|
|
|
return point.y - (point.x * derive)
|
|
|
|
def get_constant(segment: Segment) -> float:
|
|
|
|
return segment.start.x
|
|
|
|
def identify(segment: Segment) -> str:
|
|
|
|
if (segment.start.x == segment.stop.x):
|
|
return VERTICALLY
|
|
return NORMAL
|
|
|
|
def get_interception(segment1: Segment, segment2: Segment):
|
|
|
|
if (identify(segment1) == VERTICALLY and identify(segment2) == VERTICALLY):
|
|
return None
|
|
|
|
# because of in matematics world y = 10 is above y = 5 and on a display it is inverted I invert the coordonate
|
|
|
|
inverted_segment1 = Segment(Point(segment1.start.x, config.MAP_SIZE_Y - segment1.start.y), Point(segment1.stop.x, config.MAP_SIZE_Y - segment1.stop.y))
|
|
inverted_segment2 = Segment(Point(segment2.start.x, config.MAP_SIZE_Y - segment2.start.y), Point(segment2.stop.x, config.MAP_SIZE_Y - segment2.stop.y))
|
|
|
|
if (identify(segment1) == NORMAL and identify(segment2) == NORMAL):
|
|
|
|
# representation m * x + p
|
|
|
|
m1 = get_derive(inverted_segment1)
|
|
m2 = get_derive(inverted_segment2)
|
|
|
|
p1 = get_intercept(m1, inverted_segment1.start)
|
|
p2 = get_intercept(m2, inverted_segment2.start)
|
|
|
|
# m1 * x + p1 = m2 * x + p2
|
|
# m1 * x = m2 * x + p2 -p1
|
|
# m1 * x - m2 * x = p1 - p2
|
|
# x * (m1 - m2) = p1 - p2
|
|
# x = (p1 - p2) / (m1 - m2)
|
|
if (m1 == m2):
|
|
return None
|
|
|
|
# reinvert
|
|
x: float = (p1 - p2) / (m1 - m2) * (-1)
|
|
|
|
y: float = config.MAP_SIZE_Y - (m1 * x + p1)
|
|
|
|
else:
|
|
|
|
if (identify(inverted_segment1) == VERTICALLY):
|
|
constant: float = get_constant(inverted_segment1)
|
|
m: float = get_derive(inverted_segment2)
|
|
p: float = get_intercept(m, inverted_segment2.start)
|
|
else:
|
|
constant: float = get_constant(inverted_segment2)
|
|
m: float = get_derive(inverted_segment1)
|
|
p: float = get_intercept(m, inverted_segment1.start)
|
|
|
|
x: float = constant
|
|
y: float = config.MAP_SIZE_Y - (m * x + p)
|
|
|
|
impact: Point = Point(x, y)
|
|
|
|
return impact
|
|
|
|
def get_impact_data(segments: list[Segment], ball: Ball) -> dict:
|
|
|
|
cos: float = round(math.cos(ball.angle), 6)
|
|
sin: float = round(math.sin(ball.angle), 6)
|
|
|
|
inc_x: float = (-1) * get_sign(cos) * (config.STROKE_THICKNESS + config.BALL_SIZE / 2)
|
|
inc_y: float = get_sign(sin) * (config.STROKE_THICKNESS + config.BALL_SIZE / 2)
|
|
|
|
point: Point = Point(ball.position.x + cos, ball.position.y - sin)
|
|
|
|
ball_segment = Segment(ball.position, point)
|
|
|
|
closest: dict = None
|
|
|
|
for segment in segments:
|
|
|
|
segment_with_padding = segment.copy()
|
|
|
|
segment_with_padding.start.x += inc_x
|
|
segment_with_padding.stop.x += inc_x
|
|
|
|
segment_with_padding.start.y += inc_y
|
|
segment_with_padding.stop.y += inc_y
|
|
|
|
impact: Point = get_interception(segment_with_padding, ball_segment)
|
|
|
|
if (impact is None):
|
|
continue
|
|
|
|
diff_x: float = ball.position.x - impact.x
|
|
if (get_sign(diff_x) == get_sign(cos) and cos != 0):
|
|
continue
|
|
|
|
diff_y: float = (ball.position.y - impact.y)
|
|
if (get_sign(diff_y) != get_sign(sin) and sin != 0):
|
|
continue
|
|
|
|
if (closest is None or impact.distance(ball.position) < closest.get("distance")):
|
|
closest = {
|
|
"inc_x": inc_x,
|
|
"inc_y": inc_y,
|
|
"impact": impact,
|
|
"segment": segment,
|
|
"distance": impact.distance(ball.position),
|
|
}
|
|
|
|
return closest
|
|
|
|
def wall_collision(ball_angle: float, wall_angle: float) -> float:
|
|
|
|
ball_cos: float = math.cos(ball_angle)
|
|
ball_sin: float = math.sin(ball_angle)
|
|
|
|
incident_angle: float = ball_angle - wall_angle
|
|
|
|
reflection_angle: float = wall_angle - incident_angle
|
|
|
|
new_cos: float = math.cos(reflection_angle)
|
|
new_sin: float = math.sin(reflection_angle)
|
|
|
|
new_angle: float = math.atan2(new_sin, new_cos)
|
|
|
|
return new_angle
|
|
|
|
async def paddle_collision(ball: Ball, impact: Point, player: Player, inc_x: float, inc_y: float):
|
|
|
|
diff_x: float = player.rail.stop.x - player.rail.start.x
|
|
diff_y: float = player.rail.stop.y - player.rail.start.y
|
|
|
|
paddle_center_x: float = player.rail.start.x + diff_x * player.position.position
|
|
paddle_center_y: float = player.rail.start.y + diff_y * player.position.position
|
|
|
|
paddle_center: Point = Point(paddle_center_x, paddle_center_y)
|
|
|
|
rail_length: float = player.rail.length()
|
|
paddle_length: float = rail_length * config.PADDLE_RATIO;
|
|
|
|
start_x: float = paddle_center.x - (diff_x * (paddle_length / 2 / rail_length))
|
|
start_y: float = paddle_center.y - (diff_y * (paddle_length / 2 / rail_length))
|
|
stop_x: float = paddle_center.x + (diff_x * (paddle_length / 2 / rail_length))
|
|
stop_y: float = paddle_center.y + (diff_y * (paddle_length / 2 / rail_length))
|
|
|
|
start: Point = Point(start_x, start_y)
|
|
stop: Point = Point(stop_x, stop_y)
|
|
|
|
paddle: Segment = Segment(start, stop)
|
|
|
|
hit_point: Point = Point(impact.x - inc_x, impact.y - inc_y)
|
|
|
|
if (not paddle.is_on(hit_point)):
|
|
player.nb_goal += 1
|
|
await SyncToAsync(player.game.broadcast)("goal", {"player": player.user_id, "nb_goal": player.nb_goal})
|
|
return None
|
|
|
|
return ball.angle + math.pi
|
|
|
|
async def collision(game: Game, impact_data: dict) -> bool:
|
|
|
|
segment: Segment = impact_data.get("segment")
|
|
|
|
player_hitted = None
|
|
for player in game.players:
|
|
if (not player.is_connected()):
|
|
continue
|
|
if (player.rail is segment):
|
|
player_hitted = player
|
|
break
|
|
|
|
surface_angle: float = math.atan2(segment.start.y - segment.stop.y, segment.start.x - segment.stop.y)
|
|
|
|
angle: float
|
|
|
|
if (player_hitted is None):
|
|
angle = wall_collision(game.ball.angle, surface_angle)
|
|
else:
|
|
angle = await paddle_collision(game.ball, impact_data.get("impact"), player_hitted, impact_data.get("inc_x"), impact_data.get("inc_y"))
|
|
|
|
if (angle is None):
|
|
return False
|
|
|
|
game.ball.speed += config.BALL_SPEED_INC
|
|
game.ball.angle = angle
|
|
|
|
return True
|
|
|
|
async def update_ball(game: Game, impact_data: dict) -> None:
|
|
|
|
distance: float = impact_data.get("distance")
|
|
|
|
time_before_impact: float = distance / game.ball.speed
|
|
|
|
await asyncio.sleep(time_before_impact)
|
|
|
|
hit: bool = await collision(game, impact_data)
|
|
|
|
if (hit == False):
|
|
await asyncio.sleep(0.1) # delay to create frontend animation
|
|
game.ball.reset()
|
|
else:
|
|
game.ball.position = impact_data.get("impact")
|
|
|
|
await SyncToAsync(game.broadcast)("update_ball", game.ball.to_dict())
|
|
|
|
async def render_ball(game: Game):
|
|
|
|
while True:
|
|
|
|
segments: list[Segment] = [player.rail for player in game.players] + [wall.rail for wall in game.walls]
|
|
|
|
impact_data: dict = get_impact_data(segments, game.ball)
|
|
|
|
await update_ball(game, impact_data)
|
|
|
|
async def render_players(game: Game):
|
|
|
|
while True:
|
|
|
|
for player in game._updated_players:
|
|
game.broadcast("update_paddle", player.to_dict(), [player])
|
|
|
|
game._updated_players.clear()
|
|
|
|
await asyncio.sleep(1 / config.SERVER_TPS)
|
|
|
|
async def render(game: Game):
|
|
|
|
routine_ball = asyncio.create_task(render_ball(game))
|
|
routine_players = asyncio.create_task(render_players(game))
|
|
|
|
while(True):
|
|
if (game.stopped):
|
|
routine_ball.cancel()
|
|
routine_players.cancel()
|
|
return
|
|
await asyncio.sleep(0.3)
|
|
|
|
def routine(game: Game):
|
|
|
|
asyncio.run(render(game))
|