675 lines
24 KiB
Python
675 lines
24 KiB
Python
"""
|
||
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
|
||
|