""" FreeExile Server-Authoritative War-Fog Engine. Implements: 1. Field of View (FOV) calculation with Bresenham Line-of-Sight & wall occlusion. 2. Three-state Fog of War: UNEXPLORED (0), EXPLORED_FOGGED (1), VISIBLE (2). 3. Zero-Trust Server Authority: Entity visibility filtering to defeat maphack/memory-reading bots. 4. Lossless Run-Length Encoding (RLE) Fog Delta Compression for 1M CCU network efficiency. """ from __future__ import annotations import math from typing import List, Dict, Any, Tuple, Optional from server.world.map_data_types import ( FogState, MapGridData, PlayerVisibilityState, ) class WarFogEngine: """Server-Authoritative Field of View and Fog of War computing engine.""" def __init__(self, default_view_radius: int = 12): self.default_view_radius = default_view_radius def create_player_visibility( self, player_id: str, width: int, height: int, view_radius: Optional[int] = None ) -> PlayerVisibilityState: radius = view_radius or self.default_view_radius fog_matrix = [[FogState.UNEXPLORED.value for _ in range(width)] for _ in range(height)] return PlayerVisibilityState( player_id=player_id, width=width, height=height, view_radius=radius, fog_matrix=fog_matrix, explored_tiles_count=0, visible_tiles_count=0, last_pos=(0, 0), ) def update_visibility( self, vis: PlayerVisibilityState, map_data: MapGridData, player_x: int, player_y: int, view_radius: Optional[int] = None, ) -> None: radius = view_radius or vis.view_radius w, h = vis.width, vis.height # Demote previously VISIBLE cells within previous visibility bounding box prev_x, prev_y = vis.last_pos prev_min_x = max(0, prev_x - radius - 1) prev_max_x = min(w - 1, prev_x + radius + 1) prev_min_y = max(0, prev_y - radius - 1) prev_max_y = min(h - 1, prev_y + radius + 1) for y in range(prev_min_y, prev_max_y + 1): row = vis.fog_matrix[y] for x in range(prev_min_x, prev_max_x + 1): if row[x] == FogState.VISIBLE.value: row[x] = FogState.EXPLORED_FOGGED.value # Raycast to the bounding square perimeter min_x = max(0, player_x - radius) max_x = min(w - 1, player_x + radius) min_y = max(0, player_y - radius) max_y = min(h - 1, player_y + radius) # Collect target border points targets: List[Tuple[int, int]] = [] for x in range(min_x, max_x + 1): targets.append((x, min_y)) targets.append((x, max_y)) for y in range(min_y, max_y + 1): targets.append((min_x, y)) targets.append((max_x, y)) # Always mark player cell visible if 0 <= player_x < w and 0 <= player_y < h: vis.fog_matrix[player_y][player_x] = FogState.VISIBLE.value # Cast rays to perimeter targets for tx, ty in targets: self._cast_ray(vis, map_data, player_x, player_y, tx, ty, radius) # Recalculate statistics in a single fast pass vis.last_pos = (player_x, player_y) vis_count = 0 exp_count = 0 for y in range(h): row = vis.fog_matrix[y] for cell in row: if cell == FogState.VISIBLE.value: vis_count += 1 if cell != FogState.UNEXPLORED.value: exp_count += 1 vis.visible_tiles_count = vis_count vis.explored_tiles_count = exp_count def _cast_ray( self, vis: PlayerVisibilityState, map_data: MapGridData, x0: int, y0: int, x1: int, y1: int, max_dist: int, ) -> None: """Bresenham line algorithm with wall occlusion stop.""" dx = abs(x1 - x0) dy = abs(y1 - y0) sx = 1 if x0 < x1 else -1 sy = 1 if y0 < y1 else -1 err = dx - dy curr_x, curr_y = x0, y0 while True: dist = math.hypot(curr_x - x0, curr_y - y0) if dist > max_dist: break if 0 <= curr_x < vis.width and 0 <= curr_y < vis.height: vis.fog_matrix[curr_y][curr_x] = FogState.VISIBLE.value # Block further propagation if cell blocks vision (unless starting origin) if (curr_x != x0 or curr_y != y0) and map_data.blocks_vision(curr_x, curr_y): break if curr_x == x1 and curr_y == y1: break e2 = 2 * err if e2 > -dy: err -= dy curr_x += sx if e2 < dx: err += dx curr_y += sy def filter_visible_entities( self, vis: PlayerVisibilityState, entities: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: """ Server-Authoritative Anti-Maphack Guard. Strictly excludes any entity not residing within a cell with FogState.VISIBLE. Prevents client memory inspection or packet sniffing from detecting hidden mobs/loot. """ filtered: List[Dict[str, Any]] = [] w, h = vis.width, vis.height for ent in entities: ex = int(ent.get("x", -1)) ey = int(ent.get("y", -1)) if 0 <= ex < w and 0 <= ey < h: if vis.fog_matrix[ey][ex] == FogState.VISIBLE.value: filtered.append(ent) return filtered def get_explored_ratio(self, vis: PlayerVisibilityState) -> float: """Returns explored percentage (0.0 to 1.0) of the map.""" total = vis.width * vis.height if total <= 0: return 0.0 return round(vis.explored_tiles_count / total, 4) def is_cell_visible(self, vis: PlayerVisibilityState, x: int, y: int) -> bool: """Returns True if coordinate is currently inside player line-of-sight.""" if 0 <= x < vis.width and 0 <= y < vis.height: return vis.fog_matrix[y][x] == FogState.VISIBLE.value return False def is_cell_explored(self, vis: PlayerVisibilityState, x: int, y: int) -> bool: """Returns True if coordinate has been explored (visible or previously seen).""" if 0 <= x < vis.width and 0 <= y < vis.height: return vis.fog_matrix[y][x] != FogState.UNEXPLORED.value return False def compress_fog_rle(matrix: List[List[int]]) -> str: """Lossless Run-Length Encoding of 2D Fog Matrix.""" if not matrix or not matrix[0]: return "" flat = [cell for row in matrix for cell in row] runs: List[str] = [] current_val = flat[0] count = 1 for val in flat[1:]: if val == current_val: count += 1 else: runs.append(f"{current_val}x{count}") current_val = val count = 1 runs.append(f"{current_val}x{count}") return ",".join(runs) def decompress_fog_rle(rle_str: str, width: int, height: int) -> List[List[int]]: """Decompresses RLE string into a 2D integer matrix of size height x width.""" default_matrix = [[0 for _ in range(width)] for _ in range(height)] if not rle_str or width <= 0 or height <= 0: return default_matrix flat: List[int] = [] tokens = rle_str.split(",") try: for token in tokens: if not token: continue parts = token.split("x") if len(parts) != 2: return default_matrix val = int(parts[0]) count = int(parts[1]) if count < 0: return default_matrix flat.extend([val] * count) except (ValueError, OverflowError): return default_matrix if len(flat) != width * height: return default_matrix return [flat[y * width : (y + 1) * width] for y in range(height)]