"""AutoPOE2 - Monster Density Heatmap & Encounter Tracker (Doc 69 §3.4 / Stage 5). ================================================================================ Module quản trị bản đồ nhiệt mật độ quái vật và lịch sử giao tranh trong Map: - Phân tích không gian 2D theo các sector (100u x 100u) để theo dõi các bãi quái (Monster Packs). - Đo đạc số lượng quái vật đã chạm trán, số lượng đã tiêu diệt và số bãi quái độc lập (Pack Count). - Ngăn chặn triệt để hiện tượng thẩm định sai: Đánh 1 con quái rồi kết luận map đã dọn sạch. - Tích hợp với BossArenaDetector theo dõi tình trạng thanh máu Boss trên đỉnh màn hình. Tuân thủ nghiêm ngặt: - Rule 1: Python 3.11 Cold Path Tier 2. - Rule 4: Module hóa độc lập trần < 500 dòng. - Rule 16: Google Docstrings chuẩn mực. - Bất biến: INV-MAP-LEGITIMATE-CLEAR, INV-WASD-NO-BLIND-LMB. """ from __future__ import annotations from dataclasses import dataclass, field import math import time from typing import Any, Dict, List, Optional, Set, Tuple import numpy as np @dataclass class MonsterPackEncounter: """Bản ghi chạm trán một bãi quái vật trong bản đồ.""" pack_id: int center_pos: Tuple[float, float] initial_count: int current_count: int first_seen_time: float last_seen_time: float is_cleared: bool = False class MonsterDensityHeatmap: """Quản trị bản đồ nhiệt mật độ quái vật và lịch sử tuần tra Stage 5.""" def __init__( self, sector_size: float = 100.0, min_pack_monsters: int = 2, pack_merge_radius: float = 80.0, ) -> None: """Khởi tạo Monster Density Heatmap. Args: sector_size: Kích thước mỗi ô không gian sector (PoE units). min_pack_monsters: Số quái tối thiểu để coi là một bãi quái hợp lệ. pack_merge_radius: Bán kính gộp các quái vật vào cùng một bãi (units). """ self.sector_size = sector_size self.min_pack_monsters = min_pack_monsters self.pack_merge_radius = pack_merge_radius self.sectors: Dict[Tuple[int, int], int] = {} # (sec_x, sec_y) -> tổng số quái từng xuất hiện self.visited_sectors: Set[Tuple[int, int]] = set() # Các sector nhân vật đã tuần tra qua self.packs: List[MonsterPackEncounter] = [] self.total_monsters_encountered: int = 0 self.total_monsters_eliminated: int = 0 self.last_monster_count: int = 0 # Giám sát Boss self.boss_detected: bool = False self.boss_health_history: List[float] = [] self.boss_slain_optical: bool = False def reset(self) -> None: """Đặt lại bản đồ nhiệt cho phiên map mới.""" self.sectors.clear() self.visited_sectors.clear() self.packs.clear() self.total_monsters_encountered = 0 self.total_monsters_eliminated = 0 self.last_monster_count = 0 self.boss_detected = False self.boss_health_history.clear() self.boss_slain_optical = False def get_sector_coord(self, wx: float, wy: float) -> Tuple[int, int]: """Chuyển đổi tọa độ thế giới sang tọa độ sector.""" return (int(math.floor(wx / self.sector_size)), int(math.floor(wy / self.sector_size))) def record_telemetry_frame( self, player_x: float, player_y: float, monsters_near: int, now_sec: float = 0.0, ) -> None: """Ghi nhận dữ liệu vị trí người chơi và số lượng quái vật cận chiến. Args: player_x: Tọa độ X thế giới của người chơi. player_y: Tọa độ Y thế giới của người chơi. monsters_near: Số lượng quái vật sống quanh người chơi. now_sec: Timestamp hiện tại (giây). """ if now_sec <= 0.0: now_sec = time.time() if player_x != 0.0 or player_y != 0.0: p_sec = self.get_sector_coord(player_x, player_y) self.visited_sectors.add(p_sec) # Tính số quái vật bị tiêu diệt giữa các nhịp if monsters_near < self.last_monster_count: killed = self.last_monster_count - monsters_near self.total_monsters_eliminated += killed # Khi phát hiện quái vật xuất hiện if monsters_near >= self.min_pack_monsters and (player_x != 0.0 or player_y != 0.0): # Cập nhật sector heatmap p_sec = self.get_sector_coord(player_x, player_y) self.sectors[p_sec] = self.sectors.get(p_sec, 0) + monsters_near # Tìm xem có thuộc bãi quái nào đã biết gần đó không matched_pack: Optional[MonsterPackEncounter] = None for p in self.packs: dist = math.hypot(player_x - p.center_pos[0], player_y - p.center_pos[1]) if dist <= self.pack_merge_radius and not p.is_cleared: matched_pack = p break if matched_pack is not None: matched_pack.current_count = monsters_near matched_pack.last_seen_time = now_sec else: # Tạo bản ghi bãi quái mới new_pack = MonsterPackEncounter( pack_id=len(self.packs) + 1, center_pos=(player_x, player_y), initial_count=monsters_near, current_count=monsters_near, first_seen_time=now_sec, last_seen_time=now_sec, is_cleared=False, ) self.packs.append(new_pack) self.total_monsters_encountered += monsters_near # Đánh dấu bãi quái đã dọn sạch khi số quái quanh bãi tụt về 0 if monsters_near == 0 and (player_x != 0.0 or player_y != 0.0): for p in self.packs: if not p.is_cleared: dist = math.hypot(player_x - p.center_pos[0], player_y - p.center_pos[1]) if dist <= self.pack_merge_radius: p.current_count = 0 p.is_cleared = True self.last_monster_count = monsters_near def update_optical_boss_status( self, boss_bar_detected: bool, health_ratio: float = 1.0, is_phased: bool = False, ) -> None: """Cập nhật dữ liệu thị giác về thanh máu Boss từ BossArenaDetector. Args: boss_bar_detected: True nếu camera phát hiện thanh máu Boss trên đỉnh màn hình. health_ratio: Tỷ lệ máu Boss còn lại (0.0 .. 1.0). is_phased: True nếu Boss đang chuyển pha / dựng khiên bất tử. """ if boss_bar_detected: self.boss_detected = True self.boss_health_history.append(health_ratio) # Nếu thanh máu tụt xuống gần 0 (<= 0.02) if health_ratio <= 0.02: self.boss_slain_optical = True else: # Nếu trước đó đã thấy Boss và có lịch sử máu tụt thấp (< 0.15) rồi thanh máu biến mất if self.boss_detected and len(self.boss_health_history) >= 2: min_health = min(self.boss_health_history) if min_health <= 0.25: self.boss_slain_optical = True @property def total_packs_count(self) -> int: """Tổng số bãi quái độc lập đã chạm trán.""" return len(self.packs) @property def cleared_packs_count(self) -> int: """Số bãi quái đã bị dọn sạch.""" return sum(1 for p in self.packs if p.is_cleared) @property def visited_sector_count(self) -> int: """Số sector không gian nhân vật đã tuần tra qua.""" return len(self.visited_sectors) def is_clearing_legitimate(self, min_required_packs: int = 4) -> Tuple[bool, str]: """Thẩm định tính hợp lệ của quá trình dọn quái (kháng lỗi đánh 1 quái rồi dừng). Args: min_required_packs: Số bãi quái tối thiểu phải chạm trán (mặc định 4 bãi). Returns: Tuple[bool, str]: (is_legit, reason) """ # Nếu chưa gặp đủ số bãi quái tối thiểu if self.total_packs_count < min_required_packs: if self.total_monsters_eliminated >= 25: return ( True, f"Đã tiêu diệt {self.total_monsters_eliminated} quái vật (>= 25 quái qua {self.visited_sector_count} sectors)", ) return ( False, f"Chưa đạt yêu cầu tối thiểu số bãi quái: mới chạm trán {self.total_packs_count}/{min_required_packs} bãi " f"(Tổng tiêu diệt: {self.total_monsters_eliminated} quái)", ) # Nếu đã gặp >= min_required_packs và đã dọn phần lớn cleared = self.cleared_packs_count if cleared >= min_required_packs: return ( True, f"Đã dọn sạch {cleared}/{self.total_packs_count} bãi quái " f"(Tổng tiêu diệt: {self.total_monsters_eliminated} quái qua {self.visited_sector_count} sectors)" ) return ( True, f"Đã giao tranh với {self.total_packs_count} bãi quái " f"({cleared} bãi đã dọn, {self.total_monsters_eliminated} quái đã hạ)" )