""" AutoPOE2 - Autonomous Map Runner & Realtime Screenshot Sentinel (Doc 17/39) Điều phối tự động clear map, chụp ảnh màn hình định kỳ và kích hoạt khóa dừng khẩn cấp khi tử trận. """ import os import sys import time import json import ctypes import struct import subprocess import threading from datetime import datetime from pathlib import Path from ctypes import wintypes from typing import Optional, List, Dict, Any, Tuple from PIL import Image # Đảm bảo in UTF-8 không bị lỗi charmap trên Windows console if sys.stdout.encoding != 'utf-8': try: sys.stdout.reconfigure(encoding='utf-8', errors='replace') sys.stderr.reconfigure(encoding='utf-8', errors='replace') except Exception: pass # Import CoreShmBridge & MapClearEvaluator sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) try: from assistant_tool.core_shm_bridge import CoreShmBridge, CoreTelemetrySnapshot, CharacterLifecycleState except ImportError: CoreShmBridge = None CharacterLifecycleState = None try: from assistant_tool.navigation.map_clear_evaluator import MapClearEvaluator except ImportError: from src.assistant_tool.navigation.map_clear_evaluator import MapClearEvaluator class ScreenCapturerGDI: """Chụp ảnh màn hình siêu tốc bằng Windows GDI BitBlt (Hỗ trợ 1440p/4K không phụ thuộc Desktop session)""" def __init__(self): self.user32 = ctypes.windll.user32 self.gdi32 = ctypes.windll.gdi32 # Đảm bảo gắn vào Desktop Default qua hàm dùng chung đã cache try: from src.common.win32_window import attach_to_default_desktop attach_to_default_desktop() except Exception: pass def capture(self, output_path: str) -> bool: try: w = self.user32.GetSystemMetrics(0) h = self.user32.GetSystemMetrics(1) if w <= 0 or h <= 0: w, h = 2560, 1440 desk_dc = self.user32.GetDC(0) mem_dc = self.gdi32.CreateCompatibleDC(desk_dc) bmp = self.gdi32.CreateCompatibleBitmap(desk_dc, w, h) old_bmp = self.gdi32.SelectObject(mem_dc, bmp) SRCCOPY = 0x00CC0020 self.gdi32.BitBlt(mem_dc, 0, 0, w, h, desk_dc, 0, 0, SRCCOPY) class BITMAPINFOHEADER(ctypes.Structure): _fields_ = [ ('biSize', wintypes.DWORD), ('biWidth', wintypes.LONG), ('biHeight', wintypes.LONG), ('biPlanes', wintypes.WORD), ('biBitCount', wintypes.WORD), ('biCompression', wintypes.DWORD), ('biSizeImage', wintypes.DWORD), ('biXPelsPerMeter', wintypes.LONG), ('biYPelsPerMeter', wintypes.LONG), ('biClrUsed', wintypes.DWORD), ('biClrImportant', wintypes.DWORD) ] class BITMAPINFO(ctypes.Structure): _fields_ = [ ('bmiHeader', BITMAPINFOHEADER), ('bmiColors', wintypes.DWORD * 3) ] bmi = BITMAPINFO() bmi.bmiHeader.biSize = ctypes.sizeof(BITMAPINFOHEADER) bmi.bmiHeader.biWidth = w bmi.bmiHeader.biHeight = -h bmi.bmiHeader.biPlanes = 1 bmi.bmiHeader.biBitCount = 32 bmi.bmiHeader.biCompression = 0 buf_size = w * h * 4 raw_buf = ctypes.create_string_buffer(buf_size) lines = self.gdi32.GetDIBits(mem_dc, bmp, 0, h, raw_buf, ctypes.byref(bmi), 0) if lines > 0: img = Image.frombuffer("RGBA", (w, h), raw_buf, "raw", "BGRA", 0, 1) os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) img.save(output_path) res = True else: res = False self.gdi32.SelectObject(mem_dc, old_bmp) self.gdi32.DeleteObject(bmp) self.gdi32.DeleteDC(mem_dc) self.user32.ReleaseDC(0, desk_dc) return res except Exception as e: print(f"[ScreenCapturerGDI Error] {e}", flush=True) return False def check_optical_death(img_path: str) -> bool: """ Kiểm tra quang học trên ảnh chụp màn hình: Tìm dấu hiệu màn hình tử trận (Death modal ở trung tâm màn hình, hoặc thanh HP/ES rỗng). """ try: if not os.path.exists(img_path): return False with Image.open(img_path) as img: w, h = img.size # Vùng kiểm tra hộp thoại tử trận (trung tâm màn hình [0.35W..0.65W], [0.35H..0.65H]) crop_box = (int(w * 0.35), int(h * 0.35), int(w * 0.65), int(h * 0.65)) center_region = img.crop(crop_box).convert("RGB") # Vùng Life/ES Orb (góc dưới bên trái [0.01W..0.08W], [0.85H..0.98H]) orb_box = (int(w * 0.01), int(h * 0.85), int(w * 0.08), int(h * 0.98)) orb_region = img.crop(orb_box).convert("RGB") # Tính toán độ sáng và màu sắc đặc trưng # Khi chết trong PoE2, màn hình bị làm tối và xuất hiện bảng hồi sinh có viền xám/đỏ raw = center_region.tobytes() total_px = len(raw) // 3 if total_px > 0: death_btn_pixels = 0 for i in range(0, len(raw), 3): r, g, b = raw[i], raw[i+1], raw[i+2] if (r > 120 and g < 60 and b < 60) or (r > 140 and g > 110 and b < 50): death_btn_pixels += 1 ratio = death_btn_pixels / float(total_px) if ratio > 0.045: return True except Exception: pass return False def focus_poe2_window() -> bool: """Đưa cửa sổ Path of Exile 2 lên foreground để nhận input.""" try: user32 = ctypes.windll.user32 kernel32 = ctypes.windll.kernel32 from src.common.win32_window import attach_to_default_desktop attach_to_default_desktop() try: from src.common.win32_window import find_poe2_window as _find_win, ensure_window_focus as _ensure_focus hwnd = _find_win() if hwnd and _ensure_focus(hwnd): return True except Exception: hwnd = 0 if not hwnd: hwnd = user32.FindWindowW("POEWindowClass", None) if not hwnd: hwnd = user32.FindWindowW(None, "Path of Exile 2") if hwnd: if user32.IsIconic(hwnd): user32.ShowWindow(hwnd, 9) # SW_RESTORE else: user32.ShowWindow(hwnd, 5) # SW_SHOW cur_thread = kernel32.GetCurrentThreadId() fg_hwnd = user32.GetForegroundWindow() fg_thread = user32.GetWindowThreadProcessId(fg_hwnd, None) if cur_thread != fg_thread: user32.AttachThreadInput(cur_thread, fg_thread, True) user32.SetForegroundWindow(hwnd) user32.BringWindowToTop(hwnd) user32.AttachThreadInput(cur_thread, fg_thread, False) else: user32.SetForegroundWindow(hwnd) user32.BringWindowToTop(hwnd) time.sleep(0.1) return True except Exception as e: print(f"[Focus Error] {e}", flush=True) return False class AutonomousMapRunner: def __init__( self, core_exe: Optional[str] = None, capture_dir: str = r"captures\map_run", max_duration_sec: int = 240, capture_interval_sec: float = 3.0, max_respawns: int = 3, portal_exit: bool = True, max_es: int = 4284, max_mana: int = 880, max_ward: int = 241, minimap_adapter: Optional[Any] = None, wasd_sender: Optional[Any] = None, min_duration_sec: float = 90.0, target_coverage: float = 0.80, ): if not core_exe: if os.path.exists(r"bin\AutoPOE2_Core.exe"): core_exe = r"bin\AutoPOE2_Core.exe" else: core_exe = r"bin\Release\AutoPOE2_Core.exe" self.core_exe = core_exe self.capture_dir = capture_dir self.max_duration_sec = max_duration_sec self.min_duration_sec = min_duration_sec self.target_coverage = target_coverage self.capture_interval_sec = capture_interval_sec self.max_respawns = max_respawns self.portal_exit = portal_exit self.max_es = max_es self.max_mana = max_mana self.max_ward = max_ward self.minimap_adapter = minimap_adapter self.wasd_sender = wasd_sender self.respawn_count = 0 os.makedirs(self.capture_dir, exist_ok=True) self.clear_evaluator = MapClearEvaluator( min_duration_sec=min_duration_sec, target_coverage=target_coverage, min_traverse_units=1200.0, ) if MapClearEvaluator else None self.capturer = ScreenCapturerGDI() self.shm_bridge = CoreShmBridge() if CoreShmBridge else None self.proc: Optional[subprocess.Popen] = None self.is_stopped = False self.death_detected = False self.captured_files = [] def execute_stutter_step_attack(self, vector: Tuple[float, float] = (0.0, 0.0)): """ Thực thi chu kỳ Micro-Stutter-Stepping & Combat Skill Cast (ExiledBot 2 Model): - Settle pause: Dwell ngắn ~50ms sau khi nhả WASD. - Mouse Aiming: Hướng chuột về phía mục tiêu/vector di chuyển (không click chuột trái - INV-WASD-NO-BLIND-LMB). - Skill Cast: Bấm phím Primary Skill ('F' - VK_F 0x46 / SC 0x21 - INV-KEY-NO-F-FOR-INTERACT). - Dwell cast: Giữ phím ~180ms. - Release & Recovery: Nhả phím, sẵn sàng cho nhịp di chuyển kế tiếp. """ try: from src.common.win32_window import find_poe2_window as _find_win, get_client_rect as _get_rect from src.common.win32_input import post_message_hover, post_message_key hwnd = _find_win() if hwnd: _, _, cw, ch = _get_rect(hwnd) if cw > 0 and ch > 0: cx = cw // 2 cy = ch // 2 vx, vy = vector if abs(vx) > 0.01 or abs(vy) > 0.01: aim_x = int(cx + vx * (cw * 0.16)) aim_y = int(cy + vy * (ch * 0.16)) else: aim_x = cx + int(cw * 0.08) aim_y = cy - int(ch * 0.05) post_message_hover(hwnd, aim_x, aim_y, delay_after=0.02) # Cast Skill 8 / Primary Skill ('F' - VK 0x46, Scan Code 0x21) ngầm không cướp phím người dùng post_message_key(hwnd, 0x46, 0x21, dwell_ms=0.18, delay_after=0.04) print(f"[Combat] ⚔️ Stutter-Step cast skill 'F' tấn công ngầm (Vector: {vector})", flush=True) except Exception as e: print(f"[Combat Warning] Lỗi thực hiện Stutter-Step attack: {e}", flush=True) def execute_portal_exit(self) -> bool: """ Thực thi quy trình thoát map bằng Town Portal phím '.' (VK 0xBE / HID 0x37) tuân thủ INV-TOWN-PORTAL-KEY. 1. Đưa cửa sổ game lên Foreground 2. Gửi phím '.' mở Town Portal 3. Chờ 1.5s tạo portal và chụp ảnh bằng chứng 4. Click vào tâm màn hình / portal để bước vào cổng trở về Hideout 5. Chờ 3.5s màn hình tải và chụp ảnh xác nhận Hideout """ print("\n[Portal] >> KÍCH HOẠT QUY TRÌNH THOÁT MAP BẰNG TOWN PORTAL (Phím '.') <<", flush=True) time.sleep(0.3) from src.common.win32_window import find_poe2_window as _find_win, get_client_rect as _get_rect from src.common.win32_input import post_message_key, post_message_click hwnd = _find_win() VK_OEM_PERIOD = 0xBE # Phím '.' mở Town Portal (VK 190, scan code 0x34) if hwnd: post_message_key(hwnd, VK_OEM_PERIOD, 0x34, dwell_ms=0.06, delay_after=0.1) print("[Portal] Đã gửi phím '.' niệm mở Town Portal ngầm. Chờ 1.5s cổng hình thành...", flush=True) time.sleep(1.5) ts = datetime.now().strftime("%Y%m%d_%H%M%S") portal_img = os.path.join(self.capture_dir, f"portal_opened_{ts}.png") # Xác định kích thước và vị trí Client Area (hỗ trợ hoàn hảo Windowed mode) cl, ct, cw, ch = 0, 0, 2560, 1440 is_client_crop = False os.makedirs(self.capture_dir, exist_ok=True) try: from src.common.screen_capture import capture_rect_bitblt as _cap_rect if hwnd: rcl, rct, rcw, rch = _get_rect(hwnd) if rcw > 0 and rch > 0: cl, ct, cw, ch = rcl, rct, rcw, rch img_captured = _cap_rect(cl, ct, cw, ch) if img_captured: img_captured.save(portal_img) is_client_crop = True except Exception as e: print(f"[Portal] Lỗi capture client rect: {e}", flush=True) if not is_client_crop: self.capturer.capture(portal_img) self.captured_files.append(portal_img) print(f"[Portal] Ảnh bằng chứng portal mở: {portal_img} (is_client_crop={is_client_crop})", flush=True) # Thử tìm nhãn portal qua OCR dùng Single Source of Logic (Rule 4B) ocr_pt = None try: from src.common.vision_ocr import find_portal_label_by_ocr from PIL import Image img_p = Image.open(portal_img) labels = find_portal_label_by_ocr(img_p) if labels: hideout_labels = [p for p in labels if any(kw in p[2].upper() for kw in ("HIDEOUT", "SHORELINE"))] chosen = hideout_labels[0] if hideout_labels else labels[0] bx, by = chosen[0], chosen[1] ocr_pt = (bx, by) except Exception: pass if ocr_pt: portal_points = [ ocr_pt, (ocr_pt[0], ocr_pt[1] + int(ch * 0.04)), (ocr_pt[0], ocr_pt[1] + int(ch * 0.08)), ] else: portal_points = [ (int(cw * 0.50), int(ch * 0.28)), # Label (int(cw * 0.50), int(ch * 0.35)), # Vortex trên (int(cw * 0.50), int(ch * 0.42)), # Vortex thân ] if hwnd: for click_x, click_y in portal_points: post_message_click(hwnd, click_x, click_y, hold_ms=0.05, delay_after=0.2) print("[Portal] Đã click LMB bước vào Town Portal. Chờ 3.5s chuyển cảnh...", flush=True) time.sleep(3.5) return_img = os.path.join(self.capture_dir, f"map_exit_result_{ts}.png") if is_client_crop: try: from src.common.screen_capture import capture_rect_bitblt as _cap_rect img_ret = _cap_rect(cl, ct, cw, ch) if img_ret: img_ret.save(return_img) else: self.capturer.capture(return_img) except Exception: self.capturer.capture(return_img) else: self.capturer.capture(return_img) self.captured_files.append(return_img) print(f"[Portal] Ảnh xác nhận sau thoát map: {return_img}", flush=True) return True def start(self): print(f"=================================================", flush=True) print(f" AutoPOE2 Autonomous Map Clear & Sentinel ", flush=True) print(f"=================================================", flush=True) # 0. Đảm bảo cửa sổ Game ở Foreground if focus_poe2_window(): print("[Focus] Đã đưa cửa sổ 'Path of Exile 2' lên Foreground", flush=True) else: print("[Focus Warning] Không tìm thấy HWND cửa sổ 'Path of Exile 2'", flush=True) # 1. Kết nối SHM Bridge để đọc telemetry thời gian thực if self.shm_bridge: if self.shm_bridge.connect(): print("[Bridge] Đã kết nối vào khối nhớ chia sẻ IPC (SHM)", flush=True) else: print("[Bridge Warning] Chưa kết nối được SHM (sẽ tự thử lại khi Core chạy)", flush=True) # 2. Xây dựng tham số khởi chạy Core Engine # Cấu hình chuẩn xác theo thông số nhân vật thực tế (CI Staff: Life 1, ES 4284, Mana 880, Ward 241) cmd_args = [ self.core_exe, "--user-mode", "--ci", "--ci-staff", "--primary-skill", "f", "--max-hp", "1", "--max-es", str(self.max_es), "--max-mana", str(self.max_mana), "--max-ward", str(self.max_ward), "--fast-clear", "--autoquest", "--wasd", "--move", "--auto-respawn" ] print(f"[Runner] Khởi chạy Core Engine: {' '.join(cmd_args)}", flush=True) os.makedirs(r"bin\Release", exist_ok=True) log_file = open(r"bin\Release\core_map_run.log", "w", encoding="utf-8") try: self.proc = subprocess.Popen( cmd_args, stdout=log_file, stderr=subprocess.STDOUT, cwd=os.getcwd() ) except OSError as e: print(f"[Runner Warning] Không thể spawn trực tiếp Core Engine ({e}). Thử kích hoạt qua ShellExecuteW...", flush=True) try: ret = ctypes.windll.shell32.ShellExecuteW( None, "open", self.core_exe, " ".join(cmd_args[1:]), None, 1 ) if ret <= 32: ctypes.windll.shell32.ShellExecuteW( None, "runas", self.core_exe, " ".join(cmd_args[1:]), None, 1 ) except Exception as ex: print(f"[Runner Error] ShellExecute thất bại ({ex}). Tiếp tục vận hành giám sát in-map.", flush=True) start_time = time.time() last_capture_time = 0.0 step_idx = 0 try: while not self.is_stopped: now = time.time() elapsed = now - start_time # Kiểm tra tiến trình Core có còn sống không if self.proc and self.proc.poll() is not None: print(f"[Runner Warning] Core Engine kết thúc (ExitCode={self.proc.returncode}). Tiếp tục vận hành Autonomous In-Map Controller (Minimap Navigation + Stutter-Step Combat)...", flush=True) self.proc = None # Hết thời gian phiên kiểm tra -> Thoát map bằng Town Portal (INV-TOWN-PORTAL-KEY) if elapsed >= self.max_duration_sec: print(f"[Runner] Đã đạt thời lượng tối đa {self.max_duration_sec}s. Dừng chu trình kiểm tra.", flush=True) self.stop() time.sleep(0.3) if self.portal_exit and not self.death_detected: self.execute_portal_exit() break # Đọc Telemetry từ SHM cur_hp = 1 cur_es = 4713 pos_x, pos_y = 0.0, 0.0 monsters = 0 if self.shm_bridge: snap = self.shm_bridge.read_telemetry_snapshot() if snap.is_online: cur_hp = snap.player.current_hp cur_es = snap.player.current_es pos_x = snap.player.pos_x pos_y = snap.player.pos_y monsters = snap.monsters_count # THẨM ĐỊNH HOÀN TẤT BẢN ĐỒ THỰC TẾ 4 CỔNG (Doc 69 §3.4 - SSoT 2026) # Triệt tiêu hoàn toàn anti-pattern "monsters <= 30 sau 15s" gây thoát map sớm. is_map_cleared = False clear_reason = "" if self.clear_evaluator and self.shm_bridge: snap_curr = self.shm_bridge.last_snapshot core_state = snap_curr.lifecycle_state if (snap_curr and snap_curr.is_online) else None boss_slain = getattr(snap_curr, "boss_slain", False) if (snap_curr and snap_curr.is_online) else False # Đọc tỷ lệ mở sương Minimap từ adapter nếu có cov_ratio = 0.0 if self.minimap_adapter and hasattr(self.minimap_adapter, "explorer") and self.minimap_adapter.explorer: cov_ratio = getattr(self.minimap_adapter.explorer, "coverage_ratio", 0.0) self.clear_evaluator.update_telemetry( pos_x=pos_x, pos_y=pos_y, monsters_near=monsters, coverage_ratio=cov_ratio, boss_slain=boss_slain, now_sec=now, ) is_map_cleared, clear_reason = self.clear_evaluator.evaluate_clear( elapsed_sec=elapsed, core_lifecycle_state=core_state, now_sec=now, ) if is_map_cleared: print(f"\n[Runner] 🎯 XÁC NHẬN HOÀN TẤT BẢN ĐỒ CHUẨN THỰC TẾ ({clear_reason}). Kết thúc Stage 5.", flush=True) self.stop() time.sleep(0.3) if self.portal_exit and not self.death_detected: self.execute_portal_exit() break # KIỂM TRA TỬ TRẬN LỚP 1 (RAM Telemetry) # Đối với CI: Khi chết, ES tụt về 0 hoặc HP tụt về 0 is_dead_ram = False if cur_hp == 0 or (cur_es == 0 and cur_hp <= 1): # Xác thực lại để tránh trường hợp khởi động ban đầu chưa nạp xong if elapsed > 3.0: is_dead_ram = True is_dead_optical = False optical_evidence_img = "" # Chụp ảnh màn hình định kỳ if now - last_capture_time >= self.capture_interval_sec: last_capture_time = now step_idx += 1 ts_str = datetime.now().strftime("%Y%m%d_%H%M%S") img_name = f"step_{ts_str}_{step_idx:03d}.png" img_path = os.path.join(self.capture_dir, img_name) if self.capturer.capture(img_path): self.captured_files.append(img_path) print(f"[Watcher #{step_idx:02d}] Chụp {img_name} | Elapsed: {int(elapsed)}s | HP={cur_hp} ES={cur_es} | Pos=({pos_x:.1f}, {pos_y:.1f}) | Mobs={monsters}", flush=True) # DẪN ĐƯỜNG SƯƠNG MÙ BẰNG MINIMAP FOG EXPLORER (Doc 69 §3.4 / Chu Kỳ 6-7) if self.minimap_adapter: try: wasd_cmd = self.minimap_adapter.process_fullscreen_frame(img_path) if wasd_cmd and wasd_cmd.direction.value: print(f"[MinimapNav] Hướng sương mù đề xuất: {wasd_cmd.direction.value} ({wasd_cmd.angle_deg:.1f}°), Coverage: {wasd_cmd.coverage_ratio*100:.1f}%", flush=True) if self.wasd_sender: self.minimap_adapter.dispatch_to_controller(wasd_cmd, sender_func=self.wasd_sender) # Stutter-Step: Sau nhịp di chuyển WASD, thực hiện cast primary skill 'F' tấn công self.execute_stutter_step_attack(wasd_cmd.vector) else: # Fallback exploration pulse & Stutter-step cast skill self.execute_stutter_step_attack((0.0, -1.0)) except Exception as ex: print(f"[MinimapNav Warning] Lỗi tính toán hướng sương mù: {ex}", flush=True) # KIỂM TRA TỬ TRẬN LỚP 2 (Quang Học Optical Detection) if check_optical_death(img_path): is_dead_optical = True optical_evidence_img = img_path if is_dead_ram or is_dead_optical: death_reason = "RAM_TELEMETRY_ZERO_VITALS" if is_dead_ram else "OPTICAL_DEATH_MODAL" evidence_img = optical_evidence_img self.respawn_count += 1 print(f"\n[Sentinel] Phát hiện tử trận -> Chờ Core Engine tự động click hồi sinh tại Checkpoint...", flush=True) if self.respawn_count > self.max_respawns: print(f"[Sentinel] Số lần tử trận ({self.respawn_count}) vượt quá max_respawns ({self.max_respawns})! Kích hoạt kết thúc phiên.", flush=True) self.handle_death_event(death_reason, cur_hp, cur_es, pos_x, pos_y, evidence_img) break # Nếu sau 6 giây nhân vật sống lại (HP > 0 / ES > 0), log và tiếp tục phiên tự hành respawned = False wait_start = time.time() while time.time() - wait_start < 6.0 and not self.is_stopped: time.sleep(0.5) if self.proc.poll() is not None: break if self.shm_bridge: snap = self.shm_bridge.read_telemetry_snapshot() if snap.is_online: r_hp = snap.player.current_hp r_es = snap.player.current_es # Nhân vật sống lại: HP > 0 / ES > 0 if (snap.player.max_hp == 1 and r_hp > 0 and r_es > 0) or (snap.player.max_hp > 1 and r_hp > 0) or (r_hp > 1 or r_es > 0): respawned = True cur_hp = r_hp cur_es = r_es break if respawned: print(f"[Sentinel] Nhân vật đã hồi sinh thành công! Tiếp tục phiên tự hành.", flush=True) time.sleep(1.0) continue else: print(f"[Sentinel] Quá 6 giây không hồi sinh thành công! Kích hoạt kết thúc phiên.", flush=True) self.handle_death_event("RESPAWN_TIMEOUT", cur_hp, cur_es, pos_x, pos_y, evidence_img) break time.sleep(0.2) except KeyboardInterrupt: print("\n[Runner] Người dùng bấm ngắt (Ctrl+C). Dừng an toàn...", flush=True) finally: self.stop() try: log_file.close() except Exception: pass return { "death_detected": self.death_detected, "captured_files": self.captured_files, "elapsed_sec": time.time() - start_time } def handle_death_event(self, reason: str, hp: int, es: int, x: float, y: float, evidence_img: str = ""): self.death_detected = True self.is_stopped = True self.stop() ts_str = datetime.now().strftime("%Y%m%d_%H%M%S") incident_img = os.path.join("captures", f"INCIDENT_DEATH_{ts_str}.png") if evidence_img and os.path.exists(evidence_img): import shutil shutil.copyfile(evidence_img, incident_img) else: self.capturer.capture(incident_img) dossier = { "incident_type": "CHARACTER_DEATH", "timestamp": datetime.now().isoformat(), "reason": reason, "vitals": {"hp": hp, "es": es}, "position": {"x": x, "y": y}, "evidence_image": incident_img } dossier_path = os.path.join("captures", f"incident_dossier_{ts_str}.json") with open(dossier_path, "w", encoding="utf-8") as f: json.dump(dossier, f, indent=2) print(f"[SENTINEL INTERLOCK] Đã kích hoạt Kill-Switch: Tiến trình Core đã bị dừng hoàn toàn.", flush=True) print(f"[SENTINEL INTERLOCK] Hồ sơ sự cố được lưu tại: {dossier_path}", flush=True) print(f"[SENTINEL INTERLOCK] Ảnh hiện trường tử trận: {incident_img}", flush=True) def stop(self): self.is_stopped = True if self.proc and self.proc.poll() is None: print("[Runner] Gửi tín hiệu dừng Core Engine...", flush=True) try: self.proc.terminate() self.proc.wait(timeout=2.0) except Exception: try: self.proc.kill() except Exception: pass print("[Runner] Core Engine đã được giải phóng an toàn.", flush=True) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="AutoPOE2 Autonomous Map Runner") parser.add_argument("--duration", type=int, default=30, help="Thời gian chạy tối đa (giây)") parser.add_argument("--interval", type=float, default=2.5, help="Khoảng thời gian chụp ảnh (giây)") parser.add_argument("--max-respawns", type=int, default=3, help="Số lần hồi sinh tối đa trước khi dừng phiên") parser.add_argument("--portal-exit", dest="portal_exit", action="store_true", default=True, help="Tự động mở portal bằng phím '.' khi hết phiên") parser.add_argument("--no-portal-exit", dest="portal_exit", action="store_false", help="Không tự động mở portal khi hết phiên") parser.add_argument("--max-es", type=int, default=4284, help="Expected Max ES của nhân vật") parser.add_argument("--max-mana", type=int, default=880, help="Expected Max Mana của nhân vật") parser.add_argument("--max-ward", type=int, default=241, help="Expected Max Ward của nhân vật") args = parser.parse_args() runner = AutonomousMapRunner( max_duration_sec=args.duration, capture_interval_sec=args.interval, max_respawns=args.max_respawns, portal_exit=args.portal_exit, max_es=args.max_es, max_mana=args.max_mana, max_ward=args.max_ward ) res = runner.start() print(f"\n[Summary] Run finished: DeathDetected={res['death_detected']}, Captured={len(res['captured_files'])} frames, Duration={res['elapsed_sec']:.1f}s")