""" Windows Device Implementation Concrete implementation of AbstractDevice for Windows desktop applications. Uses CV mode for UI element detection. """ import logging import os import re import subprocess import sys import time from typing import Optional, Dict, Any, List, Tuple from ...core.abstract_device import AbstractDevice class WindowsApp: """ Windows 应用封装类 用于兼容 DroidBot 的 App 接口 """ def __init__(self, window_title): self._window_title = window_title self.main_activity = None self.activities = [] def get_package_name(self): return self._window_title class WindowsDevice(AbstractDevice): """ Windows 设备的具体实现（CV 模式） 使用 OmniParser 进行 UI 元素检测，pyautogui 进行输入模拟。 """ def __init__(self, window_title: str = None, exe_path: str = None, output_dir: str = None, cv_mode: bool = True, steam_game_id: str = None, # Steam 游戏 ID **kwargs): """ 初始化 Windows 设备连接 :param window_title: 目标窗口标题（支持部分匹配） :param exe_path: 可执行文件路径（用于启动应用） :param output_dir: 输出目录 :param cv_mode: 是否使用 CV 模式（默认 True） :param steam_game_id: Steam 游戏 ID，例如 '730' (CS2) :param app_path: (兼容参数) :param device_serial: (兼容参数) :param kwargs: 忽略的其他参数 """ super().__init__(output_dir) self._window_title = window_title self._exe_path = exe_path self._steam_game_id = steam_game_id self.cv_mode = cv_mode # 兼容性：初始化 _app 对象 self._app = WindowsApp(self._window_title) # 窗口句柄相关 self._hwnd = None self._window_rect = None # 显示信息缓存 self._display_info = None # 截图计数器 self._screenshot_count = 0 # 控制标志 self.pause_sending_event = False self.logger = logging.getLogger(self.__class__.__name__) def get_platform_name(self) -> str: """获取平台名称""" return "windows" # ==================== 连接管理 ==================== def set_up(self) -> None: """设置设备连接前的准备工作""" # 检查依赖库 try: import pyautogui import mss import win32gui import win32con except ImportError as e: raise ImportError( f"Missing required library: {e}. " "Please install: pip install pyautogui mss pywin32" ) # 禁用 pyautogui 的安全暂停 pyautogui.PAUSE = 0.1 pyautogui.FAILSAFE = True # 预加载必要的 win32 API try: import win32process import win32api except ImportError: pass def connect(self) -> bool: """连接到目标窗口""" self.set_up() # 查找目标窗口 self._hwnd = self._find_window() if self._hwnd is None: self.logger.warning(f"Window not found: {self._window_title}") # 如果提供了 exe_path，尝试启动应用 if self._exe_path and os.path.exists(self._exe_path): self.logger.info(f"Attempting to start app: {self._exe_path}") if self.start_app(): time.sleep(2) # 等待应用启动 self._hwnd = self._find_window() if self._hwnd: self.connected = True self._update_window_rect() self.logger.info(f"Connected to window: {self._get_window_title()}") return True else: self.connected = False self.logger.error("Failed to connect to window") return False def run_initial_setup(self) -> bool: from DroidBot.guiagent_bridge import GuiAgentBridge bridge = GuiAgentBridge(device=self, app=self._app, app_name=self._window_title) current_state = self.get_current_state() return bridge.handle_with_guiagent("game_initial", {"state": current_state}) def disconnect(self) -> None: """断开设备连接""" self._hwnd = None self._window_rect = None self.connected = False self.logger.info("Disconnected from window") def tear_down(self) -> None: """清理设备资源""" self.disconnect() def check_connectivity(self) -> bool: """检查窗口是否仍然存在 (包含句柄自愈机制)""" import win32gui # 1. 检查现有句柄是否有效 if self._hwnd and win32gui.IsWindow(self._hwnd): return True # 2. 句柄失效，尝试自愈 (重新查找窗口) # 这种情况常见于游戏崩溃重启、更新重启、或从 Launcher 切换到 Game 主窗口 self.logger.warning(f"Window handle {self._hwnd} invalid, attempting to reconnect to '{self._window_title}'...") new_hwnd = self._find_window() if new_hwnd: self._hwnd = new_hwnd self.logger.info(f"Reconnected to window: {self._hwnd}") return True return False # ==================== 窗口查找 ==================== def _find_window(self) -> Optional[int]: """查找目标窗口句柄 (优先完全匹配)""" import win32gui if self._window_title is None: return None exact_matches = [] partial_matches = [] def enum_callback(hwnd, _): if win32gui.IsWindowVisible(hwnd): title = win32gui.GetWindowText(hwnd) if not title: return True # 优先完全匹配 (忽略大小写) if self._window_title.lower() == title.lower(): exact_matches.append((hwnd, title)) # 其次部分匹配 elif self._window_title.lower() in title.lower(): partial_matches.append((hwnd, title)) return True try: win32gui.EnumWindows(enum_callback, None) except Exception as e: self.logger.error(f"EnumWindows failed: {e}") return None # 1. 优先返回完全匹配 if exact_matches: self.logger.info(f"Found exact match window: {exact_matches[0][1]} ({exact_matches[0][0]})") return exact_matches[0][0] # 2. 其次返回部分匹配 if partial_matches: self.logger.info(f"Found partial match window: {partial_matches[0][1]} ({partial_matches[0][0]})") return partial_matches[0][0] return None def _get_window_title(self) -> str: """获取当前窗口标题""" if self._hwnd is None: return "" import win32gui try: return win32gui.GetWindowText(self._hwnd) except Exception: return "" def _update_window_rect(self) -> None: """更新窗口位置和大小""" if self._hwnd is None: return import win32gui try: rect = win32gui.GetWindowRect(self._hwnd) self._window_rect = { 'left': rect[0], 'top': rect[1], 'right': rect[2], 'bottom': rect[3], 'width': rect[2] - rect[0], 'height': rect[3] - rect[1], } except Exception as e: self.logger.warning(f"Failed to get window rect: {e}") # ==================== 状态获取 ==================== def check_network(self, host: str = "8.8.8.8") -> bool: return True def get_current_state(self): """获取当前设备状态""" from .windows_device_state import WindowsDeviceState from ...cv import cv # 更新窗口位置 self._update_window_rect() # 截图 screenshot_path = self.take_screenshot() # CV 检测 cv_views = [] if screenshot_path and os.path.exists(screenshot_path): try: img = cv.load_image_from_path(screenshot_path) if img is not None: cv_views = cv.find_views(img) self.logger.info(f"CV detected {len(cv_views)} views") except Exception as e: self.logger.warning(f"CV detection failed: {e}") # 创建状态对象 state = WindowsDeviceState( device=self, cv_views=cv_views, window_title=self._get_window_title(), screenshot_path=screenshot_path ) return state def get_display_info(self, refresh: bool = False) -> Dict[str, Any]: """获取显示信息""" if self._display_info is None or refresh: self._update_window_rect() if self._window_rect: self._display_info = { 'width': self._window_rect['width'], 'height': self._window_rect['height'], 'left': self._window_rect['left'], 'top': self._window_rect['top'], } else: # 默认值 self._display_info = { 'width': 1920, 'height': 1080, 'left': 0, 'top': 0, } return self._display_info # ==================== 屏幕操作 ==================== def take_screenshot(self, path: str = None) -> Optional[str]: """截取窗口屏幕""" import mss if path is None: if self.output_dir: self._screenshot_count += 1 path = os.path.join( self.output_dir, f"screen_{self._screenshot_count}.png" ) else: path = f"screen_{int(time.time())}.png" # 确保目录存在 os.makedirs(os.path.dirname(path) if os.path.dirname(path) else '.', exist_ok=True) self._update_window_rect() try: with mss.mss() as sct: if self._window_rect: # 截取窗口区域 monitor = { 'left': self._window_rect['left'], 'top': self._window_rect['top'], 'width': self._window_rect['width'], 'height': self._window_rect['height'], } else: # 截取整个屏幕 monitor = sct.monitors[1] screenshot = sct.grab(monitor) # 保存截图 from PIL import Image img = Image.frombytes('RGB', screenshot.size, screenshot.bgra, 'raw', 'BGRX') img.save(path) self.logger.debug(f"Screenshot saved to: {path}") return path except Exception as e: self.logger.error(f"Failed to take screenshot: {e}") return None def unlock(self) -> None: """解锁屏幕（Windows 上无需实现）""" pass # ==================== 输入操作 ==================== def send_event(self, event) -> bool: """发送输入事件""" return event.send(self) def view_touch(self, x: int, y: int, button: str = 'left') -> None: """点击指定坐标""" import pyautogui # 如果是相对于窗口的坐标，转换为屏幕坐标 if self._window_rect: screen_x = self._window_rect['left'] + x screen_y = self._window_rect['top'] + y else: screen_x, screen_y = x, y pyautogui.click(screen_x, screen_y, button=button) self.logger.debug(f"Touch ({button}) at ({screen_x}, {screen_y})") def view_long_touch(self, x: int, y: int, duration: int = 2000) -> None: """长按指定坐标""" import pyautogui if self._window_rect: screen_x = self._window_rect['left'] + x screen_y = self._window_rect['top'] + y else: screen_x, screen_y = x, y # 移动到位置，按下，等待，松开 pyautogui.moveTo(screen_x, screen_y) pyautogui.mouseDown() time.sleep(duration / 1000.0) pyautogui.mouseUp() self.logger.debug(f"Long touch at ({screen_x}, {screen_y}) for {duration}ms") def view_drag(self, start_xy: Tuple[int, int], end_xy: Tuple[int, int], duration: int = 500) -> None: """拖拽操作""" import pyautogui start_x, start_y = start_xy end_x, end_y = end_xy if self._window_rect: start_x += self._window_rect['left'] start_y += self._window_rect['top'] end_x += self._window_rect['left'] end_y += self._window_rect['top'] pyautogui.moveTo(start_x, start_y) pyautogui.drag( end_x - start_x, end_y - start_y, duration=duration / 1000.0 ) self.logger.debug(f"Drag from ({start_x}, {start_y}) to ({end_x}, {end_y})") def view_set_text(self, text: str) -> None: """输入文本""" import pyautogui # 对于中文等非ASCII字符，使用 pyperclip 和 Ctrl+V try: # 检查是否包含非ASCII字符 text.encode('ascii') # 纯ASCII字符，使用 typewrite pyautogui.typewrite(text, interval=0.05) except UnicodeEncodeError: # 包含非ASCII字符，使用剪贴板 import pyperclip pyperclip.copy(text) pyautogui.hotkey('ctrl', 'v') self.logger.debug(f"Set text: {text[:20]}...") def key_press(self, key_code: str) -> None: """按键操作""" import pyautogui pyautogui.press(key_code) self.logger.debug(f"Key press: {key_code}") # ==================== 应用管理 ==================== @property def app_identifier(self) -> str: """获取应用标识符""" if self._window_title: return self._window_title if self._exe_path: return os.path.basename(self._exe_path) return "unknown" def is_foreground(self) -> bool: """检查目标窗口是否在前台""" if self._hwnd is None: return False import win32gui try: foreground_hwnd = win32gui.GetForegroundWindow() return foreground_hwnd == self._hwnd except Exception: return False def pull_back_to_app(self) -> bool: """ 将目标窗口拉到前台 (多策略尝试) """ if self._hwnd is None: # 尝试重新查找窗口 self._hwnd = self._find_window() if self._hwnd is None: self.logger.warning("Cannot find window to bring to foreground") return False import win32gui import win32con import win32process import win32api import pyautogui # 先检查窗口句柄是否仍然有效 if not win32gui.IsWindow(self._hwnd): self.logger.warning(f"Window handle {self._hwnd} is no longer valid") self._hwnd = self._find_window() if self._hwnd is None: return False # 检查当前是否已经是前台窗口 try: current_foreground = win32gui.GetForegroundWindow() if current_foreground == self._hwnd: self.logger.debug("Window is already in foreground") return True except Exception: pass self.logger.info(f"Attempting to bring window {self._hwnd} to foreground...") # === Method 1: AttachThreadInput + SetForegroundWindow === try: current_thread_id = win32api.GetCurrentThreadId() target_thread_id, target_process_id = win32process.GetWindowThreadProcessId(self._hwnd) foreground_hwnd = win32gui.GetForegroundWindow() foreground_thread_id, _ = win32process.GetWindowThreadProcessId(foreground_hwnd) attached_to_foreground = False attached_to_target = False try: # 先依附到当前前台窗口的线程（获取输入权限） if current_thread_id != foreground_thread_id: win32process.AttachThreadInput(current_thread_id, foreground_thread_id, True) attached_to_foreground = True # 再依附到目标窗口的线程 if current_thread_id != target_thread_id: win32process.AttachThreadInput(current_thread_id, target_thread_id, True) attached_to_target = True # 如果最小化了，先还原 if win32gui.IsIconic(self._hwnd): win32gui.ShowWindow(self._hwnd, win32con.SW_RESTORE) # 显示窗口 win32gui.ShowWindow(self._hwnd, win32con.SW_SHOW) # 尝试多种置顶方法 win32gui.BringWindowToTop(self._hwnd) win32gui.SetForegroundWindow(self._hwnd) finally: # 务必解除依附 if attached_to_target: try: win32process.AttachThreadInput(current_thread_id, target_thread_id, False) except Exception: pass if attached_to_foreground: try: win32process.AttachThreadInput(current_thread_id, foreground_thread_id, False) except Exception: pass time.sleep(0.3) if win32gui.GetForegroundWindow() == self._hwnd: self.logger.info("Method 1 (AttachThreadInput) succeeded") return True except Exception as e: self.logger.warning(f"Method 1 (AttachThreadInput) failed: {e}") # === Method 2: Alt Key Trick === try: self.logger.debug("Trying Alt-Key trick...") # 模拟 Alt 键按下释放，欺骗 Windows 认为有用户输入 pyautogui.keyDown('alt') time.sleep(0.02) pyautogui.keyUp('alt') time.sleep(0.02) if win32gui.IsIconic(self._hwnd): win32gui.ShowWindow(self._hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(self._hwnd) time.sleep(0.3) if win32gui.GetForegroundWindow() == self._hwnd: self.logger.info("Method 2 (Alt-Key) succeeded") return True except Exception as e: self.logger.warning(f"Method 2 (Alt-Key) failed: {e}") # === Method 3: Minimize then Restore === try: self.logger.debug("Trying Minimize-Restore trick...") win32gui.ShowWindow(self._hwnd, win32con.SW_MINIMIZE) time.sleep(0.1) win32gui.ShowWindow(self._hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(self._hwnd) time.sleep(0.3) if win32gui.GetForegroundWindow() == self._hwnd: self.logger.info("Method 3 (Minimize-Restore) succeeded") return True except Exception as e: self.logger.warning(f"Method 3 (Minimize-Restore) failed: {e}") # === Method 4: Click on Window === try: self.logger.debug("Trying direct click on window...") self._update_window_rect() if self._window_rect: # 点击窗口中心 center_x = self._window_rect['left'] + self._window_rect['width'] // 2 center_y = self._window_rect['top'] + self._window_rect['height'] // 2 pyautogui.click(center_x, center_y) time.sleep(0.3) if win32gui.GetForegroundWindow() == self._hwnd: self.logger.info("Method 4 (Direct Click) succeeded") return True except Exception as e: self.logger.warning(f"Method 4 (Direct Click) failed: {e}") self.logger.error("All methods to bring window to foreground failed.") return False def start_app(self) -> bool: """启动应用 (支持 Steam 游戏和普通应用)""" # 1. 首先尝试查找并连接现有窗口 if self._find_window(): self._hwnd = self._find_window() if self.pull_back_to_app(): self.logger.info("App is already running, brought to foreground.") return True # 2. 确定启动方式 launch_command = None # 优先使用 Steam 协议启动 if self._steam_game_id: launch_command = f"start steam://rungameid/{self._steam_game_id}" self.logger.info(f"Launching via Steam protocol: {launch_command}") elif self._exe_path: if not os.path.exists(self._exe_path): self.logger.error(f"Executable not found: {self._exe_path}") return False launch_command = self._exe_path self.logger.info(f"Launching via executable: {launch_command}") else: self.logger.warning("No steam_game_id or exe_path specified, cannot start app") return False try: # 3. 启动进程 if self._steam_game_id: # Steam 协议需要使用 shell=True subprocess.Popen(launch_command, shell=True) else: subprocess.Popen( [self._exe_path], cwd=os.path.dirname(self._exe_path), shell=True ) self.logger.info(f"Started app process") # 4. 轮询等待窗口出现 (Timeout: 120s for Steam games which may need loading) self.logger.info(f"Waiting for window '{self._window_title}' to appear...") max_retries = 120 # Steam 游戏启动可能比较慢 for i in range(max_retries): hwnd = self._find_window() if hwnd: self._hwnd = hwnd self.logger.info(f"Window appeared after {i} seconds.") time.sleep(3) # Steam 游戏窗口出现后稍等一下，等待完全初始化 return True time.sleep(1) self.logger.error("Timed out waiting for window to appear.") return False except Exception as e: self.logger.error(f"Failed to start app: {e}") return False