""" Server-Authoritative Game Engine Loop for FreeExile. Coordinates the fixed 30Hz simulation pipeline across Movement Authority, Spatial AOI Grid, Combat Engine, and Delta Snapshot generation. """ import time import math from typing import Dict, List, Set, Tuple, Optional, Any from dataclasses import dataclass, field from world.spatial_grid import SpatialGrid, Entity from world.movement_authority import MovementAuthorityEngine, PlayerCharacter from world.combat_engine import CombatEngine, CombatActor, DamageEventResult from world.martial_matrix import FiveElements from world.isometric_math import IsometricMath, Vector2D @dataclass class ClientMoveCommand: entity_id: int dir_x: float dir_y: float sequence: int timestamp_ms: int @dataclass class ClientSkillCommand: attacker_id: int target_id: int skill_id: int raw_damage: float element: FiveElements sequence: int timestamp_ms: int @dataclass class EntitySnapshot: entity_id: int x: float y: float z: float hp: float element: int is_evading: bool @dataclass class TickResult: tick_number: int tick_duration_ms: float processed_moves: int processed_skills: int entities_count: int snapshots_generated: int class ServerEngineLoop: """ Authoritative Server Game Engine Loop running at fixed 30Hz (33.33ms / tick). Ensures zero-trust simulation, strict anti-cheat validation, and efficient AOI broadcast. """ TICK_RATE_HZ: int = 30 TICK_INTERVAL_SEC: float = 1.0 / 30.0 # 0.03333s def __init__(self, cell_size: float = 64.0, stat_aggregator: Optional[Any] = None): self.tick_number: int = 0 self.current_time_ms: int = int(time.time() * 1000) # Core subsystems self.spatial_grid: SpatialGrid = SpatialGrid(cell_size=cell_size) self.movement_authority: MovementAuthorityEngine = MovementAuthorityEngine() self.combat_engine: CombatEngine = CombatEngine() self.stat_aggregator: Optional[Any] = stat_aggregator # Ingest queues self.move_queue: List[ClientMoveCommand] = [] self.skill_queue: List[ClientSkillCommand] = [] self.evasion_queue: List[Tuple[int, int]] = [] # (entity_id, timestamp_ms) # Entity registry self.active_entities: Dict[int, Entity] = {} self.entity_z_levels: Dict[int, float] = {} def register_player( self, entity_id: int, initial_x: float, initial_y: float, initial_z: float = 0.0, element: FiveElements = FiveElements.KIM, move_speed: float = 6.0, max_hp: float = 1000.0, player_id: Optional[str] = None, account_id: Optional[str] = None, character_id: Optional[str] = None, aggregated_stats: Optional[Any] = None, context_tags: Optional[Any] = None, conditions: Optional[Any] = None, context: Optional[Any] = None, ) -> CombatActor: """Registers a player into all authoritative subsystems with aggregated stats.""" resolved_hp = max_hp resolved_attack = 50.0 resolved_move_speed = move_speed resolved_crit_chance = 0.05 resolved_crit_mult = 1.50 resolved_resistances: Dict[FiveElements, float] = {} stats_obj = aggregated_stats if stats_obj is None and self.stat_aggregator is not None and player_id is not None: if hasattr(self.stat_aggregator, "aggregate_player_stats"): stats_obj = self.stat_aggregator.aggregate_player_stats( player_id=player_id, account_id=account_id or player_id, character_id=character_id or player_id, active_tags=context_tags, conditions=conditions, ) elif hasattr(self.stat_aggregator, "calculate_stats"): stats_obj = self.stat_aggregator.calculate_stats(player_id=player_id, context=context) if stats_obj is not None: resolved_hp = getattr(stats_obj, "max_hp", max_hp) resolved_attack = getattr(stats_obj, "attack_damage", getattr(stats_obj, "base_attack", 50.0)) resolved_move_speed = getattr(stats_obj, "move_speed", getattr(stats_obj, "movement_speed", move_speed)) resolved_crit_chance = getattr(stats_obj, "crit_chance", 0.05) resolved_crit_mult = getattr(stats_obj, "crit_multiplier", 1.50) res_dict = getattr(stats_obj, "resistances", {}) if isinstance(res_dict, dict): elem_mapping = { "hoa": FiveElements.HOA, "fire": FiveElements.HOA, "thuy": FiveElements.THUY, "cold": FiveElements.THUY, "kim": FiveElements.KIM, "lightning": FiveElements.KIM, "moc": FiveElements.MOC, "chaos": FiveElements.MOC, "poison": FiveElements.MOC, "tho": FiveElements.THO, "earth": FiveElements.THO, } for k, v in res_dict.items(): if isinstance(k, FiveElements): resolved_resistances[k] = float(v) elif isinstance(k, str) and k.lower() in elem_mapping: resolved_resistances[elem_mapping[k.lower()]] = float(v) elif isinstance(k, str) and hasattr(FiveElements, k.upper()): resolved_resistances[FiveElements[k.upper()]] = float(v) grid_entity = Entity(entity_id=entity_id, x=initial_x, y=initial_y) self.spatial_grid.add_entity(grid_entity) self.active_entities[entity_id] = grid_entity self.entity_z_levels[entity_id] = initial_z player_char = PlayerCharacter(entity_id=entity_id, x=initial_x, y=initial_y, move_speed=resolved_move_speed) self.movement_authority.register_player(player_char) combat_actor = CombatActor( actor_id=entity_id, name=f"Player_{entity_id}", element=element, current_hp=resolved_hp, max_hp=resolved_hp, base_attack=resolved_attack, crit_chance=resolved_crit_chance, crit_multiplier=resolved_crit_mult, resistances=resolved_resistances, is_player=True, player_id=player_id or f"player_{entity_id}", ) self.combat_engine.register_actor(combat_actor) return combat_actor def queue_move_command(self, cmd: ClientMoveCommand) -> None: self.move_queue.append(cmd) def queue_skill_command(self, cmd: ClientSkillCommand) -> None: self.skill_queue.append(cmd) def queue_evasion(self, entity_id: int, timestamp_ms: int) -> None: self.evasion_queue.append((entity_id, timestamp_ms)) def step_tick(self, dt: float = TICK_INTERVAL_SEC) -> TickResult: """ Executes one authoritative 30Hz tick cycle across all 7 phases: 1. Ingest evasions (i-frame trigger) 2. Authoritative movement displacement & anti-cheat 3. Spatial AOI grid update 4. Combat resolution & damage calculation 5. State snapshot packaging for AOI neighborhoods """ start_time = time.perf_counter() self.tick_number += 1 self.current_time_ms += int(dt * 1000) # Phase 1: Process Huyễn Ảnh Bộ Evasions (0.25s i-frame window) for entity_id, timestamp_ms in self.evasion_queue: self.combat_engine.trigger_phantom_evasion(entity_id, timestamp_ms) self.evasion_queue.clear() # Phase 2: Process Movement Commands processed_moves = 0 current_moves = self.move_queue.copy() self.move_queue.clear() for move_cmd in current_moves: success, new_x, new_y = self.movement_authority.process_move_input( entity_id=move_cmd.entity_id, dir_x=move_cmd.dir_x, dir_y=move_cmd.dir_y, dt=dt ) if success and move_cmd.entity_id in self.active_entities: entity = self.active_entities[move_cmd.entity_id] # Update spatial grid cell self.spatial_grid.update_entity_position(entity, new_x, new_y) processed_moves += 1 # Phase 3: Process Combat & Skill Commands processed_skills = 0 current_skills = self.skill_queue.copy() self.skill_queue.clear() for skill_cmd in current_skills: # Check Z-height overlap for 2.5D elevation consistency attacker_z = self.entity_z_levels.get(skill_cmd.attacker_id, 0.0) target_z = self.entity_z_levels.get(skill_cmd.target_id, 0.0) if IsometricMath.check_height_overlap(attacker_z, target_z): damage_result = self.combat_engine.calculate_damage( attacker_id=skill_cmd.attacker_id, defender_id=skill_cmd.target_id, raw_damage=skill_cmd.raw_damage, damage_element=skill_cmd.element, current_timestamp_ms=self.current_time_ms ) processed_skills += 1 # Phase 4: Generate AOI Snapshots for each active entity snapshots_generated = 0 for entity_id, entity in self.active_entities.items(): # Query 9-cell neighborhood visible_ids = self.spatial_grid.get_entities_in_aoi(entity.x, entity.y, radius_cells=1) snapshots_generated += len(visible_ids) tick_duration_ms = (time.perf_counter() - start_time) * 1000.0 return TickResult( tick_number=self.tick_number, tick_duration_ms=tick_duration_ms, processed_moves=processed_moves, processed_skills=processed_skills, entities_count=len(self.active_entities), snapshots_generated=snapshots_generated ) def get_aoi_snapshots_for_entity(self, observer_id: int) -> List[EntitySnapshot]: """ Retrieves the state snapshots of all entities visible to observer within its 9-cell AOI. """ observer = self.active_entities.get(observer_id) if not observer: return [] visible_ids = self.spatial_grid.get_entities_in_aoi(observer.x, observer.y, radius_cells=1) snapshots: List[EntitySnapshot] = [] for entity_id in visible_ids: ent = self.active_entities.get(entity_id) combat_actor = self.combat_engine.actors.get(entity_id) if not ent: continue z = self.entity_z_levels.get(entity_id, 0.0) hp = combat_actor.current_hp if combat_actor else 0.0 elem = combat_actor.element.value if combat_actor else 1 # Check if currently in evasion i-frame is_evading = False if combat_actor: elapsed = self.current_time_ms - combat_actor.last_evasion_timestamp_ms is_evading = (0 <= elapsed <= combat_actor.evasion_iframe_duration_ms) snapshots.append(EntitySnapshot( entity_id=entity_id, x=ent.x, y=ent.y, z=z, hp=hp, element=elem, is_evading=is_evading )) return snapshots