93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
"""唯一设备选择与原子 ADB 操作。绝不允许无 serial 执行手机指令。"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import subprocess
|
||
from dataclasses import dataclass
|
||
from typing import Iterable
|
||
|
||
from .platform import adb_binary
|
||
|
||
|
||
class DeviceError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Device:
|
||
serial: str
|
||
state: str
|
||
model: str = ""
|
||
|
||
|
||
def raw_adb(arguments: Iterable[str], *, text: bool = True) -> subprocess.CompletedProcess:
|
||
try:
|
||
return subprocess.run([adb_binary(), *arguments], capture_output=True, text=text, check=False)
|
||
except FileNotFoundError as exc:
|
||
raise DeviceError("未找到 adb。请将 Android Platform Tools 加入 PATH，或设置 AUTOTRAIN_ADB。") from exc
|
||
|
||
|
||
def devices() -> list[Device]:
|
||
result = raw_adb(["devices", "-l"])
|
||
if result.returncode:
|
||
raise DeviceError(result.stderr.strip() or "adb devices 执行失败")
|
||
found: list[Device] = []
|
||
for line in result.stdout.splitlines()[1:]:
|
||
pieces = line.split()
|
||
if len(pieces) < 2:
|
||
continue
|
||
attributes = dict(item.split(":", 1) for item in pieces[2:] if ":" in item)
|
||
found.append(Device(pieces[0], pieces[1], attributes.get("model", "")))
|
||
return found
|
||
|
||
|
||
def selected_device(serial: str = "") -> Device:
|
||
usable = [item for item in devices() if item.state == "device"]
|
||
if serial:
|
||
match = next((item for item in usable if item.serial == serial), None)
|
||
if match:
|
||
return match
|
||
raise DeviceError(f"指定设备 {serial} 不可用")
|
||
if len(usable) == 1:
|
||
return usable[0]
|
||
if not usable:
|
||
raise DeviceError("没有已授权的 Android 设备")
|
||
raise DeviceError("检测到多台 Android 设备；请在任务中明确指定 device_serial")
|
||
|
||
|
||
def run(serial: str, arguments: Iterable[str], *, binary: bool = False, check: bool = True) -> subprocess.CompletedProcess:
|
||
if not serial:
|
||
raise DeviceError("ADB 操作必须指定 device_serial")
|
||
result = raw_adb(["-s", serial, *arguments], text=not binary)
|
||
if check and result.returncode:
|
||
stderr = result.stderr.decode("utf-8", "ignore") if binary else result.stderr
|
||
raise DeviceError((stderr or "adb command failed").strip())
|
||
return result
|
||
|
||
|
||
def screen_size(serial: str) -> tuple[int, int]:
|
||
output = run(serial, ["shell", "wm", "size"]).stdout
|
||
if isinstance(output, bytes):
|
||
output = output.decode("utf-8", "ignore")
|
||
matches = re.findall(r"(\d+)x(\d+)", output)
|
||
if not matches:
|
||
raise DeviceError(f"无法读取屏幕尺寸：{output}")
|
||
return tuple(map(int, matches[-1])) # Override size 优先
|
||
|
||
|
||
def screen_density(serial: str) -> str:
|
||
output = str(run(serial, ["shell", "wm", "density"]).stdout)
|
||
matches = re.findall(r"(\d+)", output)
|
||
return matches[-1] if matches else ""
|
||
|
||
|
||
def fingerprint(serial: str) -> dict[str, str]:
|
||
def prop(key: str) -> str:
|
||
return str(run(serial, ["shell", "getprop", key]).stdout).strip()
|
||
width, height = screen_size(serial)
|
||
return {
|
||
"serial": serial, "manufacturer": prop("ro.product.manufacturer"),
|
||
"model": prop("ro.product.model"), "android_version": prop("ro.build.version.release"),
|
||
"screen": f"{width}x{height}", "density": screen_density(serial), "orientation": str(run(serial, ["shell", "settings", "get", "system", "user_rotation"]).stdout).strip(),
|
||
}
|