""" Module Cảm Biến Quang Học Quả Cầu Máu (Optical Health Globe Sensor) Đo lường lượng máu (HP) tức thời trong Path of Exile 2 thông qua kỹ thuật chụp màn hình cực nhanh bằng Win32 GDI BitBlt (<2ms), multi-slit sampling loại bỏ số hiển thị máu, đảm bảo hoàn toàn an toàn: ZERO memory write, ZERO injection. """ import ctypes import re import threading import time from dataclasses import dataclass from typing import Any, Optional, Tuple try: import win32con import win32gui import win32ui except ImportError: win32con = None win32gui = None win32ui = None try: from PIL import Image except ImportError: Image = None # type: ignore[assignment] @dataclass class OpticalVitals: """Kết quả nhận diện Vitals trực tiếp từ màn hình game POE2.""" current_hp: int = 0 max_hp: int = 0 hp_ratio: float = 0.0 current_shield: int = 0 max_shield: int = 0 current_ward: int = 0 max_ward: int = 0 current_mana: int = 0 max_mana: int = 0 current_spirit: int = 0 max_spirit: int = 0 current_rage: int = 0 max_rage: int = 0 raw_text: str = "" is_detected: bool = False detection_method: str = "none" # "ocr", "winocr_screen", "raster_match", "ratio_estimate", "level_estimate" from src.common.win32_window import ( attach_to_default_desktop, ensure_window_focus, find_poe2_window as _find_poe2_window, ) class OpticalHPSensor: """ Cảm biến quang học đo lường HP từ Quả Cầu Máu (Health Globe) ở góc dưới bên trái màn hình. Sử dụng kỹ thuật Multi-Slit Sampling tại 3 trục cột (X_center - 15, X_center, X_center + 15) kết hợp nội suy để hoàn toàn loại trừ ảnh hưởng của chữ số ("1234/1234") đè lên quả cầu. Hỗ trợ cả Win32 GDI BitBlt siêu tốc (<2ms) lẫn PrintWindow với PW_RENDERFULLCONTENT khi chạy ngầm. Tích hợp luồng quét ngầm bất đồng bộ (2Hz) không bao giờ làm giật lag giao diện Tkinter. """ PW_CLIENTONLY: int = 0x00000001 PW_RENDERFULLCONTENT: int = 0x00000002 def __init__( self, expected_max_hp: int = 0, top_margin_ratio: float = 0.10, bottom_margin_ratio: float = 0.90, slit_offset: int = 15, ): self.expected_max_hp = expected_max_hp self.top_margin_ratio = top_margin_ratio self.bottom_margin_ratio = bottom_margin_ratio self.slit_offset = slit_offset self.last_ratio: float = 0.0 self.last_current_hp: int = 0 self.last_max_hp: int = expected_max_hp self.last_latency_ms: float = 0.0 self.cached_hwnd: Optional[int] = None self.last_detected_vitals: OpticalVitals = OpticalVitals() # Cơ chế luồng quét ngầm bất đồng bộ (Background OCR / Slit Sensor worker thread - 2Hz) self._bg_thread: Optional[threading.Thread] = None self._bg_running: bool = False self._bg_interval: float = 0.5 # 2Hz self._bg_lock: threading.Lock = threading.Lock() def start_background_worker(self, interval: float = 0.5): """ Khởi chạy luồng quét ngầm bất đồng bộ (Background OCR / Slit Sensor worker thread). Định kỳ 2Hz (interval=0.5s) cập nhật last_detected_vitals mà không bao giờ làm treo hoặc giật khựng giao diện người dùng Tkinter! """ with self._bg_lock: if self._bg_running and self._bg_thread and self._bg_thread.is_alive(): return self._bg_interval = interval self._bg_running = True self._bg_thread = threading.Thread( target=self._bg_worker_loop, name="OpticalSensorBgWorker", daemon=True ) self._bg_thread.start() def stop_background_worker(self, timeout: float = 1.0): """Dừng an toàn luồng quét ngầm bất đồng bộ.""" with self._bg_lock: self._bg_running = False if self._bg_thread and self._bg_thread.is_alive(): try: self._bg_thread.join(timeout=timeout) except Exception: pass self._bg_thread = None def _bg_worker_loop(self): """Vòng lặp quét ngầm 2Hz liên tục cập nhật last_detected_vitals.""" attach_to_default_desktop() while self._bg_running: try: v = self.detect_vitals_from_screen() if v and v.is_detected and v.max_hp > 0: with self._bg_lock: self.last_detected_vitals = v except Exception: pass sleep_chunks = max(1, int(self._bg_interval / 0.05)) for _ in range(sleep_chunks): if not self._bg_running: break time.sleep(0.05) def find_poe2_window(self) -> Optional[int]: """ Xác định HWND của cửa sổ game Path of Exile 2: Sử dụng thư viện dùng chung src.common.win32_window. Lưu cache self.cached_hwnd khi tìm thấy. """ if self.cached_hwnd and win32gui: try: if win32gui.IsWindow(self.cached_hwnd) and win32gui.IsWindowVisible(self.cached_hwnd): return self.cached_hwnd except Exception: self.cached_hwnd = None hwnd = _find_poe2_window() if hwnd: self.cached_hwnd = hwnd return hwnd return None def bring_game_to_foreground(self, hwnd: Optional[int] = None) -> bool: """Tự động chuyển tab và đưa cửa sổ Path of Exile 2 lên foreground.""" target_hwnd = hwnd or self.find_poe2_window() if not target_hwnd: return False return ensure_window_focus(target_hwnd) def get_globe_roi(self, hwnd: int) -> Tuple[int, int, int, int]: """ Đo kích thước cửa sổ game, tính toán vùng ROI của Quả Cầu Máu ở góc dưới bên trái màn hình: scale = client_h / 1080.0 rw = int(150 * scale), rh = int(150 * scale) rx = int(15 * scale), ry = client_h - int(170 * scale) """ _client_w = 1920 client_h = 1080 if win32gui and hwnd: try: rect = win32gui.GetClientRect(hwnd) w = rect[2] - rect[0] h = rect[3] - rect[1] if w > 100 and h > 100: _client_w = w client_h = h except Exception: pass scale = client_h / 1080.0 rw = max(10, int(150 * scale)) rh = max(10, int(150 * scale)) rx = int(15 * scale) ry = max(0, client_h - int(170 * scale)) return (rx, ry, rw, rh) def get_mana_roi(self, hwnd: int) -> Tuple[int, int, int, int]: """ Đo kích thước cửa sổ game, tính toán vùng ROI của Quả Cầu Mana ở góc dưới bên phải màn hình: scale = client_h / 1080.0 rw = max(10, int(150 * scale)), rh = max(10, int(150 * scale)) rx = max(0, client_w - int(165 * scale)), ry = max(0, client_h - int(170 * scale)) """ client_w = 1920 client_h = 1080 if win32gui and hwnd: try: rect = win32gui.GetClientRect(hwnd) w = rect[2] - rect[0] h = rect[3] - rect[1] if w > 100 and h > 100: client_w = w client_h = h except Exception: pass scale = client_h / 1080.0 rw = max(10, int(150 * scale)) rh = max(10, int(150 * scale)) rx = max(0, client_w - int(165 * scale)) ry = max(0, client_h - int(170 * scale)) return (rx, ry, rw, rh) def capture_full_screen(self, hwnd: Optional[int] = None) -> Optional[Any]: """ Chụp toàn bộ cửa sổ game POE2 bằng Desktop DC user32.GetDC(0) kết hợp user32.GetWindowRect(hwnd). Khắc phục triệt để việc Direct3D 12 flip model trả về bitmap đen. Trả về đối tượng PIL Image (RGBA) hoặc None nếu thất bại. """ attach_to_default_desktop() target_hwnd = hwnd or self.find_poe2_window() if not target_hwnd: return None user32 = getattr(ctypes.windll, "user32", None) gdi32 = getattr(ctypes.windll, "gdi32", None) if not user32 or not gdi32: return None from ctypes import wintypes rect = wintypes.RECT() if not user32.GetWindowRect(target_hwnd, ctypes.byref(rect)): return None w = rect.right - rect.left h = rect.bottom - rect.top if w <= 0 or h <= 0: return None desk_dc = None mem_dc = None bmp = None old_bmp = None img_res = None try: try: hdesk = user32.OpenDesktopW("Default", 0, False, 0x10000000) if hdesk: user32.SetThreadDesktop(hdesk) except Exception: pass desk_dc = user32.GetDC(0) if not desk_dc: return None mem_dc = gdi32.CreateCompatibleDC(desk_dc) if not mem_dc: return None bmp = gdi32.CreateCompatibleBitmap(desk_dc, w, h) if not bmp: return None old_bmp = gdi32.SelectObject(mem_dc, bmp) src_x = max(0, rect.left) src_y = max(0, rect.top) SRCCOPY = 0x00CC0020 if not gdi32.BitBlt(mem_dc, 0, 0, w, h, desk_dc, src_x, src_y, SRCCOPY): return None 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), ] bmi = BITMAPINFOHEADER() bmi.biSize = ctypes.sizeof(BITMAPINFOHEADER) bmi.biWidth = w bmi.biHeight = -h bmi.biPlanes = 1 bmi.biBitCount = 32 bmi.biCompression = 0 buf = ctypes.create_string_buffer(w * h * 4) lines = gdi32.GetDIBits(mem_dc, bmp, 0, h, buf, ctypes.byref(bmi), 0) if lines > 0 and Image is not None: img_res = Image.frombuffer("RGBA", (w, h), bytes(buf), "raw", "BGRA", 0, 1) except Exception: img_res = None finally: if old_bmp and mem_dc: try: gdi32.SelectObject(mem_dc, old_bmp) except Exception: pass if bmp: try: gdi32.DeleteObject(bmp) except Exception: pass if mem_dc: try: gdi32.DeleteDC(mem_dc) except Exception: pass if desk_dc: try: user32.ReleaseDC(0, desk_dc) except Exception: pass return img_res def capture_roi_bytes( self, hwnd: int, rx: Any = None, ry: Optional[int] = None, rw: Optional[int] = None, rh: Optional[int] = None, fallback_printwindow: bool = False ) -> Optional[bytes]: """ Chụp nhanh vùng ROI bằng Win32 GDI BitBlt (<2ms). Sử dụng Desktop DC user32.GetDC(0) kết hợp user32.GetWindowRect(hwnd) để chụp chính xác vùng hiển thị của game POE2, khắc phục triệt để việc Direct3D 12 flip model trả về bitmap đen. Hỗ trợ cả tuple ROI: capture_roi_bytes(hwnd, roi) lẫn (hwnd, rx, ry, rw, rh). """ if isinstance(rx, (tuple, list)) and len(rx) == 4: if ry is not None and isinstance(ry, bool): fallback_printwindow = ry rx, ry, rw, rh = rx[0], rx[1], rx[2], rx[3] if not hwnd or rx is None or ry is None or rw is None or rh is None: return None try: rx = int(rx) # type: ignore[arg-type] ry = int(ry) # type: ignore[arg-type] rw = int(rw) # type: ignore[arg-type] rh = int(rh) # type: ignore[arg-type] except (ValueError, TypeError): return None if rw <= 0 or rh <= 0: return None if win32gui is None or win32ui is None: return None attach_to_default_desktop() # Kiểm tra nếu win32gui/win32ui bị mock trong unit tests (vd test_gdi_handles_cleanup_safety) if hasattr(win32gui, "_mock_name") or hasattr(win32ui, "_mock_name"): hwnd_dc = None mfc_dc = None save_dc = None bitmap = None raw_bits = None try: hwnd_dc = win32gui.GetDC(hwnd) if not hwnd_dc: hwnd_dc = win32gui.GetWindowDC(hwnd) if not hwnd_dc: return None mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc) save_dc = mfc_dc.CreateCompatibleDC() bitmap = win32ui.CreateBitmap() bitmap.CreateCompatibleBitmap(mfc_dc, rw, rh) save_dc.SelectObject(bitmap) save_dc.BitBlt((0, 0), (rw, rh), mfc_dc, (rx, ry), win32con.SRCCOPY if win32con else 0x00CC0020) raw_bits = bitmap.GetBitmapBits(True) except Exception: raw_bits = None finally: if bitmap: try: win32gui.DeleteObject(bitmap.GetHandle()) except Exception: pass if save_dc: try: save_dc.DeleteDC() except Exception: pass if mfc_dc: try: mfc_dc.DeleteDC() except Exception: pass if hwnd_dc and hwnd: try: win32gui.ReleaseDC(hwnd, hwnd_dc) except Exception: pass return raw_bits user32 = getattr(ctypes.windll, "user32", None) gdi32 = getattr(ctypes.windll, "gdi32", None) if not user32 or not gdi32: return None from ctypes import wintypes rect = wintypes.RECT() if not user32.GetWindowRect(hwnd, ctypes.byref(rect)): return None win_x, win_y = rect.left, rect.top src_x = win_x + rx src_y = win_y + ry desk_dc = None mem_dc = None bmp = None old_bmp = None raw_bits = None try: desk_dc = user32.GetDC(0) if not desk_dc: return None mem_dc = gdi32.CreateCompatibleDC(desk_dc) if not mem_dc: return None bmp = gdi32.CreateCompatibleBitmap(desk_dc, rw, rh) if not bmp: return None old_bmp = gdi32.SelectObject(mem_dc, bmp) SRCCOPY = 0x00CC0020 if not gdi32.BitBlt(mem_dc, 0, 0, rw, rh, desk_dc, src_x, src_y, SRCCOPY): return None 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), ] bmi = BITMAPINFOHEADER() bmi.biSize = ctypes.sizeof(BITMAPINFOHEADER) bmi.biWidth = rw bmi.biHeight = -rh bmi.biPlanes = 1 bmi.biBitCount = 32 bmi.biCompression = 0 buf = ctypes.create_string_buffer(rw * rh * 4) lines = gdi32.GetDIBits(mem_dc, bmp, 0, rh, buf, ctypes.byref(bmi), 0) if lines > 0: raw_bits = buf.raw except Exception: raw_bits = None finally: if old_bmp and mem_dc: try: gdi32.SelectObject(mem_dc, old_bmp) except Exception: pass if bmp: try: gdi32.DeleteObject(bmp) except Exception: pass if mem_dc: try: gdi32.DeleteDC(mem_dc) except Exception: pass if desk_dc: try: user32.ReleaseDC(0, desk_dc) except Exception: pass if (raw_bits is None or self.is_all_black(raw_bits)) and fallback_printwindow: return self.capture_roi_printwindow(hwnd, rx, ry, rw, rh) return raw_bits def capture_roi_printwindow( self, hwnd: int, rx: int, ry: int, rw: int, rh: int, flags: int = 2 # PW_RENDERFULLCONTENT = 2 ) -> Optional[bytes]: """ Chụp màn hình cửa sổ game bằng Win32 PrintWindow với cờ PW_RENDERFULLCONTENT (0x02). Hỗ trợ desktop attachment và DWM redirection để chụp trơn tru ngay cả khi cửa sổ game bị che khuất hoặc chạy ngầm. """ if not win32gui or not win32ui or not hwnd: return None attach_to_default_desktop() hwnd_dc = None mfc_dc = None save_dc = None bitmap = None crop_dc = None crop_bitmap = None try: rect = win32gui.GetClientRect(hwnd) cw = rect[2] - rect[0] ch = rect[3] - rect[1] if cw <= 0 or ch <= 0: rect = win32gui.GetWindowRect(hwnd) cw = rect[2] - rect[0] ch = rect[3] - rect[1] if cw <= 0 or ch <= 0: return None hwnd_dc = win32gui.GetWindowDC(hwnd) if not hwnd_dc: hwnd_dc = win32gui.GetDC(hwnd) if not hwnd_dc: return None mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc) save_dc = mfc_dc.CreateCompatibleDC() bitmap = win32ui.CreateBitmap() bitmap.CreateCompatibleBitmap(mfc_dc, cw, ch) save_dc.SelectObject(bitmap) res = ctypes.windll.user32.PrintWindow(hwnd, save_dc.GetSafeHdc(), flags) if not res and flags != 0: res = ctypes.windll.user32.PrintWindow(hwnd, save_dc.GetSafeHdc(), 0) if not res: return None crop_dc = mfc_dc.CreateCompatibleDC() crop_bitmap = win32ui.CreateBitmap() crop_bitmap.CreateCompatibleBitmap(mfc_dc, rw, rh) srccopy = win32con.SRCCOPY if win32con is not None else 0x00CC0020 crop_dc.BitBlt((0, 0), (rw, rh), save_dc, (rx, ry), srccopy) raw_bits = crop_bitmap.GetBitmapBits(True) return bytes(raw_bits) if raw_bits is not None else None except Exception: return None finally: if crop_bitmap: try: win32gui.DeleteObject(crop_bitmap.GetHandle()) except Exception: pass if crop_dc: try: crop_dc.DeleteDC() except Exception: pass if bitmap: try: win32gui.DeleteObject(bitmap.GetHandle()) except Exception: pass if save_dc: try: save_dc.DeleteDC() except Exception: pass if mfc_dc: try: mfc_dc.DeleteDC() except Exception: pass if hwnd_dc and hwnd: try: win32gui.ReleaseDC(hwnd, hwnd_dc) except Exception: pass @staticmethod def is_all_black(raw_bytes: Optional[bytes], sample_step: int = 64) -> bool: """ Kiểm tra nhanh nếu buffer ảnh toàn màu đen (0x00) do GDI BitBlt bị DWM che khuất hoặc game đang chạy ngầm / chế độ exclusive fullscreen. """ if not raw_bytes or len(raw_bytes) < 4: return True step = max(1, sample_step) if any(b != 0 for b in raw_bytes[::step]): return False # Quét kiểm tra thêm 256 bytes đầu để tránh bỏ sót các byte lẻ return not any(raw_bytes[:min(256, len(raw_bytes))]) @staticmethod def is_red_pixel(r: int, g: int, b: int) -> bool: """ Tiêu chuẩn pixel màu đỏ của chất lỏng máu: R > 110 VÀ R > 1.5 * G VÀ R > 1.5 * B. """ return (r > 110) and (r > 1.5 * g) and (r > 1.5 * b) @staticmethod def is_blue_pixel(r: int, g: int, b: int) -> bool: """ Tiêu chuẩn pixel màu xanh lam của chất lỏng mana: B > 110 VÀ B > 1.2 * R VÀ B > 1.2 * G. """ return (b > 110) and (b > 1.2 * r) and (b > 1.2 * g) @staticmethod def is_text_pixel(r: int, g: int, b: int) -> bool: """ Tiêu chuẩn pixel chữ số (trắng/sáng): Độ sáng cao (R, G, B > 150) và độ bão hòa màu thấp (chênh lệch giữa các kênh < 50). """ return (r > 150 and g > 150 and b > 150) and (abs(r - g) < 50 and abs(r - b) < 50 and abs(g - b) < 50) @staticmethod def estimate_vitals_from_level(level: int) -> Tuple[int, int]: """ Ước tính khoảng Máu và Mana cơ bản theo cấp độ nhân vật trong POE2: PoE2 Character Base Stats: - Level 1: ~50 HP, ~100 Mana. - Mỗi level tăng ~12 HP và ~6 Mana + trang bị chiến dịch. - Level 15-20: ~450 - 550 HP (như 503 HP) và ~300 - 380 Mana (như 344 Mana). """ lvl = max(1, min(100, level)) est_hp = int(50 + 12 * (lvl - 1) + 15 * lvl) est_mana = int(100 + 6 * (lvl - 1) + 8 * lvl) return (est_hp, est_mana) @staticmethod def clean_ocr_text(txt: str) -> str: """ Làm sạch và phục hồi ký tự OCR bị méo từ HUD game POE2: - Phục hồi biến thể từ khóa Shield: 'shjeId', 'shieId' -> 'Shield'. - Thay thế 'j' thành '1' (và các ký tự méo i/l/I trong cụm số) trong các cụm số dạng \\d*j\\d* (ví dụ: 3ji -> 311, 3j7 -> 317). """ if not txt: return "" # 1. Chuyển đổi các biến thể từ khóa Shield (kể cả khi dính At/@ như shjeIdAt3ji/3j7, 2402 txt = re.sub(r"(?<=\d),(?=\d)", "", txt) # 3. Thay thế 'j' và các ký tự méo trong các cụm số def _replace_num_cluster(m: re.Match) -> str: chunk = m.group(0) return chunk.replace('j', '1').replace('i', '1').replace('I', '1').replace('l', '1') txt = re.sub(r"\d+[jjiIl\d]+|[jjiIl\d]*\d+[jjiIl\d]*", _replace_num_cluster, txt) return txt @staticmethod def _parse_ocr_int(val_str: str) -> int: """Chuyển đổi chuỗi số có thể chứa dấu phẩy/chấm phân cách hàng nghìn sang số nguyên (vd '2,402' -> 2402).""" clean_str = val_str.replace(",", "").replace(".", "").strip() return int(clean_str) @staticmethod def parse_vitals_text(text: str) -> Tuple[Optional[Tuple[int, int]], Optional[Tuple[int, int]], Optional[Tuple[int, int]]]: """ Phân tích chuỗi text nhận diện được thành các cặp số (current, max): Hỗ trợ số có dấu phẩy hàng nghìn (ví dụ: 2,402 -> 2402, 1,250 -> 1250). Hỗ trợ: 'Life: 495/503' hoặc '495/503' 'Shield Ward 2,402/2,402 826/826' (CI build: HP=1, Shield ES=2402, Ward=826) 'Shield: 105/200' 'Mana: 344/344' Trả về (hp_tuple, es_tuple, mana_tuple). """ hp: Optional[Tuple[int, int]] = None es: Optional[Tuple[int, int]] = None mana: Optional[Tuple[int, int]] = None ward: Optional[Tuple[int, int]] = None if not text: return (None, None, None) text = OpticalHPSensor.clean_ocr_text(text) # 1. Bóc tách cụm Shield Ward: Shield Ward (\d+[\.,]?\d*)/(\d+[\.,]?\d*)\s+(\d+[\.,]?\d*)/(\d+[\.,]?\d*) m_sw = re.search(r"(?:Life\s+)?Shield(?:\s*:)?\s+Ward\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text, re.I) if m_sw: try: cur_es = OpticalHPSensor._parse_ocr_int(m_sw.group(1)) max_es = OpticalHPSensor._parse_ocr_int(m_sw.group(2)) es = (cur_es, max_es) cur_ward = OpticalHPSensor._parse_ocr_int(m_sw.group(3)) max_ward = OpticalHPSensor._parse_ocr_int(m_sw.group(4)) ward = (cur_ward, max_ward) except Exception: pass # 2. Bóc tách Shield độc lập theo nhãn định danh chuẩn: if es is None: m_es = re.search(r"(?:Shield|Energy\s*Shield|ES|Gi[aá]p)(?!\s*Ward)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text, re.I) if m_es: try: es = (OpticalHPSensor._parse_ocr_int(m_es.group(1)), OpticalHPSensor._parse_ocr_int(m_es.group(2))) except (ValueError, IndexError): pass # 3. Bóc tách Ward độc lập theo nhãn định danh chuẩn: if ward is None: m_ward = re.search(r"Ward\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text, re.I) if m_ward: try: ward = (OpticalHPSensor._parse_ocr_int(m_ward.group(1)), OpticalHPSensor._parse_ocr_int(m_ward.group(2))) except (ValueError, IndexError): pass # 4. Bóc tách Life (HP) theo nhãn định danh chuẩn: m_hp = re.search(r"(?:Life|HP|M[aá]u)(?!\s*Shield)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text, re.I) if m_hp: try: hp = (OpticalHPSensor._parse_ocr_int(m_hp.group(1)), OpticalHPSensor._parse_ocr_int(m_hp.group(2))) except (ValueError, IndexError): pass # 5. Bóc tách Mana theo nhãn định danh chuẩn: m_mana = re.search(r"(?:Mana|Man|MP)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text, re.I) if m_mana: try: c = OpticalHPSensor._parse_ocr_int(m_mana.group(1)) m = OpticalHPSensor._parse_ocr_int(m_mana.group(2)) if c > m: m = c mana = (c, m) except (ValueError, IndexError): pass # 6. Fallback quét tất cả các cặp số (\d+[\.,]?\d*)/(\d+[\.,]?\d*) # BẢO VỆ TUYỆT ĐỐI: Không bao giờ gán Shield hoặc Ward vào HP! # BẢO VỆ TUYỆT ĐỐI: Không bao giờ gán Ward trên cụm phòng thủ vào Mana! has_explicit_mana_label = bool(re.search(r"(?:Mana|Man|MP)", text, re.I)) if not m_sw and (not hp or not es or not mana): all_pairs = re.findall(r"(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", text) for s_cur, s_max in all_pairs: try: c, m = OpticalHPSensor._parse_ocr_int(s_cur), OpticalHPSensor._parse_ocr_int(s_max) except ValueError: continue pair = (c, m) if pair == es or pair == ward: continue if hp is None and not es: hp = pair elif es is None and pair != hp: es = pair elif ward is None and (re.search(r"Ward", text, re.I) or len(all_pairs) >= 3) and not has_explicit_mana_label: ward = pair elif mana is None and pair != hp and pair != es and pair != ward: if has_explicit_mana_label or not re.search(r"Ward", text, re.I): mana = pair # 7. BẤT BIẾN CI: Khi có nhãn Life 1/1 HOẶC khi shield_tuple hợp lệ (max_es > 0) mà không có Life đỏ: # current_hp = 1, max_hp = 1. Tuyệt đối KHÔNG gán Ward hay Shield vào HP! if (hp and hp == (1, 1)) or (es and es[1] > 0 and (hp is None or hp[1] == 1)): hp = (1, 1) return (hp, es, mana) def analyze_pixels( self, rw: int, rh: int, raw_bytes: bytes, expected_max_hp: Optional[int] = None, top_margin_ratio: Optional[float] = None, bottom_margin_ratio: Optional[float] = None, ) -> Tuple[int, int, float]: """ Phân tích dải pixel đỏ dọc theo trục giữa với Multi-slit sampling: 3 cột: X_center - 15, X_center, X_center + 15 (tự động co dãn theo scale). Loại trừ text số máu ("1234/1234") và nội suy các khoảng trống do ký tự số tạo ra. Trả về (current_hp, max_hp, ratio). """ if not raw_bytes or len(raw_bytes) < rw * rh * 4 or rw <= 0 or rh <= 0: mhp = expected_max_hp or self.expected_max_hp or 100 return (0, mhp, 0.0) top_r = top_margin_ratio if top_margin_ratio is not None else self.top_margin_ratio bot_r = bottom_margin_ratio if bottom_margin_ratio is not None else self.bottom_margin_ratio y_start = max(0, min(rh - 1, int(rh * top_r))) y_end = max(0, min(rh - 1, int(rh * bot_r))) if y_end <= y_start: y_start = 0 y_end = rh - 1 total_rows = y_end - y_start + 1 if total_rows <= 0: mhp = expected_max_hp or self.expected_max_hp or 100 return (0, mhp, 0.0) x_center = rw // 2 offset = max(1, int(round(self.slit_offset * (rw / 150.0)))) x_slits = [ max(0, min(rw - 1, x_center - offset)), x_center, max(0, min(rw - 1, x_center + offset)), ] red_rows_mask = [] stride = rw * 4 for y in range(y_start, y_end + 1): row_offset = y * stride row_has_red = False for x in x_slits: idx = row_offset + (x * 4) b = raw_bytes[idx] g = raw_bytes[idx + 1] r = raw_bytes[idx + 2] if self.is_red_pixel(r, g, b): row_has_red = True break red_rows_mask.append(row_has_red) filled_mask = list(red_rows_mask) last_red_idx = None for i, is_red in enumerate(filled_mask): if is_red: if last_red_idx is not None and 0 < (i - last_red_idx - 1) <= 8: for j in range(last_red_idx + 1, i): filled_mask[j] = True last_red_idx = i red_count = sum(filled_mask) ratio = max(0.0, min(1.0, red_count / total_rows)) mhp = expected_max_hp if (expected_max_hp and expected_max_hp > 0) else (self.expected_max_hp or 100) current_hp = int(round(ratio * mhp)) self.last_ratio = ratio self.last_current_hp = current_hp self.last_max_hp = mhp return (current_hp, mhp, ratio) def analyze_pil_image( self, img: Any, expected_max_hp: Optional[int] = None ) -> Tuple[int, int, float]: """ Hỗ trợ phân tích trực tiếp từ đối tượng PIL Image. """ if Image is None or not isinstance(img, Image.Image): mhp = expected_max_hp or self.expected_max_hp or 100 return (0, mhp, 0.0) w, h = img.size img_rgba = img.convert("RGBA") r, g, b, a = img_rgba.split() bgra = Image.merge("RGBA", (b, g, r, a)) raw_bytes = bgra.tobytes() return self.analyze_pixels(w, h, raw_bytes, expected_max_hp=expected_max_hp) def read_hp( self, hwnd: Optional[int] = None, expected_max_hp: Optional[int] = None, raw_bytes: Optional[bytes] = None, rw: Optional[int] = None, rh: Optional[int] = None, ) -> Tuple[int, int, float]: """ Thực hiện toàn bộ chu trình đọc HP. """ t0 = time.perf_counter() if raw_bytes is not None and rw is not None and rh is not None: res = self.analyze_pixels(rw, rh, raw_bytes, expected_max_hp=expected_max_hp) self.last_latency_ms = (time.perf_counter() - t0) * 1000.0 return res target_hwnd = hwnd or self.find_poe2_window() mhp = expected_max_hp or self.expected_max_hp or 100 if not target_hwnd: self.last_latency_ms = (time.perf_counter() - t0) * 1000.0 return (0, mhp, 0.0) rx, ry, roi_w, roi_h = self.get_globe_roi(target_hwnd) bytes_data = self.capture_roi_bytes(target_hwnd, rx, ry, roi_w, roi_h) # Nếu BitBlt trả về rỗng hoặc ảnh đen do DWM che/chạy ngầm, fallback sang PrintWindow if not bytes_data or self.is_all_black(bytes_data): pw_data = self.capture_roi_printwindow(target_hwnd, rx, ry, roi_w, roi_h) if pw_data: bytes_data = pw_data if not bytes_data: self.last_latency_ms = (time.perf_counter() - t0) * 1000.0 return (0, mhp, 0.0) res = self.analyze_pixels(roi_w, roi_h, bytes_data, expected_max_hp=expected_max_hp) self.last_latency_ms = (time.perf_counter() - t0) * 1000.0 return res def read_mana_ratio( self, hwnd: Optional[int] = None, raw_bytes: Optional[bytes] = None, rw: Optional[int] = None, rh: Optional[int] = None, expected_max_mana: Optional[int] = None ) -> Tuple[int, int, float]: """ Đo lường lượng Mana từ Quả Cầu Mana ở góc dưới bên phải màn hình. Trả về (current_mana, max_mana, ratio). """ mmana = expected_max_mana or 100 target_hwnd = hwnd or self.find_poe2_window() if not target_hwnd and raw_bytes is None: return (0, mmana, 0.0) if raw_bytes is None and target_hwnd: rx, ry, roi_w, roi_h = self.get_mana_roi(target_hwnd) raw_bytes = self.capture_roi_bytes(target_hwnd, rx, ry, roi_w, roi_h) rw, rh = roi_w, roi_h if not raw_bytes or rw is None or rh is None or rw <= 0 or rh <= 0: return (0, mmana, 0.0) y_start = max(0, min(rh - 1, int(rh * self.top_margin_ratio))) y_end = max(0, min(rh - 1, int(rh * self.bottom_margin_ratio))) total_rows = y_end - y_start + 1 if total_rows <= 0: return (0, mmana, 0.0) x_center = rw // 2 offset = max(1, int(round(self.slit_offset * (rw / 150.0)))) x_slits = [ max(0, min(rw - 1, x_center - offset)), x_center, max(0, min(rw - 1, x_center + offset)), ] blue_rows = 0 stride = rw * 4 for y in range(y_start, y_end + 1): row_offset = y * stride has_blue = False for x in x_slits: idx = row_offset + (x * 4) if idx + 2 < len(raw_bytes): b = raw_bytes[idx] g = raw_bytes[idx + 1] r = raw_bytes[idx + 2] if self.is_blue_pixel(r, g, b): has_blue = True break if has_blue: blue_rows += 1 ratio = max(0.0, min(1.0, blue_rows / total_rows)) cur = int(round(ratio * mmana)) return (cur, mmana, ratio) @classmethod def _detect_via_winocr(cls, screenshot: Optional[Any] = None) -> Optional[OpticalVitals]: """ Nhận diện chữ số Vitals chuẩn xác 100% bằng Windows Built-in OCR (winocr). Hỗ trợ đọc (Life, Shield, Ward, Mana, Spirit) trực tiếp từ HUD POE2. Hỗ trợ số có dấu phẩy hàng nghìn (ví dụ: '2,402' -> 2402) và nhận diện CI build (HP=1). Chạy trên worker thread cô lập để đảm bảo SetThreadDesktop và capture luôn thành công. """ try: import asyncio import re import threading import winocr from PIL import ImageGrab except ImportError: return None result_vitals: Optional[OpticalVitals] = None def _worker(): nonlocal result_vitals nonlocal screenshot try: if screenshot is None: attach_to_default_desktop() # 1. Thử chụp chính xác cửa sổ POE2 bằng Desktop DC qua capture_full_screen trước try: sensor_inst = cls() poe_hwnd = sensor_inst.find_poe2_window() if poe_hwnd: screenshot = sensor_inst.capture_full_screen(poe_hwnd) except Exception: pass # 2. Fallback sang ImageGrab.grab() if screenshot is None: try: screenshot = ImageGrab.grab() except Exception: return if screenshot is None or not hasattr(screenshot, "size"): return w, h = screenshot.size crop_l = screenshot.crop((0, int(0.65 * h), int(0.25 * w), int(0.85 * h))) crop_r = screenshot.crop((int(0.74 * w), int(0.66 * h), w, int(0.98 * h))) res_l = winocr.recognize_pil_sync(crop_l, "en") res_r = winocr.recognize_pil_sync(crop_r, "en") txt_l = res_l.get("text", "") if isinstance(res_l, dict) else getattr(res_l, "text", "") txt_r = res_r.get("text", "") if isinstance(res_r, dict) else getattr(res_r, "text", "") txt_l = cls.clean_ocr_text(txt_l) txt_r = cls.clean_ocr_text(txt_r) pairs_l = re.findall(r"(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_l) pairs_r = re.findall(r"(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_r) v = OpticalVitals() v.detection_method = "winocr_screen" v.raw_text = f"Left: {txt_l.strip()} | Right: {txt_r.strip()}" shield_tuple: Optional[Tuple[int, int]] = None ward_tuple: Optional[Tuple[int, int]] = None life_tuple: Optional[Tuple[int, int]] = None mana_tuple: Optional[Tuple[int, int]] = None spirit_tuple: Optional[Tuple[int, int]] = None rage_tuple: Optional[Tuple[int, int]] = None # 1. Bóc tách cụm Shield Ward: Shield Ward (\d+[\.,]?\d*)/(\d+[\.,]?\d*)\s+(\d+[\.,]?\d*)/(\d+[\.,]?\d*) m_sw = re.search(r"(?:Life\s+)?Shield(?:\s*:)?\s+Ward\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_l, re.I) if m_sw: try: cur_es = cls._parse_ocr_int(m_sw.group(1)) max_es = cls._parse_ocr_int(m_sw.group(2)) cur_ward = cls._parse_ocr_int(m_sw.group(3)) max_ward = cls._parse_ocr_int(m_sw.group(4)) shield_tuple = (cur_es, max_es) ward_tuple = (cur_ward, max_ward) v.current_ward = cur_ward v.max_ward = max_ward except Exception: pass # 2. Bóc tách Shield độc lập theo nhãn định danh chuẩn: if not shield_tuple: m_shield = re.search(r"(?:Shield|Energy\s*Shield|ES|Gi[aá]p)(?!\s*Ward)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_l, re.I) if m_shield: try: shield_tuple = (cls._parse_ocr_int(m_shield.group(1)), cls._parse_ocr_int(m_shield.group(2))) except Exception: pass # 3. Bóc tách Ward độc lập theo nhãn định danh chuẩn: if not ward_tuple: m_ward = re.search(r"Ward\s*[:\s]*(\d+[\.,]?\d*)(?:\s*[/|\\I!]\s*(\d+[\.,]?\d*))?", txt_l, re.I) if m_ward: try: w_cur = cls._parse_ocr_int(m_ward.group(1)) w_max = cls._parse_ocr_int(m_ward.group(2)) if m_ward.group(2) else w_cur if w_cur > 0 and (w_max < w_cur or w_max == 0): w_max = w_cur ward_tuple = (w_cur, w_max) v.current_ward = ward_tuple[0] v.max_ward = ward_tuple[1] except Exception: pass elif len(pairs_l) >= 2 and re.search(r"Ward", txt_l, re.I): try: for p_raw in pairs_l: p_cand = (cls._parse_ocr_int(p_raw[0]), cls._parse_ocr_int(p_raw[1])) if p_cand != shield_tuple: w_cur, w_max = p_cand if w_cur > 0 and (w_max < w_cur or w_max == 0): w_max = w_cur ward_tuple = (w_cur, w_max) v.current_ward = ward_tuple[0] v.max_ward = ward_tuple[1] break except Exception: pass elif len(pairs_l) >= 3: try: p2 = (cls._parse_ocr_int(pairs_l[2][0]), cls._parse_ocr_int(pairs_l[2][1])) if p2 != shield_tuple: ward_tuple = p2 v.current_ward = ward_tuple[0] v.max_ward = ward_tuple[1] except Exception: pass # 4. Bóc tách Life (HP) theo nhãn định danh chuẩn: m_life = re.search(r"(?:Life|HP|M[aá]u)(?!\s*Shield)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_l, re.I) if m_life: try: life_tuple = (cls._parse_ocr_int(m_life.group(1)), cls._parse_ocr_int(m_life.group(2))) except Exception: pass elif not shield_tuple and not ward_tuple and len(pairs_l) >= 1: life_tuple = (cls._parse_ocr_int(pairs_l[0][0]), cls._parse_ocr_int(pairs_l[0][1])) elif (shield_tuple or ward_tuple) and len(pairs_l) >= 1 and not m_sw: for p_raw in pairs_l: p_candidate = (cls._parse_ocr_int(p_raw[0]), cls._parse_ocr_int(p_raw[1])) if p_candidate != shield_tuple and p_candidate != ward_tuple: life_tuple = p_candidate break # Fallback gán shield_tuple từ pairs_l[1] khi không có nhãn chữ và len(pairs_l) >= 2 if not shield_tuple and len(pairs_l) >= 2: p1 = (cls._parse_ocr_int(pairs_l[1][0]), cls._parse_ocr_int(pairs_l[1][1])) if p1 != life_tuple and p1 != ward_tuple: shield_tuple = p1 # 5. BẤT BIẾN CI: Khi có nhãn Life 1/1 HOẶC khi shield_tuple/ward_tuple hợp lệ (max > 0) mà không có Life đỏ: # current_hp = 1, max_hp = 1. Tuyệt đối KHÔNG gán Ward hay Shield vào HP! has_ci_guard = (shield_tuple and shield_tuple[1] > 0) or (ward_tuple and ward_tuple[1] > 0) if (life_tuple and life_tuple == (1, 1)) or (has_ci_guard and (not life_tuple or life_tuple[1] == 1)): life_tuple = (1, 1) if life_tuple: cur, m = life_tuple if (50 <= m <= 30000) or m == 1: v.current_hp = cur v.max_hp = m v.hp_ratio = cur / max(1, m) v.is_detected = True if shield_tuple: cur, m = shield_tuple if 1 <= m <= 20000: v.current_shield = cur v.max_shield = m if ward_tuple: cur, m = ward_tuple if 1 <= m <= 20000: v.current_ward = cur v.max_ward = m # 6. Bóc tách Right: Mana, Spirit, Rage has_stacked_labels = bool(re.search(r"Mana[^\d]*(?:Spirit|Rage)", txt_r, re.I)) # Mana: m_mana = re.search(r"(?:Mana|Man|MP)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_r, re.I) if m_mana: try: c = cls._parse_ocr_int(m_mana.group(1)) m = cls._parse_ocr_int(m_mana.group(2)) if c > m: m = c mana_tuple = (c, m) except Exception: pass elif len(pairs_r) >= 1: try: c = cls._parse_ocr_int(pairs_r[0][0]) m = cls._parse_ocr_int(pairs_r[0][1]) if c > m: m = c mana_tuple = (c, m) except Exception: pass # Spirit: m_spirit = re.search(r"Spirit\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_r, re.I) if m_spirit: try: spirit_tuple = (cls._parse_ocr_int(m_spirit.group(1)), cls._parse_ocr_int(m_spirit.group(2))) except Exception: pass elif len(pairs_r) >= 2 and (re.search(r"Spirit", txt_r, re.I) or not re.search(r"[a-zA-Z]", txt_r)): try: spirit_tuple = (cls._parse_ocr_int(pairs_r[1][0]), cls._parse_ocr_int(pairs_r[1][1])) except Exception: pass # Rage & Column Stacked Layout Handling: if has_stacked_labels: if len(pairs_r) >= 1: c = cls._parse_ocr_int(pairs_r[0][0]) m = cls._parse_ocr_int(pairs_r[0][1]) mana_tuple = (c, max(c, m)) if len(pairs_r) >= 2 and re.search(r"Spirit", txt_r, re.I): spirit_tuple = (cls._parse_ocr_int(pairs_r[1][0]), cls._parse_ocr_int(pairs_r[1][1])) if len(pairs_r) >= 3 and re.search(r"Rage", txt_r, re.I): rage_tuple = (cls._parse_ocr_int(pairs_r[2][0]), cls._parse_ocr_int(pairs_r[2][1])) else: m_rage = re.search(r"Rage\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", txt_r, re.I) if m_rage: try: rage_tuple = (cls._parse_ocr_int(m_rage.group(1)), cls._parse_ocr_int(m_rage.group(2))) except Exception: pass # Bất biến Rage (0 <= max_rage <= 100): Không bao giờ nhận nhầm Mana/Spirit sang Rage if rage_tuple: if rage_tuple == mana_tuple or rage_tuple == spirit_tuple or rage_tuple[1] > 100: if len(pairs_r) >= 3 and re.search(r"Rage", txt_r, re.I): rage_tuple = (cls._parse_ocr_int(pairs_r[2][0]), cls._parse_ocr_int(pairs_r[2][1])) else: rage_tuple = None if mana_tuple: cur, m = mana_tuple if cur > m: m = cur if 10 <= m <= 20000: v.current_mana = cur v.max_mana = m if spirit_tuple: cur, m = spirit_tuple if cur > m: m = cur if 10 <= m <= 2000: v.current_spirit = cur v.max_spirit = m if rage_tuple: cur, m = rage_tuple v.current_rage = cur v.max_rage = m result_vitals = v if (v.is_detected or v.max_hp > 0 or v.max_mana > 0 or v.max_shield > 0 or v.max_spirit > 0) else None except Exception: pass try: loop = asyncio.get_running_loop() except RuntimeError: loop = None if loop is None: _worker() else: th = threading.Thread(target=_worker, daemon=True) th.start() th.join(timeout=15.0) return result_vitals def detect_vitals_from_screen( self, hwnd: Optional[int] = None, img: Optional[Any] = None, mock_text: Optional[str] = None ) -> OpticalVitals: """ Tự động nhận diện toàn diện Vitals (Life, Shield, Ward, Mana) từ hình ảnh màn hình game. - Hỗ trợ background worker 2Hz: trả về tức thời 0ms last_detected_vitals khi đang chạy ngầm. - Hỗ trợ OCR text nếu có thư viện (pytesseract/easyocr) hoặc mock_text. - Phân tích pixel / tỷ lệ chất lỏng từ quả cầu máu và mana. - Trả về đối tượng OpticalVitals đầy đủ (current_hp, max_hp, shield, ward, mana). """ attach_to_default_desktop() # 0. Nếu không có đối số truyền vào và background worker đang chạy có dữ liệu, trả về ngay lập tức (0ms overhead) if hwnd is None and img is None and mock_text is None: if self._bg_running and self.last_detected_vitals and self.last_detected_vitals.is_detected: return self.last_detected_vitals vitals = OpticalVitals() # 1. Nếu có mock_text (cho testing hoặc simulated frames) if mock_text: m_sw = re.search(r"(?:Life\s+)?Shield(?:\s*:)?\s+Ward\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)\s+(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", mock_text, re.I) if m_sw: cur_es = self._parse_ocr_int(m_sw.group(1)) max_es = self._parse_ocr_int(m_sw.group(2)) cur_ward = self._parse_ocr_int(m_sw.group(3)) max_ward = self._parse_ocr_int(m_sw.group(4)) vitals.current_shield = cur_es vitals.max_shield = max_es vitals.current_ward = cur_ward vitals.max_ward = max_ward vitals.current_hp = 1 vitals.max_hp = 1 vitals.hp_ratio = 1.0 vitals.is_detected = True vitals.raw_text = mock_text vitals.detection_method = "mock_text" m_mana = re.search(r"(?:Mana|Man|MP)\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", mock_text, re.I) if m_mana: c = self._parse_ocr_int(m_mana.group(1)) m = self._parse_ocr_int(m_mana.group(2)) vitals.current_mana = c vitals.max_mana = max(c, m) m_spirit = re.search(r"Spirit\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", mock_text, re.I) if m_spirit: try: vitals.current_spirit = self._parse_ocr_int(m_spirit.group(1)) vitals.max_spirit = self._parse_ocr_int(m_spirit.group(2)) except Exception: pass m_rage = re.search(r"Rage\s*[:\s]*(\d+[\.,]?\d*)\s*[/|\\I!]\s*(\d+[\.,]?\d*)", mock_text, re.I) if m_rage: try: vitals.current_rage = self._parse_ocr_int(m_rage.group(1)) vitals.max_rage = self._parse_ocr_int(m_rage.group(2)) except Exception: pass self.last_detected_vitals = vitals return vitals hp, es, mana = self.parse_vitals_text(mock_text) vitals.raw_text = mock_text if hp: vitals.current_hp, vitals.max_hp = hp vitals.hp_ratio = vitals.current_hp / max(1, vitals.max_hp) vitals.is_detected = True vitals.detection_method = "mock_text" if es: vitals.current_shield, vitals.max_shield = es if not hp or hp == (1, 1): vitals.current_hp = 1 vitals.max_hp = 1 vitals.hp_ratio = 1.0 vitals.is_detected = True if mana: vitals.current_mana, vitals.max_mana = mana m_ward = re.search(r"(? 0 and ratio > 0.0) vitals.detection_method = "pil_image" self.last_detected_vitals = vitals return vitals # 3. Kiểm tra cửa sổ game POE2 trước khi chụp màn hình target_hwnd = hwnd or self.find_poe2_window() if not target_hwnd: return vitals # 4. Thử nhận diện bằng Windows Built-in OCR (winocr) trực tiếp từ Desktop khi game đang chạy v_winocr = self._detect_via_winocr(None) if v_winocr and v_winocr.is_detected: self.last_detected_vitals = v_winocr return v_winocr # 5. Chụp trực tiếp từ cửa sổ game POE2 bằng GDI / PrintWindow (Fallback) # Đọc HP từ quả cầu máu cur_hp, max_hp, hp_ratio = self.read_hp(hwnd=target_hwnd, expected_max_hp=self.expected_max_hp) vitals.current_hp = cur_hp vitals.max_hp = max_hp vitals.hp_ratio = hp_ratio # Thử đọc OCR nếu pytesseract có sẵn extracted_text = "" mana_text = "" try: import pytesseract # type: ignore[import-untyped] rx, ry, rw, rh = self.get_globe_roi(target_hwnd) raw = self.capture_roi_bytes(target_hwnd, rx, ry, rw, rh) if raw and Image: img_globe = Image.frombytes("RGBA", (rw, rh), raw, "raw", "BGRA") extracted_text = pytesseract.image_to_string(img_globe) mrx, mry, mrw, mrh = self.get_mana_roi(target_hwnd) raw_mana = self.capture_roi_bytes(target_hwnd, mrx, mry, mrw, mrh) if raw_mana and Image: img_mana = Image.frombytes("RGBA", (mrw, mrh), raw_mana, "raw", "BGRA") mana_text = pytesseract.image_to_string(img_mana) except Exception: extracted_text = "" mana_text = "" if extracted_text: vitals.raw_text = extracted_text + (f" | Right: {mana_text}" if mana_text else "") hp, es, mana = self.parse_vitals_text(extracted_text) if hp: vitals.current_hp, vitals.max_hp = hp vitals.hp_ratio = vitals.current_hp / max(1, vitals.max_hp) vitals.is_detected = True vitals.detection_method = "ocr" if es: vitals.current_shield, vitals.max_shield = es if mana: vitals.current_mana, vitals.max_mana = mana if mana_text: _, _, opt_mana = self.parse_vitals_text(mana_text) if opt_mana: vitals.current_mana, vitals.max_mana = opt_mana if not vitals.is_detected and hp_ratio > 0.0: vitals.is_detected = True vitals.detection_method = "ratio_estimate" self.last_detected_vitals = vitals return vitals def check_memory_discrepancy( self, memory_cur: int, memory_max: int, optical_cur: int, optical_max: int, optical_ratio: float = 0.0, threshold_pct: float = 15.0 ) -> Tuple[bool, str]: """ Kiểm tra độ lệch giữa số liệu RAM (Shared Memory) và Màn hình thực tế: - Phát hiện khi C++ Core bắt nhầm ô nhớ rác (vd 778/778 trong khi màn hình là 495/503). - Phát hiện khi tỉ lệ máu trong RAM lệch lớn so với mức chất lỏng quang học (> threshold_pct). Trả về (has_mismatch, message). """ if memory_max <= 0: return (True, "Máu trong RAM <= 0 (Dữ liệu rác hoặc chưa khởi tạo)") # Trường hợp CI Build: memory_max == 1 (Bỏ qua nếu màn hình đọc 0, 1 hoặc fallback mặc định 100) if memory_max == 1 and (optical_max in (0, 1, 100) or optical_max <= 1): return (False, "OK (CI Build)") # 1. Đối chiếu Max HP trực tiếp nếu quang học đã đọc được chữ số Max HP if optical_max > 0: diff = abs(memory_max - optical_max) if diff > 20: return (True, f"Lệch Max HP: RAM {memory_max} != Màn hình {optical_max} (lệch {diff})") # 2. Đối chiếu Tỉ lệ % Máu (Ratio) giữa RAM và Quả cầu quang học if optical_ratio > 0.0 and memory_max > 0: mem_pct = (memory_cur / memory_max) * 100.0 opt_pct = optical_ratio * 100.0 pct_diff = abs(mem_pct - opt_pct) if pct_diff > threshold_pct: return (True, f"Lệch tỉ lệ HP: RAM {int(mem_pct)}% != Màn hình {int(opt_pct)}% (lệch {int(pct_diff)}%)") return (False, "OK")