From 991b8f3660fb0ed717c443827136293e90d44f57 Mon Sep 17 00:00:00 2001 From: Vladyslav Doloman Date: Tue, 7 Oct 2025 20:27:43 +0300 Subject: [PATCH] Server: per-tick simulation skeleton (inputs, movement, blocking/shrink, apple growth, wrap behavior) and basic state delta broadcast MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- server/model.py | 22 +++++++++ server/server.py | 120 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 141 insertions(+), 1 deletion(-) diff --git a/server/model.py b/server/model.py index a2cee20..5d86151 100644 --- a/server/model.py +++ b/server/model.py @@ -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: diff --git a/server/server.py b/server/server.py index c60c5ce..df8ea74 100644 --- a/server/server.py +++ b/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]: