Merge pull request #10 from z060142/Refactoring

Refactor keyword detection with dual-template matching and coordinate correction
This commit is contained in:
z060142 2025-05-01 22:16:14 +08:00 committed by GitHub
commit 90b3a492d7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 554 additions and 173 deletions

View File

@ -175,7 +175,32 @@ Wolf Chat 是一個基於 MCP (Modular Capability Provider) 框架的聊天機
- **`llm_interaction.py`** - **`llm_interaction.py`**
- 修改了 `parse_structured_response` 函數中構建結果字典的順序。 - 修改了 `parse_structured_response` 函數中構建結果字典的順序。
- 現在,當成功解析來自 LLM 的有效 JSON 時,輸出的字典鍵順序將優先排列 `commands` - 現在,當成功解析來自 LLM 的有效 JSON 時,輸出的字典鍵順序將優先排列 `commands`
- **效果**:標準化了 JSON 回應的結構順序,有助於下游處理,並可能間接幫助 LLM 更清晰地組織其輸出,尤其是在涉及工具調用和特定指令時。 - **效果**:標準化了 JSON 回應的結構順序,有助於下游處理,並可能間接幫助 LLM 更清晰地組織其輸出,尤其是在涉及工具調用和特定指令時。
## 最近改進2025-05-01
### 關鍵字檢測重構 (雙重方法 + 座標校正)
- **目的**:根據 "Wolf 關鍵詞檢測方法深度重構指南",重構 `ui_interaction.py` 中的關鍵字檢測邏輯,以提高辨識穩定性並確保座標系統一致性。
- **`ui_interaction.py` (`DetectionModule`)**
- **新增雙重檢測方法 (`find_keyword_dual_method`)**
- 使用 OpenCV (`cv2.matchTemplate`) 進行模板匹配。
- 同時在**灰度圖**和 **CLAHE 增強圖**上進行匹配,並處理**反相**情況 (`cv2.bitwise_not`)。
- **座標校正**`cv2.matchTemplate` 返回的座標是相對於截圖區域 (`region`) 的。在返回結果前,已將其轉換為絕對螢幕座標 (`absolute_x = region_x + relative_x`, `absolute_y = region_y + relative_y`),以確保與 `pyautogui` 點擊座標一致。
- **結果合併**
1. 優先選擇灰度與 CLAHE 方法結果**重合**且距離接近 (`MATCH_DISTANCE_THRESHOLD`) 的匹配。
2. 若無重合,則選擇單一方法中置信度**非常高** (`DUAL_METHOD_HIGH_CONFIDENCE_THRESHOLD`) 的結果。
3. 若仍無結果,則回退到單一方法中置信度**較高** (`DUAL_METHOD_FALLBACK_CONFIDENCE_THRESHOLD`) 的結果。
- **核心模板**:僅使用三個核心模板 (`keyword_wolf_lower`, `keyword_Wolf_upper`, `keyword_wolf_reply`) 進行檢測。
- **效能統計**:添加了計數器以追蹤檢測次數、成功率、各方法使用分佈、平均時間和反相匹配率 (`print_detection_stats` 方法)。
- **除錯視覺化**:在高除錯級別 (`DEBUG_LEVEL >= 3`) 下,會保存預處理圖像和標記了檢測點的結果圖像。
- **舊方法保留 (`_find_keyword_legacy`)**:原有的基於 `pyautogui.locateAllOnScreen` 和多模板的 `find_keyword_in_region` 邏輯被移至此私有方法,用於向後兼容或調試比較。
- **包裝器方法 (`find_keyword_in_region`)**:現在作為一個包裝器,根據 `use_dual_method` 標誌(預設為 `True`)調用新的雙重方法或舊的 legacy 方法。
- **初始化更新**`__init__` 方法更新以支持 `use_dual_method` 標誌、CLAHE 初始化、核心模板提取和效能計數器。
- **`ui_interaction.py` (`run_ui_monitoring_loop`)**
- **模板字典**:初始化時區分 `essential_templates``legacy_templates`,並合併後傳遞給 `DetectionModule`
- **模塊實例化**:以 `use_dual_method=True` 實例化 `DetectionModule`
- **效果**:預期能提高關鍵字檢測在不同光照、對比度或 UI 主題下的魯棒性,同時確保檢測到的座標能被 `pyautogui` 正確用於點擊。簡化了需要維護的關鍵字模板數量。
## 配置與部署 ## 配置與部署
@ -262,6 +287,32 @@ Wolf Chat 是一個基於 MCP (Modular Capability Provider) 框架的聊天機
- 在「技術實現」的「發送者識別」部分強調了點擊位置是相對於觸發泡泡計算的,並註明了新的偏移量。 - 在「技術實現」的「發送者識別」部分強調了點擊位置是相對於觸發泡泡計算的,並註明了新的偏移量。
- 添加了此「最近改進」條目。 - 添加了此「最近改進」條目。
### 關鍵字檢測重構 (雙重方法與座標校正) (2025-05-01)
- **目的**:提高關鍵字 ("wolf", "Wolf", 回覆指示符) 檢測的穩定性和對視覺變化的魯棒性,並確保檢測到的座標能準確對應 `pyautogui` 的點擊座標。
- **`ui_interaction.py` (`DetectionModule`)**
- **重構 `find_keyword_in_region`**:此方法現在作為一個包裝器 (wrapper)。
- **新增 `find_keyword_dual_method`**
- 成為預設的關鍵字檢測方法 (由 `use_dual_method` 標誌控制,預設為 `True`)。
- **核心邏輯**
1. 對目標區域截圖。
2. 同時準備灰度 (grayscale) 和 CLAHE (對比度限制自適應直方圖均衡化) 增強的圖像版本。
3. 對三種核心關鍵字模板 (`keyword_wolf_lower`, `keyword_Wolf_upper`, `keyword_wolf_reply`) 也進行灰度與 CLAHE 預處理。
4. 使用 `cv2.matchTemplate` 分別在灰度圖和 CLAHE 圖上進行模板匹配 (包括正向和反向匹配 `cv2.bitwise_not`)。
5. **座標校正**:將 `cv2.matchTemplate` 返回的 **相對** 於截圖區域的座標,通過加上截圖區域的左上角座標 (`region_x`, `region_y`),轉換為 **絕對** 螢幕座標,確保與 `pyautogui` 使用的座標系統一致。
6. **結果合併策略**
- 優先選擇灰度與 CLAHE 方法結果**重合** (中心點距離小於 `MATCH_DISTANCE_THRESHOLD`) 且置信度最高的匹配。
- 若無重合,則回退到單一方法中置信度最高的匹配 (需高於特定閾值 `DUAL_METHOD_FALLBACK_CONFIDENCE_THRESHOLD`)。
- **效能統計**:增加了計數器 (`performance_stats`) 來追蹤檢測總數、成功數、各方法成功數、反相匹配數和總耗時。新增 `print_detection_stats` 方法用於輸出統計。
- **除錯增強**:在高除錯級別 (`DEBUG_LEVEL >= 3`) 下,會保存預處理圖像和標記了檢測結果的圖像。
- **新增 `_find_keyword_legacy`**:包含原 `find_keyword_in_region` 的邏輯,使用 `pyautogui.locateAllOnScreen` 遍歷所有(包括已棄用的)關鍵字模板,用於向後兼容或除錯比較。
- **常量整理**:將核心關鍵字模板標記為活躍,其他類型標記為棄用,並添加了 CLAHE 和雙重方法相關的新常量。
- **初始化更新**`__init__` 方法更新以支持新標誌、初始化 CLAHE 物件和效能計數器。
- **`ui_interaction.py` (`run_ui_monitoring_loop`)**
- 更新了 `templates` 字典的創建方式,區分核心模板和舊模板。
- 在實例化 `DetectionModule` 時傳遞 `use_dual_method=True`
- **效果**:預期能更可靠地在不同光照、對比度或顏色主題下檢測到關鍵字,同時確保點擊位置的準確性。
### 聊天泡泡重新定位以提高穩定性 ### 聊天泡泡重新定位以提高穩定性
- **UI 互動模塊 (`ui_interaction.py`)** - **UI 互動模塊 (`ui_interaction.py`)**

View File

