""" AutoPOE2 Control Center - Giao Diện Điều Khiển Trung Tâm Đồ Họa (GUI) Tích hợp nút Bắt đầu / Dừng, các checkbox tùy chọn tính năng, quét bộ nhớ thời gian thực và tự động chụp màn hình ghép telemetry để phân tích. """ import os import subprocess import sys import threading import time from datetime import datetime from typing import Any, Callable, Dict, Optional, Tuple import customtkinter as ctk from src.agentic_engine.coordinator import AgenticCoordinator from src.assistant_tool.auto_quester import AutoQuester from src.assistant_tool.config import Config from src.assistant_tool.core_controller import CoreController from src.assistant_tool.core_shm_bridge import CoreShmBridge, CoreTelemetrySnapshot from src.assistant_tool.gem_manager import GemManager from src.assistant_tool.screen_capturer import ScreenCapturer from src.assistant_tool.tactical_brain import TacticalBrain, TacticalRecommendation from src.assistant_tool import ui_theme as theme from src.assistant_tool.ui import ( create_header, create_hud_strip, build_right_telemetry_and_logs, load_user_settings, save_user_settings, get_expected_max_hp_from_str, is_valid_vitals, parse_character_state, build_tab_quest, build_tab_combat, build_tab_client, build_tab_intel, ) from src.common.audio_feedback import beep_bot_toggle, beep_panic, beep_notification from src.assistant_tool.system_tray_manager import SystemTrayManager try: from src.assistant_tool.optical_hp_sensor import OpticalHPSensor except ImportError: try: from optical_hp_sensor import OpticalHPSensor # type: ignore[no-redef] except ImportError: OpticalHPSensor = None # type: ignore[assignment,misc] try: from src.agentic_engine.invariant_sentinel import InvariantSentinel from src.agentic_engine.flight_recorder import MultimodalIncidentRecorder except ImportError: try: from src.assistant_tool.agent_watcher import InvariantSentinel, MultimodalIncidentRecorder except ImportError: InvariantSentinel = None MultimodalIncidentRecorder = None try: from src.assistant_tool.modifier_filter import ModifierFilter, ModReport, evaluate_modifiers except ImportError: try: from modifier_filter import ModifierFilter, ModReport, evaluate_modifiers # type: ignore[no-redef] except ImportError: ModifierFilter = None # type: ignore[assignment,misc] ModReport = None # type: ignore[assignment,misc] evaluate_modifiers = None # type: ignore[assignment,misc] try: from src.assistant_tool.item_hover_watcher import ItemHoverWatcher from src.assistant_tool.item_tooltip_overlay import ItemTooltipOverlay except ImportError: ItemHoverWatcher = None ItemTooltipOverlay = None class ControlCenterApp(ctk.CTk): def __init__(self, start_in_tray: bool = False): super().__init__() # Thiết lập giao diện tổng thể self.title("AutoPOE2 Control Center") self.geometry(theme.window_geometry()) self.minsize(Config.CONTROL_CENTER_MIN_WIDTH, Config.CONTROL_CENTER_MIN_HEIGHT) self.configure(fg_color=theme.c("bg_dark")) ctk.set_appearance_mode("Dark") ctk.set_default_color_theme("blue") self._last_lifecycle_log = "" # Khởi tạo các phân hệ self.core_bridge = CoreShmBridge(low_hp_threshold=35.0) self.core_controller = CoreController() self.screen_capturer = ScreenCapturer(output_dir="captures") self.gem_manager = GemManager() self.auto_quester = AutoQuester() self.tactical_brain = TacticalBrain(self.core_bridge) self.agentic_coordinator = AgenticCoordinator.get_instance() self.default_expected_max_hp = 0 self.optical_sensor = OpticalHPSensor(expected_max_hp=0) if OpticalHPSensor is not None else None if self.optical_sensor is not None: try: self.optical_sensor.start_background_worker(interval=0.5) except Exception: pass self._last_optical_read_time = 0.0 self._last_optical_crosscheck = time.time() + 999999.0 self._optical_correction_count = 0 self._optical_verified = False self._last_burst_capture = 0.0 self._last_periodic_capture = 0.0 self._last_hp = 0 self._last_toggle_time = 0.0 self._last_pos: Optional[tuple[float, float, float]] = None self._last_pos_time: float = 0.0 # Lắng nghe phím tắt: F8 / F6 (Bật/Tắt Bot), Pause (Panic Stop) & F10 (Chiến Thuật) self.bind("", lambda e: self._on_bot_toggle_hotkey()) self.bind("", lambda e: self._on_bot_toggle_hotkey()) self.bind("", lambda e: self._on_panic_stop_hotkey()) self.bind("", lambda e: self._on_tactical_brain_clicked()) # Xây dựng bố cục giao diện self._create_header() self._create_hud_strip() self._create_main_content() # Đăng ký sự kiện từ CoreController và CoreShmBridge self.core_controller.register_log_callback(lambda msg: self._safe_after(self._append_log, msg)) self.core_controller.register_state_callback(lambda running: self._safe_after(self._on_core_state_changed, running)) self.core_bridge.register_listener(lambda snap: self._safe_after(self._on_telemetry_snapshot, snap)) self.core_bridge.start() # Khởi chạy LogWatcher để đồng bộ sự kiện game (Zone, Death, Items) vào Agentic Coordinator self.log_watcher = None try: from src.assistant_tool.log_watcher import LogWatcher log_path = Config.find_poe2_client_log() if log_path and os.path.exists(log_path): self.log_watcher = LogWatcher(log_path=log_path) self.log_watcher.register_listener(lambda evt: self._safe_after(self._on_log_event, evt)) self.log_watcher.start() self._append_log(f"[LogWatcher] Đã gắn kết theo dõi nhật ký game: {log_path}") except Exception as e: self._append_log(f"[LogWatcher Note] Không thể khởi động LogWatcher: {e}") # Khởi tạo Bộ Điều Phối Vòng Đời Tự Hành (Autonomous Lifecycle Manager) try: from src.assistant_tool.autonomous_lifecycle import AutonomousLifecycleManager except ImportError: from autonomous_lifecycle import AutonomousLifecycleManager self.lifecycle_manager = AutonomousLifecycleManager( core_bridge=self.core_bridge, core_controller=self.core_controller, config={} ) self.lifecycle_manager.register_listener( lambda phase, text: self._safe_after(self._on_lifecycle_update, phase, text) ) # Seed SCENE chơi thật gần nhất SAU khi lifecycle đã sẵn sàng nhận sự kiện. # Tail bắt đầu ở EOF nên nếu không seed, tool mở giữa map (Hive Fortress) # sẽ kẹt "chờ xác thực vị trí bản đồ" dù Client.txt đã có [SCENE] Set Source. if self.log_watcher is not None: try: seeded = self.log_watcher.seed_latest_playable_zone() if seeded is not None: self._on_log_event(seeded) zone = seeded.area_name or seeded.message self._append_log(f"[LogWatcher] Đã xác thực khu vực hiện tại từ đuôi Client.txt: {zone}") self._refresh_area_hud_from_lifecycle() else: self._append_log("[LogWatcher] Chưa thấy SCENE chơi thật trong đuôi Client.txt — chờ dòng log mới.") except Exception as e: self._append_log(f"[LogWatcher Note] Không seed được zone hiện tại: {e}") # Khởi chạy luồng phím nóng toàn cục (Global Hotkey F8 / Pause / F10 / F9) self._start_global_hotkey_listener() # Định thời kiểm tra định kỳ (Periodic check 1s) self.after(1000, self._periodic_task) self._watcher_sentinel = None self._watcher_recorder = None self._watcher_violation_count = 0 if InvariantSentinel is not None and MultimodalIncidentRecorder is not None: try: self._watcher_sentinel = InvariantSentinel() self._watcher_recorder = MultimodalIncidentRecorder( output_dir="debug_harness/incidents", screen_capturer=self.screen_capturer ) self.after(500, lambda: self._append_log("[AgentWatcher] ✅ Invariant Sentinel đã kích hoạt giám sát bất biến kiến trúc (Rule 13)")) except Exception as e: err_msg = str(e) self.after(500, lambda m=err_msg: self._append_log(f"[AgentWatcher] Không thể khởi tạo Watcher: {m}")) # Không tự động mở game khi vừa mở Control Center (chỉ mở khi người dùng yêu cầu) # self.after(1500, self._check_and_auto_launch_game_on_startup) # Ghi log phát hiện cài đặt game vào Log Box if hasattr(self, "game_cfg_path") and self.game_cfg_path: self._append_log(f"[AutoDetect] Đã tìm thấy tệp cấu hình game: {os.path.basename(self.game_cfg_path)}") self._append_log(f" ↳ Tự động kích hoạt chế độ di chuyển: {'⌨️ BÀN PHÍM (WASD)' if self.detected_move_mode == 'wasd' else '🖱️ CHUỘT (Click-to-Move)'}") # Khởi tạo System Tray Manager (mặc định thu nhỏ về tray để không chiếm màn hình làm việc) self.tray_manager = SystemTrayManager( app=self, on_exit_callback=self.quit_app_completely, on_toggle_bot_callback=self._on_bot_toggle_hotkey, ) self.tray_manager.start() if start_in_tray: self.withdraw() def _safe_after(self, func: Callable, *args: Any) -> None: self.after(0, func, *args) # ========================================================== # 1. HEADER & THANH ĐIỀU KHIỂN CHÍNH # ========================================================== # ========================================================== # 1. GIAO DIỆN ĐỒ HỌA MODULAR (DELEGATED TO src.assistant_tool.ui) # ========================================================== def _create_header(self): create_header(self, self) def _create_hud_strip(self): create_hud_strip(self, self) def _create_main_content(self): container = ctk.CTkFrame(self, fg_color="transparent") container.pack(fill="both", expand=True, padx=8, pady=6) left_col = ctk.CTkFrame(container, fg_color=theme.c("card_bg"), corner_radius=8, width=430) left_col.pack(side="left", fill="both", expand=True, padx=(0, 4)) right_col = ctk.CTkFrame(container, fg_color=theme.c("card_bg"), corner_radius=8, width=460) right_col.pack(side="right", fill="both", expand=True, padx=(4, 0)) self._build_left_features(left_col) self._build_right_telemetry_and_logs(right_col) def _build_left_features(self, parent): detected_mode, cfg_path = Config.detect_poe2_movement_mode() self.detected_move_mode = detected_mode self.game_cfg_path = cfg_path tabs = ctk.CTkTabview( parent, fg_color=theme.c("card_bg"), segmented_button_selected_color=theme.c("start"), segmented_button_selected_hover_color=theme.c("ok_hover"), segmented_button_unselected_color=theme.c("card_bg_alt"), text_color=theme.c("text_primary"), height=420 ) tabs.pack(fill="both", expand=True, padx=4, pady=4) self.feature_tabs = tabs def _tab_scroll(name: str): frame = tabs.add(name) sc = ctk.CTkScrollableFrame(frame, fg_color="transparent") sc.pack(fill="both", expand=True, padx=2, pady=2) return sc quest_scroll = _tab_scroll("Nhiệm vụ") combat_scroll = _tab_scroll("Chiến đấu") client_scroll = _tab_scroll("Client") intel_scroll = _tab_scroll("Tác vụ") self._build_tab_quest(quest_scroll) self._build_tab_combat(combat_scroll) self._build_tab_client(client_scroll, detected_mode, cfg_path) self._build_tab_intel(intel_scroll) def _build_tab_quest(self, scroll): build_tab_quest(self, scroll) def _build_tab_combat(self, scroll): build_tab_combat(self, scroll) def _build_tab_client(self, scroll, detected_mode, cfg_path): build_tab_client(self, scroll, detected_mode, cfg_path) def _build_tab_intel(self, scroll): build_tab_intel(self, scroll) def _build_right_telemetry_and_logs(self, parent): build_right_telemetry_and_logs(self, parent) def _create_section_label(self, parent, text: str): lbl = ctk.CTkLabel( parent, text=text, font=ctk.CTkFont(family="Segoe UI", size=11, weight="bold"), text_color=theme.c("accent_gold") ) return lbl def _resolve_playable_area(self, snap) -> tuple[str, int]: """ Tên khu vực hiển thị = SHM area_name (LogSensor) hoặc seed Client.txt. Overlay Act/Atlas/(null) không được hiện như map chơi. """ from src.assistant_tool.autonomous_lifecycle import AutonomousLifecycleManager snap_name = (getattr(snap, "area_name", None) or "").strip() snap_level = int(getattr(snap, "area_level", 0) or 0) if snap_name and not AutonomousLifecycleManager.is_ui_scene(snap_name): return snap_name, snap_level life_name = "" if hasattr(self, "lifecycle_manager"): life_name = (self.lifecycle_manager.current_area_name or "").strip() if life_name and not AutonomousLifecycleManager.is_ui_scene(life_name): return life_name, snap_level return "--", snap_level def _refresh_area_hud_from_lifecycle(self) -> None: if not hasattr(self, "lbl_area"): return snap = getattr(getattr(self, "core_bridge", None), "last_snapshot", None) if snap and getattr(snap, "is_online", False): sh_name = (getattr(snap, "area_name", "") or "").strip() from src.assistant_tool.autonomous_lifecycle import AutonomousLifecycleManager if sh_name and not AutonomousLifecycleManager.is_ui_scene(sh_name): return name = "" if hasattr(self, "lifecycle_manager"): name = (self.lifecycle_manager.current_area_name or "").strip() if name: self.lbl_area.configure(text=f"Khu vực: {name}") if hasattr(self, "lbl_movement_plan"): town = bool(getattr(self.lifecycle_manager, "is_in_town", False)) plan = f"Làng {name}" if town else f"Chiến đấu · {name}" self.lbl_movement_plan.configure(text=f"Kế hoạch: {plan}") # ========================================================== # 3. XỬ LÝ SỰ KIỆN TƯƠNG TÁC # ========================================================== def _on_burst_slider_change(self, val): self.lbl_burst.configure(text=f" ↳ Ngưỡng sốc máu kích hoạt né: {int(val)}% maxHP") def _on_flask_slider_change(self, val): self.lbl_flask_hp.configure(text=f" ↳ Bơm bình Máu khi HP <= {int(val)}%") def _on_flask_mana_slider_change(self, val): self.lbl_flask_mana.configure(text=f" ↳ Bơm bình Mana khi Mana <= {int(val)}%") def _on_loot_radius_change(self, val): self.lbl_loot_radius.configure(text=f" ↳ Bán kính nhặt đồ: {int(val)} đơn vị") def _on_open_visual_tool(self): self._append_log("[VisualTool] Đang kiểm tra ứng dụng Poe2VisualTool độc lập...") # 1. Kiểm tra script / launcher của dự án độc lập Poe2VisualTool external_launchers = [ r"C:\Projects\Poe2VisualTool\run_tool.bat", r"C:\Projects\Poe2VisualTool\bin\Release\Poe2VisualTool.exe", ] target_launcher = None for launcher in external_launchers: if os.path.isfile(launcher): target_launcher = launcher break if not target_launcher: self._append_log(r"[VisualTool] Poe2VisualTool chưa được cài đặt tại C:\Projects\Poe2VisualTool") return # 2. Dùng Win32 API tìm cửa sổ Poe2VisualTool để focus lên trước nếu đã chạy try: import win32con import win32gui found_hwnd = None target_titles = [ "Poe2VisualTool — Companion & Visual Engine", "Poe2VisualTool", ] target_classes = [ "Poe2VisualToolOverlay", ] for cls_name in target_classes: hwnd = win32gui.FindWindow(cls_name, None) if hwnd: found_hwnd = hwnd break if not found_hwnd: for title in target_titles: hwnd = win32gui.FindWindow(None, title) if hwnd: found_hwnd = hwnd break if not found_hwnd: def _enum_cb(hwnd, extra): nonlocal found_hwnd if not win32gui.IsWindow(hwnd) or not win32gui.IsWindowVisible(hwnd): return True txt = win32gui.GetWindowText(hwnd) for t in target_titles: if t.lower() in txt.lower(): found_hwnd = hwnd return False return True win32gui.EnumWindows(_enum_cb, None) if found_hwnd: win32gui.ShowWindow(found_hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(found_hwnd) self._append_log("[VisualTool] Cửa sổ Poe2VisualTool đã mở — đưa lên trước.") return except Exception: pass # 3. Nếu chưa chạy, gọi subprocess.Popen để mở ứng dụng độc lập try: subprocess.Popen( [target_launcher], cwd=os.path.dirname(target_launcher), ) self._append_log(f"[VisualTool] Đã khởi chạy Poe2VisualTool từ: {target_launcher}") except Exception as exc: self._append_log(f"[VisualTool] Lỗi khi mở {target_launcher}: {exc}") def _on_auto_detect_toggle(self): if self.chk_auto_detect_move.get(): mode, path = Config.detect_poe2_movement_mode() self.var_move_mode.set(mode) badge_text = "⚡ ĐÃ TỰ ĐỘNG NHẬN DIỆN TỪ GAME: ⌨️ BÀN PHÍM (WASD)" if mode == "wasd" else "⚡ ĐÃ TỰ ĐỘNG NHẬN DIỆN TỪ GAME: 🖱️ CHUỘT (Click-to-Move)" badge_color = theme.c("info_cyan") if mode == "wasd" else theme.c("ok") self.lbl_move_badge.configure(text=badge_text, text_color=badge_color) self._append_log(f"[AutoDetect] Tự động đồng bộ với tệp cài đặt game: {'⌨️ BÀN PHÍM (WASD)' if mode == 'wasd' else '🖱️ CHUỘT (Click-to-Move)'}") def _on_login_delay_change(self, val): self.lbl_login_delay.configure(text=f" ↳ Thời gian chờ trước khi kết nối lại: {int(val)} giây") def _on_char_slot_change(self, choice): slot_num = choice.split()[0] self.var_char_slot.set(slot_num) def _on_launch_game_clicked(self): exe_path = self.ent_exe_path.get().strip() self._append_log(f"[ClientLauncher] Đang khởi chạy PathOfExile.exe từ: {exe_path}...") try: from src.assistant_tool.client_launcher import Poe2ClientLauncher except ImportError: from client_launcher import Poe2ClientLauncher import threading def _bg(): ok = Poe2ClientLauncher.launch_client(exe_path, wait_timeout_sec=30) if ok: self.after(0, lambda: self._append_log("[ClientLauncher] Khởi chạy game thành công! Cửa sổ POE2 đã sẵn sàng.")) else: self.after(0, lambda: self._append_log("[ClientLauncher Error] Không thể mở game hoặc hết thời gian chờ.")) threading.Thread(target=_bg, daemon=True).start() def _check_and_auto_launch_game_on_startup(self): """Tự động kiểm tra và mở game khi khởi động Control Center nếu chưa chạy.""" try: from src.assistant_tool.client_launcher import Poe2ClientLauncher except ImportError: from client_launcher import Poe2ClientLauncher if bool(self.chk_auto_launch.get()) and not Poe2ClientLauncher.is_client_running(): self._append_log("[AutoLaunch] Phát hiện game chưa chạy -> Đang tự động mở PathOfExile.exe...") self._on_launch_game_clicked() def _on_quest_act_changed(self, choice: str): """Cập nhật danh sách map khi người dùng chuyển Act.""" if "Act 1" in choice: zones = [ "The Riverbank (Entrance)", "The Ogham Farmlands (Boss Crowbell - Skill Book)", "The Hunting Grounds (Waypoint)", "The Crypt (Crypt Boss)" ] elif "Act 2" in choice: zones = [ "Vastiri Outskirts (Waypoint)", "The Mastodon Sands (Boss Goliath)", "The Traitor's Passage (Waypoint)", "Dreadnought (Boss Jamanra - Skill Book)" ] elif "Act 3" in choice: zones = [ "Sandswept Marsh (Waypoint)", "Phrey Jungle (Boss / Transition)", "The Pools of Chakhir (Boss - Skill Point)", "The Matatl Grounds (Waypoint)", "The Ziggurat / Aggorat (Boss - Skill Book)" ] elif "Atlas" in choice: zones = [ "Atlas Map (Boss Rush / Progression)", "Citadel / Pinnacle Boss", "Breach / Ritual / Expedition Map" ] else: zones = [ "Sandswept Marsh (Waypoint)", "Phrey Jungle (Boss / Transition)", "The Pools of Chakhir (Boss - Skill Point)", "The Matatl Grounds (Waypoint)", "The Ziggurat / Aggorat (Boss - Skill Book)" ] self.cmb_quest_zone.configure(values=zones) self.var_quest_zone.set(zones[0]) self._append_log(f"[AutoQuest] Đã chọn {choice}. Danh mục phụ bản đã được cập nhật.") def _get_current_map_scenario_code(self) -> int: val = getattr(self, "var_map_scenario", None) text = val.get() if val else "" if "Boss Rush" in text: return 0 elif "Atlas Quest" in text: return 1 elif "Fast Clear" in text: return 2 elif "Full Clear" in text: return 3 return 0 def _on_map_scenario_changed(self, choice: str): code = self._get_current_map_scenario_code() auto_p = bool(self.chk_auto_portal.get()) if hasattr(self, "chk_auto_portal") else True if hasattr(self, "core_bridge") and self.core_bridge and hasattr(self.core_bridge, "send_set_map_scenario"): self.core_bridge.send_set_map_scenario(code, auto_p) if hasattr(self, "lifecycle_manager") and self.lifecycle_manager and hasattr(self.lifecycle_manager, "quester"): try: from src.assistant_tool.auto_quester import MapScenario self.lifecycle_manager.quester.set_map_scenario(MapScenario(code)) except Exception: pass self._append_log(f"[Map Scenario] Đã chuyển kịch bản: {choice} (Mã: {code}, AutoPortal: {auto_p})") def _on_auto_portal_toggle(self): code = self._get_current_map_scenario_code() auto_p = bool(self.chk_auto_portal.get()) if hasattr(self, "chk_auto_portal") else True if hasattr(self, "core_bridge") and self.core_bridge and hasattr(self.core_bridge, "send_set_map_scenario"): self.core_bridge.send_set_map_scenario(code, auto_p) self._append_log(f"[Map Scenario] Tự động mở Portal sau Boss: {'BẬT' if auto_p else 'TẮT'}") def _on_trigger_quest_travel_clicked(self): """Kích hoạt dịch chuyển World Map qua phím U tới phụ bản nhiệm vụ đã chọn.""" lm = getattr(self, "lifecycle_manager", None) if lm is not None: if not getattr(lm, "has_real_map_log", False): self._append_log("[AutoQuest] Chặn phím U: chưa xác thực khu vực từ Client.txt / SHM.") return if not getattr(lm, "is_in_town", False): self._append_log( f"[AutoQuest] Chặn phím U: '{lm.current_area_name}' không phải làng/hideout." ) return act_text = self.var_quest_act.get() act = 1 if "Act 1" in act_text else (3 if "Act 3" in act_text else 2) zone_text = self.var_quest_zone.get() cur_values = self.cmb_quest_zone.cget("values") zone_index = 0 if zone_text in cur_values: zone_index = cur_values.index(zone_text) self._append_log(f"[AutoQuest] 🚀 Kích hoạt dịch chuyển World Map 'U': Act {act}, Phụ bản: {zone_text} (Index #{zone_index})...") if self.core_bridge and hasattr(self.core_bridge, "send_trigger_world_map_travel"): if self.core_bridge.send_trigger_world_map_travel(act, zone_index): self._append_log("[AutoQuest] >> Đã gửi lệnh TRIGGER_WORLD_MAP_TRAVEL (OpCode 96) thành công! Core đang gửi phím 'U'...") self.lbl_active_quest.configure( text=f" 👉 [Đang Dịch Chuyển]: Act {act} -> {zone_text} (Đang mở World Map U)", text_color=theme.c("info_cyan") ) else: self._append_log("[AutoQuest Warning] Không thể gửi lệnh tới Core. Hàng đợi đầy hoặc Core chưa chạy.") else: self._append_log("[AutoQuest Warning] Cầu nối Shared Memory chưa sẵn sàng. Vui lòng bật bot trước.") def _on_test_login_clicked(self): self._append_log("[AutoLogin] Gửi yêu cầu kiểm tra quy trình tự động đăng nhập xuống Core...") if self.core_bridge and self.core_bridge.send_trigger_auto_login(): self._append_log("[AutoLogin] Đã gửi lệnh TRIGGER_AUTO_LOGIN thành công!") else: self._append_log("[AutoLogin Warning] Chưa kết nối được tới Core Engine. Vui lòng bật bot trước.") def _load_user_settings(self) -> Dict[str, Any]: return load_user_settings() def _save_user_settings(self): text = self.entry_max_hp.get().strip() if hasattr(self, "entry_max_hp") else "Auto" save_user_settings({"expected_max_hp": text if text else "Auto"}) def _get_expected_max_hp(self) -> int: if hasattr(self, "entry_max_hp"): return get_expected_max_hp_from_str(self.entry_max_hp.get().strip()) return 0 def _on_expected_hp_changed(self): val = self._get_expected_max_hp() if hasattr(self, "optical_sensor") and self.optical_sensor: self.optical_sensor.expected_max_hp = val self._save_user_settings() is_valid_vitals = staticmethod(is_valid_vitals) parse_character_state = staticmethod(parse_character_state) def _on_auto_detect_vitals_clicked(self): """ Kích hoạt quét quang học màn hình game POE2 tức thì: 1. Tự động nhận diện Life (Cur/Max), Shield (Cur/Max), Mana (Cur/Max). 2. Kiểm tra tính hợp lý (không gửi nếu max_hp < 50, current_hp > max_hp, max_hp > 30000, lặp bất thường). 3. Tự động điền Max HP vào ô nhập liệu và lưu cấu hình. 4. Nếu C++ Core đang chạy, gửi lệnh hiệu chuẩn tức thời qua Shared Memory IPC (opCode 98). """ if not hasattr(self, "optical_sensor") or not self.optical_sensor: return self._append_log("[Vitals Calibrator] 🔍 Đang quét màn hình game POE2 để nhận diện chỉ số sinh tồn...") vitals = self.optical_sensor.detect_vitals_from_screen() # Nếu nhận diện được Max HP if vitals.max_hp > 0: target_hp = vitals.max_hp target_mana = vitals.max_mana if vitals.max_mana > 0 else 344 target_es = vitals.max_shield if (hasattr(vitals, "max_shield") and vitals.max_shield > 0) else 0 # Kiểm tra tính hợp lý của Vitals trước khi gửi recalibrate sang C++ Core if not self.is_valid_vitals(target_hp, vitals.current_hp): self._append_log(f"[Vitals Calibrator] ⚠️ Bỏ qua chỉ số không hợp lý hoặc lặp bất thường: HP={vitals.current_hp}/{target_hp}") return if hasattr(self, "entry_max_hp"): self.entry_max_hp.delete(0, "end") self.entry_max_hp.insert(0, str(target_hp)) self._save_user_settings() # Gửi lệnh đồng bộ sang C++ Core qua IPC nếu đang kết nối synced_core = False if hasattr(self, "core_bridge") and self.core_bridge and hasattr(self.core_bridge, "send_recalibrate_vitals"): synced_core = self.core_bridge.send_recalibrate_vitals(target_hp, target_mana, target_es) status_msg = f"Life {vitals.current_hp}/{target_hp}" if vitals.max_shield > 0: status_msg += f", Shield {vitals.current_shield}/{vitals.max_shield}" if vitals.max_mana > 0: status_msg += f", Mana {vitals.current_mana}/{target_mana}" self._append_log(f"[Vitals Calibrator] 🎯 Nhận diện thành công ({vitals.detection_method}): {status_msg}") if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text=f"🟢 HP Sensor: ĐÃ ĐỒNG BỘ ({target_hp} HP)", text_color=theme.c("ok") ) if synced_core: self._append_log(f"[Vitals Calibrator] ✅ Đã gửi lệnh hiệu chuẩn sang C++ Core: Max HP = {target_hp}!") else: # Nếu chưa đọc được chữ số cụ thể từ màn hình, thử đọc tỷ lệ quang học và ước tính cur, mhp, ratio = self.optical_sensor.read_hp() if ratio > 0.0: self._append_log(f"[Vitals Calibrator] ⚠️ Nhận diện tỷ lệ quả cầu máu: {int(ratio * 100)}%. (Gợi ý: Bật 'Always Show Values' trong UI game để tự đọc số chuẩn)") else: self._append_log("[Vitals Calibrator] ⚠️ Chưa tìm thấy cửa sổ POE2 hoặc màn hình đang tải/bị che khuất.") def _collect_config(self): exp_hp = self._get_expected_max_hp() det_vitals = getattr(getattr(self, "optical_sensor", None), "last_detected_vitals", None) if (not det_vitals or det_vitals.max_hp <= 0) and hasattr(self, "optical_sensor") and self.optical_sensor: try: det_vitals = self.optical_sensor.detect_vitals_from_screen() except Exception: det_vitals = None detected_max_hp = det_vitals.max_hp if (det_vitals and det_vitals.max_hp > 0 and self.is_valid_vitals(det_vitals.max_hp, det_vitals.current_hp)) else 0 detected_max_mana = det_vitals.max_mana if (det_vitals and det_vitals.max_mana > 0) else 0 detected_max_es = det_vitals.max_shield if (det_vitals and hasattr(det_vitals, "max_shield") and det_vitals.max_shield > 0) else 0 return { "move_mode": self.var_move_mode.get(), "auto_dodge": bool(self.chk_dodge.get()), "burst_hp": int(self.sld_burst.get()), "smart_flask": bool(self.chk_flask.get()), "flask_hp": int(self.sld_flask_hp.get()), "flask_mana": int(self.sld_flask_mana.get()), "max_mana": detected_max_mana if detected_max_mana > 0 else 0, "max_es": detected_max_es, "max_hp": exp_hp, "expected_max_hp": exp_hp, "detected_max_hp": detected_max_hp, "detected_max_mana": detected_max_mana, "detected_max_es": detected_max_es, "optical_inject_max_hp": False if exp_hp == 1 else True, "kiting": bool(self.chk_kiting.get()), "emergency_logout": bool(self.chk_chicken.get()), "auto_loot": bool(self.chk_loot.get()), "loot_radius": float(self.sld_loot_radius.get()), "auto_combo": bool(self.chk_combo.get()), "primary_skill": "rmb", "is_ranged": True, "auto_quest": bool(self.chk_quest.get()), "auto_patrol": bool(self.chk_patrol.get()), "dynamic_quest": bool(self.chk_dynamic_quest.get()), "auto_launch_client": bool(self.chk_auto_launch.get()), "poe2_exe_path": self.ent_exe_path.get().strip(), "league_name": self.var_league.get().split()[0], "auto_login": bool(self.chk_autologin.get()), "login_delay": int(self.sld_login_delay.get() * 1000), "char_slot": int(self.var_char_slot.get().split()[0]), "login_pass": self.ent_login_pass.get().strip(), "map_scenario": self._get_current_map_scenario_code(), "auto_portal_after_boss": bool(self.chk_auto_portal.get()) if hasattr(self, "chk_auto_portal") else True, } def _start_global_hotkey_listener(self): """Lắng nghe phím nóng toàn cầu: F8 (Bật/Tắt Bot), Pause/Break hoặc Ctrl+Shift+F12 (Dừng Khẩn Cấp) & F10 (Chiến Thuật).""" def _hotkey_loop(): import ctypes VK_PAUSE = 0x13 VK_F6 = 0x75 VK_F8 = 0x77 VK_F10 = 0x79 VK_F12 = 0x7B VK_SHIFT = 0x10 VK_CONTROL = 0x11 try: ctypes.windll.user32.GetAsyncKeyState(VK_PAUSE) ctypes.windll.user32.GetAsyncKeyState(VK_F6) ctypes.windll.user32.GetAsyncKeyState(VK_F8) ctypes.windll.user32.GetAsyncKeyState(VK_F10) ctypes.windll.user32.GetAsyncKeyState(VK_F12) except Exception: pass last_f8_time = 0.0 last_panic_time = 0.0 last_f10_time = 0.0 while getattr(self, "_running_hotkeys", True): try: now = time.time() # 1. Panic Killswitch: Pause/Break hoặc tổ hợp Ctrl + Shift + F12 # (Tránh kích hoạt nhầm khi người dùng bấm F12 chụp ảnh màn hình Steam) state_pause = ctypes.windll.user32.GetAsyncKeyState(VK_PAUSE) state_ctrl = ctypes.windll.user32.GetAsyncKeyState(VK_CONTROL) state_shift = ctypes.windll.user32.GetAsyncKeyState(VK_SHIFT) state_f12 = ctypes.windll.user32.GetAsyncKeyState(VK_F12) is_panic = (state_pause & 0x8000) or ((state_ctrl & 0x8000) and (state_shift & 0x8000) and (state_f12 & 0x8000)) if is_panic and (now - last_panic_time > 0.5): last_panic_time = now self.after(0, self._on_panic_stop_hotkey) # 2. Điều khiển Bật/Tắt Bot chuyên biệt qua phím F8 hoặc F6 (tránh trùng hotkey chụp màn hình trong game) state_f8 = ctypes.windll.user32.GetAsyncKeyState(VK_F8) state_f6 = ctypes.windll.user32.GetAsyncKeyState(VK_F6) is_toggle = bool((state_f8 & 0x8000) or (state_f6 & 0x8000)) if is_toggle and (now - last_f8_time > 0.5): last_f8_time = now self.after(0, self._on_bot_toggle_hotkey) state_f10 = ctypes.windll.user32.GetAsyncKeyState(VK_F10) if (state_f10 & 0x8000) and (now - last_f10_time > 0.5): last_f10_time = now self.after(0, self._on_tactical_brain_clicked) except Exception: pass time.sleep(0.04) self._running_hotkeys = True self._hk_thread = threading.Thread(target=_hotkey_loop, daemon=True) self._hk_thread.start() def _on_panic_stop_hotkey(self): """Xử lý ngắt khẩn cấp toàn bộ hệ thống (Panic Killswitch).""" self._append_log("[Panic Hotkey] 🛑 DỪNG KHẨN CẤP TOÀN BỘ HỆ THỐNG (Pause / Ctrl+Shift+F12)!") beep_panic() self.core_controller.stop() def _on_bot_toggle_hotkey(self): """Xử lý bật/tắt bot tự hành qua phím F8 hoặc giao diện.""" now = time.time() if now - getattr(self, "_last_toggle_time", 0.0) < 1.2: return self._last_toggle_time = now # Vô hiệu hóa nút bấm tạm thời (disabled 500ms) khi click để chống double click buttons_to_toggle = [] if hasattr(self, "btn_toggle_bot"): buttons_to_toggle.append(self.btn_toggle_bot) if hasattr(self, "btn_master_toggle"): buttons_to_toggle.append(self.btn_master_toggle) for btn in buttons_to_toggle: try: btn.configure(state="disabled") except Exception: pass def _re_enable_buttons(): for btn in buttons_to_toggle: try: btn.configure(state="normal") except Exception: pass try: self.after(500, _re_enable_buttons) except Exception: pass is_running = self.core_controller.is_running() or (hasattr(self, "lifecycle_manager") and self.lifecycle_manager.is_active) if is_running: self._append_log("[Hotkey F8 / F6] ⏹ DỪNG TỰ HÀNH (OFF)!") beep_bot_toggle(False) if hasattr(self, "lifecycle_manager"): self.lifecycle_manager.stop() self.core_controller.stop() if hasattr(self, "lbl_lifecycle_badge"): self.lbl_lifecycle_badge.configure(text="⏸ ĐANG TẠM DỪNG (SẴN SÀNG)", text_color="#94a3b8") if hasattr(self, "btn_master_toggle"): self.btn_master_toggle.configure(text="▶ BẬT TỰ HÀNH [F8 / F6]", fg_color=theme.c("start"), hover_color=theme.c("ok_hover")) if hasattr(self, "btn_toggle_bot"): self.btn_toggle_bot.configure(text="▶ BẮT ĐẦU BOT [F8 / F6]", fg_color=theme.c("start"), hover_color=theme.c("ok_hover")) else: self._append_log("[Hotkey F8 / F6] ▶ BẬT TỰ HÀNH TOÀN DIỆN (ON)!") beep_bot_toggle(True) cfg = self._collect_config() self.core_controller.start(cfg) if hasattr(self, "lifecycle_manager"): self.lifecycle_manager.start(cfg) if hasattr(self, "lbl_lifecycle_badge"): self.lbl_lifecycle_badge.configure(text="🟢 ĐANG TỰ HÀNH TOÀN DIỆN (1-CLICK RUN)", text_color="#34d399") if hasattr(self, "btn_master_toggle"): self.btn_master_toggle.configure(text="⏹ DỪNG TỰ HÀNH [F8 / F6]", fg_color=theme.c("alert_red"), hover_color=theme.c("alert_hover")) if hasattr(self, "btn_toggle_bot"): self.btn_toggle_bot.configure(text="⏹ DỪNG BOT [F8 / F6]", fg_color=theme.c("alert_red"), hover_color=theme.c("alert_hover")) def _on_toggle_bot_clicked(self): self._on_bot_toggle_hotkey() def _on_lifecycle_update(self, phase, text: str): """Cập nhật tiến trình vòng đời tự hành lên giao diện.""" if hasattr(self, "lbl_lifecycle_action"): self.lbl_lifecycle_action.configure(text=f" ↳ Hành động: {text}") if hasattr(self, "lbl_lifecycle_phase"): phase_val = getattr(phase, "value", str(phase)) if phase_val == "CAMPAIGN_GRIND": self.lbl_lifecycle_phase.configure( text="🛡️ GIAI ĐOẠN 1: CÀY CUỐC & LÀM QUEST CHIẾN DỊCH (PRE-ENDGAME)", text_color="#6ee7b7" ) elif phase_val == "ENDGAME_EARNING": self.lbl_lifecycle_phase.configure( text="💰 GIAI ĐOẠN 2: EARNING & FARM MAPS (ENDGAME)", text_color=theme.c("gold_light") ) elif phase_val == "AUTO_LAUNCH_CLIENT": self.lbl_lifecycle_phase.configure( text="🎮 ĐANG TỰ ĐỘNG KHỞI CHẠY CLIENT POE2...", text_color=theme.c("info_cyan") ) else: self.lbl_lifecycle_phase.configure( text="⏸ CHẾ ĐỘ CHỜ (BẤM F8 / F6 ĐỂ TIẾP TỤC)", text_color=theme.c("text_muted") ) log_key = f"{getattr(phase, 'value', phase)}|{text}" if log_key != getattr(self, "_last_lifecycle_log", ""): self._last_lifecycle_log = log_key self._append_log(f"[Tự Hành] {text}") def _on_core_state_changed(self, is_running: bool): if is_running: self._last_optical_crosscheck = time.time() self.btn_toggle_bot.configure( text="⏹ DỪNG BOT [F8 / F6]", fg_color="#ef4444", hover_color="#b91c1c" ) if hasattr(self, "btn_master_toggle"): self.btn_master_toggle.configure(text="⏹ DỪNG TỰ HÀNH [F8 / F6]", fg_color=theme.c("alert_red"), hover_color=theme.c("alert_hover")) if hasattr(self, "lbl_lifecycle_badge"): self.lbl_lifecycle_badge.configure(text="🟢 ĐANG TỰ HÀNH TOÀN DIỆN (1-CLICK RUN)", text_color="#34d399") else: self._last_optical_crosscheck = time.time() + 999999.0 self.btn_toggle_bot.configure( text="▶ BẮT ĐẦU BOT [F8 / F6]", fg_color=theme.c("start"), hover_color=theme.c("ok_hover") ) if hasattr(self, "btn_master_toggle"): self.btn_master_toggle.configure(text="▶ BẬT TỰ HÀNH [F8 / F6]", fg_color=theme.c("start"), hover_color=theme.c("ok_hover")) if hasattr(self, "lbl_lifecycle_badge"): self.lbl_lifecycle_badge.configure(text="⏸ ĐANG TẠM DỪNG (SẴN SÀNG)", text_color="#94a3b8") def _build_telemetry_dict(self, event_name: str, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: telem = { "event": event_name, "hp": self.lbl_hp.cget("text"), "mana": self.lbl_mana.cget("text"), "xyz": self.lbl_xyz.cget("text"), "area": self.lbl_area.cget("text") } if self.core_bridge and self.core_bridge.last_snapshot and self.core_bridge.last_snapshot.is_online: p = self.core_bridge.last_snapshot.player telem.update({ "current_hp": p.current_hp, "max_hp": p.max_hp, "current_mana": p.current_mana, "max_mana": p.max_mana, "current_es": p.current_es, "max_es": p.max_es, "current_spirit": p.current_spirit, "max_spirit": p.max_spirit, "pos_x": p.pos_x, "pos_y": p.pos_y, "pos_z": p.pos_z, "player_addr": p.player_addr }) if extra: telem.update(extra) return telem def _on_recalibrate_clicked(self): self._append_log("[Calibrate XYZ] Đã gửi lệnh Calibrate Tọa Độ (F11) tới Core Engine...") self._append_log(" 👉 Vui lòng nhìn lên Overlay trong game và làm theo 3 bước:") self._append_log(" 1. Chạy vòng quanh map trong 5-10 giây.") self._append_log(" 2. Đứng yên buông chuột 3 giây.") self._append_log(" 3. Bước thêm vài bước chân để Core chốt tọa độ mới.") if self.core_controller.is_running(): self.core_controller.recalibrate_xyz(self.core_bridge) else: self._append_log("[Calibrate Cảnh Báo] Bot chưa chạy! Hãy bấm '▶ BẮT ĐẦU BOT' trước.") def _on_panic_clicked(self): self._append_log("[PANIC] Nút khẩn cấp được bấm! Gửi lệnh dừng toàn diện...") self.core_bridge.send_panic_stop() self.core_controller.stop() def _append_log(self, text: str): now_str = datetime.now().strftime("%H:%M:%S") self.log_textbox.insert("end", f"[{now_str}] {text}\n") num_lines = int(self.log_textbox.index("end-1c").split(".")[0]) if num_lines > 1000: self.log_textbox.delete("1.0", f"{num_lines - 1000}.0") self.log_textbox.see("end") def _clear_log(self): self.log_textbox.delete("1.0", "end") def _open_captures_folder(self): d = os.path.abspath(self.screen_capturer.output_dir) os.makedirs(d, exist_ok=True) if os.name == "nt": os.startfile(d) def recalibrate_vitals(self, max_hp: int, max_mana: int = 0, max_es: int = 0) -> bool: """Gửi đầy đủ lệnh IPC hiệu chuẩn Vitals (Max HP, Max Mana, Max ES) sang C++ Core qua opCode 98.""" if hasattr(self, "core_bridge") and self.core_bridge and hasattr(self.core_bridge, "send_recalibrate_vitals"): return self.core_bridge.send_recalibrate_vitals(max_hp, max_mana, max_es) return False def _on_evaluate_map_mods_clicked(self, sample_text: Optional[str] = None): """Đánh giá thuộc tính bản đồ từ clipboard hoặc mẫu văn bản (100% Zero-Disk Zero-Ban).""" text_to_eval = sample_text if not text_to_eval: try: text_to_eval = self.clipboard_get() except Exception: text_to_eval = "" if not text_to_eval or not text_to_eval.strip(): text_to_eval = ( "Area has patches of Desecrated Ground\n" "30% increased Quantity of Items found in this Area\n" "20% increased Monster Pack Size" ) report = evaluate_modifiers(text_to_eval) if evaluate_modifiers else None if report: self._update_mod_filter_display(report) self._append_log(f"[ModFilter] {report.danger_summary} | Điểm an toàn: {report.safety_score}/100") if report.red_mods: self._append_log(f" ↳ 🛑 Mod Đỏ nguy hiểm: {', '.join(report.red_mods)}") if report.green_mods: self._append_log(f" ↳ 🟢 Mod Xanh thưởng cao: {', '.join(report.green_mods)}") return report def _update_mod_filter_display(self, report: Any): """Cập nhật giao diện banner Mod Filter.""" if not report: return if hasattr(self, "lbl_mod_safety_badge"): if getattr(report, "is_brick", False): self.lbl_mod_safety_badge.configure( text=f"🛑 BRICK ({report.safety_score}/100)", text_color="#fca5a5", fg_color="#7f1d1d" ) elif report.safety_score < 60: self.lbl_mod_safety_badge.configure( text=f"⚠️ NGUY HIỂM ({report.safety_score}/100)", text_color="#fde047", fg_color="#713f12" ) elif report.safety_score < 85: self.lbl_mod_safety_badge.configure( text=f"⚡ CẨN TRỌNG ({report.safety_score}/100)", text_color="#93c5fd", fg_color="#1e3a8a" ) else: self.lbl_mod_safety_badge.configure( text=f"🟢 AN TOÀN ({report.safety_score}/100)", text_color=theme.c("ok"), fg_color="#064e3b" ) if hasattr(self, "lbl_mod_report"): summary_parts = [] if report.red_mods: summary_parts.append(f"🛑 Đỏ ({len(report.red_mods)}): {'; '.join(report.red_mods[:2])}") if report.green_mods: summary_parts.append(f"🟢 Xanh ({len(report.green_mods)}): {'; '.join(report.green_mods[:2])}") if not summary_parts: summary_parts.append("Không phát hiện mod nguy hiểm hoặc tăng thưởng.") self.lbl_mod_report.configure(text=" ↳ " + " | ".join(summary_parts)) def _on_telemetry_snapshot(self, snap: CoreTelemetrySnapshot): if not snap.is_online: return cur_hp = snap.player.current_hp max_hp = snap.player.max_hp pct = snap.player.hp_percentage expected_hp = self._get_expected_max_hp() # Kiểm tra đối chiếu chéo với Cảm biến quang học (Optical Sensor Cross-Validation) has_discrepancy = False discrepancy_msg = "" opt_vitals = getattr(self.optical_sensor, "last_detected_vitals", None) if hasattr(self, "optical_sensor") and self.optical_sensor else None if hasattr(self, "optical_sensor") and self.optical_sensor and (max_hp > 0): if opt_vitals and opt_vitals.is_detected and opt_vitals.max_hp > 0: has_discrepancy, discrepancy_msg = self.optical_sensor.check_memory_discrepancy( memory_cur=cur_hp, memory_max=max_hp, optical_cur=opt_vitals.current_hp, optical_max=opt_vitals.max_hp, optical_ratio=opt_vitals.hp_ratio ) # Chế độ Chaos Inoculation (CI): max_hp == 1 và expected_hp == 1 (hoặc auto 0) # Sinh tồn phụ thuộc hoàn toàn vào ES/Ward, Max HP bất biến là 1. if max_hp == 1 and (expected_hp in (0, 1)): has_discrepancy = False # Chế độ Tự Động (expected_hp == 0): # Mọi snapshot hợp lệ từ SHM (max_hp > 0 và cur_hp > 0) đều được công nhận là VERIFIED, # TRỪ KHI phát hiện lệch lớn so với hình ảnh thực tế từ Cảm biến quang học! if expected_hp == 0: is_verified = (max_hp > 0 and cur_hp > 0 and not has_discrepancy) else: is_verified = (max_hp == expected_hp and max_hp > 0 and not has_discrepancy) if is_verified: if max_hp == 1: self.lbl_hp.configure( text=f"❤️ HP: {cur_hp} / {max_hp} (CI)", text_color=theme.c("es") ) else: self.lbl_hp.configure( text=f"❤️ HP: {cur_hp} / {max_hp} ({int(pct)}%)", text_color=theme.c("hp") ) self.bar_hp.set(max(0.0, min(1.0, pct / 100.0))) if max_hp == 1: self.bar_hp.configure(progress_color=theme.c("es")) elif pct < 35.0: self.bar_hp.configure(progress_color=theme.c("alert_red")) elif pct < 60.0: self.bar_hp.configure(progress_color=theme.c("warn")) else: self.bar_hp.configure(progress_color=theme.c("hp")) if hasattr(self, "lbl_hp_sensor_badge"): if expected_hp == 0: self.lbl_hp_sensor_badge.configure( text=f"🟢 HP Sensor: MEMORY RPM (AUTO-DETECTED: {cur_hp}/{max_hp})", text_color=theme.c("ok") ) else: self.lbl_hp_sensor_badge.configure( text="🏷️ HP Sensor: MEMORY RPM (VERIFIED)", text_color=theme.c("ok") ) elif has_discrepancy: # Phát hiện ô nhớ RAM bị lệch so với hình ảnh thực tế (vd 778/778 vs 495/503) opt_max = opt_vitals.max_hp if opt_vitals else 0 opt_cur = opt_vitals.current_hp if opt_vitals else 0 self.lbl_hp.configure( text=f"❤️ HP: {cur_hp} / {max_hp} ({int(pct)}%) ⚠️ LỆCH BỘ NHỚ (MÀN HÌNH: {opt_cur}/{opt_max})", text_color=theme.c("alert_red") ) self.bar_hp.set(max(0.0, min(1.0, pct / 100.0 if max_hp > 0 else 0.0))) self.bar_hp.configure(progress_color=theme.c("alert_red")) if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text=f"⚠️ LỆCH RAM: {max_hp} != MÀN HÌNH {opt_max} [CLICK ĐỂ ĐỒNG BỘ]", text_color=theme.c("alert_red") ) else: # Shared Memory báo max_hp không khớp expected_max_hp hoặc max_hp <= 0 (vd số rác 2/250, 125/0, 2/256): # Kích hoạt Optical Sensor làm nguồn xác thực độc lập! optical_data = None if hasattr(self, "optical_sensor") and self.optical_sensor: try: cur_opt, mhp_opt, ratio_opt = self.optical_sensor.read_hp(expected_max_hp=expected_hp) if mhp_opt > 0 and (ratio_opt > 0.0 or cur_opt > 0): optical_data = (cur_opt, mhp_opt, ratio_opt) except Exception: pass if optical_data: cur_opt, mhp_opt, ratio_opt = optical_data pct_opt = ratio_opt * 100.0 self.lbl_hp.configure( text=f"❤️ HP: {cur_opt} / {mhp_opt} ({int(pct_opt)}%) ⚠️ UNVERIFIED SHM ({cur_hp}/{max_hp} != {expected_hp})", text_color=theme.c("alert_red") ) self.bar_hp.set(max(0.0, min(1.0, ratio_opt))) self.bar_hp.configure(progress_color=theme.c("alert_red")) if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text=f"🏷️ HP Sensor: MEMORY RPM ⚠️ UNVERIFIED (OPTICAL: {cur_opt}/{mhp_opt})", text_color=theme.c("alert_red") ) else: reason = f"{max_hp} != {expected_hp}" if max_hp > 0 else "Máu <= 0" self.lbl_hp.configure( text=f"❤️ HP: {cur_hp} / {max_hp} ({int(pct)}%) ⚠️ UNVERIFIED (Kỳ vọng: {expected_hp})", text_color=theme.c("alert_red") ) self.bar_hp.set(max(0.0, min(1.0, pct / 100.0 if max_hp > 0 else 0.0))) self.bar_hp.configure(progress_color=theme.c("alert_red")) if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text=f"🏷️ HP Sensor: MEMORY RPM ⚠️ UNVERIFIED ({reason})", text_color=theme.c("alert_red") ) cur_mana = snap.player.current_mana max_mana = snap.player.max_mana mana_pct = (cur_mana / max_mana * 100.0) if max_mana > 0 else 100.0 cur_es = snap.player.current_es max_es = snap.player.max_es cur_ward = getattr(snap.player, "current_ward", 0) max_ward = getattr(snap.player, "max_ward", 0) if opt_vitals and (max_ward == 0 and getattr(opt_vitals, "max_ward", 0) > 0): cur_ward = opt_vitals.current_ward max_ward = opt_vitals.max_ward if opt_vitals and (max_es == 0 and getattr(opt_vitals, "max_shield", 0) > 0): cur_es = opt_vitals.current_shield max_es = opt_vitals.max_shield cur_sp = snap.player.current_spirit max_sp = snap.player.max_spirit if hasattr(self, "lbl_vitals_es_ward"): self.lbl_vitals_es_ward.configure(text=f"🛡️ ES: {cur_es}/{max_es} | 🎯 Ward: {cur_ward}/{max_ward}") self.lbl_mana.configure(text=f"💧 Mana: {cur_mana} / {max_mana} ({int(mana_pct)}%)") if hasattr(self, "lbl_spirit"): self.lbl_spirit.configure(text=f"Spirit: {cur_sp}/{max_sp}") if hasattr(self, "bar_mana"): self.bar_mana.set(max(0.0, min(1.0, mana_pct / 100.0))) px = snap.player.pos_x py = snap.player.pos_y pz = snap.player.pos_z now = time.time() is_stale = False if self._last_pos is not None: dx = px - self._last_pos[0] dy = py - self._last_pos[1] dz = pz - self._last_pos[2] if (dx * dx + dy * dy + dz * dz) > 0.04: self._last_pos_time = now elif now - self._last_pos_time > 4.0 and cur_hp > 0: is_stale = True else: self._last_pos_time = now self._last_pos = (px, py, pz) if is_stale: self.lbl_xyz.configure( text=f"📍 XYZ: ({px:.1f}, {py:.1f}, {pz:.1f}) ⚠️ ĐỨNG IM (Bấm '🎯 Định Vị XYZ')", text_color=theme.c("alert_red") ) else: self.lbl_xyz.configure( text=f"📍 XYZ: ({px:.1f}, {py:.1f}, {pz:.1f}) 🟢 Realtime", text_color=theme.c("start") ) # Realtime Character State từ snap.player.movement_flags mov_flags = getattr(snap.player, "movement_flags", 0) state_icon, state_name, state_color = self.parse_character_state(mov_flags) if hasattr(self, "lbl_char_state"): self.lbl_char_state.configure( text=f"{state_icon} Trạng thái: [{state_name}]", text_color=state_color ) area_name, area_level = self._resolve_playable_area(snap) if hasattr(self, "lbl_area"): lvl = f" (Lv {area_level})" if area_level else "" self.lbl_area.configure(text=f"Khu vực: {area_name}{lvl}") # Đếm quái vật: entities_monster_count quái gần (monsters_alive_total tổng) entities_monster_count = getattr(snap, "entities_monster_count", snap.monsters_count) monsters_alive_total = getattr(snap, "monsters_alive_total", 0) if monsters_alive_total == 0: monsters_alive_total = entities_monster_count if hasattr(self, "lbl_monsters"): self.lbl_monsters.configure( text=f"👾 Quái vật: {entities_monster_count} quái gần ({monsters_alive_total} tổng)", text_color=theme.c("alert_red") if entities_monster_count > 5 else ("#f59e0b" if entities_monster_count > 0 else "#34d399") ) # Kế hoạch di chuyển: Banner mục tiêu quester_active_objective = "Tự do thám hiểm" plan_area = area_name if area_name != "--" else "" if plan_area and hasattr(self, "auto_quester"): try: quester = self.auto_quester if quester.is_town_area("", plan_area): target_z = self.var_quest_zone.get() if hasattr(self, "var_quest_zone") else "Map nhiệm vụ" quester_active_objective = f"[Làng {plan_area}] Mở World Map (U) -> Vào {target_z}" else: obj = quester.resolve_active_objective("", plan_area, area_level) if obj: quester_active_objective = f"[{obj.area_name}] {obj.action_type}: {obj.target_name} ({obj.description})" else: quester_active_objective = f"[{plan_area}] Thám hiểm và dọn quái" except Exception: quester_active_objective = f"[{plan_area}] Tiến tới mục tiêu bản đồ" if hasattr(self, "lbl_movement_plan"): self.lbl_movement_plan.configure(text=f"Kế hoạch: {quester_active_objective}") if plan_area and hasattr(self, "lbl_active_quest"): try: quester = self.auto_quester if quester.is_town_area("", plan_area): target_z = self.var_quest_zone.get() if hasattr(self, "var_quest_zone") else "Map nhiệm vụ" self.lbl_active_quest.configure( text=f" 👉 [LÀNG]: Đang ở {plan_area} -> Bấm '🚀 MỞ WORLD MAP U' để vào {target_z}!", text_color=theme.c("gold_light") ) else: q_banner = quester.get_status_banner("", plan_area, area_level) self.lbl_active_quest.configure(text=f" 👉 {q_banner}", text_color="#34d399") except Exception: pass radar_text = f"📡 Radar: {snap.monsters_count} Quái | {snap.items_count} Đồ rơi | {snap.portals_count} Cổng ({snap.entity_count} tổng)" if snap.entity_count > 0: self.lbl_radar.configure(text=radar_text, text_color="#38bdf8") else: self.lbl_radar.configure(text="📡 Radar: Đang quét thực thể quanh người chơi...", text_color="#f59e0b") # Tự động chụp màn hình khi dính burst damage lớn if self._last_hp > 0 and cur_hp < self._last_hp and self.chk_cap_burst.get(): dmg = self._last_hp - cur_hp if max_hp > 0 and (dmg / max_hp) >= (self.sld_burst.get() / 100.0): now = time.time() if now - self._last_burst_capture > 2.0: self._last_burst_capture = now telem = self._build_telemetry_dict("BURST_DAMAGE", {"damage": dmg}) path = self.screen_capturer.capture_screenshot("BURST_DAMAGE", telem) if path: self._append_log(f"[AutoCapture] Phát hiện sốc sát thương (-{dmg} HP) -> Đã lưu ảnh: {os.path.basename(path)}") self._last_hp = cur_hp # Cập nhật Telemetry sang Agentic Brain Coordinator và Autonomous Lifecycle Manager if hasattr(self, "agentic_coordinator"): self.agentic_coordinator.on_telemetry_snapshot(snap) if hasattr(self, "lifecycle_manager"): self.lifecycle_manager.on_telemetry_snapshot(snap) if self._watcher_sentinel is not None: try: watcher_data = { "current_hp": snap.player.current_hp, "max_hp": snap.player.max_hp, "current_es": snap.player.current_es, "max_es": snap.player.max_es, "current_mana": snap.player.current_mana, "max_mana": snap.player.max_mana, "current_spirit": snap.player.current_spirit, "max_spirit": snap.player.max_spirit, "pos_x": snap.player.pos_x, "pos_y": snap.player.pos_y, "pos_z": snap.player.pos_z, "monsters_count": snap.monsters_count, } violations = self._watcher_sentinel.check_all(watcher_data) if violations: self._watcher_violation_count += 1 if self._watcher_violation_count <= 5 or self._watcher_violation_count % 50 == 0: self._append_log(f"[AgentWatcher] ⚠️ Vi phạm bất biến #{self._watcher_violation_count}: {', '.join(violations[:3])}") except Exception: pass def _on_agentic_toggle(self): enabled = bool(self.chk_agentic_brain.get()) if hasattr(self, "agentic_coordinator"): self.agentic_coordinator.set_enabled(enabled) status_txt = "ĐÃ BẬT" if enabled else "ĐÃ TẮT" self._append_log(f"[AgenticBrain] Chế độ tự học & thích ứng: {status_txt}") def _on_tactical_brain_clicked(self): self._append_log("[TacticalBrain] 🧠 Đang phân tích bối cảnh: 'Tốt nhất bây giờ mình nên làm gì?'...") self.btn_tactical_brain.configure(text="⏳ Đang suy nghĩ...", state="disabled") def _bg(): rec = self.tactical_brain.advise_best_action_sync(include_screen=True) self.after(0, lambda: self._display_tactical_recommendation(rec)) threading.Thread(target=_bg, daemon=True).start() def _display_tactical_recommendation(self, rec: TacticalRecommendation): self.btn_tactical_brain.configure(text="🧠 Nên Làm Gì? (F10)", state="normal") color_map = { "LOW": "#10b981", "MODERATE": "#f59e0b", "HIGH": "#ea580c", "CRITICAL": "#ef4444" } color = color_map.get(rec.danger_level.value, "#8b5cf6") self._append_log(f"[TacticalBrain] 🎯 {rec.primary_directive.value} (Nguy cơ: {rec.danger_level.value}): {rec.action_summary}") self._append_log(f" ↳ Lý do: {rec.detailed_reasoning}") self.lbl_tactical.configure( text=f"⚡ {rec.primary_directive.value} [{rec.danger_level.value}]: {rec.action_summary}", text_color=color ) def _on_log_event(self, evt): if hasattr(self, "agentic_coordinator"): self.agentic_coordinator.on_log_event(evt) if hasattr(self, "lifecycle_manager") and getattr(evt, "event_type", "") in ("ZONE_CHANGE", "SCENE_CHANGED"): clean_name = ( getattr(evt, "area_name", "") or getattr(evt, "message", "").replace("Đã chuyển đến khu vực: ", "").strip() ) self.lifecycle_manager.on_area_changed(getattr(evt, "area_code", "") or "", clean_name) elif hasattr(self, "lifecycle_manager") and getattr(evt, "event_type", "") == "AREA_GENERATED": code = getattr(evt, "area_code", "") or getattr(evt, "message", "") if code: self.lifecycle_manager.current_area_code = code if evt.event_type in ("PLAYER_DEATH", "ZONE_CHANGE"): self._append_log(f"[{evt.event_type}] {evt.message}") elif "Items identified" in evt.message or "Item identified" in evt.message: self._append_log(f"[ItemLifecycle] {evt.message}") def _on_test_item_appraisal_clicked(self): if not hasattr(self, "agentic_coordinator"): return self._append_log("[WhatIsBetter] 🔍 Đang tiến hành thẩm định so sánh trang bị mẫu cho võ sư Monk (Martial Artist)...") res = self.agentic_coordinator.simulate_sample_item_appraisal() self._append_log(f"[WhatIsBetter] Phán quyết: {res.headline_verdict}") self._append_log(f" ↳ Quyết định: {res.decision} (DPS: {res.dps_delta_pct:+.1f}%, EHP: {res.ehp_delta_pct:+.1f}%)") self._append_log(f" ↳ Chi tiết: {res.detailed_reasoning}") if res.resist_impact_warning: self._append_log(f" ↳ ⚠️ {res.resist_impact_warning}") self._append_log(f" ↳ Xử lý đồ cũ: {res.salvage_or_trade_advice}") if hasattr(self, "lbl_item_eval"): self.lbl_item_eval.configure( text=f" ↳ Phán quyết: {res.decision} (DPS: {res.dps_delta_pct:+.1f}%, EHP: {res.ehp_delta_pct:+.1f}%)", text_color="#34d399" if res.decision == "EQUIP_UPGRADE" else "#f59e0b" ) def _on_test_item_triage_clicked(self): if not hasattr(self, "agentic_coordinator"): return self._append_log("[ItemTriage] 📜 Đang thực hiện sàng lọc tiền giám định cho 3 phôi đồ mẫu rơi trong Map...") # 1. Spiked Fists (S-Tier phôi găng tay Martial Artist) r1 = self.agentic_coordinator.triage_unidentified_item( base_name="Spiked Fists", item_level=79, rarity="Rare", slot="gloves", scrolls_available=20, in_combat_zone=True ) self._append_log(f" 1. [S-TIER GLOVES] {r1.item_base_name} (iLvl {r1.item_level}): Kênh ➔ {r1.recommended_channel.value} ({r1.priority_reason})") # 2. Amethyst Ring (S-Tier phôi nhẫn kháng Chaos) r2 = self.agentic_coordinator.triage_unidentified_item( base_name="Amethyst Ring", item_level=76, rarity="Rare", slot="ring", scrolls_available=20, in_combat_zone=True ) self._append_log(f" 2. [S-TIER JEWELLERY] {r2.item_base_name} (iLvl {r2.item_level}): Kênh ➔ {r2.recommended_channel.value} ({r2.priority_reason})") # 3. Phôi thông thường (Varnished Coat) r3 = self.agentic_coordinator.triage_unidentified_item( base_name="Varnished Coat", item_level=68, rarity="Rare", slot="body_armour", scrolls_available=20, in_combat_zone=True ) self._append_log(f" 3. [COMMON BODY] {r3.item_base_name} (iLvl {r3.item_level}): Kênh ➔ {r3.recommended_channel.value} ({r3.priority_reason})") if hasattr(self, "lbl_triage_status"): self.lbl_triage_status.configure( text=" ↳ Sàng lọc phôi: Đã phân luồng Spiked Fists (Scroll) & Amethyst Ring / Coat (The Hooded One)", text_color=theme.c("info_cyan") ) def _on_toggle_item_inspector_clicked(self): """Kích hoạt hoặc dừng Universal Auto-Hover Item Inspector HUD (WS_EX_NOACTIVATE).""" if hasattr(self, "_item_inspector_proc") and self._item_inspector_proc and self._item_inspector_proc.poll() is None: self._append_log("[ItemInspector] ⏹️ Đang dừng Auto-Hover Item Inspector HUD...") try: self._item_inspector_proc.terminate() self._item_inspector_proc.wait(timeout=1.0) except Exception: pass self._item_inspector_proc = None self.btn_toggle_item_inspector.configure( text="👁️ Bật Hover Inspector", fg_color="#059669" ) self._append_log("[ItemInspector] Đã tắt Auto-Hover Item Inspector.") else: self._append_log("[ItemInspector] ▶️ Đang kích hoạt Universal Auto-Hover Item Inspector HUD (WS_EX_NOACTIVATE)...") try: import subprocess self._item_inspector_proc = subprocess.Popen( [sys.executable, "run_item_inspector.py"], cwd=os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) ) self.btn_toggle_item_inspector.configure( text="⏹️ Tắt Hover Inspector", fg_color="#dc2626" ) self._append_log("[ItemInspector] ✅ Đã khởi chạy Auto-Hover Item Inspector thành công (Zero-Focus Theft).") except Exception as e: self._append_log(f"[ItemInspector] Lỗi khởi chạy: {e}") def _hot_reload_agent_modules(self): """Hot-reload các module liên quan đến Agent Watcher & Optical Sensor (Rule 13).""" import importlib try: import src.agentic_engine.invariant_sentinel as is_module importlib.reload(is_module) import src.agentic_engine.flight_recorder as fr_module importlib.reload(fr_module) import src.assistant_tool.agent_watcher as aw_module importlib.reload(aw_module) from src.agentic_engine.invariant_sentinel import InvariantSentinel from src.agentic_engine.flight_recorder import MultimodalIncidentRecorder self._watcher_sentinel = InvariantSentinel() self._watcher_recorder = MultimodalIncidentRecorder( output_dir="debug_harness/incidents", screen_capturer=self.screen_capturer ) import src.assistant_tool.optical_hp_sensor as ohs_module importlib.reload(ohs_module) from src.assistant_tool.optical_hp_sensor import OpticalHPSensor if hasattr(self, "optical_sensor") and self.optical_sensor: try: self.optical_sensor.stop_background_worker() except Exception: pass self.optical_sensor = OpticalHPSensor() self.optical_sensor.start_background_worker(interval=0.5) self._append_log("[HotReload] ♻️ Đã nạp lại thành công InvariantSentinel, IncidentRecorder & OpticalHPSensor (winocr).") except Exception as e: self._append_log(f"[HotReload] Lỗi khi nạp lại module: {e}") def _periodic_task(self): """Công việc định kỳ: kiểm tra chụp ảnh interval, nhịp tim và tự động thích ứng cấu hình game.""" now = time.time() if (hasattr(self, "optical_sensor") and self.optical_sensor and (self.core_controller.is_running() or self._last_optical_crosscheck == 0.0) and now - self._last_optical_crosscheck >= 1.5): self._last_optical_crosscheck = now try: vitals = self.optical_sensor.detect_vitals_from_screen() if vitals and vitals.max_hp > 0: optical_max_hp = vitals.max_hp optical_cur_hp = vitals.current_hp optical_mana = vitals.max_mana if hasattr(vitals, 'max_mana') and vitals.max_mana > 0 else 0 optical_es = vitals.max_shield if hasattr(vitals, 'max_shield') and vitals.max_shield > 0 else 0 optical_ward = getattr(vitals, 'max_ward', 0) # Kiểm tra tính hợp lý của Vitals trước khi gửi recalibrate sang C++ Core: # Không gửi nếu max_hp < 50 (trừ CI max_hp=1) hoặc current_hp > max_hp hoặc max_hp > 30000 hoặc lặp bất thường (ví dụ 7793) if self.is_valid_vitals(optical_max_hp, optical_cur_hp): shm_max_hp = 0 shm_max_mana = 0 shm_max_es = 0 last_snap = getattr(self.core_bridge, "last_snapshot", None) if last_snap and last_snap.is_online: shm_max_hp = last_snap.player.max_hp shm_max_mana = last_snap.player.max_mana shm_max_es = last_snap.player.max_es # Kiểm tra trường hợp CI (Chaos Inoculation): max_hp == 1 và max_shield > 0 if optical_max_hp == 1 and optical_es > 0: if shm_max_hp != 1 or shm_max_es != optical_es or (optical_mana > 0 and shm_max_mana != optical_mana): self.core_bridge.send_recalibrate_vitals(1, optical_mana, optical_es) self._optical_correction_count += 1 self._optical_verified = False if hasattr(self, "entry_max_hp"): self.entry_max_hp.delete(0, "end") self.entry_max_hp.insert(0, "CI") self._append_log( f"[Proactive Auto-Detect] ⚡ Tự động hiệu chuẩn Vitals từ màn hình sang Core: " f"HP=1, Mana={optical_mana}, ES={optical_es} | [CI] Shield={optical_es}, Ward={optical_ward}" ) # Hiển thị lên GUI: [CI] Shield={optical_max_shield}, Ward={optical_max_ward} if hasattr(self, "lbl_vitals_es_ward"): self.lbl_vitals_es_ward.configure(text=f"🛡️ ES: {optical_es}/{optical_es} | 🎯 Ward: {optical_ward}/{optical_ward}") if hasattr(self, "lbl_hp") and shm_max_hp == 0: self.lbl_hp.configure( text=f"❤️ HP: 1 / 1 (CI) | 🛡️ [CI] Shield={optical_es}, Ward={optical_ward}", text_color=theme.c("info_cyan") ) if hasattr(self, "lbl_hp_sensor_badge"): rpm_tag = " MEMORY RPM (VERIFIED)" if shm_max_hp > 0 else "" self.lbl_hp_sensor_badge.configure( text=f"🏷️ HP Sensor:{rpm_tag} [CI] Shield={optical_es}, Ward={optical_ward}", text_color=theme.c("info_cyan") ) else: # Chủ động tự điều chỉnh Vitals (Proactive Auto-Calibration): # Khi shm_max_hp == 0 (Core chưa khóa được máu người chơi!) HOẶC shm_max_hp != optical_max_hp: if hasattr(self, "lbl_vitals_es_ward") and (optical_es > 0 or optical_ward > 0): self.lbl_vitals_es_ward.configure(text=f"🛡️ ES: {optical_es}/{optical_es} | 🎯 Ward: {optical_ward}/{optical_ward}") hp_desync = (shm_max_hp == 0 or shm_max_hp != optical_max_hp) mana_desync = optical_mana > 0 and (shm_max_mana == 0 or shm_max_mana != optical_mana) es_desync = optical_es > 0 and (shm_max_es == 0 or shm_max_es != optical_es) if hp_desync or mana_desync or es_desync: success = self.core_bridge.send_recalibrate_vitals(optical_max_hp, optical_mana, optical_es) self._optical_correction_count += 1 self._optical_verified = False # Tự động điền optical_max_hp vào ô nhập entry_max_hp (nếu max_hp == 1 thì điền "CI") if hasattr(self, "entry_max_hp"): self.entry_max_hp.delete(0, "end") self.entry_max_hp.insert(0, "CI" if optical_max_hp == 1 else str(optical_max_hp)) # Ghi log chuẩn theo yêu cầu: self._append_log( f"[Proactive Auto-Detect] ⚡ Tự động hiệu chuẩn Vitals từ màn hình sang Core: " f"HP={optical_max_hp}, Mana={optical_mana}, ES={optical_es}" ) elif not self._optical_verified: self._optical_verified = True self._append_log( f"[Optical CrossCheck] ✅ Vitals đồng bộ: HP={shm_max_hp}, Mana={shm_max_mana}, ES={shm_max_es} | Verified!" ) except Exception: pass signal_file = os.path.join(os.path.dirname(__file__), "../../bin/Release/.reload_signal") if os.path.exists(signal_file): try: os.remove(signal_file) self._hot_reload_agent_modules() except Exception: pass if self.chk_cap_periodic.get() and self.core_controller.is_running(): if now - self._last_periodic_capture >= 30.0: self._last_periodic_capture = now telem = self._build_telemetry_dict("PERIODIC_ANALYSIS") path = self.screen_capturer.capture_screenshot("PERIODIC_ANALYSIS", telem) if path: self._append_log(f"[AutoCapture] Chụp định kỳ 30s để tích lũy dữ liệu: {os.path.basename(path)}") # Tự động thích ứng khi người dùng đổi thiết lập trong game if self.chk_auto_detect_move.get(): cfg_path = Config.find_poe2_config_path() if cfg_path and os.path.exists(cfg_path): try: mtime = os.path.getmtime(cfg_path) if not hasattr(self, "_last_cfg_mtime"): self._last_cfg_mtime = mtime elif mtime != self._last_cfg_mtime: self._last_cfg_mtime = mtime mode, _ = Config.detect_poe2_movement_mode() if mode != self.var_move_mode.get(): self.var_move_mode.set(mode) badge_text = "⚡ ĐÃ TỰ ĐỘNG THÍCH ỨNG: ⌨️ BÀN PHÍM (WASD)" if mode == "wasd" else "⚡ ĐÃ TỰ ĐỘNG THÍCH ỨNG: 🖱️ CHUỘT (Click-to-Move)" badge_color = theme.c("info_cyan") if mode == "wasd" else theme.c("ok") self.lbl_move_badge.configure(text=badge_text, text_color=badge_color) self._append_log(f"[LiveAdapt] Phát hiện thay đổi trong game! Tự động chuyển sang: {'⌨️ BÀN PHÍM (WASD)' if mode == 'wasd' else '🖱️ CHUỘT (Click-to-Move)'}") except Exception: pass # Cập nhật trạng thái hiển thị của Agentic Brain if hasattr(self, "agentic_coordinator") and hasattr(self, "lbl_ai_action"): try: ai_state = self.agentic_coordinator.get_dashboard_state() self.lbl_ai_action.configure( text=f" ↳ Khuyến nghị: 🎯 {ai_state['recommended_action']} | DPS thực chiến: {ai_state['current_dps']:.0f}" ) self.lbl_ai_tuner.configure( text=f" ↳ Phản xạ: Flask {int(ai_state['tuned_flask_hp']*100)}% | Dodge {int(ai_state['tuned_dodge_burst']*100)}% | Kite {ai_state['tuned_kiting_dist']:.1f}m" ) if ai_state.get("hooded_one_notice"): self.lbl_hooded_notice.configure( text=f" ↳ The Hooded One: ⚡ {ai_state['hooded_one_notice'][:45]}...", text_color=theme.c("info_cyan") ) if hasattr(self, "lbl_triage_status") and ai_state.get("unidentified_items_count") is not None: unid_cnt = ai_state.get("unidentified_items_count", 0) self.lbl_triage_status.configure( text=f" ↳ Đồ chưa giám định: {unid_cnt} món | Kênh: Tự động gom về The Hooded One (Làng)", text_color="#c7d2fe" if unid_cnt == 0 else "#fbbf24" ) except Exception: pass # Kiểm tra và cập nhật cảm biến HP quang học (Optical HP Sensor) nếu SHM không hoạt động hoặc unverified if hasattr(self, "optical_sensor") and self.optical_sensor: expected_hp = self._get_expected_max_hp() self.optical_sensor.expected_max_hp = expected_hp last_snap = getattr(self.core_bridge, "last_snapshot", None) if self.core_bridge else None if expected_hp == 0: is_shm_verified = bool( last_snap and last_snap.is_online and last_snap.player.max_hp > 0 and last_snap.player.current_hp > 0 ) else: is_shm_verified = bool( last_snap and last_snap.is_online and last_snap.player.max_hp == expected_hp and last_snap.player.max_hp > 0 ) # Nếu SHM chưa online HOẶC online nhưng báo max_hp sai lệch (unverified / rác 250, 256, 0): # Kích hoạt Optical Sensor làm nguồn xác thực độc lập! if not is_shm_verified: try: cur, mhp, ratio = self.optical_sensor.read_hp(expected_max_hp=expected_hp) if mhp > 0 and (ratio > 0.0 or cur > 0): pct = ratio * 100.0 if expected_hp > 0 and last_snap and last_snap.is_online and last_snap.player.max_hp != expected_hp: self.lbl_hp.configure( text=f"❤️ HP: {cur} / {mhp} ({int(pct)}%) ⚠️ [OPTICAL ACTIVE | SHM {last_snap.player.current_hp}/{last_snap.player.max_hp} MISMATCH]", text_color=theme.c("alert_red") ) self.bar_hp.configure(progress_color=theme.c("alert_red")) if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text="🏷️ HP Sensor: OPTICAL (Instant 2ms) ⚠️ SHM UNVERIFIED", text_color=theme.c("alert_red") ) else: if mhp == 1: self.lbl_hp.configure( text=f"❤️ HP: {cur} / {mhp} (CI)", text_color="#f1f5f9" ) else: self.lbl_hp.configure( text=f"❤️ HP: {cur} / {mhp} ({int(pct)}%)", text_color="#f1f5f9" ) if pct < 35.0: self.bar_hp.configure(progress_color=theme.c("alert_red")) elif pct < 60.0: self.bar_hp.configure(progress_color="#f59e0b") else: self.bar_hp.configure(progress_color="#10b981") if hasattr(self, "lbl_hp_sensor_badge"): self.lbl_hp_sensor_badge.configure( text="🏷️ HP Sensor: OPTICAL (Instant 2ms)", text_color=theme.c("info_cyan") ) self.bar_hp.set(max(0.0, min(1.0, ratio))) except Exception: pass self.after(1000, self._periodic_task) def on_window_x_clicked(self): """Khi bấm nút X trên cửa sổ, thu nhỏ về System Tray thay vì tắt hẳn.""" if hasattr(self, "tray_manager") and self.tray_manager and self.tray_manager.is_running(): self.withdraw() else: self.quit_app_completely() def quit_app_completely(self): """Thoát hoàn toàn ứng dụng và dọn dẹp các tiến trình phụ.""" if hasattr(self, "tray_manager") and self.tray_manager: try: self.tray_manager.stop() except Exception: pass self._running_hotkeys = False if hasattr(self, "_item_inspector_proc") and self._item_inspector_proc and self._item_inspector_proc.poll() is None: try: self._item_inspector_proc.terminate() except Exception: pass if hasattr(self, "optical_sensor") and self.optical_sensor: try: self.optical_sensor.stop_background_worker() except Exception: pass if hasattr(self, "log_watcher") and self.log_watcher: self.log_watcher.stop() if self.core_controller.is_running(): self.core_controller.stop() self.core_bridge.stop() self.destroy() def on_closing(self): self.quit_app_completely() def main(start_in_tray: Optional[bool] = None): if start_in_tray is None: start_in_tray = "--show-gui" not in sys.argv app = ControlCenterApp(start_in_tray=start_in_tray) app.protocol("WM_DELETE_WINDOW", app.on_window_x_clicked) app.mainloop() if __name__ == "__main__": main()