"""AutoPOE2 - Minimap Fog of War & Frontier Grid Exploration Engine (Doc 69 §3.4). =================================================================================== Module nhận diện sương mù Minimap và định tuyến thám hiểm biên mở rộng: - Phân tích ảnh Minimap / lưới địa hình để trích xuất vùng chưa mở (Fog of War). - Xác định và phân cụm các mép biên thám hiểm (Frontier Clusters). - Định tuyến mục tiêu khám phá tối ưu qua hàm tiện ích tổng hợp (Information Gain, Distance, Visit Heatmap Penalty, và Quán tính di chuyển Frontier Momentum). - Cắt tỉa triệt để ngõ cụt đã dọn sạch (Dead-End Topological Pruning) để chống backtrack. - Tự động đánh giá tiêu chí hoàn thành dọn Map (Coverage >= 95% monsters/tiles). - Tương thích toàn diện với Dynamic Resolution Scaler (Windowed 1189x891 PID 22632, 1440p QHD, 1080p SSoT, 720p). Tuân thủ nghiêm ngặt: - Rule 1: Python 3.11 Cold Path Tier 2. - Rule 4: Module hóa độc lập tại src/assistant_tool/navigation/. - Rule 5: 100% artifact thật từ captures/. - Rule 16: Google Docstrings chuẩn mực. - Bất biến: INV-NAV-FRONTIER-MOMENTUM, INV-SCALE-IDEMPOTENT-1PX, INV-WASD-NO-BLIND-LMB. """ from __future__ import annotations from collections import deque from dataclasses import dataclass, field from enum import IntEnum import math from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Union import numpy as np try: from PIL import Image except ImportError: Image = None # type: ignore[assignment] from src.common.coordinate_transform import ( CoordinateScaler, Resolution, ScalingMode, RES_1080P, RES_1440P, RES_WINDOWED_DEFAULT, resolve_resolution, ) class FogCellState(IntEnum): """Trạng thái tế bào trên lưới thám hiểm sương mù.""" UNEXPLORED = 0 # Sương mù tối / Chưa bao giờ quét tới EXPLORED = 1 # Đã mở / Mặt đất có thể đi được WALL_OBSTACLE = 2 # Tường / Chướng ngại vật không thể đi qua FRONTIER = 3 # Mép biên tiếp giáp sương mù chưa mở DEAD_END = 4 # Ngõ cụt đã dọn sạch, bị cắt tỉa chống backtrack @dataclass class FrontierCluster: """Cụm các điểm biên giới tiếp giáp sương mù chưa khám phá.""" cluster_id: int cells: List[Tuple[int, int]] center_grid: Tuple[float, float] size: int information_gain: float = 0.0 # Số ô sương mù chưa mở trong bán kính lân cận avg_visit_count: float = 0.0 # Số lần ghé thăm trung bình từ heatmap @dataclass class ExplorationTarget: """Mục tiêu thám hiểm được lựa chọn tối ưu.""" target_grid_xy: Tuple[int, int] target_minimap_xy: Tuple[int, int] target_screen_xy: Tuple[int, int] direction_vector: Tuple[float, float] # Vector đơn vị chuẩn hóa (dx, dy) utility_score: float coverage_ratio: float # Tỷ lệ bản đồ đã mở (0.0 .. 1.0) is_fully_explored: bool # True nếu coverage >= 0.95 hoặc 0 frontier active_clusters_count: int dead_ends_pruned_count: int # Tọa độ bounding box của Corner Minimap trên màn hình 1080p SSoT # Vùng góc trên bên phải: x1=1460, y1=20, x2=1900, y2=360 (440x340 px) CANONICAL_MINIMAP_BBOX_1080P = (1460.0, 20.0, 1900.0, 360.0) class ExplorationGrid: """Lưới tế bào quản trị trạng thái sương mù, bản đồ nhiệt và cắt tỉa ngõ cụt.""" def __init__( self, grid_width: int = 64, grid_height: int = 64, dark_threshold: float = 15.0, wall_threshold: float = 180.0, ) -> None: """Khởi tạo lưới thám hiểm. Args: grid_width: Số cột của lưới tế bào. grid_height: Số hàng của lưới tế bào. dark_threshold: Ngưỡng độ sáng tối đa để phân loại là sương mù chưa mở. wall_threshold: Ngưỡng độ sáng tối thiểu để phân loại là tường/vật cản. """ self.gw = grid_width self.gh = grid_height self.dark_threshold = dark_threshold self.wall_threshold = wall_threshold self.cells = np.full((self.gh, self.gw), FogCellState.UNEXPLORED, dtype=np.uint8) self.visit_counts = np.zeros((self.gh, self.gw), dtype=np.uint16) self.dead_ends = np.zeros((self.gh, self.gw), dtype=bool) def load_from_minimap_image( self, img_input: Union[np.ndarray, Image.Image], ) -> None: """Phân tích ảnh Minimap và cập nhật trạng thái các tế bào trên lưới.""" if isinstance(img_input, Image.Image): arr = np.array(img_input) else: arr = img_input ih, iw = arr.shape[:2] if arr.ndim == 3: rgb = arr[..., :3] gray = np.mean(rgb, axis=2) else: gray = arr.astype(float) # Tối ưu hóa hiệu năng cao bằng PIL resize + Vectorized NumPy (< 3ms thay vì 600ms) if Image is not None: if isinstance(img_input, Image.Image): arr = np.array(img_input) else: arr = img_input if arr.ndim == 3: gray = np.mean(arr[..., :3], axis=2).astype(np.uint8) else: gray = arr.astype(np.uint8) pil_gray = Image.fromarray(gray).resize((self.gw, self.gh), Image.Resampling.BOX) gray_grid = np.asarray(pil_gray, dtype=np.float32) unexplored_mask = gray_grid < self.dark_threshold wall_mask = gray_grid > self.wall_threshold explored_mask = (~unexplored_mask) & (~wall_mask) self.cells[unexplored_mask] = FogCellState.UNEXPLORED self.cells[wall_mask] = FogCellState.WALL_OBSTACLE self.cells[explored_mask] = FogCellState.EXPLORED return # Fallback lặp thủ công nếu PIL không khả dụng block_h = max(1, ih // self.gh) block_w = max(1, iw // self.gw) for gy in range(self.gh): y_start = gy * block_h y_end = min(ih, (gy + 1) * block_h) if y_start >= y_end: continue for gx in range(self.gw): x_start = gx * block_w x_end = min(iw, (gx + 1) * block_w) if x_start >= x_end: continue patch = gray[y_start:y_end, x_start:x_end] mean_brightness = float(np.mean(patch)) # Phân loại trạng thái tế bào if mean_brightness < self.dark_threshold: self.cells[gy, gx] = FogCellState.UNEXPLORED elif mean_brightness > self.wall_threshold: self.cells[gy, gx] = FogCellState.WALL_OBSTACLE else: self.cells[gy, gx] = FogCellState.EXPLORED def record_visit(self, gx: int, gy: int, radius: int = 1) -> None: """Ghi nhận vết bước chân vào bản đồ nhiệt.""" for y in range(max(0, gy - radius), min(self.gh, gy + radius + 1)): for x in range(max(0, gx - radius), min(self.gw, gx + radius + 1)): self.visit_counts[y, x] = min(65535, self.visit_counts[y, x] + 1) if self.cells[y, x] == FogCellState.UNEXPLORED: self.cells[y, x] = FogCellState.EXPLORED def get_coverage_ratio(self) -> float: """Tính toán tỷ lệ đã khám phá trên tổng số ô có thể đi được.""" explored = np.sum(self.cells == FogCellState.EXPLORED) unexplored = np.sum(self.cells == FogCellState.UNEXPLORED) total_traversable = explored + unexplored if total_traversable == 0: return 1.0 return float(explored / total_traversable) def extract_frontier_cells(self) -> List[Tuple[int, int]]: """Xác định danh sách các ô Frontier (ô explored tiếp giáp unexplored).""" is_unexp = (self.cells == FogCellState.UNEXPLORED) has_unexp = np.zeros((self.gh, self.gw), dtype=bool) has_unexp[1:, :] |= is_unexp[:-1, :] has_unexp[:-1, :] |= is_unexp[1:, :] has_unexp[:, 1:] |= is_unexp[:, :-1] has_unexp[:, :-1] |= is_unexp[:, 1:] is_frontier = (self.cells == FogCellState.EXPLORED) & (~self.dead_ends) & has_unexp ys, xs = np.where(is_frontier) return list(zip(xs.tolist(), ys.tolist())) def cluster_frontiers( self, frontier_cells: List[Tuple[int, int]], max_dist: float = 2.5, ) -> List[FrontierCluster]: """Gom nhóm các ô Frontier kề nhau thành các cụm (Connected Components).""" if not frontier_cells: return [] cell_set = set(frontier_cells) visited: Set[Tuple[int, int]] = set() clusters: List[FrontierCluster] = [] cluster_id = 0 for start_cell in frontier_cells: if start_cell in visited: continue current_cluster_cells: List[Tuple[int, int]] = [] queue = deque([start_cell]) visited.add(start_cell) while queue: cx, cy = queue.popleft() current_cluster_cells.append((cx, cy)) # Duyệt lân cận 8 hướng (Moore) for dy in (-1, 0, 1): for dx in (-1, 0, 1): if dx == 0 and dy == 0: continue nbr = (cx + dx, cy + dy) if nbr in cell_set and nbr not in visited: visited.add(nbr) queue.append(nbr) # Tính tâm cụm avg_x = sum(c[0] for c in current_cluster_cells) / float(len(current_cluster_cells)) avg_y = sum(c[1] for c in current_cluster_cells) / float(len(current_cluster_cells)) # Tính Information Gain (số ô sương mù trong bán kính 3 ô quanh cụm) unexplored_count = 0 neighbor_boxes: Set[Tuple[int, int]] = set() for cx, cy in current_cluster_cells: for dy in range(-2, 3): for dx in range(-2, 3): nx, ny = cx + dx, cy + dy if 0 <= nx < self.gw and 0 <= ny < self.gh: neighbor_boxes.add((nx, ny)) for nx, ny in neighbor_boxes: if self.cells[ny, nx] == FogCellState.UNEXPLORED: unexplored_count += 1 # Tính độ nhiệt trung bình total_visits = sum(self.visit_counts[c[1], c[0]] for c in current_cluster_cells) avg_heat = total_visits / float(len(current_cluster_cells)) clusters.append( FrontierCluster( cluster_id=cluster_id, cells=current_cluster_cells, center_grid=(avg_x, avg_y), size=len(current_cluster_cells), information_gain=float(unexplored_count), avg_visit_count=avg_heat, ) ) cluster_id += 1 return clusters def prune_dead_ends(self) -> int: """Cắt tỉa ngõ cụt topological: Tìm các ô đã khám phá không còn đường mở. Returns: int: Số lượng ô ngõ cụt mới được cắt tỉa. """ new_pruned = 0 is_unexp = (self.cells == FogCellState.UNEXPLORED) has_unexp = np.zeros((self.gh, self.gw), dtype=bool) has_unexp[1:, :] |= is_unexp[:-1, :] has_unexp[:-1, :] |= is_unexp[1:, :] has_unexp[:, 1:] |= is_unexp[:, :-1] has_unexp[:, :-1] |= is_unexp[:, 1:] new_pruned = 0 for _ in range(50): is_open = (self.cells == FogCellState.EXPLORED) & (~self.dead_ends) nbr_count = np.zeros((self.gh, self.gw), dtype=np.int8) nbr_count[1:, :] += is_open[:-1, :] nbr_count[:-1, :] += is_open[1:, :] nbr_count[:, 1:] += is_open[:, :-1] nbr_count[:, :-1] += is_open[:, 1:] to_prune = is_open & (~has_unexp) & (nbr_count <= 1) cnt = int(np.count_nonzero(to_prune)) if cnt == 0: break self.dead_ends |= to_prune new_pruned += cnt return new_pruned class MinimapFogExplorer: """Bộ điều phối thám hiểm sương mù Minimap & Frontier Grid Exploration.""" def __init__( self, grid_width: int = 64, grid_height: int = 64, target_resolution: Union[Resolution, Tuple[int, int]] = RES_1080P, weight_info: float = 2.0, weight_dist: float = 1.0, weight_heat: float = 3.0, weight_momentum: float = 1.5, target_coverage: float = 0.95, ) -> None: """Khởi tạo bộ thám hiểm. Args: grid_width: Kích thước chiều ngang lưới thám hiểm. grid_height: Kích thước chiều dọc lưới thám hiểm. target_resolution: Độ phân giải màn hình đang chạy. weight_info: Trọng số ưu tiên mở rộng thông tin sương mù mới. weight_dist: Trọng số phạt khoảng cách xa. weight_heat: Trọng số phạt đi vào vùng đã dẫm chân nhiều lần (chống loop). weight_momentum: Trọng số ưu tiên giữ hướng di chuyển (chống giật cục). target_coverage: Ngưỡng hoàn thành bản đồ (mặc định 95%). """ self.grid = ExplorationGrid(grid_width=grid_width, grid_height=grid_height) self.resolution = resolve_resolution(target_resolution) self.scaler = CoordinateScaler(source_res=RES_1080P, target_res=self.resolution) self.w_info = weight_info self.w_dist = weight_dist self.w_heat = weight_heat self.w_mom = weight_momentum self.target_coverage = target_coverage # Tính bounding box của Minimap trên độ phân giải hiện tại bx1, by1, bx2, by2 = CANONICAL_MINIMAP_BBOX_1080P self.minimap_screen_bbox = self.scaler.transform_bbox(bx1, by1, bx2, by2) def set_target_resolution(self, res_input: Union[Resolution, Tuple[int, int]]) -> None: """Cập nhật độ phân giải mục tiêu mới (hỗ trợ resize cửa sổ game runtime).""" self.resolution = resolve_resolution(res_input) self.scaler = CoordinateScaler(source_res=RES_1080P, target_res=self.resolution) bx1, by1, bx2, by2 = CANONICAL_MINIMAP_BBOX_1080P self.minimap_screen_bbox = self.scaler.transform_bbox(bx1, by1, bx2, by2) def select_best_frontier( self, clusters: List[FrontierCluster], player_grid_xy: Tuple[float, float], player_vel: Tuple[float, float] = (0.0, 0.0), ) -> Optional[Tuple[FrontierCluster, float]]: """Lựa chọn cụm Frontier tối ưu dựa trên hàm tiện ích tổng hợp. Hàm tiện ích: U(C) = w_info * InfoGain(C) - w_dist * Dist(P, C) - w_heat * AvgHeat(C) + w_mom * Momentum(P, v, C) """ if not clusters: return None best_cluster: Optional[FrontierCluster] = None best_utility = -float("inf") px, py = player_grid_xy vx, vy = player_vel v_norm = math.hypot(vx, vy) for cl in clusters: cx, cy = cl.center_grid dx = cx - px dy = cy - py dist = math.hypot(dx, dy) if dist < 1e-4: dist = 0.1 # Đánh giá Momentum hướng tiến momentum_score = 0.0 if v_norm > 1e-3: cos_theta = (dx * vx + dy * vy) / (dist * v_norm) momentum_score = max(-1.0, min(1.0, cos_theta)) # Tính hàm tiện ích tổng hợp utility = ( (self.w_info * math.log1p(cl.information_gain)) - (self.w_dist * (dist / max(1.0, self.grid.gw))) - (self.w_heat * cl.avg_visit_count) + (self.w_mom * momentum_score) ) if utility > best_utility: best_utility = utility best_cluster = cl if best_cluster is not None: return best_cluster, best_utility return None def analyze_minimap_and_route( self, minimap_img: Union[np.ndarray, Image.Image], player_grid_pos: Optional[Tuple[float, float]] = None, player_velocity: Tuple[float, float] = (0.0, 0.0), ) -> ExplorationTarget: """Quy trình khép kín: Đọc Minimap -> Trích xuất Frontier -> Cắt tỉa ngõ cụt -> Định tuyến. Args: minimap_img: Ảnh cắt vùng Minimap hoặc toàn bộ Minimap canvas. player_grid_pos: Vị trí người chơi trên lưới (mặc định ở trung tâm lưới). player_velocity: Vector vận tốc di chuyển hiện tại (vx, vy). Returns: ExplorationTarget: Mục tiêu thám hiểm tối ưu đã chuẩn hóa tọa độ màn hình. """ # 1. Cập nhật trạng thái sương mù trên lưới self.grid.load_from_minimap_image(minimap_img) # Mặc định người chơi ở giữa Minimap nếu không cung cấp if player_grid_pos is None: player_grid_pos = (self.grid.gw * 0.5, self.grid.gh * 0.5) pgx, pgy = int(round(player_grid_pos[0])), int(round(player_grid_pos[1])) self.grid.record_visit(pgx, pgy, radius=2) # 2. Cắt tỉa các ngõ cụt đã dọn sạch (Anti-backtracking) pruned_count = self.grid.prune_dead_ends() # 3. Trích xuất các ô biên Frontier và gom cụm frontiers = self.grid.extract_frontier_cells() clusters = self.grid.cluster_frontiers(frontiers) # 4. Kiểm tra độ bao phủ toàn bản đồ coverage = self.grid.get_coverage_ratio() is_complete = (coverage >= self.target_coverage) or (len(clusters) == 0) # 5. Lựa chọn Frontier tối ưu best_selection = self.select_best_frontier( clusters, player_grid_xy=player_grid_pos, player_vel=player_velocity ) # Kích thước ảnh Minimap if isinstance(minimap_img, Image.Image): mi_w, mi_h = minimap_img.size else: mi_h, mi_w = minimap_img.shape[:2] if best_selection is not None and not is_complete: best_cl, score = best_selection tgt_gx = int(round(best_cl.center_grid[0])) tgt_gy = int(round(best_cl.center_grid[1])) # Ánh xạ từ tọa độ lưới sang pixel trên ảnh Minimap mm_x = int(round((tgt_gx / float(self.grid.gw)) * mi_w)) mm_y = int(round((tgt_gy / float(self.grid.gh)) * mi_h)) # Tính vector hướng di chuyển từ player đến target p_center_x = mi_w * 0.5 p_center_y = mi_h * 0.5 dir_x = float(mm_x - p_center_x) dir_y = float(mm_y - p_center_y) d_norm = math.hypot(dir_x, dir_y) if d_norm > 1e-5: direction_vector = (dir_x / d_norm, dir_y / d_norm) else: direction_vector = (0.0, 0.0) else: # Không còn frontier hoặc đã dọn xong tgt_gx, tgt_gy = pgx, pgy mm_x, mm_y = int(mi_w * 0.5), int(mi_h * 0.5) direction_vector = (0.0, 0.0) score = 0.0 # Ánh xạ từ Minimap sang Screen Coordinates tuyệt đối bx1, by1, bx2, by2 = self.minimap_screen_bbox bw = max(1, bx2 - bx1) bh = max(1, by2 - by1) screen_x = int(round(bx1 + (mm_x / float(mi_w)) * bw)) screen_y = int(round(by1 + (mm_y / float(mi_h)) * bh)) return ExplorationTarget( target_grid_xy=(tgt_gx, tgt_gy), target_minimap_xy=(mm_x, mm_y), target_screen_xy=(screen_x, screen_y), direction_vector=direction_vector, utility_score=score, coverage_ratio=coverage, is_fully_explored=is_complete, active_clusters_count=len(clusters), dead_ends_pruned_count=pruned_count, )