"""AutoPOE2 - Atlas Graph Data Structure (Doc 56 / Doc 65 / Doc 69). =================================================================== Mô hình hóa toàn bộ cây bản đồ Atlas POE2 v0.5.5+ dưới dạng Đồ thị Vô hướng G = (V, E). Quản lý trạng thái tiến trình (LOCKED -> REACHABLE_FRONTIER -> COMPLETED / BONUS_COMPLETED), liên kết kề cận, phân vùng và mở rộng biên (Frontier Progression). Tuân thủ nghiêm ngặt: - Rule 1: Cold Path Tier 2 (Python 3.11). - Rule 4: Module hóa, trần file < 500 dòng. - Rule 8: Bất biến kiến trúc INV-ATLAS-FARM-NODE & INV-ATLAS-QUEST-PROG. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Set, Tuple from src.assistant_tool.atlas_knowledge_base import AtlasKnowledgeBase, AtlasNodeType, AtlasNodeProfile class NodeProgressionState(str, Enum): """Trạng thái tiến trình mở khóa của Node trên Atlas.""" LOCKED = "LOCKED" # Nằm trong sương mù hoặc chưa có đường nối REACHABLE_FRONTIER = "REACHABLE_FRONTIER" # Tiếp giáp với node đã hoàn thành, sẵn sàng kích hoạt mở khóa COMPLETED = "COMPLETED" # Đã hoàn thành map nhưng chưa đạt điều kiện thưởng Bonus BONUS_COMPLETED = "BONUS_COMPLETED" # Đã hoàn thành và nhận +1 Điểm Nội Tại Atlas @dataclass class AtlasGraphNode: """Đỉnh của Đồ thị Atlas đại diện cho một bản đồ hoặc địa điểm đặc biệt.""" node_id: str name: str tier: int region: str = "Central" screen_x: float = 960.0 screen_y: float = 540.0 node_type: AtlasNodeType = AtlasNodeType.REGULAR_WAYSTONE_MAP progression_state: NodeProgressionState = NodeProgressionState.LOCKED encounters: List[str] = field(default_factory=list) # Breach, Ritual, Delirium, Expedition tower_coverage_count: int = 0 # Số Precursor Tower bao phủ neighbors: Set[str] = field(default_factory=set) # Danh sách node_id kết nối trực tiếp bonus_condition: str = "Complete with Rare Waystone" atlas_points_reward: int = 1 weight_modifier: float = 1.0 # Hệ số ưu tiên đường đi @property def is_completed(self) -> bool: return self.progression_state in ( NodeProgressionState.COMPLETED, NodeProgressionState.BONUS_COMPLETED, ) @property def is_bonus_completed(self) -> bool: return self.progression_state == NodeProgressionState.BONUS_COMPLETED @property def is_frontier(self) -> bool: return self.progression_state == NodeProgressionState.REACHABLE_FRONTIER def calculate_weight( self, preferred_maps: Optional[List[str]] = None, ignored_maps: Optional[List[str]] = None, ) -> float: """Tính điểm trọng số ưu tiên lựa chọn bản đồ cho Monk CI Tier 15 (Doc 56 / Doc 70 / INV-ATLAS-FARM-NODE).""" return calculate_node_weight(self, preferred_maps=preferred_maps, ignored_maps=ignored_maps) def calculate_node_weight( node: AtlasGraphNode, preferred_maps: Optional[List[str]] = None, ignored_maps: Optional[List[str]] = None, ) -> float: """Tính điểm trọng số ưu tiên lựa chọn bản đồ cho Monk CI Tier 15 (Doc 56 / Doc 70 / INV-ATLAS-FARM-NODE). Quy chuẩn trọng số SSoT: 1. Node bị cấm (The Burning Monolith / Arbiter of Ash / Pinnacle Arenas / không nhận Waystone): Trả về -1000.0 (Fail-Closed, tuyệt đối cấm). 2. Danh sách đen (ignored_maps: Vaal Pyramid, Overgrown Ruin, Maze...): Trả về -500.0. 3. Trật tự ưu tiên layout mở an toàn cho Monk CI (Deserted > Fortress > Mineshaft > Creek > Marsh > Dunes): - Deserted: +100.0 - Fortress: +80.0 - Mineshaft: +60.0 - Creek: +40.0 - Marsh: +25.0 - Dunes: +15.0 - Các map khác trong preferred_maps: +10.0 4. Trọng số Tier: - Tier 15: +30.0 (Tier vàng cho Monk CI Endgame) - Tier 14: +20.0 - Tier 16: +10.0 - Tier khác: tier * 1.0 5. Cơ chế có giá trị (Breach, Delirium, Ritual, Expedition): +5.0 mỗi cơ chế. 6. Tháp Tiền Thân bao phủ (tower_coverage_count): +8.0 mỗi tháp. 7. Điểm thưởng nội tại Atlas (atlas_points_reward > 1 hoặc Quest Node): +20.0 incentive. 8. Node chưa lấy bonus (is_bonus_completed == False): +15.0 incentive. """ # 1. Chốt chặn 2 tầng bất biến INV-ATLAS-FARM-NODE: Cấm tuyệt đối The Burning Monolith / Arbiter of Ash name_upper = node.name.upper().strip() if ( node.node_type == AtlasNodeType.BURNING_MONOLITH or "BURNING MONOLITH" in name_upper or "ARBITER OF ASH" in name_upper or AtlasKnowledgeBase.is_forbidden_for_waystone_farming(node.name) ): return -1000.0 # 2. Danh sách đen bản đồ nguy hiểm if ignored_maps: for ign in ignored_maps: if ign.lower() in node.name.lower(): return -500.0 # 3. Trật tự ưu tiên map mở an toàn cho Monk CI Tier 15 score = 0.0 name_lower = node.name.lower() if "deserted" in name_lower: score += 100.0 elif "fortress" in name_lower: score += 80.0 elif "stronghold" in name_lower: score += 75.0 elif "crypt" in name_lower: score += 70.0 elif "mineshaft" in name_lower: score += 60.0 elif "creek" in name_lower: score += 40.0 elif "marsh" in name_lower: score += 25.0 elif "dunes" in name_lower: score += 15.0 elif preferred_maps and any(p.lower() in name_lower for p in preferred_maps): score += 10.0 # 4. Trọng số Tier if node.tier == 15: score += 30.0 elif node.tier == 14: score += 20.0 elif node.tier == 16: score += 10.0 else: score += float(node.tier) # 5. Cơ chế có giá trị (Breach, Delirium, Ritual, Expedition) score += len(node.encounters) * 5.0 # 6. Tháp Tiền Thân bao phủ score += node.tower_coverage_count * 8.0 # 7. Điểm thưởng nội tại Atlas (Quest Node hoặc Unique Map) if node.atlas_points_reward > 1 or node.node_type.name.startswith("QUEST_"): score += 20.0 # 8. Node chưa lấy bonus if not node.is_bonus_completed: score += 15.0 return round(score * node.weight_modifier, 2) class AtlasGraph: """ Đồ thị Atlas G = (V, E) quản trị tiến trình mở khóa toàn bộ cây bản đồ (Full Node Progression). """ def __init__(self): self._nodes: Dict[str, AtlasGraphNode] = {} self._gateway_nodes: Set[str] = set() @property def total_nodes(self) -> int: return len(self._nodes) def add_node(self, node: AtlasGraphNode) -> None: """Thêm một node vào đồ thị. Tự động liên kết hồ sơ tri thức từ AtlasKnowledgeBase.""" profile = AtlasKnowledgeBase.classify_node(node.name) if node.node_type == AtlasNodeType.REGULAR_WAYSTONE_MAP and profile.node_type != AtlasNodeType.UNKNOWN: node.node_type = profile.node_type node.atlas_points_reward = profile.atlas_points_reward if profile.atlas_points_reward > 0 else 1 self._nodes[node.node_id] = node if node.node_type == AtlasNodeType.GATEWAY_CHECKPOINT or node.tier == 1: self._gateway_nodes.add(node.node_id) def add_edge(self, u_id: str, v_id: str, bidirectional: bool = True) -> None: """Thiết lập cạnh kết nối giữa hai node.""" if u_id not in self._nodes or v_id not in self._nodes: raise KeyError(f"Không thể tạo cạnh giữa '{u_id}' và '{v_id}': Node không tồn tại trong đồ thị.") self._nodes[u_id].neighbors.add(v_id) if bidirectional: self._nodes[v_id].neighbors.add(u_id) def get_node(self, node_id: str) -> Optional[AtlasGraphNode]: return self._nodes.get(node_id) def get_neighbors(self, node_id: str) -> List[AtlasGraphNode]: node = self.get_node(node_id) if not node: return [] return [self._nodes[nbr_id] for nbr_id in node.neighbors if nbr_id in self._nodes] def mark_node_completed(self, node_id: str, bonus: bool = True) -> None: """ Đánh dấu hoàn thành một node và kích hoạt cơ chế Mở rộng biên (Frontier Expansion): Mọi node lân cận đang ở trạng thái LOCKED sẽ tự động được thăng hạng thành REACHABLE_FRONTIER. """ node = self.get_node(node_id) if not node: return node.progression_state = ( NodeProgressionState.BONUS_COMPLETED if bonus else NodeProgressionState.COMPLETED ) # Mở rộng biên: Thăng cấp hàng xóm LOCKED -> REACHABLE_FRONTIER for nbr_id in node.neighbors: nbr = self.get_node(nbr_id) if nbr and nbr.progression_state == NodeProgressionState.LOCKED: nbr.progression_state = NodeProgressionState.REACHABLE_FRONTIER def get_frontier_nodes(self) -> List[AtlasGraphNode]: """Lấy toàn bộ các node biên sẵn sàng được cày để mở rộng bản đồ.""" return [n for n in self._nodes.values() if n.is_frontier] def get_uncompleted_nodes(self) -> List[AtlasGraphNode]: """Lấy toàn bộ các node chưa hoàn thành hoặc chưa đạt điểm thưởng bonus.""" return [n for n in self._nodes.values() if not n.is_bonus_completed] def get_completed_nodes(self) -> List[AtlasGraphNode]: """Lấy toàn bộ các node đã hoàn thành ít nhất 1 lần.""" return [n for n in self._nodes.values() if n.is_completed] def get_nodes_by_tier(self, tier: int) -> List[AtlasGraphNode]: return [n for n in self._nodes.values() if n.tier == tier] def get_completion_stats(self) -> Dict[str, Any]: """Tính toán các chỉ số định lượng về tiến trình mở khóa toàn bộ Atlas.""" total = len(self._nodes) if total == 0: return { "total_nodes": 0, "completed_count": 0, "bonus_completed_count": 0, "frontier_count": 0, "locked_count": 0, "completion_rate": 0.0, "bonus_completion_rate": 0.0, "total_atlas_passive_points": 0, } completed = sum(1 for n in self._nodes.values() if n.is_completed) bonus_comp = sum(1 for n in self._nodes.values() if n.is_bonus_completed) frontier = sum(1 for n in self._nodes.values() if n.is_frontier) locked = sum(1 for n in self._nodes.values() if n.progression_state == NodeProgressionState.LOCKED) points = sum(n.atlas_points_reward for n in self._nodes.values() if n.is_bonus_completed) return { "total_nodes": total, "completed_count": completed, "bonus_completed_count": bonus_comp, "frontier_count": frontier, "locked_count": locked, "completion_rate": round((completed / total) * 100.0, 1), "bonus_completion_rate": round((bonus_comp / total) * 100.0, 1), "total_atlas_passive_points": points, } def export_state_dict(self) -> Dict[str, Any]: """Xuất trạng thái đồ thị để lưu trữ hoặc phục vụ telemetry.""" return { "nodes": { node_id: { "name": n.name, "tier": n.tier, "state": n.progression_state.value, "neighbors": list(n.neighbors), "encounters": n.encounters, "tower_coverage": n.tower_coverage_count, "screen_x": n.screen_x, "screen_y": n.screen_y, } for node_id, n in self._nodes.items() }, "stats": self.get_completion_stats(), } def load_state_dict(self, data: Dict[str, Any]) -> None: """Nạp trạng thái đồ thị từ dict (phục hồi phiên làm việc).""" nodes_data = data.get("nodes", {}) for node_id, n_info in nodes_data.items(): if node_id in self._nodes: self._nodes[node_id].progression_state = NodeProgressionState( n_info.get("state", NodeProgressionState.LOCKED.value) ) self._nodes[node_id].encounters = n_info.get("encounters", []) self._nodes[node_id].tower_coverage_count = int(n_info.get("tower_coverage", 0)) @classmethod def create_canonical_atlas_graph(cls, total_target_nodes: int = 115) -> "AtlasGraph": """ Khởi tạo cây bản đồ Atlas POE2 v0.5.5+ chuẩn mực với 115+ nodes: - Phân bổ từ Tier 1 tới Tier 16 trên các vùng lãnh thổ (Regions). - Tích hợp các Quest Nodes (+1 Point), Unique Maps (+2 Points), Citadels và Special Encounters. - Node khởi đầu (Beach T1) mặc định BONUS_COMPLETED; các node kề cận tự động là REACHABLE_FRONTIER. """ graph = cls() regions = ["Oasis", "Tundra", "Jungle", "Volcano", "Citadel Core"] base_map_names = [ "Beach", "Dunes", "Graveyard", "Lookout", "Arid Lake", "Flooded Strand", "Marshes", "Peninsula", "Port", "Strand", "Volcano", "Canyon", "Chasm", "Deserted", "Fortress", "Creek", "Sandswept Marsh", "Riverbank", "Barrows", "Bog", "Burial Chambers", "Cemetery", "Coral Ruins", "Crater", "Crypt", "Excavation", "Geode", "Glacier", "Iceberg", "Infested Valley", "Laboratory", "Lava Lake", "Lair", "Mausoleum", "Mesa", "Mud Geyser", "Orchard", "Overgrown Shrine", "Phantasmagoria", "Plaza", "Racecourse", "Ramparts", "Reef", "Shore", "Sunken City", "Temple", "Thicket", "Underground River", "Vaal Pyramid", "Vault", "Waste Pool", "Waterways", "Wharf", "Ashen Wood", "Atoll", "Belfry", "Bramble Valley", "Caldera", "Carcass", "Castle Ruins", "Chateau", "Colosseum", "Colonnade", "Core", "Coves", "Crimson Temple", "Dark Forest", "Defiled Cathedral", "Desert Spring", "Dig", "Dungeon", "Estuary", "Fields", "Foundry", "Gardens", "Ghetto", "Grotto", "Haunted Mansion", "Ivory Temple", "Jungle Valley", "Lava Chamber", "Leyline", "Lighthouse", "Malformation", "Mineral Pools", "Moon Temple", "Museum", "Necropolis", "Park", "Pillar", "Pit", "Plateau", "Primordial Blocks", "Promenade", "Relic Chambers", "Residence", "Scrip", "Sepulchre", "Shrine", "Siege", "Silo", "Spider Forest", "Spider Lair", "Stagnation", "Sulphur Vents", "Summit", "Terrace", "Tower", "Toxic Sewer", "Underground Sea", "Vaal Temple", "Villa", "Wand", "Forbidden Woods", ] # 1. Tạo các node bản đồ cơ bản (100 nodes thường) created_count = 0 node_ids_by_tier: Dict[int, List[str]] = {t: [] for t in range(1, 17)} for idx, name in enumerate(base_map_names): if created_count >= total_target_nodes - 15: break # Tính Tier tăng dần từ 1 đến 15 tier = min(15, max(1, (idx // 7) + 1)) region = regions[idx % len(regions)] node_id = f"node_{name.lower().replace(' ', '_')}_{tier}" encounters = [] if idx % 9 == 0: encounters.append("Breach") elif idx % 9 == 3: encounters.append("Delirium") elif idx % 9 == 6: encounters.append("Ritual") elif idx % 11 == 0: encounters.append("Expedition") coverage = 1 if (idx % 8 == 0) else 0 # Node Beach T1 đầu tiên là điểm xuất phát initial_state = NodeProgressionState.BONUS_COMPLETED if idx == 0 else NodeProgressionState.LOCKED node = AtlasGraphNode( node_id=node_id, name=name, tier=tier, region=region, progression_state=initial_state, encounters=encounters, tower_coverage_count=coverage, ) graph.add_node(node) node_ids_by_tier[tier].append(node_id) created_count += 1 # 2. Bổ sung 5 Atlas Quest Nodes (+1 Point) quest_specs = [ ("node_enigma_chamber", "Enigma Chamber", 4, AtlasNodeType.QUEST_ENIGMA_CHAMBER), ("node_corrupted_nexus", "Corrupted Nexus", 8, AtlasNodeType.QUEST_CORRUPTED_NEXUS), ("node_great_beast_lair", "Great Beast Lair", 11, AtlasNodeType.QUEST_GREAT_BEAST), ("node_withered_willow", "The Withered Willow", 13, AtlasNodeType.QUEST_WITHERED_WILLOW), ("node_kingsmarch_ruins", "Ruins of Kingsmarch", 14, AtlasNodeType.QUEST_KINGSMARCH_RUINS), ] for q_id, q_name, q_tier, q_type in quest_specs: q_node = AtlasGraphNode( node_id=q_id, name=q_name, tier=q_tier, region="Central Quest", progression_state=NodeProgressionState.LOCKED, node_type=q_type, atlas_points_reward=1, ) graph.add_node(q_node) node_ids_by_tier[q_tier].append(q_id) # 3. Bổ sung 5 Unique Maps (+2 Points) unique_specs = [ ("node_untainted_paradise", "Untainted Paradise", 7), ("node_moment_of_zen", "Moment of Zen", 9), ("node_vaults_of_kamasa", "Vaults of Kamasa", 10), ("node_the_fractured_lake", "The Fractured Lake", 12), ("node_the_silent_cave", "The Silent Cave", 14), ] for u_id, u_name, u_tier in unique_specs: u_node = AtlasGraphNode( node_id=u_id, name=u_name, tier=u_tier, region="Unique Cluster", progression_state=NodeProgressionState.LOCKED, node_type=AtlasNodeType.UNIQUE_MAP, atlas_points_reward=2, ) graph.add_node(u_node) node_ids_by_tier[u_tier].append(u_id) # 4. Bổ sung 5 Citadels & Pinnacle Nodes (Tier 16) pinnacle_specs = [ ("node_iron_citadel", "Iron Citadel", 16, AtlasNodeType.CITADEL_BOSS), ("node_copper_citadel", "Copper Citadel", 16, AtlasNodeType.CITADEL_BOSS), ("node_stone_citadel", "Stone Citadel", 16, AtlasNodeType.CITADEL_BOSS), ("node_burning_monolith", "The Burning Monolith", 16, AtlasNodeType.BURNING_MONOLITH), ("node_origin_tower", "Origin Tower", 16, AtlasNodeType.ORIGIN_TOWER), ] for p_id, p_name, p_tier, p_type in pinnacle_specs: p_node = AtlasGraphNode( node_id=p_id, name=p_name, tier=p_tier, region="Citadel Core", progression_state=NodeProgressionState.LOCKED, node_type=p_type, ) graph.add_node(p_node) node_ids_by_tier[16].append(p_id) # 5. Tạo mạng lưới cạnh kết nối (Edges) liên kết các tier # - Nối các node trong cùng tier thành chuỗi/vòng for t in range(1, 17): t_nodes = node_ids_by_tier[t] for i in range(len(t_nodes) - 1): graph.add_edge(t_nodes[i], t_nodes[i + 1]) # - Nối giữa các Tier liên tiếp (Tier t -> Tier t + 1) for t in range(1, 16): curr_nodes = node_ids_by_tier[t] next_nodes = node_ids_by_tier[t + 1] for i, c_id in enumerate(curr_nodes): if next_nodes: target_next = next_nodes[i % len(next_nodes)] graph.add_edge(c_id, target_next) # Tự động cập nhật Frontier cho node xuất phát (Beach) first_node = list(graph._nodes.values())[0] graph.mark_node_completed(first_node.node_id, bonus=True) return graph