@ -15,6 +15,7 @@ import json # Added for color config loading
import queue import queue
from typing import List, Tuple, Optional, Dict, Any from typing import List, Tuple, Optional, Dict, Any
import threading # Import threading for Lock if needed, or just use a simple flag import threading # Import threading for Lock if needed, or just use a simple flag
import math # Added for distance calculation in dual method
# --- Global Pause Flag --- # --- Global Pause Flag ---
# Using a simple mutable object (list) for thread-safe-like access without explicit lock # Using a simple mutable object (list) for thread-safe-like access without explicit lock
@ -44,7 +45,7 @@ def load_bubble_colors(config_path='bubble_colors.json'):
return [ return [
{ {
"name": "normal_user", "name": "normal_user",
"is_bot": false, "is_bot": False, # Corrected boolean value
"hsv_lower": [6, 0, 240], "hsv_lower": [6, 0, 240],
"hsv_upper": [18, 23, 255], "hsv_upper": [18, 23, 255],
"min_area": 2500, "min_area": 2500,
@ -52,7 +53,7 @@ def load_bubble_colors(config_path='bubble_colors.json'):
}, },
{ {
"name": "bot", "name": "bot",
"is_bot": true, "is_bot": True, # Corrected boolean value
"hsv_lower": [105, 9, 208], "hsv_lower": [105, 9, 208],
"hsv_upper": [116, 43, 243], "hsv_upper": [116, 43, 243],
"min_area": 2500, "min_area": 2500,
@ -69,6 +70,7 @@ os.makedirs(TEMPLATE_DIR, exist_ok=True)
DEBUG_SCREENSHOT_DIR = os.path.join(SCRIPT_DIR, "debug_screenshots") DEBUG_SCREENSHOT_DIR = os.path.join(SCRIPT_DIR, "debug_screenshots")
MAX_DEBUG_SCREENSHOTS = 8 MAX_DEBUG_SCREENSHOTS = 8
os.makedirs(DEBUG_SCREENSHOT_DIR, exist_ok=True) os.makedirs(DEBUG_SCREENSHOT_DIR, exist_ok=True)
DEBUG_LEVEL = 1 # 0=Off, 1=Basic Info, 2=Detailed, 3=Visual Debug
# --- End Debugging --- # --- End Debugging ---
# --- Template Paths (Consider moving to config.py or loading dynamically) --- # --- Template Paths (Consider moving to config.py or loading dynamically) ---
@ -97,21 +99,21 @@ BOT_CORNER_BR_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "bot_corner_br_type2.png")
BOT_CORNER_TL_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "bot_corner_tl_type3.png") BOT_CORNER_TL_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "bot_corner_tl_type3.png")
BOT_CORNER_BR_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "bot_corner_br_type3.png") BOT_CORNER_BR_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "bot_corner_br_type3.png")
# --- End Additional Types --- # --- End Additional Types ---
# Keywords # Keywords (Refactored based on guide)
KEYWORD_wolf_LOWER_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower.png") KEYWORD_wolf_LOWER_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower.png") # Active Core
KEYWORD_Wolf_UPPER_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper.png") KEYWORD_Wolf_UPPER_IMG = os.path.join(TEMPLATE_DIR, "keyword_Wolf_upper.png") # Active Core
KEYWORD_wolf_LOWER_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type2.png") # Added for type3 bubbles KEYWORD_WOLF_REPLY_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply.png") # Active Core
KEYWORD_Wolf_UPPER_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type2.png") # Added for type3 bubbles
KEYWORD_wolf_LOWER_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type3.png") # Added for type3 bubbles # Deprecated but kept for potential legacy fallback or reference
KEYWORD_Wolf_UPPER_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type3.png") # Added for type3 bubbles KEYWORD_wolf_LOWER_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type2.png") # Deprecated
KEYWORD_wolf_LOWER_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type4.png") # Added for type4 bubbles KEYWORD_Wolf_UPPER_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type2.png") # Deprecated
KEYWORD_Wolf_UPPER_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type4.png") # Added for type4 bubbles KEYWORD_wolf_LOWER_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type3.png") # Deprecated
# --- Reply Keywords --- KEYWORD_Wolf_UPPER_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type3.png") # Deprecated
KEYWORD_WOLF_REPLY_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply.png") # Added for reply detection KEYWORD_wolf_LOWER_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_lower_type4.png") # Deprecated
KEYWORD_WOLF_REPLY_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type2.png") # Added for reply detection type2 KEYWORD_Wolf_UPPER_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_upper_type4.png") # Deprecated
KEYWORD_WOLF_REPLY_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type3.png") # Added for reply detection type3 KEYWORD_WOLF_REPLY_TYPE2_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type2.png") # Deprecated
KEYWORD_WOLF_REPLY_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type4.png") # Added for reply detection type4 KEYWORD_WOLF_REPLY_TYPE3_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type3.png") # Deprecated
# --- End Reply Keywords --- KEYWORD_WOLF_REPLY_TYPE4_IMG = os.path.join(TEMPLATE_DIR, "keyword_wolf_reply_type4.png") # Deprecated
# UI Elements # UI Elements
COPY_MENU_ITEM_IMG = os.path.join(TEMPLATE_DIR, "copy_menu_item.png") COPY_MENU_ITEM_IMG = os.path.join(TEMPLATE_DIR, "copy_menu_item.png")
PROFILE_OPTION_IMG = os.path.join(TEMPLATE_DIR, "profile_option.png") PROFILE_OPTION_IMG = os.path.join(TEMPLATE_DIR, "profile_option.png")
@ -171,6 +173,14 @@ BUBBLE_RELOCATE_FALLBACK_CONFIDENCE = 0.6 # Lower confidence for fallback attemp
BBOX_SIMILARITY_TOLERANCE = 10 BBOX_SIMILARITY_TOLERANCE = 10
RECENT_TEXT_HISTORY_MAXLEN = 5 # This state likely belongs in the coordinator RECENT_TEXT_HISTORY_MAXLEN = 5 # This state likely belongs in the coordinator
# --- New Constants for Dual Method ---
CLAHE_CLIP_LIMIT = 2.0 # CLAHE enhancement parameter
CLAHE_TILE_SIZE = (8, 8) # CLAHE grid size
MATCH_DISTANCE_THRESHOLD = 10 # Threshold for considering detections as overlapping (pixels)
DUAL_METHOD_CONFIDENCE_THRESHOLD = 0.85 # Confidence threshold for individual methods in dual mode
DUAL_METHOD_HIGH_CONFIDENCE_THRESHOLD = 0.85 # Threshold for accepting single method result directly
DUAL_METHOD_FALLBACK_CONFIDENCE_THRESHOLD = 0.8 # Threshold for accepting single method result in fallback
# --- Helper Function (Module Level) --- # --- Helper Function (Module Level) ---
def are_bboxes_similar(bbox1: Optional[Tuple[int, int, int, int]], def are_bboxes_similar(bbox1: Optional[Tuple[int, int, int, int]],
bbox2: Optional[Tuple[int, int, int, int]], bbox2: Optional[Tuple[int, int, int, int]],
@ -189,18 +199,41 @@ class DetectionModule:
def __init__(self, templates: Dict[str, str], confidence: float = CONFIDENCE_THRESHOLD, def __init__(self, templates: Dict[str, str], confidence: float = CONFIDENCE_THRESHOLD,
state_confidence: float = STATE_CONFIDENCE_THRESHOLD, state_confidence: float = STATE_CONFIDENCE_THRESHOLD,
region: Optional[Tuple[int, int, int, int]] = SCREENSHOT_REGION): region: Optional[Tuple[int, int, int, int]] = SCREENSHOT_REGION,
use_dual_method: bool = True): # Added use_dual_method flag
# --- Hardcoded Settings (as per user instruction) --- # --- Hardcoded Settings (as per user instruction) ---
self.use_color_detection: bool = True # Set to True to enable color detection by default self.use_color_detection: bool = True # Set to True to enable color detection by default
self.color_config_path: str = "bubble_colors.json" self.color_config_path: str = "bubble_colors.json"
# --- End Hardcoded Settings --- # --- End Hardcoded Settings ---
self.templates = templates self.templates = templates
self.confidence = confidence self.confidence = confidence # Default confidence for legacy methods
self.state_confidence = state_confidence self.state_confidence = state_confidence
self.region = region self.region = region
self._warned_paths = set() self._warned_paths = set()
# --- Dual Method Specific Initialization ---
self.use_dual_method = use_dual_method
self.clahe = cv2.createCLAHE(clipLimit=CLAHE_CLIP_LIMIT, tileGridSize=CLAHE_TILE_SIZE)
self.core_keyword_templates = {k: v for k, v in templates.items()
if k in ['keyword_wolf_lower', 'keyword_Wolf_upper', 'keyword_wolf_reply']}
self.last_detection_method = None
self.last_detection_confidence = 0.0
self.DEBUG_LEVEL = DEBUG_LEVEL # Use global debug level
# Performance Stats
self.performance_stats = {
'total_detections': 0,
'successful_detections': 0,
'gray_only_detections': 0,
'clahe_only_detections': 0,
'dual_method_detections': 0,
'fallback_detections': 0, # Added for fallback tracking
'total_detection_time': 0.0,
'inverted_matches': 0
}
# --- End Dual Method Specific Initialization ---
# Load color configuration if color detection is enabled # Load color configuration if color detection is enabled
self.bubble_colors = [] self.bubble_colors = []
if self.use_color_detection: if self.use_color_detection:
@ -208,10 +241,27 @@ class DetectionModule:
if not self.bubble_colors: if not self.bubble_colors:
print("Warning: Color detection enabled, but failed to load any color configurations. Color detection might not work.") print("Warning: Color detection enabled, but failed to load any color configurations. Color detection might not work.")
print(f"DetectionModule initialized. Color Detection: {'Enabled' if self.use_color_detection else 'Disabled'}") print(f"DetectionModule initialized. Color Detection: {'Enabled' if self.use_color_detection else 'Disabled'}. Dual Keyword Method: {'Enabled' if self.use_dual_method else 'Disabled'}")
def _apply_clahe(self, image):
"""Apply CLAHE to enhance image contrast."""
if image is None:
print("Warning: _apply_clahe received None image.")
return None
try:
if len(image.shape) == 3:
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
else:
gray = image.copy() # Assume already grayscale
enhanced = self.clahe.apply(gray)
return enhanced
except Exception as e:
print(f"Error applying CLAHE: {e}")
# Return original grayscale image on error
return gray if 'gray' in locals() else image
def _find_template(self, template_key: str, confidence: Optional[float] = None, region: Optional[Tuple[int, int, int, int]] = None, grayscale: bool = False) -> List[Tuple[int, int]]: def _find_template(self, template_key: str, confidence: Optional[float] = None, region: Optional[Tuple[int, int, int, int]] = None, grayscale: bool = False) -> List[Tuple[int, int]]:
"""Internal helper to find a template by its key. Returns list of CENTER coordinates.""" """Internal helper to find a template by its key using PyAutoGUI. Returns list of CENTER coordinates (absolute)."""
template_path = self.templates.get(template_key) template_path = self.templates.get(template_key)
if not template_path: if not template_path:
print(f"Error: Template key '{template_key}' not found in provided templates.") print(f"Error: Template key '{template_key}' not found in provided templates.")
@ -524,83 +574,375 @@ class DetectionModule:
print(f"Color detection found {len(all_bubbles_info)} bubbles.") print(f"Color detection found {len(all_bubbles_info)} bubbles.")
return all_bubbles_info return all_bubbles_info
def find_keyword_in_region(self, region: Tuple[int, int, int, int]) -> Optional[Tuple[int, int]]: def _find_keyword_legacy(self, region: Tuple[int, int, int, int]) -> Optional[Tuple[int, int]]:
"""Look for keywords within a specified region. Returns center coordinates.""" """
Original find_keyword_in_region implementation using multiple templates and PyAutoGUI.
Kept for backward compatibility or fallback. Returns absolute center coordinates or None.
"""
if region[2] <= 0 or region[3] <= 0: return None # Invalid region width/height if region[2] <= 0 or region[3] <= 0: return None # Invalid region width/height
# Try original lowercase with color matching # Define the order of templates to check (legacy approach)
locations_lower = self._find_template('keyword_wolf_lower', region=region, grayscale=True) # Changed grayscale to False legacy_keyword_templates = [
if locations_lower: # Original keywords first
print(f"Found keyword (lowercase, color) in region {region}, position: {locations_lower[0]}") # Updated log message 'keyword_wolf_lower', 'keyword_wolf_upper',
return locations_lower[0] # Deprecated keywords next (order might matter based on visual similarity)
'keyword_wolf_lower_type2', 'keyword_wolf_upper_type2',
'keyword_wolf_lower_type3', 'keyword_wolf_upper_type3',
'keyword_wolf_lower_type4', 'keyword_wolf_upper_type4',
# Reply keywords last
'keyword_wolf_reply', 'keyword_wolf_reply_type2',
'keyword_wolf_reply_type3', 'keyword_wolf_reply_type4'
]
# Try original uppercase with color matching for key in legacy_keyword_templates:
locations_upper = self._find_template('keyword_wolf_upper', region=region, grayscale=True) # Changed grayscale to False # Determine grayscale based on key (example logic, adjust as needed)
if locations_upper: # Original logic seemed to use grayscale=True for lower/upper, False otherwise. Let's replicate that.
print(f"Found keyword (uppercase, color) in region {region}, position: {locations_upper[0]}") # Updated log message use_grayscale = ('lower' in key or 'upper' in key) and 'type' not in key and 'reply' not in key
return locations_upper[0] # Use the default confidence defined in __init__ for legacy checks
locations = self._find_template(key, region=region, grayscale=use_grayscale, confidence=self.confidence)
if locations:
print(f"Legacy method found keyword ('{key}') in region {region}, position: {locations[0]}")
return locations[0] # Return the first match found
# Try type2 lowercase (white text, no grayscale) return None # No keyword found using legacy method
locations_lower_type2 = self._find_template('keyword_wolf_lower_type2', region=region, grayscale=False) # Added type2 check
if locations_lower_type2:
print(f"Found keyword (lowercase, type2) in region {region}, position: {locations_lower_type2[0]}")
return locations_lower_type2[0]
# Try type2 uppercase (white text, no grayscale) def find_keyword_dual_method(self, region: Tuple[int, int, int, int]) -> Optional[Tuple[int, int]]:
locations_upper_type2 = self._find_template('keyword_wolf_upper_type2', region=region, grayscale=False) # Added type2 check """
if locations_upper_type2: Find keywords using grayscale and CLAHE preprocessed images with OpenCV template matching.
print(f"Found keyword (uppercase, type2) in region {region}, position: {locations_upper_type2[0]}") Applies coordinate correction to return absolute screen coordinates.
return locations_upper_type2[0] Returns absolute center coordinates tuple (x, y) or None.
"""
if region is None or len(region) != 4 or region[2] <= 0 or region[3] <= 0:
print(f"Error: Invalid region provided to find_keyword_dual_method: {region}")
return None
# Try type3 lowercase (white text, no grayscale) - Corrected start_time = time.time()
locations_lower_type3 = self._find_template('keyword_wolf_lower_type3', region=region, grayscale=False) region_x, region_y, region_w, region_h = region
if locations_lower_type3:
print(f"Found keyword (lowercase, type3) in region {region}, position: {locations_lower_type3[0]}")
return locations_lower_type3[0]
# Try type3 uppercase (white text, no grayscale) - Corrected try:
locations_upper_type3 = self._find_template('keyword_wolf_upper_type3', region=region, grayscale=False) screenshot = pyautogui.screenshot(region=region)
if locations_upper_type3: if screenshot is None:
print(f"Found keyword (uppercase, type3) in region {region}, position: {locations_upper_type3[0]}") print("Error: Failed to capture screenshot for dual method detection.")
return locations_upper_type3[0] return None
img = np.array(screenshot)
img_bgr = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
except Exception as e:
print(f"Error capturing or converting screenshot in region {region}: {e}")
return None
# Try type4 lowercase (white text, no grayscale) - Added type4 img_gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY)
locations_lower_type4 = self._find_template('keyword_wolf_lower_type4', region=region, grayscale=False) img_clahe = self._apply_clahe(img_gray) # Use helper method
if locations_lower_type4:
print(f"Found keyword (lowercase, type4) in region {region}, position: {locations_lower_type4[0]}")
return locations_lower_type4[0]
# Try type4 uppercase (white text, no grayscale) - Added type4 if img_clahe is None:
locations_upper_type4 = self._find_template('keyword_wolf_upper_type4', region=region, grayscale=False) print("Error: CLAHE preprocessing failed. Cannot proceed with CLAHE matching.")
if locations_upper_type4: # Optionally, could proceed with only grayscale matching here, but for simplicity, we return None.
print(f"Found keyword (uppercase, type4) in region {region}, position: {locations_upper_type4[0]}") return None
return locations_upper_type4[0]
# Try reply keyword (normal) gray_results = []
locations_reply = self._find_template('keyword_wolf_reply', region=region, grayscale=False) clahe_results = []
if locations_reply: template_types = { # Map core template keys to types
print(f"Found keyword (reply) in region {region}, position: {locations_reply[0]}") 'keyword_wolf_lower': 'standard',
return locations_reply[0] 'keyword_Wolf_upper': 'standard',
'keyword_wolf_reply': 'reply'
}
# Try reply keyword (type2) for key, template_path in self.core_keyword_templates.items():
locations_reply_type2 = self._find_template('keyword_wolf_reply_type2', region=region, grayscale=False) if not os.path.exists(template_path):
if locations_reply_type2: if template_path not in self._warned_paths:
print(f"Found keyword (reply, type2) in region {region}, position: {locations_reply_type2[0]}") print(f"Warning: Core keyword template not found: {template_path}")
return locations_reply_type2[0] self._warned_paths.add(template_path)
continue
# Try reply keyword (type3) template_bgr = cv2.imread(template_path)
locations_reply_type3 = self._find_template('keyword_wolf_reply_type3', region=region, grayscale=False) if template_bgr is None:
if locations_reply_type3: if template_path not in self._warned_paths:
print(f"Found keyword (reply, type3) in region {region}, position: {locations_reply_type3[0]}") print(f"Warning: Failed to load core keyword template: {template_path}")
return locations_reply_type3[0] self._warned_paths.add(template_path)
continue
# Try reply keyword (type4) template_gray = cv2.cvtColor(template_bgr, cv2.COLOR_BGR2GRAY)
locations_reply_type4 = self._find_template('keyword_wolf_reply_type4', region=region, grayscale=False) template_clahe = self._apply_clahe(template_gray) # Use helper method
if locations_reply_type4:
print(f"Found keyword (reply, type4) in region {region}, position: {locations_reply_type4[0]}") if template_clahe is None:
return locations_reply_type4[0] print(f"Warning: CLAHE preprocessing failed for template {key}. Skipping CLAHE match for this template.")
continue # Skip CLAHE part for this template
h_gray, w_gray = template_gray.shape[:2]
h_clahe, w_clahe = template_clahe.shape[:2]
# --- Grayscale Matching ---
try:
gray_res = cv2.matchTemplate(img_gray, template_gray, cv2.TM_CCOEFF_NORMED)
gray_inv_res = cv2.matchTemplate(img_gray, cv2.bitwise_not(template_gray), cv2.TM_CCOEFF_NORMED)
gray_combined = np.maximum(gray_res, gray_inv_res)
_, gray_max_val, _, gray_max_loc = cv2.minMaxLoc(gray_combined)
if gray_max_val >= DUAL_METHOD_CONFIDENCE_THRESHOLD:
# Calculate relative center
relative_center_x = gray_max_loc[0] + w_gray // 2
relative_center_y = gray_max_loc[1] + h_gray // 2
# *** COORDINATE CORRECTION ***
absolute_center_x = region_x + relative_center_x
absolute_center_y = region_y + relative_center_y
# Check inversion
gray_orig_val = gray_res[gray_max_loc[1], gray_max_loc[0]] # Get value at max_loc from original match
is_inverted = (gray_orig_val < gray_max_val - 0.05)
gray_results.append({
'template': key,
'center': (absolute_center_x, absolute_center_y), # Store absolute coords
'confidence': gray_max_val,
'is_inverted': is_inverted,
'type': template_types.get(key, 'standard')
})
except cv2.error as e:
print(f"OpenCV Error during Grayscale matching for {key}: {e}")
except Exception as e:
print(f"Unexpected Error during Grayscale matching for {key}: {e}")
# --- CLAHE Matching ---
try:
clahe_res = cv2.matchTemplate(img_clahe, template_clahe, cv2.TM_CCOEFF_NORMED)
clahe_inv_res = cv2.matchTemplate(img_clahe, cv2.bitwise_not(template_clahe), cv2.TM_CCOEFF_NORMED)
clahe_combined = np.maximum(clahe_res, clahe_inv_res)
_, clahe_max_val, _, clahe_max_loc = cv2.minMaxLoc(clahe_combined)
if clahe_max_val >= DUAL_METHOD_CONFIDENCE_THRESHOLD:
# Calculate relative center
relative_center_x = clahe_max_loc[0] + w_clahe // 2
relative_center_y = clahe_max_loc[1] + h_clahe // 2
# *** COORDINATE CORRECTION ***
absolute_center_x = region_x + relative_center_x
absolute_center_y = region_y + relative_center_y
# Check inversion
clahe_orig_val = clahe_res[clahe_max_loc[1], clahe_max_loc[0]] # Get value at max_loc from original match
is_inverted = (clahe_orig_val < clahe_max_val - 0.05)
clahe_results.append({
'template': key,
'center': (absolute_center_x, absolute_center_y), # Store absolute coords
'confidence': clahe_max_val,
'is_inverted': is_inverted,
'type': template_types.get(key, 'standard')
})
except cv2.error as e:
print(f"OpenCV Error during CLAHE matching for {key}: {e}")
except Exception as e:
print(f"Unexpected Error during CLAHE matching for {key}: {e}")
# --- Result Merging and Selection ---
elapsed_time = time.time() - start_time
self.performance_stats['total_detections'] += 1
self.performance_stats['total_detection_time'] += elapsed_time
best_match = None
final_result_coords = None
detection_type = "None" # For stats
if not gray_results and not clahe_results:
if self.DEBUG_LEVEL > 1:
print(f"[Dual Method] No keywords found by either method. Time: {elapsed_time:.3f}s")
self.last_detection_method = None
self.last_detection_confidence = 0.0
return None
# Strategy 1: High-confidence single method result
best_gray = max(gray_results, key=lambda x: x['confidence']) if gray_results else None
best_clahe = max(clahe_results, key=lambda x: x['confidence']) if clahe_results else None
if best_gray and not best_clahe and best_gray['confidence'] >= DUAL_METHOD_HIGH_CONFIDENCE_THRESHOLD:
final_result_coords = best_gray['center']
self.last_detection_method = "Gray" + (" (Inv)" if best_gray['is_inverted'] else "")
self.last_detection_confidence = best_gray['confidence']
detection_type = "Gray Only (High Conf)"
self.performance_stats['gray_only_detections'] += 1
if best_gray['is_inverted']: self.performance_stats['inverted_matches'] += 1
print(f"[Dual Method] Using high-confidence Gray result: {best_gray['template']} at {final_result_coords} (Conf: {best_gray['confidence']:.2f})")
elif best_clahe and not best_gray and best_clahe['confidence'] >= DUAL_METHOD_HIGH_CONFIDENCE_THRESHOLD:
final_result_coords = best_clahe['center']
self.last_detection_method = "CLAHE" + (" (Inv)" if best_clahe['is_inverted'] else "")
self.last_detection_confidence = best_clahe['confidence']
detection_type = "CLAHE Only (High Conf)"
self.performance_stats['clahe_only_detections'] += 1
if best_clahe['is_inverted']: self.performance_stats['inverted_matches'] += 1
print(f"[Dual Method] Using high-confidence CLAHE result: {best_clahe['template']} at {final_result_coords} (Conf: {best_clahe['confidence']:.2f})")
# Strategy 2: Find overlapping results if no high-confidence single result yet
if final_result_coords is None:
best_overlap_match = None
highest_overlap_confidence = 0
for gray_match in gray_results:
for clahe_match in clahe_results:
# Check if templates match (or maybe just type?) - let's stick to same template for now
if gray_match['template'] == clahe_match['template']:
dist = math.sqrt((gray_match['center'][0] - clahe_match['center'][0])**2 +
(gray_match['center'][1] - clahe_match['center'][1])**2)
if dist < MATCH_DISTANCE_THRESHOLD:
# Use average confidence or max? Let's use average.
combined_confidence = (gray_match['confidence'] + clahe_match['confidence']) / 2
if combined_confidence > highest_overlap_confidence:
highest_overlap_confidence = combined_confidence
avg_center = (
(gray_match['center'][0] + clahe_match['center'][0]) // 2,
(gray_match['center'][1] + clahe_match['center'][1]) // 2
)
best_overlap_match = {
'template': gray_match['template'],
'center': avg_center,
'confidence': combined_confidence,
'dist': dist,
'is_inverted': gray_match['is_inverted'] or clahe_match['is_inverted'],
'type': gray_match['type'] # Type should be same
}
if best_overlap_match:
final_result_coords = best_overlap_match['center']
self.last_detection_method = "Dual Overlap" + (" (Inv)" if best_overlap_match['is_inverted'] else "")
self.last_detection_confidence = best_overlap_match['confidence']
detection_type = "Dual Overlap"
self.performance_stats['dual_method_detections'] += 1
if best_overlap_match['is_inverted']: self.performance_stats['inverted_matches'] += 1
print(f"[Dual Method] Using overlapping result: {best_overlap_match['template']} at {final_result_coords} (Conf: {best_overlap_match['confidence']:.2f}, Dist: {best_overlap_match['dist']:.1f}px)")
# Strategy 3: Fallback to best single result if no overlap found
if final_result_coords is None:
all_results = gray_results + clahe_results
if all_results:
best_overall = max(all_results, key=lambda x: x['confidence'])
# Use a slightly lower threshold for fallback
if best_overall['confidence'] >= DUAL_METHOD_FALLBACK_CONFIDENCE_THRESHOLD:
final_result_coords = best_overall['center']
method_name = "Gray Fallback" if best_overall in gray_results else "CLAHE Fallback"
method_name += " (Inv)" if best_overall['is_inverted'] else ""
self.last_detection_method = method_name
self.last_detection_confidence = best_overall['confidence']
detection_type = "Fallback"
self.performance_stats['fallback_detections'] += 1 # Track fallbacks
if best_overall in gray_results: self.performance_stats['gray_only_detections'] += 1
else: self.performance_stats['clahe_only_detections'] += 1
if best_overall['is_inverted']: self.performance_stats['inverted_matches'] += 1
print(f"[Dual Method] Using fallback result ({method_name}): {best_overall['template']} at {final_result_coords} (Conf: {best_overall['confidence']:.2f})")
# --- Final Result Handling & Debug ---
if final_result_coords:
self.performance_stats['successful_detections'] += 1
if self.DEBUG_LEVEL >= 3:
# --- Visual Debugging ---
try:
# Create side-by-side comparison of gray and clahe
debug_processed_path = os.path.join(DEBUG_SCREENSHOT_DIR, f"dual_processed_{int(time.time())}.png")
# Ensure images have same height for hstack
h_gray_img, w_gray_img = img_gray.shape[:2]
h_clahe_img, w_clahe_img = img_clahe.shape[:2]
max_h = max(h_gray_img, h_clahe_img)
# Resize if needed (convert to BGR for stacking color images if necessary)
img_gray_bgr = cv2.cvtColor(cv2.resize(img_gray, (int(w_gray_img * max_h / h_gray_img), max_h)), cv2.COLOR_GRAY2BGR)
img_clahe_bgr = cv2.cvtColor(cv2.resize(img_clahe, (int(w_clahe_img * max_h / h_clahe_img), max_h)), cv2.COLOR_GRAY2BGR)
debug_img_processed = np.hstack([img_gray_bgr, img_clahe_bgr])
cv2.imwrite(debug_processed_path, debug_img_processed)
# Draw results on original BGR image
result_img = img_bgr.copy()
# Draw relative centers for visualization within the region
for result in gray_results:
rel_x = result['center'][0] - region_x
rel_y = result['center'][1] - region_y
cv2.circle(result_img, (rel_x, rel_y), 5, (0, 0, 255), -1) # Red = Gray
for result in clahe_results:
rel_x = result['center'][0] - region_x
rel_y = result['center'][1] - region_y
cv2.circle(result_img, (rel_x, rel_y), 5, (0, 255, 0), -1) # Green = CLAHE
# Mark final chosen point (relative)
final_rel_x = final_result_coords[0] - region_x
final_rel_y = final_result_coords[1] - region_y
cv2.circle(result_img, (final_rel_x, final_rel_y), 8, (255, 0, 0), 2) # Blue circle = Final
debug_result_path = os.path.join(DEBUG_SCREENSHOT_DIR, f"dual_result_{int(time.time())}.png")
cv2.imwrite(debug_result_path, result_img)
print(f"[Dual Method Debug] Saved processed image to {debug_processed_path}")
print(f"[Dual Method Debug] Saved result image to {debug_result_path}")
except Exception as debug_e:
print(f"Error during visual debugging image generation: {debug_e}")
# --- End Visual Debugging ---
return final_result_coords # Return absolute coordinates
else:
if self.DEBUG_LEVEL > 0: # Log failure only if debug level > 0
print(f"[Dual Method] No sufficiently confident match found. Time: {elapsed_time:.3f}s")
self.last_detection_method = None
self.last_detection_confidence = 0.0
return None
def find_keyword_in_region(self, region: Tuple[int, int, int, int]) -> Optional[Tuple[int, int]]:
"""
Wrapper method to find keywords in a region.
Uses either the new dual method or the legacy method based on the 'use_dual_method' flag.
Returns absolute center coordinates or None.
"""
if region is None or len(region) != 4 or region[2] <= 0 or region[3] <= 0:
print(f"Error: Invalid region provided to find_keyword_in_region: {region}")
return None
if self.use_dual_method:
result = self.find_keyword_dual_method(region)
# Debug Fallback Check
if result is None and self.DEBUG_LEVEL >= 3:
print("[DEBUG] Dual method failed. Trying legacy method for comparison...")
legacy_result = self._find_keyword_legacy(region)
if legacy_result:
print(f"[DEBUG] Legacy method succeeded where dual method failed. Legacy Coords: {legacy_result}")
else:
print("[DEBUG] Legacy method also failed.")
return result # Return the result from the dual method
else:
# Use legacy method if dual method is disabled
return self._find_keyword_legacy(region)
def print_detection_stats(self):
"""Prints the collected keyword detection performance statistics."""
stats = self.performance_stats
total = stats['total_detections']
successful = stats['successful_detections']
if total == 0:
print("\n=== Keyword Detection Performance Stats ===")
print("No detections recorded yet.")
return
print("\n=== Keyword Detection Performance Stats ===")
print(f"Total Detection Attempts: {total}")
success_rate = (successful / total * 100) if total > 0 else 0
print(f"Successful Detections: {successful} ({success_rate:.1f}%)")
avg_time = (stats['total_detection_time'] / total * 1000) if total > 0 else 0
print(f"Average Detection Time: {avg_time:.2f} ms")
if successful > 0:
dual_pct = stats['dual_method_detections'] / successful * 100
gray_pct = stats['gray_only_detections'] / successful * 100
clahe_pct = stats['clahe_only_detections'] / successful * 100
fallback_pct = stats['fallback_detections'] / successful * 100 # Percentage of successful that were fallbacks
print("\nDetection Method Distribution (Successful Detections):")
print(f" - Dual Overlap: {stats['dual_method_detections']} ({dual_pct:.1f}%)")
print(f" - Gray Only: {stats['gray_only_detections']} ({gray_pct:.1f}%)")
print(f" - CLAHE Only: {stats['clahe_only_detections']} ({clahe_pct:.1f}%)")
# Note: Gray Only + CLAHE Only might include high-confidence singles and fallbacks.
# Fallback count is a subset of Gray/CLAHE Only.
print(f" - Fallback Used:{stats['fallback_detections']} ({fallback_pct:.1f}%)")
if stats['inverted_matches'] > 0:
inv_pct = stats['inverted_matches'] / successful * 100
print(f"\nInverted Matches Detected: {stats['inverted_matches']} ({inv_pct:.1f}%)")
print("==========================================")
return None
def calculate_avatar_coords(self, bubble_tl_coords: Tuple[int, int], offset_x: int = AVATAR_OFFSET_X) -> Tuple[int, int]: def calculate_avatar_coords(self, bubble_tl_coords: Tuple[int, int], offset_x: int = AVATAR_OFFSET_X) -> Tuple[int, int]:
""" """
@ -1263,41 +1605,29 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
print("\n--- Starting UI Monitoring Loop (Thread) ---") print("\n--- Starting UI Monitoring Loop (Thread) ---")
# --- Initialization (Instantiate modules within the thread) --- # --- Initialization (Instantiate modules within the thread) ---
# Load templates directly using constants defined in this file for now # --- Template Dictionary Setup (Refactored) ---
# Consider passing config or a template loader object in the future essential_templates = {
templates = { # Bubble Corners (All types needed for legacy/color fallback)
# Regular Bubble (Original + Skins) - Keys match those used in find_dialogue_bubbles
'corner_tl': CORNER_TL_IMG, 'corner_br': CORNER_BR_IMG, 'corner_tl': CORNER_TL_IMG, 'corner_br': CORNER_BR_IMG,
'corner_tl_type2': CORNER_TL_TYPE2_IMG, 'corner_br_type2': CORNER_BR_TYPE2_IMG, 'corner_tl_type2': CORNER_TL_TYPE2_IMG, 'corner_br_type2': CORNER_BR_TYPE2_IMG,
'corner_tl_type3': CORNER_TL_TYPE3_IMG, 'corner_br_type3': CORNER_BR_TYPE3_IMG, 'corner_tl_type3': CORNER_TL_TYPE3_IMG, 'corner_br_type3': CORNER_BR_TYPE3_IMG,
'corner_tl_type4': CORNER_TL_TYPE4_IMG, 'corner_br_type4': CORNER_BR_TYPE4_IMG, # Added type4 'corner_tl_type4': CORNER_TL_TYPE4_IMG, 'corner_br_type4': CORNER_BR_TYPE4_IMG, # Added type4
# Bot Bubble (Single Type)
'bot_corner_tl': BOT_CORNER_TL_IMG, 'bot_corner_br': BOT_CORNER_BR_IMG, 'bot_corner_tl': BOT_CORNER_TL_IMG, 'bot_corner_br': BOT_CORNER_BR_IMG,
# Keywords & UI Elements # Core Keywords (for dual method)
'keyword_wolf_lower': KEYWORD_wolf_LOWER_IMG, 'keyword_wolf_lower': KEYWORD_wolf_LOWER_IMG,
'keyword_wolf_upper': KEYWORD_Wolf_UPPER_IMG, 'keyword_Wolf_upper': KEYWORD_Wolf_UPPER_IMG,
'keyword_wolf_lower_type2': KEYWORD_wolf_LOWER_TYPE2_IMG,
'keyword_wolf_upper_type2': KEYWORD_Wolf_UPPER_TYPE2_IMG,
'keyword_wolf_lower_type3': KEYWORD_wolf_LOWER_TYPE3_IMG,
'keyword_wolf_upper_type3': KEYWORD_Wolf_UPPER_TYPE3_IMG,
'keyword_wolf_lower_type4': KEYWORD_wolf_LOWER_TYPE4_IMG, # Added type4
'keyword_wolf_upper_type4': KEYWORD_Wolf_UPPER_TYPE4_IMG, # Added type4
# --- Add Reply Keywords ---
'keyword_wolf_reply': KEYWORD_WOLF_REPLY_IMG, 'keyword_wolf_reply': KEYWORD_WOLF_REPLY_IMG,
'keyword_wolf_reply_type2': KEYWORD_WOLF_REPLY_TYPE2_IMG, # Essential UI Elements
'keyword_wolf_reply_type3': KEYWORD_WOLF_REPLY_TYPE3_IMG,
'keyword_wolf_reply_type4': KEYWORD_WOLF_REPLY_TYPE4_IMG,
# --- End Reply Keywords ---
'copy_menu_item': COPY_MENU_ITEM_IMG, 'profile_option': PROFILE_OPTION_IMG, 'copy_menu_item': COPY_MENU_ITEM_IMG, 'profile_option': PROFILE_OPTION_IMG,
'copy_name_button': COPY_NAME_BUTTON_IMG, 'send_button': SEND_BUTTON_IMG, 'copy_name_button': COPY_NAME_BUTTON_IMG, 'send_button': SEND_BUTTON_IMG,
'chat_input': CHAT_INPUT_IMG, 'profile_name_page': PROFILE_NAME_PAGE_IMG, 'chat_input': CHAT_INPUT_IMG, 'profile_name_page': PROFILE_NAME_PAGE_IMG,
'profile_page': PROFILE_PAGE_IMG, 'chat_room': CHAT_ROOM_IMG, 'profile_page': PROFILE_PAGE_IMG, 'chat_room': CHAT_ROOM_IMG,
'base_screen': BASE_SCREEN_IMG, 'world_map_screen': WORLD_MAP_IMG, # Added for navigation 'base_screen': BASE_SCREEN_IMG, 'world_map_screen': WORLD_MAP_IMG,
'world_chat': WORLD_CHAT_IMG, 'private_chat': PRIVATE_CHAT_IMG, 'world_chat': WORLD_CHAT_IMG, 'private_chat': PRIVATE_CHAT_IMG,
# Add position templates # Position templates
'development_pos': POS_DEV_IMG, 'interior_pos': POS_INT_IMG, 'science_pos': POS_SCI_IMG, 'development_pos': POS_DEV_IMG, 'interior_pos': POS_INT_IMG, 'science_pos': POS_SCI_IMG,
'security_pos': POS_SEC_IMG, 'strategy_pos': POS_STR_IMG, 'security_pos': POS_SEC_IMG, 'strategy_pos': POS_STR_IMG,
# Add capitol templates # Capitol templates
'capitol_button': CAPITOL_BUTTON_IMG, 'president_title': PRESIDENT_TITLE_IMG, 'capitol_button': CAPITOL_BUTTON_IMG, 'president_title': PRESIDENT_TITLE_IMG,
'pos_btn_dev': POS_BTN_DEV_IMG, 'pos_btn_int': POS_BTN_INT_IMG, 'pos_btn_sci': POS_BTN_SCI_IMG, 'pos_btn_dev': POS_BTN_DEV_IMG, 'pos_btn_int': POS_BTN_INT_IMG, 'pos_btn_sci': POS_BTN_SCI_IMG,
'pos_btn_sec': POS_BTN_SEC_IMG, 'pos_btn_str': POS_BTN_STR_IMG, 'pos_btn_sec': POS_BTN_SEC_IMG, 'pos_btn_str': POS_BTN_STR_IMG,
@ -1305,16 +1635,34 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
'page_sec': PAGE_SEC_IMG, 'page_str': PAGE_STR_IMG, 'page_sec': PAGE_SEC_IMG, 'page_str': PAGE_STR_IMG,
'dismiss_button': DISMISS_BUTTON_IMG, 'confirm_button': CONFIRM_BUTTON_IMG, 'dismiss_button': DISMISS_BUTTON_IMG, 'confirm_button': CONFIRM_BUTTON_IMG,
'close_button': CLOSE_BUTTON_IMG, 'back_arrow': BACK_ARROW_IMG, 'close_button': CLOSE_BUTTON_IMG, 'back_arrow': BACK_ARROW_IMG,
'reply_button': REPLY_BUTTON_IMG # Added reply button template key 'reply_button': REPLY_BUTTON_IMG
} }
# Use default confidence/region settings from constants legacy_templates = {
# Detector now loads its own color settings internally based on hardcoded values # Deprecated Keywords (for legacy method fallback)
detector = DetectionModule(templates, 'keyword_wolf_lower_type2': KEYWORD_wolf_LOWER_TYPE2_IMG,
confidence=CONFIDENCE_THRESHOLD, 'keyword_wolf_upper_type2': KEYWORD_Wolf_UPPER_TYPE2_IMG,
'keyword_wolf_lower_type3': KEYWORD_wolf_LOWER_TYPE3_IMG,
'keyword_wolf_upper_type3': KEYWORD_Wolf_UPPER_TYPE3_IMG,
'keyword_wolf_lower_type4': KEYWORD_wolf_LOWER_TYPE4_IMG,
'keyword_wolf_upper_type4': KEYWORD_Wolf_UPPER_TYPE4_IMG,
'keyword_wolf_reply_type2': KEYWORD_WOLF_REPLY_TYPE2_IMG,
'keyword_wolf_reply_type3': KEYWORD_WOLF_REPLY_TYPE3_IMG,
'keyword_wolf_reply_type4': KEYWORD_WOLF_REPLY_TYPE4_IMG,
}
# Combine dictionaries
all_templates = {**essential_templates, **legacy_templates}
# --- End Template Dictionary Setup ---
# --- Instantiate Modules ---
detector = DetectionModule(all_templates,
confidence=CONFIDENCE_THRESHOLD, # Default for legacy pyautogui calls
state_confidence=STATE_CONFIDENCE_THRESHOLD, state_confidence=STATE_CONFIDENCE_THRESHOLD,
region=SCREENSHOT_REGION) region=SCREENSHOT_REGION,
# Use default input coords/keys from constants use_dual_method=True) # Enable new method by default
interactor = InteractionModule(detector, input_coords=(CHAT_INPUT_CENTER_X, CHAT_INPUT_CENTER_Y), input_template_key='chat_input', send_button_key='send_button') interactor = InteractionModule(detector,
input_coords=(CHAT_INPUT_CENTER_X, CHAT_INPUT_CENTER_Y),
input_template_key='chat_input',
send_button_key='send_button')
# --- State Management (Local to this monitoring thread) --- # --- State Management (Local to this monitoring thread) ---
last_processed_bubble_info = None # Store the whole dict now last_processed_bubble_info = None # Store the whole dict now
@ -1322,8 +1670,13 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
screenshot_counter = 0 # Initialize counter for debug screenshots screenshot_counter = 0 # Initialize counter for debug screenshots
main_screen_click_counter = 0 # Counter for consecutive main screen clicks main_screen_click_counter = 0 # Counter for consecutive main screen clicks
loop_counter = 0 # Add loop counter for debugging
while True: while True:
loop_counter += 1
# print(f"\n--- UI Loop Iteration #{loop_counter} ---") # DEBUG REMOVED
# --- Process ALL Pending Commands First --- # --- Process ALL Pending Commands First ---
# print("[DEBUG] UI Loop: Processing command queue...") # DEBUG REMOVED
commands_processed_this_cycle = False commands_processed_this_cycle = False
try: try:
while True: # Loop to drain the queue while True: # Loop to drain the queue
@ -1401,22 +1754,26 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
except queue.Empty: except queue.Empty:
# No more commands in the queue for this cycle # No more commands in the queue for this cycle
if commands_processed_this_cycle: # if commands_processed_this_cycle: # DEBUG REMOVED
print("UI Thread: Finished processing commands for this cycle.") # print("UI Thread: Finished processing commands for this cycle.") # DEBUG REMOVED
pass pass
except Exception as cmd_err: except Exception as cmd_err:
print(f"UI Thread: Error processing command queue: {cmd_err}") print(f"UI Thread: Error processing command queue: {cmd_err}")
# Consider if pausing is needed on error, maybe not # Consider if pausing is needed on error, maybe not
# --- Now, Check Pause State --- # --- Now, Check Pause State ---
# print("[DEBUG] UI Loop: Checking pause state...") # DEBUG REMOVED
if monitoring_paused_flag[0]: if monitoring_paused_flag[0]:
# print("[DEBUG] UI Loop: Monitoring is paused. Sleeping...") # DEBUG REMOVED
# If paused, sleep and skip UI monitoring part # If paused, sleep and skip UI monitoring part
time.sleep(0.1) # Sleep briefly while paused time.sleep(0.1) # Sleep briefly while paused
continue # Go back to check commands again continue # Go back to check commands again
# --- If not paused, proceed with UI Monitoring --- # --- If not paused, proceed with UI Monitoring ---
# print("[DEBUG] UI Loop: Monitoring is active. Proceeding...") # DEBUG REMOVED
# --- Check for Main Screen Navigation --- # --- Check for Main Screen Navigation ---
# print("[DEBUG] UI Loop: Checking for main screen navigation...") # DEBUG REMOVED
try: try:
base_locs = detector._find_template('base_screen', confidence=0.8) base_locs = detector._find_template('base_screen', confidence=0.8)
map_locs = detector._find_template('world_map_screen', confidence=0.8) map_locs = detector._find_template('world_map_screen', confidence=0.8)
@ -1447,53 +1804,8 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
# Decide if you want to continue or pause after error # Decide if you want to continue or pause after error
main_screen_click_counter = 0 # Reset counter on error too main_screen_click_counter = 0 # Reset counter on error too
# --- Process Commands Second (Non-blocking) ---
# This block seems redundant now as commands are processed at the start of the loop.
# Keeping it commented out for now, can be removed later if confirmed unnecessary.
# try:
# command_data = command_queue.get_nowait() # Check for commands without blocking
# action = command_data.get('action')
# if action == 'send_reply':
# text_to_send = command_data.get('text')
# # reply_context_activated = command_data.get('reply_context_activated', False) # Check if reply context was set
#
# if not text_to_send:
# print("UI Thread: Received send_reply command with no text.")
# continue # Skip if no text
#
# print(f"UI Thread: Received command to send reply: '{text_to_send[:50]}...'")
# # The reply context (clicking bubble + reply button) is now handled *before* putting into queue.
# # So, we just need to send the message directly here.
# # The input field should already be focused and potentially have @Username prefix if reply context was activated.
# interactor.send_chat_message(text_to_send)
#
# elif action == 'remove_position': # <--- Handle new command
# region = command_data.get('trigger_bubble_region')
# if region:
# print(f"UI Thread: Received command to remove position triggered by bubble region: {region}")
# # Call the new UI function
# success = remove_user_position(detector, interactor, region) # Call synchronous function
# print(f"UI Thread: Position removal attempt finished. Success: {success}")
# # Note: No need to send result back unless main thread needs confirmation
# else:
# print("UI Thread: Received remove_position command without trigger_bubble_region.")
# elif action == 'pause': # <--- Handle pause command
# print("UI Thread: Received pause command. Pausing monitoring.")
# monitoring_paused_flag[0] = True
# continue # Immediately pause after receiving command
# elif action == 'resume': # <--- Handle resume command (might be redundant if checked above, but safe)
# print("UI Thread: Received resume command. Resuming monitoring.")
# monitoring_paused_flag[0] = False
# else:
# print(f"UI Thread: Received unknown command: {action}")
# except queue.Empty:
# pass # No command waiting, continue with monitoring
# except Exception as cmd_err:
# print(f"UI Thread: Error processing command queue: {cmd_err}")
# # This block is now part of the command processing loop above
# pass
# --- Verify Chat Room State Before Bubble Detection (Only if NOT paused) --- # --- Verify Chat Room State Before Bubble Detection (Only if NOT paused) ---
# print("[DEBUG] UI Loop: Verifying chat room state...") # DEBUG REMOVED
try: try:
# Use a slightly lower confidence maybe, or state_confidence # Use a slightly lower confidence maybe, or state_confidence
chat_room_locs = detector._find_template('chat_room', confidence=detector.state_confidence) chat_room_locs = detector._find_template('chat_room', confidence=detector.state_confidence)
@ -1505,8 +1817,8 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
print("UI Thread: Continuing loop after attempting chat room cleanup.") print("UI Thread: Continuing loop after attempting chat room cleanup.")
time.sleep(0.5) # Small pause after cleanup attempt time.sleep(0.5) # Small pause after cleanup attempt
continue continue
# else: # Optional: Log if chat room is confirmed # else: # Optional: Log if chat room is confirmed # DEBUG REMOVED
# print("UI Thread: Chat room state confirmed.") # print("[DEBUG] UI Thread: Chat room state confirmed.") # DEBUG REMOVED
except Exception as state_check_err: except Exception as state_check_err:
print(f"UI Thread: Error checking for chat room state: {state_check_err}") print(f"UI Thread: Error checking for chat room state: {state_check_err}")
@ -1515,24 +1827,34 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
# --- Then Perform UI Monitoring (Bubble Detection) --- # --- Then Perform UI Monitoring (Bubble Detection) ---
# print("[DEBUG] UI Loop: Starting bubble detection...") # DEBUG REMOVED
try: try:
# 1. Detect Bubbles # 1. Detect Bubbles
all_bubbles_data = detector.find_dialogue_bubbles() # Returns list of dicts all_bubbles_data = detector.find_dialogue_bubbles() # Returns list of dicts
if not all_bubbles_data: time.sleep(2); continue if not all_bubbles_data:
# print("[DEBUG] UI Loop: No bubbles detected.") # DEBUG REMOVED
time.sleep(2); continue
# Filter out bot bubbles # Filter out bot bubbles
other_bubbles_data = [b_info for b_info in all_bubbles_data if not b_info['is_bot']] other_bubbles_data = [b_info for b_info in all_bubbles_data if not b_info['is_bot']]
if not other_bubbles_data: time.sleep(0.2); continue if not other_bubbles_data:
# print("[DEBUG] UI Loop: No non-bot bubbles detected.") # DEBUG REMOVED
time.sleep(0.2); continue
# print(f"[DEBUG] UI Loop: Found {len(other_bubbles_data)} non-bot bubbles. Sorting...") # DEBUG REMOVED
# Sort bubbles from bottom to top (based on bottom Y coordinate) # Sort bubbles from bottom to top (based on bottom Y coordinate)
sorted_bubbles = sorted(other_bubbles_data, key=lambda b_info: b_info['bbox'][3], reverse=True) sorted_bubbles = sorted(other_bubbles_data, key=lambda b_info: b_info['bbox'][3], reverse=True)
# Iterate through sorted bubbles (bottom to top) # Iterate through sorted bubbles (bottom to top)
for target_bubble_info in sorted_bubbles: # print("[DEBUG] UI Loop: Iterating through sorted bubbles...") # DEBUG REMOVED
for i, target_bubble_info in enumerate(sorted_bubbles):
# print(f"[DEBUG] UI Loop: Processing bubble #{i+1}") # DEBUG REMOVED
target_bbox = target_bubble_info['bbox'] target_bbox = target_bubble_info['bbox']
bubble_region = (target_bbox[0], target_bbox[1], target_bbox[2]-target_bbox[0], target_bbox[3]-target_bbox[1]) # Ensure bubble_region uses standard ints
bubble_region = (int(target_bbox[0]), int(target_bbox[1]), int(target_bbox[2]-target_bbox[0]), int(target_bbox[3]-target_bbox[1]))
# 3. Detect Keyword in Bubble # 3. Detect Keyword in Bubble
# print(f"[DEBUG] UI Loop: Detecting keyword in region {bubble_region}...") # DEBUG REMOVED
keyword_coords = detector.find_keyword_in_region(bubble_region) keyword_coords = detector.find_keyword_in_region(bubble_region)
if keyword_coords: if keyword_coords:
@ -1567,6 +1889,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
print("Warning: SCREENSHOT_REGION not defined, searching full screen for bubble snapshot.") print("Warning: SCREENSHOT_REGION not defined, searching full screen for bubble snapshot.")
# --- Take Snapshot for Re-location --- # --- Take Snapshot for Re-location ---
# print("[DEBUG] UI Loop: Taking bubble snapshot...") # DEBUG REMOVED
try: try:
bubble_region_tuple = (int(bubble_region[0]), int(bubble_region[1]), int(bubble_region[2]), int(bubble_region[3])) bubble_region_tuple = (int(bubble_region[0]), int(bubble_region[1]), int(bubble_region[2]), int(bubble_region[3]))
if bubble_region_tuple[2] <= 0 or bubble_region_tuple[3] <= 0: if bubble_region_tuple[2] <= 0 or bubble_region_tuple[3] <= 0:
@ -1594,7 +1917,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
continue # Skip to next bubble continue # Skip to next bubble
# 4. Re-locate bubble *before* copying text # 4. Re-locate bubble *before* copying text
print("Attempting to re-locate bubble before copying text...") # print("[DEBUG] UI Loop: Re-locating bubble before copying text...") # DEBUG REMOVED
new_bubble_box_for_copy = None new_bubble_box_for_copy = None
if bubble_snapshot: if bubble_snapshot:
try: try:
@ -1610,11 +1933,12 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
continue # Skip to the next bubble in the outer loop continue # Skip to the next bubble in the outer loop
print(f"Successfully re-located bubble for copy at: {new_bubble_box_for_copy}") print(f"Successfully re-located bubble for copy at: {new_bubble_box_for_copy}")
# Define the region based on the re-located bubble to find the keyword again # Define the region based on the re-located bubble, casting to int
copy_bubble_region = (new_bubble_box_for_copy.left, new_bubble_box_for_copy.top, copy_bubble_region = (int(new_bubble_box_for_copy.left), int(new_bubble_box_for_copy.top),
new_bubble_box_for_copy.width, new_bubble_box_for_copy.height) int(new_bubble_box_for_copy.width), int(new_bubble_box_for_copy.height))
# Find the keyword *again* within the *new* bubble region to get current coords # Find the keyword *again* within the *new* bubble region to get current coords
# print("[DEBUG] UI Loop: Finding keyword again in re-located region...") # DEBUG REMOVED
current_keyword_coords = detector.find_keyword_in_region(copy_bubble_region) current_keyword_coords = detector.find_keyword_in_region(copy_bubble_region)
if not current_keyword_coords: if not current_keyword_coords:
print("Warning: Keyword not found in the re-located bubble region. Skipping this bubble.") print("Warning: Keyword not found in the re-located bubble region. Skipping this bubble.")
@ -1634,6 +1958,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
print(f"Detected keyword is not a reply type (current location). Click target: {click_coords_current}") print(f"Detected keyword is not a reply type (current location). Click target: {click_coords_current}")
# Interact: Get Bubble Text using current coordinates # Interact: Get Bubble Text using current coordinates
# print("[DEBUG] UI Loop: Copying text...") # DEBUG REMOVED
bubble_text = interactor.copy_text_at(click_coords_current) bubble_text = interactor.copy_text_at(click_coords_current)
if not bubble_text: if not bubble_text:
print("Error: Could not get dialogue content for this bubble (after re-location).") print("Error: Could not get dialogue content for this bubble (after re-location).")
@ -1641,6 +1966,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
continue # Skip to next bubble continue # Skip to next bubble
# Check recent text history # Check recent text history
# print("[DEBUG] UI Loop: Checking recent text history...") # DEBUG REMOVED
if bubble_text in recent_texts: if bubble_text in recent_texts:
print(f"Content '{bubble_text[:30]}...' in recent history, skipping this bubble.") print(f"Content '{bubble_text[:30]}...' in recent history, skipping this bubble.")
continue # Skip to next bubble continue # Skip to next bubble
@ -1650,6 +1976,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
recent_texts.append(bubble_text) recent_texts.append(bubble_text)
# 5. Interact: Get Sender Name (uses re-location internally via retrieve_sender_name_interaction) # 5. Interact: Get Sender Name (uses re-location internally via retrieve_sender_name_interaction)
# print("[DEBUG] UI Loop: Retrieving sender name...") # DEBUG REMOVED
sender_name = None sender_name = None
try: try:
# --- Bubble Re-location Logic --- # --- Bubble Re-location Logic ---
@ -1715,6 +2042,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
continue # Skip to next bubble continue # Skip to next bubble
# 6. Perform Cleanup # 6. Perform Cleanup
# print("[DEBUG] UI Loop: Performing cleanup after getting name...") # DEBUG REMOVED
cleanup_successful = perform_state_cleanup(detector, interactor) cleanup_successful = perform_state_cleanup(detector, interactor)
if not cleanup_successful: if not cleanup_successful:
print("Error: Failed to return to chat screen after getting name. Skipping this bubble.") print("Error: Failed to return to chat screen after getting name. Skipping this bubble.")
@ -1725,6 +2053,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
continue # Skip to next bubble continue # Skip to next bubble
# --- Attempt to activate reply context --- # --- Attempt to activate reply context ---
# print("[DEBUG] UI Loop: Attempting to activate reply context...") # DEBUG REMOVED
reply_context_activated = False reply_context_activated = False
try: try:
print("Attempting to activate reply context...") print("Attempting to activate reply context...")
@ -1801,6 +2130,7 @@ def run_ui_monitoring_loop(trigger_queue: queue.Queue, command_queue: queue.Queu
# If the loop finished without breaking (i.e., no trigger processed), wait the full interval. # If the loop finished without breaking (i.e., no trigger processed), wait the full interval.
# If it broke, the sleep still happens here before the next cycle. # If it broke, the sleep still happens here before the next cycle.
# print("[DEBUG] UI Loop: Finished bubble iteration or broke early. Sleeping...") # DEBUG REMOVED
time.sleep(1.5) # Polling interval after checking all bubbles or processing one time.sleep(1.5) # Polling interval after checking all bubbles or processing one
except KeyboardInterrupt: except KeyboardInterrupt: