"""AutoPOE2 - Multimodal Telemetry Correlator (Doc 50 / Doc 61 / Rule 2 / Rule 5). ================================================================================ Module chuyên trách hợp nhất đa phương thức (Multimodal Fusion): 1. Chụp/đọc Screenshot hiện trường (.png) + phân tích quang học (WinOCR & Color Clustering). 2. Đọc RAM snapshot thô (.json). 3. Trích xuất tail log game Client.txt (khu vực hiện tại, loading screen, chat). 4. Đối soát mã bất biến SSoT (INV-CI-01, INV-FSM-TOWN-NO-EXPLORE, INV-KEY-*). 5. Xuất ra cấu trúc JSON duy nhất, chuẩn mực cho Orchestrator và Sub-Agents. """ from __future__ import annotations import glob import json import os import re import sys try: if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") except Exception: pass from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) from PIL import Image import src.common as common class MultimodalTelemetryCorrelator: """Hợp nhất và tương quan dữ liệu Screenshot + RAM + Client.txt thành JSON chuẩn.""" def __init__( self, client_log_path: str = r"C:\Program Files (x86)\Grinding Gear Games\Path of Exile 2\logs\Client.txt", captures_dir: str = "captures", ): self.client_log_path = client_log_path self.captures_dir = captures_dir def get_latest_capture_pair(self) -> Tuple[Optional[str], Optional[str]]: """Tìm cặp ảnh screenshot và RAM snapshot mới nhất trong thư mục captures.""" png_files = sorted(glob.glob(os.path.join(self.captures_dir, "*.png")), key=os.path.getmtime, reverse=True) valid_pngs = [ f for f in png_files if not f.endswith("_thumb.png") and not f.endswith("_view.png") and not f.endswith("_CORRELATED.png") ] if not valid_pngs: return None, None latest_png = valid_pngs[0] base_no_ext = os.path.splitext(latest_png)[0] ram_json = f"{base_no_ext}_RAM.json" if not os.path.exists(ram_json): all_rams = sorted(glob.glob(os.path.join(self.captures_dir, "*_RAM.json")), key=os.path.getmtime, reverse=True) ram_json = all_rams[0] if all_rams else None return latest_png, ram_json def extract_recent_game_log(self, max_lines: int = 500) -> Dict[str, Any]: """Đọc và trích xuất thông tin trọng yếu từ Client.txt.""" info = { "current_zone": "Unknown", "is_hideout": False, "is_map": False, "last_loading_duration": 0.0, "last_log_timestamp": "", "recent_events": [], } if not os.path.exists(self.client_log_path): return info try: with open(self.client_log_path, "r", encoding="utf-8", errors="ignore") as f: lines = f.readlines()[-max_lines:] zone_pattern = re.compile(r"\[SCENE\] Set Source \[(.*?)\]") load_pattern = re.compile(r"\[LOADING SCREEN\] \((.*?)\) Duration = ([\d\.]+) seconds") for line in lines: m_load = load_pattern.search(line) if m_load: info["current_zone"] = m_load.group(1).strip() info["last_loading_duration"] = float(m_load.group(2)) m_zone = zone_pattern.search(line) if m_zone and m_zone.group(1).strip() not in ("(null)", ""): info["current_zone"] = m_zone.group(1).strip() current_lower = info["current_zone"].lower() info["is_hideout"] = "hideout" in current_lower info["is_map"] = ("map" in current_lower or "chamber" in current_lower or "ruins" in current_lower) and not info["is_hideout"] if lines: info["last_log_timestamp"] = lines[-1][:19] if len(lines[-1]) >= 19 else "" info["recent_events"] = [l.strip() for l in lines[-10:] if "Set Source" in l or "LOADING" in l or "Connect" in l] except Exception as e: info["error"] = str(e) return info def parse_ram_snapshot(self, ram_json_path: Optional[str]) -> Dict[str, Any]: """Phân tích nội dung RAM Snapshot thô.""" vitals = { "hp": 0, "max_hp": 0, "es": 0, "max_es": 0, "mana": 0, "max_mana": 0, "pos": [0.0, 0.0, 0.0], "alive": False, "is_ci": False, "monsters_count": 0, "items_count": 0, "portals_count": 0, "portals": [], } if ram_json_path and os.path.exists(ram_json_path): try: with open(ram_json_path, "r", encoding="utf-8") as f: data = json.load(f) player = data.get("player", {}) vitals["hp"] = player.get("hp", 0) vitals["max_hp"] = player.get("max_hp", 0) vitals["es"] = player.get("es", 0) vitals["max_es"] = player.get("max_es", 0) vitals["mana"] = player.get("mana", 0) vitals["max_mana"] = player.get("max_mana", 0) vitals["pos"] = [player.get("pos_x", 0.0), player.get("pos_y", 0.0), player.get("pos_z", 0.0)] vitals["monsters_count"] = len(data.get("monsters", [])) vitals["items_count"] = len(data.get("items", [])) vitals["portals"] = data.get("portals", []) vitals["portals_count"] = len(vitals["portals"]) if vitals["portals"] else data.get("portals_count", 0) vitals["is_ci"] = vitals["max_hp"] == 1 vitals["alive"] = (vitals["es"] > 0 if vitals["is_ci"] else vitals["hp"] > 0) except Exception as e: vitals["error"] = str(e) # Bóc tách thực thể tức thời từ CoreShmBridge (IPC Lock-free Seqlock) try: from src.assistant_tool.core_shm_bridge import CoreShmBridge bridge = CoreShmBridge() snap = bridge.read_telemetry_snapshot() if snap and snap.is_online: if snap.portals_count > 0: vitals["portals_count"] = max(vitals["portals_count"], snap.portals_count) for ent in snap.entities: if ent.get("type") == 3: vitals["portals"].append(ent) if snap.monsters_count > 0: vitals["monsters_count"] = max(vitals["monsters_count"], snap.monsters_count) if snap.max_hp > 0 or snap.max_es > 0: vitals["hp"] = snap.hp vitals["max_hp"] = snap.max_hp vitals["es"] = snap.es vitals["max_es"] = snap.max_es vitals["is_ci"] = (snap.max_hp == 1) vitals["alive"] = (snap.es > 0 if vitals["is_ci"] else snap.hp > 0) vitals["pos"] = [snap.pos_x, snap.pos_y, snap.pos_z] except Exception: pass return vitals def analyze_vision(self, img_path: Optional[str]) -> Dict[str, Any]: """Phân tích thị giác máy tính trên screenshot thật.""" vision = { "image_path": img_path, "resolution": "Unknown", "is_inventory_open": False, "is_atlas_open": False, "portal_detected": False, "portal_pos": None, "portal_label": None, "map_device_pos": None, } if not img_path or not os.path.exists(img_path): return vision try: img = Image.open(img_path) w, h = img.size vision["resolution"] = f"{w}x{h}" sx, sy = w / 2560.0, h / 1440.0 vision["is_inventory_open"] = common.is_inventory_open(img) vision["is_atlas_open"] = common.is_atlas_ui_open(img) vision["map_device_pos"] = common.find_map_device_screen_pos(img, scale_x=sx, scale_y=sy) labels = common.find_portal_label_by_ocr(img, scale_x=sx, scale_y=sy) if labels: vision["portal_detected"] = True vision["portal_pos"] = [labels[0][0], labels[0][1]] vision["portal_label"] = labels[0][2] else: wp_centers = [] try: import winocr ocr_res = winocr.recognize_pil_sync(img) for line in ocr_res.get("lines", []): if "WAYPOINT" in line.get("text", "").upper(): for w_info in line.get("words", []): if "WAYPOINT" in w_info.get("text", "").upper(): bx = w_info["bounding_rect"]["x"] + w_info["bounding_rect"]["width"] / 2 by = w_info["bounding_rect"]["y"] + w_info["bounding_rect"]["height"] / 2 wp_centers.append((bx, by)) except Exception: pass if not wp_centers: wp_centers.append((960 * sx, 650 * sy)) portal_pts = [] for y in range(int(h * 0.05), int(h * 0.55), 3): for x in range(int(w * 0.05), int(w * 0.95), 3): if any((x - wx) ** 2 + (y - wy) ** 2 < (280 * sx) ** 2 for wx, wy in wp_centers): continue r, g, b = img.getpixel((x, y))[:3] if b > 180 and b > r + 30 and b > g + 10: portal_pts.append((x, y)) if portal_pts: clusters = [] for pt in portal_pts: matched = False for c in clusters: if abs(pt[0] - c["cx"]) < 90 * sx and abs(pt[1] - c["cy"]) < 120 * sy: c["pts"].append(pt) c["cx"] = sum(p[0] for p in c["pts"]) // len(c["pts"]) c["cy"] = sum(p[1] for p in c["pts"]) // len(c["pts"]) matched = True break if not matched: clusters.append({"pts": [pt], "cx": pt[0], "cy": pt[1]}) valid = [c for c in clusters if len(c["pts"]) >= 30] if valid: best = max(valid, key=lambda c: len(c["pts"])) vision["portal_detected"] = True vision["portal_pos"] = [best["cx"], best["cy"]] vision["portal_label"] = "VORTEX_BLUE" except Exception as e: vision["error"] = str(e) return vision def correlate( self, img_path: Optional[str] = None, ram_path: Optional[str] = None, ) -> Dict[str, Any]: """Tạo báo cáo tổng hợp JSON toàn diện.""" if not img_path or not ram_path: latest_img, latest_ram = self.get_latest_capture_pair() img_path = img_path or latest_img ram_path = ram_path or latest_ram log_data = self.extract_recent_game_log() ram_data = self.parse_ram_snapshot(ram_path) vision_data = self.analyze_vision(img_path) # Hợp nhất quang học (Optical Vitals Fusion) khi RAM snapshot trống / zeroed if ram_data.get("max_hp", 0) == 0 and ram_data.get("es", 0) == 0 and img_path and os.path.exists(img_path): try: from src.assistant_tool.optical_hp_sensor import OpticalHPSensor sensor = OpticalHPSensor() raw_img = Image.open(img_path) v = sensor._detect_via_winocr(raw_img) if v and v.is_detected and (v.max_hp > 0 or v.max_shield > 0): ram_data["hp"] = v.current_hp ram_data["max_hp"] = v.max_hp ram_data["es"] = v.current_shield ram_data["max_es"] = v.max_shield ram_data["ward"] = v.current_ward ram_data["max_ward"] = v.max_ward ram_data["mana"] = v.current_mana ram_data["max_mana"] = v.max_mana ram_data["spirit"] = v.current_spirit ram_data["max_spirit"] = v.max_spirit ram_data["is_ci"] = (v.max_hp == 1) ram_data["alive"] = (v.current_shield > 0 if ram_data["is_ci"] else v.current_hp > 0) ram_data["vitals_source"] = "OPTICAL_WINOCR" except Exception as e: ram_data["optical_fusion_error"] = str(e) is_ci = ram_data.get("is_ci", False) es_val = ram_data.get("es", 0) hp_val = ram_data.get("hp", 0) max_hp_val = ram_data.get("max_hp", 0) if is_ci: ci_status = "PASSED" if (es_val > 0 and max_hp_val == 1) else "FAILED" elif max_hp_val > 0: ci_status = "PASSED" if hp_val > 0 else "FAILED" else: ci_status = "PENDING_TELEMETRY" invariants = { "INV-CI-01": ci_status, "INV-FSM-TOWN-NO-EXPLORE": "PASSED" if log_data.get("is_hideout") else "NOT_IN_HIDEOUT", "INV-INPUT-ABS-MOUSE": "COMPLIANT", "INV-KEY-NO-F-FOR-INTERACT": "COMPLIANT", } suggested_stage = 1 reason = "Mặc định từ đầu quy trình Stage 1." if log_data.get("is_map"): suggested_stage = 5 reason = f"Đang ở trong map '{log_data.get('current_zone')}': Tiếp tục Stage 5 In-Map Clear." elif log_data.get("is_hideout"): has_portal = vision_data.get("portal_detected") or ram_data.get("portals_count", 0) > 0 if has_portal: suggested_stage = 4 src_label = "RAM" if ram_data.get("portals_count", 0) > 0 else "Quang học (OCR/Vision)" pos_info = vision_data.get("portal_pos") or "Gần Map Device" reason = f"Ở Hideout và đã phát hiện Portal mở ({src_label} tại {pos_info}): Bỏ qua Stage 1-3, nhảy thẳng vào Stage 4!" elif vision_data.get("is_atlas_open"): suggested_stage = 3 reason = "Giao diện Atlas đang mở: Tiến hành Stage 3 chọn node và nạp waystone." elif vision_data.get("is_inventory_open"): suggested_stage = 1 reason = "Túi đồ đang mở: Hoàn tất Stage 1 quét waystone." else: suggested_stage = 1 reason = "Ở Hideout và chưa mở UI: Bắt đầu từ Stage 1." return { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "zone": log_data, "player": ram_data, "vision": vision_data, "invariants": invariants, "recommendation": { "suggested_stage": suggested_stage, "reason": reason, }, } def main(): correlator = MultimodalTelemetryCorrelator() report = correlator.correlate() print(json.dumps(report, indent=2, ensure_ascii=False)) if __name__ == "__main__": main()