kushidou/padsleep_adb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加无线 ADB 连接支持

Browse Source 
- 扩展配置：添加 adb_wireless_host/port/auto_connect 字段
- AdbManager: 新增无线连接方法 (connect_wireless, disconnect_wireless)
- AdbManager: 实现自动重连机制 (ensure_wireless_connection)
- AdbManager: USB 优先策略，无线作为备选
- IPC: 新增 connect_wireless/disconnect_wireless 命令
- TUI: 新增无线连接对话框 (W 键)
- TUI: 设备列表显示连接类型 (USB/WiFi)
This commit is contained in:
kushidou
2026-07-10 09:39:45 +08:00
parent fa0a315ba2
commit 1cde7a5d81
2 changed files with 243 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
padsleep.py
+180 -10
View File
@@ -35,7 +35,9 @@ DEFAULT_CONFIG = {
"check_interval_seconds": 30,
"auto_sleep_delay_minutes": 5,
"adb_device_serial": "",
"adb_path": "adb",
"adb_wireless_host": "",
"adb_wireless_port": 5555,
"adb_wireless_auto_connect": False,
}
# ── 日志 ────────────────────────────────────────────────────────
@@ -182,6 +184,8 @@ class AdbManager:
self._screen_on_since: float | None = None
self._last_known_screen_on = False
self._last_adb_check = 0
self._wireless_addr: str = "" # 当前无线连接地址 (IP:PORT)
self._wireless_retries: int = 0 # 无线重连尝试次数
# ── 连接管理 ──
@@ -209,6 +213,107 @@ class AdbManager:
self._last_error = str(e)
return None
# ── 无线连接管理 ──
@staticmethod
def is_wireless_serial(serial: str) -> bool:
"""判断 serial 是否为无线格式（IP:PORT）。"""
parts = serial.split(":")
if len(parts) != 2:
return False
try:
port = int(parts[1])
ip_parts = parts[0].split(".")
if len(ip_parts) == 4 and all(0 <= int(p) <= 255 for p in ip_parts):
return True
except (ValueError, IndexError):
pass
return False
def connect_wireless(self, host: str, port: int = 5555, timeout: float = 5.0) -> tuple[bool, str]:
"""连接无线 ADB 设备。返回 (success, message)。"""
addr = f"{host}:{port}"
try:
client = self._ensure_client()
result = client.connect(addr, timeout=timeout)
if "connected" in result.lower():
self._wireless_addr = addr
self._wireless_retries = 0
log.info("无线 ADB 已连接: %s", addr)
return True, result
return False, result
except Exception as e:
self._last_error = str(e)
log.error("无线连接失败: %s", e)
return False, str(e)
def disconnect_wireless(self, addr: str = "") -> bool:
"""断开无线 ADB 连接。"""
target = addr or self._wireless_addr
if not target:
return False
try:
client = self._ensure_client()
client.disconnect(target)
log.info("无线 ADB 已断开: %s", target)
if target == self._wireless_addr:
self._wireless_addr = ""
return True
except Exception as e:
self._last_error = str(e)
log.error("断开无线连接失败: %s", e)
return False
def get_connection_type(self) -> str:
"""返回当前连接类型：'usb' / 'wireless' / 'none'。"""
if self._device is None:
return "none"
if self.is_wireless_serial(self._device.serial):
return "wireless"
return "usb"
def ensure_wireless_connection(self) -> bool:
"""确保无线连接存活，失败时尝试重连。返回是否成功。"""
config = self._config_getter()
wireless_host = config.get("adb_wireless_host", "")
wireless_port = config.get("adb_wireless_port", 5555)
if not wireless_host:
return False
addr = f"{wireless_host}:{wireless_port}"
# 检查是否已有连接
if self._device and self._device.serial == addr:
try:
self._device.getprop("ro.build.version.sdk")
return True
except Exception:
# 连接已断开，尝试重连
pass
# 尝试重连（最多 3 次）
max_retries = 3
for i in range(max_retries):
self._wireless_retries += 1
log.info("尝试重连无线设备 (%d/%d): %s", i + 1, max_retries, addr)
success, msg = self.connect_wireless(wireless_host, wireless_port, timeout=3.0)
if success:
# 重新获取设备
try:
client = self._ensure_client()
self._device = client.device(addr)
self._device.getprop("ro.build.version.sdk")
self._ready = True
self._last_error = ""
return True
except Exception:
pass
time.sleep(2)
self._last_error = f"无线重连失败: {addr}"
return False
# ── 设备管理 ──
def check_available(self) -> bool:
@@ -222,11 +327,19 @@ class AdbManager:
return False
def list_devices(self) -> list[dict]:
"""返回 [{"serial": "...", "status": "device"}]。"""
"""返回 [{"serial": "...", "status": "device", "connection_type": "usb"/"wireless"}]。"""
try:
client = self._ensure_client()
devices = client.device_list()
return [{"serial": d.serial, "status": "device"} for d in devices]
result = []
for d in devices:
conn_type = "wireless" if self.is_wireless_serial(d.serial) else "usb"
result.append({
"serial": d.serial,
"status": "device",
"connection_type": conn_type,
})
return result
except Exception as e:
self._last_error = str(e)
return []
@@ -246,22 +359,45 @@ class AdbManager:
config = self._config_getter()
target_serial = config.get("adb_device_serial", "")
wireless_host = config.get("adb_wireless_host", "")
try:
client = self._ensure_client()
devices = client.device_list()
# USB 连接优先
for d in devices:
if not target_serial or d.serial == target_serial:
# 尝试获取设备属性以验证连接
d.getprop("ro.build.version.sdk")
self._device = d
self._ready = True
self._last_error = ""
if not self.is_wireless_serial(d.serial):
if not target_serial or d.serial == target_serial:
d.getprop("ro.build.version.sdk")
self._device = d
self._ready = True
self._last_error = ""
return True
# 尝试无线连接
if wireless_host:
wireless_port = config.get("adb_wireless_port", 5555)
wireless_addr = f"{wireless_host}:{wireless_port}"
# 检查是否已在设备列表中
for d in devices:
if d.serial == wireless_addr:
d.getprop("ro.build.version.sdk")
self._device = d
self._ready = True
self._last_error = ""
self._wireless_addr = wireless_addr
return True
# 不在列表中，尝试连接
if self.ensure_wireless_connection():
return True
if not devices:
if not devices and not wireless_host:
self._last_error = "未检测到 ADB 设备，请连接平板"
elif not devices:
self._last_error = f"无线设备未连接: {wireless_host}:{config.get('adb_wireless_port', 5555)}"
else:
self._last_error = f"未找到指定设备: {target_serial}"
except Exception as e:
@@ -456,6 +592,9 @@ class PadSleepApp:
cfg = self.config_mgr.get()
status["periods_count"] = len(cfg.get("screen_on_periods", []))
status["check_interval"] = cfg.get("check_interval_seconds", 30)
status["connection_type"] = self.adb.get_connection_type()
status["wireless_host"] = cfg.get("adb_wireless_host", "")
status["wireless_port"] = cfg.get("adb_wireless_port", 5555)
return {"ok": True, "data": status}
if cmd == "get_config":
@@ -498,6 +637,25 @@ class PadSleepApp:
ok = self.adb.screen_on()
return {"ok": ok}
if cmd == "connect_wireless":
host = request.get("host", "")
port = request.get("port", 5555)
if not host:
return {"ok": False, "error": "请提供 IP 地址"}
success, msg = self.adb.connect_wireless(host, int(port))
if success:
# 更新配置
self.config_mgr.update({
"adb_wireless_host": host,
"adb_wireless_port": int(port),
})
return {"ok": success, "data": {"message": msg}}
if cmd == "disconnect_wireless":
addr = request.get("addr", "")
ok = self.adb.disconnect_wireless(addr)
return {"ok": ok}
return {"ok": False, "error": f"未知命令: {cmd}"}
# ── 主循环 ──
@@ -517,6 +675,18 @@ class PadSleepApp:
else:
log.info("ADB 可执行文件已就绪")
# 自动连接无线设备
config = self.config_mgr.get()
if config.get("adb_wireless_auto_connect") and config.get("adb_wireless_host"):
host = config["adb_wireless_host"]
port = config.get("adb_wireless_port", 5555)
log.info("尝试自动连接无线设备: %s:%d", host, port)
success, msg = self.adb.connect_wireless(host, port, timeout=5.0)
if success:
log.info("无线设备已连接: %s", msg)
else:
log.warning("无线设备连接失败: %s", msg)
# 启动 IPC
self.ipc.start()
padsleep_tui.py
+63 -3
View File
@@ -197,6 +197,9 @@ class TuiApp:
elif key == ord("s") or key == ord("S"):
self._dlg_select_device()
elif key == ord("w") or key == ord("W"):
self._dlg_connect_wireless()
elif key == ord("h") or key == ord("H"):
self._show_help()
@@ -309,23 +312,39 @@ class TuiApp:
if self.status:
dev = self.status.get("device") or "未选择"
conn_type = self.status.get("connection_type", "none")
conn_icon = "USB" if conn_type == "usb" else "WiFi" if conn_type == "wireless" else "---"
self._clear_line(y)
try:
self.stdscr.addstr(y, 4, f"当前设备: {dev}")
self.stdscr.addstr(f" [{conn_icon}]")
except curses.error:
pass
y += 1
# 显示无线连接配置
wireless_host = self.status.get("wireless_host", "")
wireless_port = self.status.get("wireless_port", 5555)
if wireless_host:
self._clear_line(y)
try:
self.stdscr.addstr(y, 4, f"无线配置: {wireless_host}:{wireless_port}", curses.A_DIM)
except curses.error:
pass
y += 1
if self.devices:
for d in self.devices:
if y >= h:
break
self._clear_line(y)
icon = "✓" if d["status"] == "device" else "✗"
conn_type = d.get("connection_type", "usb")
conn_icon = "USB" if conn_type == "usb" else "WiFi"
pair = 3 if d["status"] == "device" else 2
try:
self.stdscr.addstr(y, 4, f"{icon} {d['serial']}", curses.color_pair(pair))
self.stdscr.addstr(f" ({d['status']})")
self.stdscr.addstr(f" [{conn_icon}]")
except curses.error:
pass
y += 1
@@ -341,7 +360,7 @@ class TuiApp:
if y < h:
self._clear_line(y)
try:
self.stdscr.addstr(y, 4, "[S]选择设备 [R]刷新", curses.A_DIM)
self.stdscr.addstr(y, 4, "[S]选择设备 [W]无线连接 [R]刷新", curses.A_DIM)
except curses.error:
pass
return y + 1
@@ -385,6 +404,9 @@ class TuiApp:
try:
self.stdscr.addstr(y, 4, "ADB: ")
self.stdscr.addstr(f"{ar}", curses.color_pair(ar_color) | curses.A_BOLD)
conn_type = s.get("connection_type", "none")
if conn_type != "none":
self.stdscr.addstr(f" ({conn_type})")
except curses.error:
pass
y += 1
@@ -423,7 +445,7 @@ class TuiApp:
pass # show message below separator
else:
if self.connected:
msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 H帮助"
msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 W无线 H帮助"
else:
msg = "未检测到守护进程 | T 尝试连接"
try:
@@ -603,6 +625,43 @@ class TuiApp:
else:
self.msg(f"选择失败: {r.get('error', '')}")
def _dlg_connect_wireless(self):
"""无线连接对话框。"""
# 获取当前无线配置作为默认值
wireless_host = ""
wireless_port = "5555"
if self.status:
wireless_host = self.status.get("wireless_host", "")
wireless_port = str(self.status.get("wireless_port", 5555))
# 输入 IP 地址
host = self._dlg_input("无线设备 IP 地址:", wireless_host)
if not host:
return
# 输入端口
port_str = self._dlg_input("端口号:", wireless_port)
if not port_str:
return
try:
port = int(port_str)
if not (1 <= port <= 65535):
self.msg("端口范围错误 (1-65535)")
return
except ValueError:
self.msg("端口格式错误")
return
# 发送连接请求
self.msg(f"正在连接 {host}:{port} ...")
r = self.client.send("connect_wireless", host=host, port=port)
if r.get("ok"):
self.msg(f"无线设备已连接: {host}:{port}")
self.refresh()
else:
self.msg(f"连接失败: {r.get('error', '未知错误')}")
@staticmethod
def _validate_time(tstr: str) -> bool:
try:
@@ -631,6 +690,7 @@ class TuiApp:
" E 编辑选中时间段",
" D 删除选中时间段",
" S 选择 ADB 设备",
" W 无线连接设备",
" R 刷新状态",
" L 重载守护进程配置",
" F 强制息屏",