"""AutoPOE2 - Dynamic Loot Value Density Clusterer (Doc 69 §3.4 / SSoT 2026). ============================================================================= Module gom cụm vật phẩm rơi theo mật độ giá trị kinh tế (Value Density Score - VDS): - Triệt tiêu 100% hiện tượng nhân vật đi giật lùi (Backtracking) để nhặt rác giá trị thấp: - Khi một vật phẩm nằm ngược hướng tiến trình bản đồ (cos θ < -0.30) và có mật độ giá trị thấp (< 1.5 Chaos/giây), thuật toán tự động gắn cờ REJECT_BACKTRACK. - Rào chắn bảo vệ vật phẩm tối thượng (Critical Item Interlock): Vật phẩm bậc SSS (Divine Orb, Mirror of Kalandra, Waystone T15+) được miễn trừ phạt ngược hướng, lập tức kích hoạt đường vòng nhặt an toàn. - Gom cụm không gian (Spatial Clustering): Gom các món đồ trong bán kính 35u thành 1 cụm để tối ưu hóa chuỗi nhặt đồ liên hoàn (Traveling Salesperson cục bộ). Tuân thủ nghiêm ngặt: - Rule 1: Dual-Track Operational Doctrine, Python 3.11 Tier 2. - Rule 4: Module hóa độc lập, trần file < 500 dòng. - Rule 5: 100% test case kiểm thử thực nghiệm trên taxonomy POE2 thật. - Bất biến: INV-WASD-NO-BLIND-LMB, INV-WAYSTONE-ROW1-ONLY. """ from __future__ import annotations from dataclasses import dataclass, field import math from typing import Any, Dict, List, Optional, Set, Tuple @dataclass class GroundItemInfo: """Thông tin một vật phẩm rơi trên mặt đất.""" entity_id: int name: str x: float y: float rarity: int = 0 # 0=Normal, 1=Magic, 2=Rare, 3=Unique, 4=Currency estimated_chaos_value: float = 0.0 is_critical: bool = False # True nếu là Divine, Mirror, T15+ Waystone @dataclass class LootClusterResult: """Kết quả phân tích một cụm vật phẩm trên sàn.""" cluster_id: int center_pos: Tuple[float, float] items: List[GroundItemInfo] total_chaos_value: float has_critical_item: bool distance_to_player: float detour_angle_deg: float is_backtracking: bool kinetic_cost_sec: float value_density_score: float # Chaos / Giây should_pickup: bool reason: str = "" class DynamicLootDensityClusterer: """Động cơ gom cụm vật phẩm và đánh giá mật độ kinh tế chống quay đầu.""" # Bảng giá trị cơ sở tạm tính theo Chaos Orb (Taxonomy POE2 v0.5.5+) CANONICAL_VALUE_TABLE: Dict[str, float] = { "mirror of kalandra": 50000.0, "divine orb": 200.0, "greater jeweller's orb": 45.0, "perfect jeweller's orb": 180.0, "exalted orb": 20.0, "chaos orb": 1.0, "orb of alchemy": 0.20, "regal orb": 0.50, "orb of chance": 0.05, "scroll of wisdom": 0.001, "portal scroll": 0.002, "orb of transmutation": 0.01, "orb of augmentation": 0.02, "blacksmith's whetstone": 0.02, "armourer's scrap": 0.02, "waystone (tier 15)": 15.0, "waystone (tier 16)": 25.0, "waystone (tier 1)": 0.10, } def __init__( self, cluster_radius: float = 35.0, player_speed: float = 38.0, pickup_dwell_sec: float = 0.08, min_vds_threshold: float = 0.50, backtrack_penalty_factor: float = 2.5, min_backtrack_chaos: float = 2.0, ) -> None: """Khởi tạo bộ phân cụm giá trị chiến lợi phẩm. Args: cluster_radius: Bán kính gộp các món đồ vào cùng cụm (units). player_speed: Tốc độ di chuyển ước tính của nhân vật (units/sec). pickup_dwell_sec: Thời gian trễ mỗi thao tác nhặt (giây). min_vds_threshold: Ngưỡng mật độ giá trị tối thiểu (Chaos/giây) để chấp nhận nhặt. backtrack_penalty_factor: Hệ số phạt thời gian khi di chuyển ngược hướng map. min_backtrack_chaos: Giá trị tối thiểu để cho phép quay đầu nhặt đồ. """ self.cluster_radius = cluster_radius self.player_speed = player_speed self.pickup_dwell_sec = pickup_dwell_sec self.min_vds = min_vds_threshold self.backtrack_penalty = backtrack_penalty_factor self.min_backtrack_chaos = min_backtrack_chaos @classmethod def estimate_item_value(cls, item_name: str, rarity: int = 0) -> Tuple[float, bool]: """Ước tính giá trị kinh tế của vật phẩm theo Chaos Orb. Returns: Tuple[float, bool]: (Giá trị Chaos, True nếu là Critical Item). """ lower = item_name.lower().strip() for k, v in cls.CANONICAL_VALUE_TABLE.items(): if k in lower: is_crit = v >= 15.0 return v, is_crit # Đồ Unique hoặc Waystone nói chung if rarity == 3: return 2.0, False if "waystone" in lower: return 1.0, False if rarity == 2: # Rare gear return 0.10, False return 0.01, False def cluster_and_evaluate( self, ground_items: List[GroundItemInfo], player_pos: Tuple[float, float], map_progression_dir: Tuple[float, float], ) -> List[LootClusterResult]: """Gom cụm danh sách vật phẩm và ra quyết định nhặt hay bỏ qua. Args: ground_items: Danh sách vật phẩm đang hiển thị trên sàn đấu. player_pos: Tọa độ (X, Y) thế giới hiện tại của người chơi. map_progression_dir: Vector định hướng tiến trình của map (dx, dy). Returns: List[LootClusterResult]: Danh sách các cụm vật phẩm kèm điểm số VDS. """ if not ground_items: return [] # 1. Gom cụm không gian (Single-Linkage Clustering theo bán kính) clusters: List[List[GroundItemInfo]] = [] for item in ground_items: assigned = False for cl in clusters: # Kiểm tra khoảng cách với tâm tạm thời của cụm cx = sum(it.x for it in cl) / len(cl) cy = sum(it.y for it in cl) / len(cl) if math.hypot(item.x - cx, item.y - cy) <= self.cluster_radius: cl.append(item) assigned = True break if not assigned: clusters.append([item]) # Chuẩn hóa vector tiến trình px, py = player_pos p_dx, p_dy = map_progression_dir p_norm = math.hypot(p_dx, p_dy) if p_norm > 1e-4: p_dx /= p_norm p_dy /= p_norm else: p_dx, p_dy = 1.0, 0.0 results: List[LootClusterResult] = [] # 2. Đánh giá từng cụm for c_id, cl_items in enumerate(clusters, start=1): cl_x = sum(it.x for it in cl_items) / len(cl_items) cl_y = sum(it.y for it in cl_items) / len(cl_items) total_val = sum(it.estimated_chaos_value for it in cl_items) has_crit = any(it.is_critical for it in cl_items) # Tính vector từ người chơi tới cụm vx = cl_x - px vy = cl_y - py dist = math.hypot(vx, vy) # Tính góc lệch với hướng tiến trình (Detour Angle) if dist > 1e-4: ux = vx / dist uy = vy / dist cos_theta = ux * p_dx + uy * p_dy cos_theta = max(-1.0, min(1.0, cos_theta)) angle_deg = math.degrees(math.acos(cos_theta)) else: angle_deg = 0.0 cos_theta = 1.0 # Cụm được coi là đi ngược đường nếu cos_theta < -0.30 (lệch > 107 độ) is_backtracking = cos_theta < -0.30 # Tính chi phí thời gian động lực học (Kinetic Cost) travel_time = dist / max(1.0, self.player_speed) pickup_time = len(cl_items) * self.pickup_dwell_sec kinetic_cost = travel_time + pickup_time if is_backtracking and not has_crit: # Phạt nặng thời gian khi phải quay đầu nhặt đồ kinetic_cost *= self.backtrack_penalty # Điểm mật độ giá trị VDS = Chaos / Giây vds = total_val / max(0.1, kinetic_cost) # 3. Ra quyết định nhặt hay bỏ qua should_pickup = False reason = "" if has_crit: # BẬT TỰ ĐỘNG: Đồ tối thượng (Divine/Mirror/T15+) luôn được nhặt bất kể góc lệch should_pickup = True reason = f"Chứa vật phẩm tối thượng (Giá trị cụm: {total_val:.1f} Chaos)" elif is_backtracking: if total_val >= self.min_backtrack_chaos and vds >= self.min_vds: should_pickup = True reason = f"Đi ngược đường hợp lệ (Giá trị {total_val:.1f}C >= {self.min_backtrack_chaos}C, VDS={vds:.2f} C/s)" else: should_pickup = False reason = f"BỎ QUA CHỐNG QUAY ĐẦU: Góc lệch {angle_deg:.0f}° ngược hướng, giá trị {total_val:.2f}C < ngưỡng {self.min_backtrack_chaos}C" else: # Cùng hướng tiến trình hoặc góc lệch nhỏ if vds >= self.min_vds or total_val >= 0.20: should_pickup = True reason = f"Cùng hướng tiến trình (VDS={vds:.2f} C/s, Góc={angle_deg:.0f}°)" else: should_pickup = False reason = f"Bỏ qua rác giá trị quá thấp (VDS={vds:.2f} C/s)" results.append( LootClusterResult( cluster_id=c_id, center_pos=(cl_x, cl_y), items=cl_items, total_chaos_value=total_val, has_critical_item=has_crit, distance_to_player=dist, detour_angle_deg=angle_deg, is_backtracking=is_backtracking, kinetic_cost_sec=kinetic_cost, value_density_score=vds, should_pickup=should_pickup, reason=reason, ) ) # Sắp xếp các cụm nên nhặt theo VDS giảm dần results.sort(key=lambda r: (r.has_critical_item, r.value_density_score), reverse=True) return results def select_best_cluster(self, clusters: List[LootClusterResult]) -> Optional[LootClusterResult]: """Chọn cụm tối ưu nhất để nhặt trong nhịp hiện tại.""" for cl in clusters: if cl.should_pickup: return cl return None