From d3ae0fd6965b231be881201abee1113ee5904e22 Mon Sep 17 00:00:00 2001 From: yltx <2326439151@qq.com> Date: Wed, 18 Feb 2026 03:25:53 +0800 Subject: [PATCH 1/2] fix: WSL-only support and emulator detection --- autowsgr/configs.py | 50 +++++---- autowsgr/timer/controllers/os_controller.py | 112 +++++++++++++++++++- autowsgr/types.py | 19 +++- 3 files changed, 160 insertions(+), 21 deletions(-) diff --git a/autowsgr/configs.py b/autowsgr/configs.py index 60b2e61c..23f54395 100644 --- a/autowsgr/configs.py +++ b/autowsgr/configs.py @@ -268,25 +268,37 @@ def __post_init__(self) -> None: object.__setattr__(self, 'os_type', OSType.auto()) # 模拟器 - if self.emulator_name is None: - object.__setattr__( - self, - 'emulator_name', - self.emulator_type.default_emulator_name(self.os_type), - ) - if self.emulator_start_cmd is None: - object.__setattr__( - self, - 'emulator_start_cmd', - self.emulator_type.auto_emulator_path(self.os_type), - ) - assert self.emulator_start_cmd is not None - if self.emulator_process_name is None: - object.__setattr__( - self, - 'emulator_process_name', - os.path.basename(self.emulator_start_cmd), - ) + if self.os_type == OSType.linux: + if self.emulator_name is None: + raise ValueError('WSL 需要显式设置 emulator_name') + if self.emulator_start_cmd is None: + raise ValueError('WSL 需要显式设置 emulator_start_cmd') + if self.emulator_process_name is None: + object.__setattr__( + self, + 'emulator_process_name', + os.path.basename(self.emulator_start_cmd), + ) + else: + if self.emulator_name is None: + object.__setattr__( + self, + 'emulator_name', + self.emulator_type.default_emulator_name(self.os_type), + ) + if self.emulator_start_cmd is None: + object.__setattr__( + self, + 'emulator_start_cmd', + self.emulator_type.auto_emulator_path(self.os_type), + ) + assert self.emulator_start_cmd is not None + if self.emulator_process_name is None: + object.__setattr__( + self, + 'emulator_process_name', + os.path.basename(self.emulator_start_cmd), + ) # 游戏 object.__setattr__(self, 'app_name', self.game_app.app_name) diff --git a/autowsgr/timer/controllers/os_controller.py b/autowsgr/timer/controllers/os_controller.py index be6ef84f..b785edd5 100644 --- a/autowsgr/timer/controllers/os_controller.py +++ b/autowsgr/timer/controllers/os_controller.py @@ -1,6 +1,7 @@ import json import os import re +import shlex import subprocess import time from typing import Protocol @@ -12,7 +13,7 @@ from autowsgr.configs import UserConfig from autowsgr.constants.custom_exceptions import CriticalErr -from autowsgr.types import EmulatorType +from autowsgr.types import EmulatorType, OSType from autowsgr.utils.logger import Logger @@ -300,3 +301,112 @@ def __get_mumu_info(self) -> dict: except Exception as e: self.logger.error(f'{cmd} {e}') return {} + + +class LinuxController(OSController): + def __init__(self, config: UserConfig, logger: Logger) -> None: + self.logger = logger + self.is_wsl = OSType._is_wsl() + + self.emulator_type = config.emulator_type + if config.emulator_name is None: + raise CriticalErr('WSL 需要显式设置 emulator_name') + if config.emulator_start_cmd is None: + raise CriticalErr('WSL 需要显式设置 emulator_start_cmd') + if config.emulator_process_name is None: + raise CriticalErr('WSL 需要显式设置 emulator_process_name') + self.emulator_name = config.emulator_name + self.emulator_start_cmd = config.emulator_start_cmd + self.emulator_process_name = config.emulator_process_name + self.dev_name = f'Android:///{self.emulator_name}' + + def is_android_online(self) -> bool: + """判断 timer 给定的设备是否在线""" + devices = self._adb_devices() + if self.emulator_name in devices: + return True + if self.is_wsl: + return self._is_windows_process_running() + try: + subprocess.run( + ['pgrep', '-f', self.emulator_process_name], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + return True + except subprocess.CalledProcessError: + return False + + def kill_android(self) -> None: + """强制终止模拟器进程""" + try: + if self.is_wsl: + result = subprocess.run( + ['taskkill.exe', '/f', '/im', self.emulator_process_name], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise CriticalErr(result.stderr.strip() or result.stdout.strip()) + else: + subprocess.run( + ['pkill', '-9', '-f', self.emulator_process_name], + check=True, + ) + self.logger.info(f'已终止模拟器进程: {self.emulator_process_name}') + except Exception as e: + raise CriticalErr(f'终止模拟器进程失败: {e}') + + def start_android(self) -> None: + """启动模拟器""" + try: + subprocess.Popen(self._split_command(self.emulator_start_cmd)) + self.logger.info(f'正在启动模拟器: {self.emulator_start_cmd}') + start_time = time.time() + while not self.is_android_online(): + time.sleep(1) + if time.time() - start_time > 120: + raise TimeoutError('模拟器启动超时！') + except Exception as e: + raise CriticalErr(f'启动模拟器失败: {e}') + + def _adb_devices(self) -> list[str]: + try: + from airtest.core.android.adb import ADB + + adb = ADB().get_adb_path() + result = subprocess.run( + [adb, 'devices'], + capture_output=True, + text=True, + check=True, + ) + except Exception as e: + self.logger.error(f'adb devices 失败: {e}') + return [] + + devices = [] + for line in result.stdout.splitlines(): + line = line.strip() + if not line or line.startswith('List of devices'): + continue + parts = line.split() + if len(parts) >= 2 and parts[1] == 'device': + devices.append(parts[0]) + return devices + + def _is_windows_process_running(self) -> bool: + result = subprocess.run( + ['tasklist.exe', '/fi', f'IMAGENAME eq {self.emulator_process_name}'], + capture_output=True, + text=True, + ) + output = (result.stdout or '').lower() + if 'no tasks' in output: + return False + return self.emulator_process_name.lower() in output + + @staticmethod + def _split_command(command: str) -> list[str]: + return shlex.split(command) diff --git a/autowsgr/types.py b/autowsgr/types.py index 70935ea4..5df17088 100644 --- a/autowsgr/types.py +++ b/autowsgr/types.py @@ -46,7 +46,7 @@ class OcrBackend(StrEnum): class OSType(StrEnum): windows = 'Windows' - linux = 'Linux' + linux = 'linux' macos = 'macOS' @classmethod @@ -55,8 +55,25 @@ def auto(cls) -> 'OSType': return OSType.windows if sys.platform == 'darwin': return OSType.macos + if sys.platform.startswith('linux'): + if cls._is_wsl(): + return OSType.linux + raise ValueError('暂不支持非 WSL 的 Linux 系统') raise ValueError(f'不支持的操作系统 {sys.platform}') + @staticmethod + def _is_wsl() -> bool: + if os.environ.get('WSL_DISTRO_NAME') or os.environ.get('WSL_INTEROP'): + return True + for path in ('/proc/sys/kernel/osrelease', '/proc/version'): + try: + with open(path, 'r', encoding='utf-8', errors='ignore') as handle: + if 'microsoft' in handle.read().lower(): + return True + except OSError: + continue + return False + class EmulatorType(StrEnum): leidian = '雷电' From 65e1682fb6c373bb6e7fd08b3970353ac6a5ef01 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 17 Feb 2026 19:28:14 +0000 Subject: [PATCH 2/2] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- autowsgr/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/autowsgr/types.py b/autowsgr/types.py index 5df17088..ddd9b412 100644 --- a/autowsgr/types.py +++ b/autowsgr/types.py @@ -67,7 +67,7 @@ def _is_wsl() -> bool: return True for path in ('/proc/sys/kernel/osrelease', '/proc/version'): try: - with open(path, 'r', encoding='utf-8', errors='ignore') as handle: + with open(path, encoding='utf-8', errors='ignore') as handle: if 'microsoft' in handle.read().lower(): return True except OSError: