139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Android 真机管理模块。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import subprocess
|
||
import time
|
||
from typing import Optional
|
||
|
||
from utils_android.device_config import ensure_android_wireless_connected, get_android_device_serial
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PhysicalDeviceManager:
|
||
"""真机设备管理器。
|
||
|
||
支持通过配置指定 USB serial 或无线调试 ip:port。
|
||
"""
|
||
|
||
def __init__(self, config):
|
||
self.config = config
|
||
self.serial = get_android_device_serial(config)
|
||
self.boot_timeout_seconds = int(self.config.get("PHYSICAL_BOOT_TIMEOUT_SECONDS", 180))
|
||
self.boot_poll_interval_seconds = int(self.config.get("PHYSICAL_BOOT_POLL_INTERVAL_SECONDS", 5))
|
||
if self.serial:
|
||
logger.info(f"真机 ADB 目标: {self.serial}")
|
||
|
||
@property
|
||
def device_type(self) -> str:
|
||
return "physical"
|
||
|
||
def _run_adb(self, args, timeout: int = 15) -> subprocess.CompletedProcess[str]:
|
||
cmd = ["adb"]
|
||
if self.serial:
|
||
cmd.extend(["-s", self.serial])
|
||
cmd.extend(args)
|
||
return subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
encoding="utf-8",
|
||
errors="ignore",
|
||
timeout=timeout,
|
||
)
|
||
|
||
def _read_output(self, result: subprocess.CompletedProcess[str]) -> str:
|
||
return (result.stdout or result.stderr or "").strip()
|
||
|
||
def _ensure_connected(self) -> bool:
|
||
return ensure_android_wireless_connected(self.config, self.serial)
|
||
|
||
def _wait_for_boot_complete(self, timeout_seconds: Optional[int] = None) -> bool:
|
||
deadline = time.time() + float(timeout_seconds or self.boot_timeout_seconds)
|
||
while time.time() < deadline:
|
||
try:
|
||
self._ensure_connected()
|
||
state_result = self._run_adb(["get-state"], timeout=10)
|
||
state = self._read_output(state_result).lower()
|
||
if state_result.returncode != 0 or state != "device":
|
||
time.sleep(self.boot_poll_interval_seconds)
|
||
continue
|
||
|
||
boot_result = self._run_adb(["shell", "getprop", "sys.boot_completed"], timeout=10)
|
||
boot_completed = self._read_output(boot_result)
|
||
if boot_result.returncode == 0 and boot_completed == "1":
|
||
sdk_result = self._run_adb(["shell", "getprop", "ro.build.version.sdk"], timeout=10)
|
||
if sdk_result.returncode == 0 and self._read_output(sdk_result):
|
||
logger.info("真机重启完成，ADB 已恢复可用")
|
||
return True
|
||
except subprocess.TimeoutExpired:
|
||
logger.warning("等待真机启动完成超时，继续轮询")
|
||
except Exception as exc:
|
||
logger.warning(f"等待真机启动完成时出现异常: {exc}")
|
||
|
||
time.sleep(self.boot_poll_interval_seconds)
|
||
|
||
logger.error("等待真机启动完成超时")
|
||
return False
|
||
|
||
def check_status(self) -> bool:
|
||
try:
|
||
if not self._ensure_connected():
|
||
return False
|
||
state_result = self._run_adb(["get-state"], timeout=10)
|
||
state = self._read_output(state_result).lower()
|
||
if state_result.returncode != 0 or state != "device":
|
||
logger.warning(f"真机 ADB 状态异常: {self._read_output(state_result) or state or 'unknown'}")
|
||
return False
|
||
|
||
verify_result = self._run_adb(["shell", "getprop", "ro.build.version.sdk"], timeout=10)
|
||
if verify_result.returncode == 0 and self._read_output(verify_result):
|
||
return True
|
||
|
||
logger.warning(f"真机 ADB 校验失败: {self._read_output(verify_result)}")
|
||
return False
|
||
except subprocess.TimeoutExpired:
|
||
logger.warning("检查真机状态超时")
|
||
return False
|
||
except FileNotFoundError:
|
||
logger.error("未找到 adb，可执行文件不可用")
|
||
return False
|
||
except Exception as exc:
|
||
logger.warning(f"检查真机状态失败: {exc}")
|
||
return False
|
||
|
||
def ensure_running(self) -> tuple[bool, bool]:
|
||
return (self.check_status(), False)
|
||
|
||
def restart(self) -> bool:
|
||
logger.info("正在重启真机...")
|
||
try:
|
||
result = self._run_adb(["reboot"], timeout=20)
|
||
if result.returncode != 0:
|
||
logger.error(f"真机重启命令失败: {self._read_output(result)}")
|
||
return False
|
||
time.sleep(3)
|
||
return self._wait_for_boot_complete()
|
||
except subprocess.TimeoutExpired:
|
||
logger.warning("adb reboot 等待超时，继续等待设备上线")
|
||
return self._wait_for_boot_complete()
|
||
except Exception as exc:
|
||
logger.error(f"真机重启失败: {exc}")
|
||
return False
|
||
|
||
def recover(self) -> bool:
|
||
logger.info("真机模式不执行 recover，直接跳过")
|
||
return True
|
||
|
||
def recover_from_frozen(self) -> bool:
|
||
logger.warning("检测到真机可能卡死，尝试重启恢复")
|
||
return self.restart()
|
||
|
||
def is_player_frozen(self) -> bool:
|
||
return not self.check_status()
|