butubb/auto_control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

阶段2: USB 远程 adb server 打通（220 host 网络 5037 全接口监听，无需改 220）+ 自建远程看屏（MJPEG 流 + u2 触控 tap/swipe/key/text，ws-scrcpy 不在 npm 改自建）

Browse Source
This commit is contained in:
butubb
2026-08-17 16:50:12 +08:00
parent 5354fe6379
commit 645938b305
10 changed files with 466 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/adb_helper.py
+55
View File
@@ -31,6 +31,26 @@ def _adb(*args):
except subprocess.TimeoutExpired:
_log.warning(f"adb {' '.join(args)} 超时 {_ADB_TIMEOUT}s")
return ""
return _decode(r)
def _adb_remote(host, port, *args):
"""在远程 adb server（如 220）上执行 adb 命令，返回文本。
USB 设备插在 220，本机 adb server 看不到，必须 -H/-P 指向 220 的 adb server
（host 网络模式，5037 全接口监听，Tailscale 可达）。超时语义同 _adb。
"""
try:
r = subprocess.run([ADB_PATH, "-H", host, "-P", str(port), *args],
capture_output=True, timeout=_ADB_TIMEOUT)
except subprocess.TimeoutExpired:
_log.warning(f"adb -H {host} -P {port} {' '.join(args)} 超时 {_ADB_TIMEOUT}s")
return ""
return _decode(r)
def _decode(r):
"""subprocess 结果解码（Windows GBK 兼容）。"""
out = (r.stdout or b"") + (r.stderr or b"")
try:
return out.decode("utf-8", errors="replace").strip()
@@ -77,6 +97,41 @@ def get_foreground_app(remote_adb_url, timeout=3):
return None
def get_foreground_app_remote(host, port, serial, timeout=3):
"""远程 adb server（220）上设备的前台 App（USB 设备用）。
与 get_foreground_app 同逻辑，只是 adb 指向远程 server（-H/-P）。
"""
if not serial:
return None
try:
r = subprocess.run(
[ADB_PATH, "-H", host, "-P", str(port), "-s", serial, "shell",
"dumpsys window | grep mCurrentFocus"],
capture_output=True, timeout=timeout
)
out = ((r.stdout or b"") + (r.stderr or b"")).decode("utf-8", errors="replace").strip()
if not out:
return None
low = out.lower()
if "device offline" in low or "device not found" in low or "error:" in low:
return None
for line in out.split("\n"):
if "mCurrentFocus" not in line:
continue
m = re.search(r'\bu0\s+(\S+)', line)
if m:
token = m.group(1).rstrip("}")
if "/" in token:
return token.split("/")[0]
return token or None
return None
except subprocess.TimeoutExpired:
return None
except Exception:
return None
def adb_connect(remote_adb_url, retries=5):
"""adb connect（带重试）。返回 True/False。
core/device_pool.py
+21 -2
View File
@@ -17,7 +17,8 @@ import time
from sqlalchemy import text
from core.adb_helper import _adb
from config import USB_ADB_HOST, USB_ADB_PORT
from core.adb_helper import _adb, _adb_remote
from core.logger import get_logger
from core.models import db, Device
@@ -61,13 +62,31 @@ def list_devices():
# ================== 在线状态 ==================
def list_online():
"""本机 adb devices 中 state=device 的设备（实时）。"""
"""本机 adb 中 state=device 的设备 + 池内 USB 设备（经 220 远程 adb server）。
池内无 USB 设备（serial 无冒号）时不查远程 server，零额外开销。
"""
out = _adb("devices") or ""
serials = []
for line in out.splitlines()[1:]:
parts = line.split()
if len(parts) >= 2 and parts[0] and parts[1] == "device":
serials.append(parts[0])
# 池内有 USB 设备时，补查远程 adb server（220，host 网络 5037 全接口监听）
try:
usb = [s for s in list_configured() if ":" not in s]
except Exception:
usb = []
if usb:
try:
out = _adb_remote(USB_ADB_HOST, USB_ADB_PORT, "devices") or ""
for line in out.splitlines()[1:]:
parts = line.split()
if len(parts) >= 2 and parts[0] and parts[1] == "device":
if parts[0] not in serials:
serials.append(parts[0])
except Exception:
pass
return serials
core/device_worker.py
+57 -32
View File
@@ -1,9 +1,9 @@
"""设备生命周期 + 通用 Worker 基类 + 全局状态注册表 + 心跳看门狗。
STFDevice — 单设备生命周期：acquire（占用+adb连接）→ release（断开+释放）
BaseWorker — 通用 worker 基类：封装 STF 生命周期 + u2 连接 + 状态上报 + stop
STFDevice — 单设备生命周期：acquire（直连/远程 server）→ release
BaseWorker — 通用 worker 基类：封装设备生命周期 + u2 连接 + 状态上报 + stop
+ 生命周期钩子(setup/teardown/on_error) + 心跳上报
子类只需实现 run_task(d)，可选实现钩子，不用管 STF/adb/状态/异常
子类只需实现 run_task(d)，可选实现钩子，不用管 adb/状态/异常
_WORKERS — 全局状态注册表，供 web_server 读取实时状态
_watchdog — 心跳看门狗线程，检测 worker 卡死
@@ -16,10 +16,11 @@ from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeou
import uiautomator2 as u2
from config import USB_ADB_HOST, USB_ADB_PORT
from core.logger import get_logger
from core.stf_client import DeviceOfflineError, STFError
from core import device_pool
from .adb_helper import adb_connect, adb_disconnect
from .adb_helper import adb_connect, _adb_remote
_log = get_logger("core.worker")
@@ -35,15 +36,15 @@ _U2_INFO_TIMEOUT = 10
class STFDevice:
"""单设备生命周期：acquire（直连）→ release（无操作）。
阶段 1：已摘除 STF occupy/release——单实例互斥由 TaskManager._running 保证，
设备健康由 device_pool（本地 adb 在线状态）判断。保留 STF 引用仅为
非 IP:port（USB 序列号）设备的 remoteConnect 桥接，阶段 2 换成 220 adb server。
阶段 2：已摘除 STF occupy/release 与 remoteConnect 桥接——
单实例互斥由 TaskManager._running 保证；非 IP:port（USB 序列号）设备
通过 220 的远程 adb server（-H/-P）驱动，不再依赖 STF。
"""
def __init__(self, stf_client=None, serial=None):
self.stf = stf_client # 仅 USB 桥接用（阶段 2 移除）
def __init__(self, serial=None):
self.serial = serial
self.remote_adb_url = None
self.remote_adb_url = None # IP:5555 设备：设备地址
self.usb_server = None # USB 设备：远程 adb server 地址（host:port）
def acquire(self, serial=None):
self.serial = serial or self.serial or self._pick_free()
@@ -51,18 +52,21 @@ class STFDevice:
# 直连优先：serial 本身是 IP:5555（如 Tailscale 网络设备）时直接 adb connect。
# 只 connect、绝不 disconnect/kill-server（红线），不影响 STF provider 的连接。
# 仅当 serial 不是 IP:port（如 USB 序列号）时才走 STF 桥接（阶段 2 替换）。
if ":" in self.serial:
self.remote_adb_url = self.serial
_log.info(f"[{self.serial}] 直连设备: {self.remote_adb_url}")
if not adb_connect(self.remote_adb_url):
raise STFError(f"adb connect 失败: {self.remote_adb_url}", "offline")
else:
if self.stf is None:
raise STFError(f"USB 设备 {self.serial} 暂不支持（未启用 STF 桥接）", "offline")
self.remote_adb_url = self.stf.remote_connect(self.serial)
_log.info(f"[{self.serial}] STF 桥接: {self.remote_adb_url}")
if not adb_connect(self.remote_adb_url):
raise STFError(f"adb connect 失败: {self.remote_adb_url}", "offline")
# USB 设备：插在 220 上，本机 adb 看不到，经远程 adb server 驱动。
# 无需 adb connect——设备已在远程 server 上，直接校验可见性。
self.usb_server = f"{USB_ADB_HOST}:{USB_ADB_PORT}"
_log.info(f"[{self.serial}] USB 设备，远程 adb server: {self.usb_server}")
out = _adb_remote(USB_ADB_HOST, USB_ADB_PORT, "devices") or ""
seen = {line.split()[0] for line in out.splitlines()[1:]
if len(line.split()) >= 2}
if self.serial not in seen:
raise STFError(f"USB 设备 {self.serial} 不在远程 adb server", "offline")
time.sleep(2)
def _pick_free(self):
@@ -72,15 +76,8 @@ class STFDevice:
return free[0]
def release(self):
if not self.serial:
return
# 直连模式（remote_adb_url == serial）：不 disconnect（红线，STF 共享 transport）
if self.remote_adb_url and self.remote_adb_url != self.serial:
adb_disconnect(self.remote_adb_url)
try:
self.stf.remote_disconnect(self.serial)
except Exception:
pass
# 直连模式：不 disconnect（红线，STF 共享 transport）；USB：无连接可断
pass
# ================== 全局 worker 状态注册表（供 web_server 读取） ==================
@@ -349,15 +346,20 @@ class BaseWorker(threading.Thread):
# ================== 基类主循环（不要重写） ==================
def run(self):
"""基类主循环：不要重写。子类实现 run_task + 可选钩子。"""
device = STFDevice(self.stf, serial=self.serial)
device = STFDevice(serial=self.serial)
_update_status(self.serial, status="connecting", stf_occupied=False,
last_error="", remote_adb_url="", model="",
last_error="", remote_adb_url="", usb_server="", model="",
current_action="")
try:
device.acquire()
remote = device.remote_adb_url
# u2.connect 可能 hang 住（atx-agent 无响应），用线程池+超时保护
self.d = self._u2_connect_with_timeout(remote)
usb_server = device.usb_server
# u2.connect 可能 hang 住（atx-agent 无响应），用线程池+超时保护；
# USB 设备（无 IP:5555）经远程 adb server 建 u2 连接
if usb_server:
self.d = self._u2_connect_usb_with_timeout(usb_server)
else:
self.d = self._u2_connect_with_timeout(remote)
if self.d is None:
raise STFError(f"u2.connect 超时 {_U2_CONNECT_TIMEOUT}s，atx-agent 无响应", "offline")
_log.info(f"[{self.serial}] u2 连接成功")
@@ -369,7 +371,8 @@ class BaseWorker(threading.Thread):
except Exception:
pass
_update_status(self.serial, status="running", stf_occupied=True,
remote_adb_url=remote, model=model)
remote_adb_url=remote or "", usb_server=usb_server or "",
model=model)
# setup 钩子
try:
@@ -435,6 +438,28 @@ class BaseWorker(threading.Thread):
except FuturesTimeout:
_log.error(f"[{self.serial}] u2.connect 超时 {_U2_CONNECT_TIMEOUT}s")
return None
def _u2_connect_usb_with_timeout(self, server):
"""USB 设备：经远程 adb server（220）建 u2 连接，带超时保护。
u2 3.7 底层 adbutils 支持远程 adb server：AdbClient(host, port).device(serial)，
再交给 u2.connect(AdbDevice)（与本地连接同语义，无需 adb connect）。
"""
host, _, port = server.partition(":")
with ThreadPoolExecutor(max_workers=1) as pool:
def _connect():
from adbutils import AdbClient
dev = AdbClient(host=host, port=int(port)).device(self.serial)
return u2.connect(dev)
fut = pool.submit(_connect)
try:
return fut.result(timeout=_U2_CONNECT_TIMEOUT)
except FuturesTimeout:
_log.error(f"[{self.serial}] u2 远程连接超时 {_U2_CONNECT_TIMEOUT}s")
return None
except Exception as e:
_log.error(f"[{self.serial}] u2 远程连接失败: {e}")
return None
except Exception as e:
_log.error(f"[{self.serial}] u2.connect 异常: {e}")
return None
core/task_manager.py
+37 -9
View File
@@ -32,7 +32,7 @@ from core.logger import get_logger
from core.models import db, DeviceGroup as GroupRow, TaskJob as JobRow
from core import device_pool
from .stf_client import STFClient, DeviceOfflineError
from .adb_helper import get_foreground_app, adb_connect_light, adb_disconnect
from .adb_helper import get_foreground_app, get_foreground_app_remote
from .device_worker import (
get_all_worker_status, _update_status, _remove_worker,
_WORKERS, _WORKERS_LOCK,
@@ -260,22 +260,26 @@ class _ForegroundScanner:
worker_status = {w["serial"]: w for w in get_all_worker_status()}
# 分类设备
have_conn = {} # serial -> remote_adb_url（worker 运行中，已有 adb 连接）
have_conn = {} # serial -> remote_adb_url（IP:5555 运行中设备）
usb_conn = {} # serial -> usb_server（USB 运行中设备，经远程 adb server）
free_serials = [] # 空闲设备
for serial in online:
w = worker_status.get(serial, {})
url = w.get("remote_adb_url")
if url and w.get("status") in ("running", "connecting"):
# 1. 设备正在执行任务，已有 adb 连接，直接查询
have_conn[serial] = url
if w.get("status") not in ("running", "connecting"):
free_serials.append(serial)
elif w.get("remote_adb_url"):
# 1. IP:5555 设备正在执行任务，已有 adb 连接，直接查询
have_conn[serial] = w["remote_adb_url"]
elif w.get("usb_server"):
# 1b. USB 设备运行中：经远程 adb server 查询
usb_conn[serial] = w["usb_server"]
else:
# 2. 空闲设备
free_serials.append(serial)
results = {}
_log.info("前台 App 扫描分类: 运行中=%d, 空闲=%d",
len(have_conn), len(free_serials))
_log.info("前台 App 扫描分类: 运行中=%d, USB运行中=%d, 空闲=%d",
len(have_conn), len(usb_conn), len(free_serials))
# 1. worker 运行中设备：用已有 remote_adb_url 查询（并发 10）
if have_conn:
@@ -289,6 +293,18 @@ class _ForegroundScanner:
except Exception:
results[s] = None
# 1b. USB 运行中设备：经远程 adb server 查询（并发 5）
if usb_conn:
with ThreadPoolExecutor(max_workers=min(5, len(usb_conn))) as pool:
futures = {pool.submit(self._scan_usb_conn, s): s
for s, server in usb_conn.items()}
for fut in as_completed(futures, timeout=15):
s = futures[fut]
try:
results[s] = fut.result()
except Exception:
results[s] = None
# 2. 空闲设备：不打扰，直接返回"空闲"
for s in free_serials:
results[s] = "空闲"
@@ -303,6 +319,18 @@ class _ForegroundScanner:
self._scanning.clear()
_log.info("前台 App 扫描完成: %d 台设备", len(results))
def _scan_usb_conn(self, serial):
"""USB 运行中设备：经远程 adb server（220）查询前台 App。"""
try:
with _WORKERS_LOCK:
server = _WORKERS.get(serial, {}).get("usb_server", "")
if not server:
return None
host, _, port = server.partition(":")
return get_foreground_app_remote(host, int(port), serial)
except Exception:
return None
# ================== 任务管理器 ==================
class TaskManager: