Server: per-tick simulation skeleton (inputs, movement, blocking/shrink, apple growth, wrap behavior) and basic state delta broadcast
- Input buffer rules: enqueue+consume with 180° guard (len>1) - Movement: wrap per config; allow moving into own tail when it vacates - Blocking: head holds, tail shrinks to min 1 - Apples: eat= grow; ensure target apples after tick - Broadcast: send current snakes/apples as delta (placeholder for real deltas)
This commit is contained in:
Vladyslav Doloman
@@ -16,11 +16,33 @@ class Snake:
|
||||
head: Coord
|
||||
direction: Direction
|
||||
body: Deque[Coord] = field(default_factory=deque) # includes head at index 0
|
||||
input_buf: Deque[Direction] = field(default_factory=deque)
|
||||
blocked: bool = False
|
||||
|
||||
@property
|
||||
def length(self) -> int:
|
||||
return len(self.body)
|
||||
|
||||
def enqueue_direction(self, new_dir: Direction, capacity: int = 3) -> None:
|
||||
"""Apply input buffer rules: size<=capacity, replace last on overflow/opposite, drop duplicates."""
|
||||
last_dir = self.input_buf[-1] if self.input_buf else self.direction
|
||||
# Drop duplicates
|
||||
if int(new_dir) == int(last_dir):
|
||||
return
|
||||
# Opposite of last? replace last
|
||||
if (int(new_dir) ^ int(last_dir)) == 2: # 0^2,1^3,2^0,3^1 are opposites
|
||||
if self.input_buf:
|
||||
self.input_buf[-1] = new_dir
|
||||
else:
|
||||
# No buffered inputs; just add new_dir (consumption will handle 180° rule)
|
||||
self.input_buf.append(new_dir)
|
||||
return
|
||||
# Normal append with overflow replacement
|
||||
if len(self.input_buf) >= capacity:
|
||||
self.input_buf[-1] = new_dir
|
||||
else:
|
||||
self.input_buf.append(new_dir)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlayerSession:
|
||||
|
||||
120
server/server.py
120
server/server.py
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from collections import deque
|
||||
|
||||
from .config import ServerConfig
|
||||
from .model import GameState, PlayerSession, Snake, Coord
|
||||
@@ -15,6 +16,9 @@ from .protocol import (
|
||||
build_config_update,
|
||||
build_input_broadcast,
|
||||
build_state_full,
|
||||
BodyTLV,
|
||||
build_state_full,
|
||||
quic_varint_encode,
|
||||
pack_header,
|
||||
parse_input,
|
||||
parse_join,
|
||||
@@ -104,7 +108,8 @@ class GameServer:
|
||||
next_cfg_resend = self._config_update_interval_ticks
|
||||
while True:
|
||||
start = asyncio.get_event_loop().time()
|
||||
# TODO: process inputs, update snakes, collisions, apples, deltas
|
||||
# process inputs, update snakes, collisions, apples, deltas
|
||||
await self._simulate_tick()
|
||||
r.state.tick = (r.state.tick + 1) & 0xFFFFFFFF
|
||||
|
||||
next_cfg_resend -= 1
|
||||
@@ -115,6 +120,119 @@ class GameServer:
|
||||
elapsed = asyncio.get_event_loop().time() - start
|
||||
await asyncio.sleep(max(0.0, tick_duration - elapsed))
|
||||
|
||||
# --- Simulation ---
|
||||
|
||||
def _consume_input_for_snake(self, s: Snake) -> None:
|
||||
# Consume at most one input; skip 180° turns when length>1
|
||||
while s.input_buf:
|
||||
nd = s.input_buf[0]
|
||||
# 180-degree check
|
||||
if s.length > 1 and ((int(nd) ^ int(s.direction)) == 2):
|
||||
s.input_buf.popleft()
|
||||
continue
|
||||
# Accept
|
||||
s.direction = nd
|
||||
s.input_buf.popleft()
|
||||
break
|
||||
|
||||
def _step_from(self, x: int, y: int, d: Direction) -> Tuple[int, int, bool]:
|
||||
dx = 1 if d == Direction.RIGHT else -1 if d == Direction.LEFT else 0
|
||||
dy = 1 if d == Direction.DOWN else -1 if d == Direction.UP else 0
|
||||
nx, ny = x + dx, y + dy
|
||||
st = self.runtime.state
|
||||
wrap = self.runtime.config.wrap_edges
|
||||
wrapped = False
|
||||
if wrap:
|
||||
if nx < 0:
|
||||
nx = st.width - 1; wrapped = True
|
||||
elif nx >= st.width:
|
||||
nx = 0; wrapped = True
|
||||
if ny < 0:
|
||||
ny = st.height - 1; wrapped = True
|
||||
elif ny >= st.height:
|
||||
ny = 0; wrapped = True
|
||||
return nx, ny, wrapped
|
||||
|
||||
async def _simulate_tick(self) -> None:
|
||||
st = self.runtime.state
|
||||
cfg = self.runtime.config
|
||||
apples_eaten: List[Coord] = []
|
||||
# Prepare snapshot of tails to allow moving into own tail when it vacates
|
||||
tails: Dict[int, Coord] = {}
|
||||
for sid, s in st.snakes.items():
|
||||
if s.length > 1:
|
||||
tails[sid] = s.body[-1]
|
||||
# Process snakes
|
||||
for sid, s in st.snakes.items():
|
||||
# Consume one input if available
|
||||
self._consume_input_for_snake(s)
|
||||
hx, hy = s.body[0]
|
||||
nx, ny, wrapped = self._step_from(hx, hy, s.direction)
|
||||
# Check wall if no wrap
|
||||
obstacle = False
|
||||
if not self.runtime.config.wrap_edges and not st.in_bounds(nx, ny):
|
||||
obstacle = True
|
||||
else:
|
||||
# Bounds correction already handled by _step_from
|
||||
# Occupancy check
|
||||
occ = st.occupancy.get((nx, ny))
|
||||
if occ is not None:
|
||||
# Allow moving into own tail if it will vacate (not growing)
|
||||
own_tail_ok = (s.length > 1 and (nx, ny) == tails.get(sid))
|
||||
if not own_tail_ok:
|
||||
obstacle = True
|
||||
if obstacle:
|
||||
s.blocked = True
|
||||
# shrink tail by 1 (to min 1)
|
||||
if s.length > 1:
|
||||
tx, ty = s.body.pop()
|
||||
st.occupancy.pop((tx, ty), None)
|
||||
continue
|
||||
# Move or grow
|
||||
s.blocked = False
|
||||
will_grow = (nx, ny) in st.apples
|
||||
# Add new head
|
||||
s.body.appendleft((nx, ny))
|
||||
st.occupancy[(nx, ny)] = (sid, 0)
|
||||
if will_grow:
|
||||
# eat apple; no tail removal this tick
|
||||
st.apples.remove((nx, ny))
|
||||
apples_eaten.append((nx, ny))
|
||||
else:
|
||||
# normal move: remove tail (unless length==0)
|
||||
tx, ty = s.body.pop()
|
||||
if (tx, ty) in st.occupancy:
|
||||
st.occupancy.pop((tx, ty), None)
|
||||
# Replenish apples to target
|
||||
self._ensure_apples()
|
||||
# Broadcast a basic delta (currently full snakes + apples as delta)
|
||||
snakes_dirs: List[Tuple[int, int, int, int, List[Direction]]] = []
|
||||
for s in st.snakes.values():
|
||||
dirs: List[Direction] = []
|
||||
coords = list(s.body)
|
||||
for i in range(len(coords) - 1):
|
||||
x0, y0 = coords[i]
|
||||
x1, y1 = coords[i + 1]
|
||||
if x1 == x0 and y1 == y0 - 1:
|
||||
dirs.append(Direction.UP)
|
||||
elif x1 == x0 + 1 and y1 == y0:
|
||||
dirs.append(Direction.RIGHT)
|
||||
elif x1 == x0 and y1 == y0 + 1:
|
||||
dirs.append(Direction.DOWN)
|
||||
elif x1 == x0 - 1 and y1 == y0:
|
||||
dirs.append(Direction.LEFT)
|
||||
hx, hy = coords[0]
|
||||
snakes_dirs.append((s.snake_id, s.length, hx, hy, dirs))
|
||||
delta = build_state_full(
|
||||
version=self.runtime.version,
|
||||
seq=self.runtime.next_seq(),
|
||||
tick=st.tick,
|
||||
snakes=snakes_dirs,
|
||||
apples=st.apples,
|
||||
)
|
||||
for session in list(self.sessions.values()):
|
||||
await self.transport.send(delta, TransportPeer(session.peer))
|
||||
|
||||
# --- Join / Spawn ---
|
||||
|
||||
def _allocate_player_id(self) -> Optional[int]:
|
||||
|
||||
Reference in New Issue
Block a user