From 203d5d15ef1376f938d68a662e81e3ca74627293 Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 10 Jul 2026 09:11:43 +0800 Subject: [PATCH 1/5] =?UTF-8?q?=E5=B0=86=20ADB=20=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E4=BB=8E=20subprocess=20=E6=94=B9=E9=80=A0=E4=B8=BA=20adbutils?= =?UTF-8?q?=20=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 subprocess.run 调用，改用 adbutils 实现持久连接 - 添加 adbutils>=2.12.0 到 requirements.txt - 更新 .gitignore 添加 Python 包相关条目 - 保持原有公共接口不变，确保兼容性 --- .gitignore | 25 +++++++ padsleep.py | 190 +++++++++++++++++++++++++++-------------------- requirements.txt | 3 +- 3 files changed, 135 insertions(+), 83 deletions(-) diff --git a/.gitignore b/.gitignore index 36b13f1..5d6c2fd 100644 --- a/.gitignore +++ b/.gitignore @@ -27,6 +27,10 @@ share/python-wheels/ *.egg MANIFEST +# PyPI +*.whl +*.tar.gz + # PyInstaller # Usually these files are written by a python script from a template # before PyInstaller builds the exe, so as to inject date/other infos into it. @@ -174,3 +178,24 @@ cython_debug/ # PyPI configuration file .pypirc +# Virtual environments +.venv/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Logs +*.log + +# 开发环境 +.vscode +.mimocode +.codewhale + diff --git a/padsleep.py b/padsleep.py index f496b58..c588ea7 100644 --- a/padsleep.py +++ b/padsleep.py @@ -14,14 +14,14 @@ import logging import os import signal import socket -import subprocess -import sys import threading import time from datetime import datetime from pathlib import Path from socketserver import ThreadingUnixStreamServer, StreamRequestHandler +from adbutils import AdbClient, AdbDevice + # ── 常量 ──────────────────────────────────────────────────────── SCRIPT_DIR = Path(__file__).parent.resolve() @@ -167,68 +167,72 @@ class TimeChecker: # ═══════════════════════════════════════════════════════════════════ -# AdbManager — ADB 操作封装 +# AdbManager — ADB 操作封装（使用 adbutils 库） # ═══════════════════════════════════════════════════════════════════ class AdbManager: - """封装所有 ADB CLI 操作，所有异常都被捕获，不会导致进程退出。""" + """封装所有 ADB 操作，使用 adbutils 库实现持久连接。""" def __init__(self, config_getter): self._config_getter = config_getter + self._client: AdbClient | None = None + self._device: AdbDevice | None = None self._ready = False self._last_error = "" self._screen_on_since: float | None = None self._last_known_screen_on = False - self._last_adb_check = 0 # 限频用 + self._last_adb_check = 0 - # ── 底层命令执行 ── + # ── 连接管理 ── - def _exec(self, args: list, timeout: int = 10) -> tuple: - """执行 ADB 命令，返回 (returncode, stdout, stderr)。永不抛出。""" + def _ensure_client(self) -> AdbClient: + """懒初始化 ADB 客户端连接。""" + if self._client is None: + self._client = AdbClient(host="127.0.0.1", port=5037) + return self._client + + def _ensure_device(self) -> AdbDevice | None: + """获取目标设备，支持串号过滤。""" config = self._config_getter() - adb_path = config.get("adb_path", "adb") - serial = config.get("adb_device_serial", "") - - cmd = [adb_path] - if serial: - cmd.extend(["-s", serial]) - cmd.extend(args) + target_serial = config.get("adb_device_serial", "") try: - r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - return r.returncode, r.stdout.strip(), r.stderr.strip() - except FileNotFoundError: - self._ready = False - self._last_error = f"ADB 未找到（{adb_path}），请安装: sudo apt install adb" - except subprocess.TimeoutExpired: - self._ready = False - self._last_error = "ADB 命令超时" + client = self._ensure_client() + if target_serial: + self._device = client.device(target_serial) + else: + devices = client.device_list() + self._device = devices[0] if devices else None + return self._device except Exception as e: - self._ready = False + self._device = None self._last_error = str(e) - return -1, "", self._last_error + return None # ── 设备管理 ── def check_available(self) -> bool: - """检查 ADB 是否可执行。""" - rc, _, _ = self._exec(["--version"], timeout=5) - return rc == 0 + """检查 ADB Server 是否可连接。""" + try: + client = self._ensure_client() + client.device_list() + return True + except Exception as e: + self._last_error = f"ADB Server 不可用: {e}" + return False def list_devices(self) -> list[dict]: - """返回 [{"serial": "...", "status": "..."}]。""" - rc, out, _ = self._exec(["devices"]) - if rc != 0: + """返回 [{"serial": "...", "status": "device"}]。""" + try: + client = self._ensure_client() + devices = client.device_list() + return [{"serial": d.serial, "status": "device"} for d in devices] + except Exception as e: + self._last_error = str(e) return [] - devices = [] - for line in out.split("\n"): - parts = line.strip().split("\t") - if len(parts) >= 2 and not line.startswith("List"): - devices.append({"serial": parts[0], "status": parts[1]}) - return devices def get_ready_device(self) -> str | None: - """返回第一个状态为 device 的串号，无则返回 None。""" + """返回第一个已连接设备的串号，无则返回 None。""" for d in self.list_devices(): if d["status"] == "device": return d["serial"] @@ -237,57 +241,67 @@ class AdbManager: def is_ready(self) -> bool: """检查是否有已授权的 ADB 设备可用。""" now = time.time() - # 限频：每秒最多检测一次 if now - self._last_adb_check < 1 and self._ready: return True self._last_adb_check = now config = self._config_getter() - target = config.get("adb_device_serial", "") + target_serial = config.get("adb_device_serial", "") - devices = self.list_devices() - for d in devices: - if d["status"] == "device" and (not target or d["serial"] == target): - self._ready = True - self._last_error = "" - return True + try: + client = self._ensure_client() + devices = client.device_list() + + 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 = "" + return True + + if not devices: + self._last_error = "未检测到 ADB 设备，请连接平板" + else: + self._last_error = f"未找到指定设备: {target_serial}" + except Exception as e: + self._last_error = str(e) - # 诊断 - unauthorized = [d for d in devices if d["status"] == "unauthorized"] - if unauthorized: - self._last_error = f"设备未授权: {[d['serial'] for d in unauthorized]}，请在平板上确认授权" - elif not devices: - self._last_error = "未检测到 ADB 设备，请连接平板" - else: - self._last_error = "没有已就绪的 ADB 设备" self._ready = False + self._device = None return False # ── 屏幕控制 ── def get_screen_state(self) -> tuple[bool, str]: """返回 (is_on, wakefulness)，失败时返回上次已知状态。""" - rc, out, _ = self._exec(["shell", "dumpsys", "power"]) - if rc != 0: - return self._last_known_screen_on, "unknown" + try: + device = self._ensure_device() + if device is None: + return self._last_known_screen_on, "unknown" - for line in out.split("\n"): - line = line.strip() - if "mWakefulness" in line: - state = line.split("=")[-1].strip() - is_on = state == "Awake" - self._last_known_screen_on = is_on - if is_on and self._screen_on_since is None: - self._screen_on_since = time.time() - elif not is_on: - self._screen_on_since = None - return is_on, state + output = device.shell("dumpsys power") + for line in output.split("\n"): + line = line.strip() + if "mWakefulness" in line: + state = line.split("=")[-1].strip() + is_on = state == "Awake" + self._last_known_screen_on = is_on + if is_on and self._screen_on_since is None: + self._screen_on_since = time.time() + elif not is_on: + self._screen_on_since = None + return is_on, state + + # fallback + for line in output.split("\n"): + if "Display Power" in line or "mScreenOn" in line: + is_on = "ON" in line.upper() or "=true" in line.lower() + return is_on, "unknown" + except Exception as e: + self._last_error = str(e) - # fallback - for line in out.split("\n"): - if "Display Power" in line or "mScreenOn" in line: - is_on = "ON" in line.upper() or "=true" in line.lower() - return is_on, "unknown" return self._last_known_screen_on, "unknown" def get_screen_on_duration(self) -> float: @@ -299,23 +313,35 @@ class AdbManager: def screen_off(self) -> bool: """发送息屏命令。""" log.info("执行息屏操作") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) - if rc == 0: + try: + device = self._ensure_device() + if device is None: + log.error("息屏失败: 无可用设备") + return False + device.shell("input keyevent 26") self._screen_on_since = None return True - log.error("息屏失败: %s", err) - return False + except Exception as e: + self._last_error = str(e) + log.error("息屏失败: %s", e) + return False def screen_on(self) -> bool: """发送亮屏并解锁命令。""" log.info("执行亮屏操作") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) - if rc == 0: + try: + device = self._ensure_device() + if device is None: + log.error("亮屏失败: 无可用设备") + return False + device.shell("input keyevent 26") self._screen_on_since = time.time() - self._exec(["shell", "input", "keyevent", "82"]) # 解锁（MENU） + device.shell("input keyevent 82") # 解锁（MENU） return True - log.error("亮屏失败: %s", err) - return False + except Exception as e: + self._last_error = str(e) + log.error("亮屏失败: %s", e) + return False # ═══════════════════════════════════════════════════════════════════ diff --git a/requirements.txt b/requirements.txt index 8ba4822..a9b23dd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ # padsleep_adb 依赖 -# 本程序仅使用 Python 标准库，无需额外安装第三方包。 +adbutils>=2.12.0 + # 系统依赖：adb (Android Debug Bridge) # 安装: sudo apt-get install adb # From 9f489e8c40376f7b5a3b2abedf11c20bfbea0287 Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 10 Jul 2026 09:14:09 +0800 Subject: [PATCH 2/5] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20get=5Fready=5Fdevice?= =?UTF-8?q?=20=E4=BB=8E=E5=B7=B2=E8=BF=9E=E6=8E=A5=E8=AE=BE=E5=A4=87?= =?UTF-8?q?=E8=8E=B7=E5=8F=96=20serial?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 不再每次调用都重新查询设备列表 - 直接从 self._device.serial 获取，复用已验证的连接 --- padsleep.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/padsleep.py b/padsleep.py index c588ea7..217b989 100644 --- a/padsleep.py +++ b/padsleep.py @@ -232,10 +232,9 @@ class AdbManager: return [] def get_ready_device(self) -> str | None: - """返回第一个已连接设备的串号，无则返回 None。""" - for d in self.list_devices(): - if d["status"] == "device": - return d["serial"] + """返回当前已连接设备的串号，无则返回 None。""" + if self._device is not None: + return self._device.serial return None def is_ready(self) -> bool: From fa0a315ba2cbdf0879cdd76154a931d05dfe6323 Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 10 Jul 2026 09:17:10 +0800 Subject: [PATCH 3/5] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20TUI=20=E7=95=8C?= =?UTF-8?q?=E9=9D=A2=E9=97=AA=E7=83=81=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 clear() 调用，避免瞬间清空屏幕 - 添加 _clear_line() 辅助方法，逐行清除残留内容 - 每个面板在写入前先清空对应行，确保无残留字符 --- padsleep_tui.py | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/padsleep_tui.py b/padsleep_tui.py index 8e4f117..1e508f1 100644 --- a/padsleep_tui.py +++ b/padsleep_tui.py @@ -135,7 +135,6 @@ class TuiApp: while True: self.height, self.width = self.stdscr.getmaxyx() - self.stdscr.clear() self._draw() self.stdscr.refresh() @@ -224,6 +223,14 @@ class TuiApp: y = self._draw_status_panel(y) self._draw_bottombar() + def _clear_line(self, y: int): + """清空指定行，防止残留内容。""" + if 0 <= y < self.height: + try: + self.stdscr.addstr(y, 0, " " * self.width) + except curses.error: + pass + def _draw_titlebar(self): h, w = self.height, self.width title = " padsleep 配置工具 " @@ -258,6 +265,7 @@ class TuiApp: periods = (self.config or {}).get("screen_on_periods", []) if not periods: if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "(暂无时间段 — 按 A 新增)") except curses.error: @@ -266,6 +274,7 @@ class TuiApp: else: # 表头 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, " # │ 开始 │ 结束 │", curses.A_UNDERLINE) except curses.error: @@ -274,6 +283,7 @@ class TuiApp: for i, p in enumerate(periods): if y >= h: break + self._clear_line(y) sel = "▸" if i == self.selected else " " line = f"{sel}{i+1:>2} │ {p['start']} │ {p['end']} " try: @@ -286,6 +296,7 @@ class TuiApp: y += 1 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "[A]新增 [E]编辑 [D]删除 ↑↓选择", curses.A_DIM) except curses.error: @@ -298,6 +309,7 @@ class TuiApp: if self.status: dev = self.status.get("device") or "未选择" + self._clear_line(y) try: self.stdscr.addstr(y, 4, f"当前设备: {dev}") except curses.error: @@ -308,6 +320,7 @@ class TuiApp: for d in self.devices: if y >= h: break + self._clear_line(y) icon = "✓" if d["status"] == "device" else "✗" pair = 3 if d["status"] == "device" else 2 try: @@ -318,6 +331,7 @@ class TuiApp: y += 1 else: if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "(无设备 — 请连接平板)") except curses.error: @@ -325,6 +339,7 @@ class TuiApp: y += 1 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "[S]选择设备 [R]刷新", curses.A_DIM) except curses.error: @@ -348,6 +363,7 @@ class TuiApp: wf = s.get("wakefulness", "?") if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "屏幕: ") self.stdscr.addstr(f"{sc}", curses.color_pair(sc_color) | curses.A_BOLD) @@ -356,6 +372,7 @@ class TuiApp: pass y += 1 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "时段内: ") self.stdscr.addstr(f"{wp}", curses.color_pair(wp_color) | curses.A_BOLD) @@ -364,6 +381,7 @@ class TuiApp: pass y += 1 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "ADB: ") self.stdscr.addstr(f"{ar}", curses.color_pair(ar_color) | curses.A_BOLD) @@ -372,6 +390,7 @@ class TuiApp: y += 1 else: if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "(未连接守护进程 — 按 T 重连)") except curses.error: @@ -398,6 +417,7 @@ class TuiApp: bar_y += 1 if bar_y >= h: return + self._clear_line(bar_y) msg = self.message if msg and time.time() - self.msg_time < 5: pass # show message below separator From 1cde7a5d81e93bc21a3b8fa4606e8ee87af2ff91 Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 10 Jul 2026 09:39:45 +0800 Subject: [PATCH 4/5] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=97=A0=E7=BA=BF=20ADB?= =?UTF-8?q?=20=E8=BF=9E=E6=8E=A5=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 扩展配置：添加 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) --- padsleep.py | 190 +++++++++++++++++++++++++++++++++++++++++++++--- padsleep_tui.py | 66 ++++++++++++++++- 2 files changed, 243 insertions(+), 13 deletions(-) diff --git a/padsleep.py b/padsleep.py index 217b989..bf25556 100644 --- a/padsleep.py +++ b/padsleep.py @@ -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() diff --git a/padsleep_tui.py b/padsleep_tui.py index 1e508f1..56c5bd0 100644 --- a/padsleep_tui.py +++ b/padsleep_tui.py @@ -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 强制息屏", From 3dc30db3d6c4abadc84e698f63ee43ac25aa751d Mon Sep 17 00:00:00 2001 From: kushidou Date: Fri, 10 Jul 2026 10:52:06 +0800 Subject: [PATCH 5/5] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20README=20=E6=96=87?= =?UTF-8?q?=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 完善功能特性说明 - 添加安装和使用指南 - 添加配置字段详细说明 - 添加无线 ADB 连接说明 - 添加 TUI 快捷键表格 - 添加文件结构说明 --- README.md | 165 +++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 152 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 288223b..1283f03 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,166 @@ # padsleep_adb -基于树莓派，通过adb控制安卓平板亮屏和息屏 +基于树莓派，通过 ADB 控制安卓平板自动亮屏和息屏的守护程序。 -## 项目原理 +## 功能特性 -当触发设定的条件时，通过adb命令，模拟锁屏按键，控制pad自动亮屏和息屏。实现ha终端的省电和延长屏幕寿命。 +- **定时控制**：根据配置的时间段自动亮屏/息屏 +- **超时保护**：屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏 +- **无线 ADB**：支持通过 TCP/IP 连接安卓设备（无线调试） +- **TUI 管理**：提供终端界面工具，方便查看状态和修改配置 +- **IPC 通信**：通过 Unix Socket 与 TUI 工具通信 +- **自动重连**：无线连接断开后自动尝试重新连接 -## 开发阶段 +## 系统要求 -第一阶段： +- 树莓派（或其他 Linux 设备） +- Python 3.10+ +- ADB (Android Debug Bridge) +- 安卓平板（已开启 USB 调试或无线调试） -根据时间，定制自动亮屏和自动息屏的时间；当检测到亮屏时间大于5分钟且不在设定的亮屏时间段内，则自动息屏。 +## 安装 -第二阶段： +### 1. 安装系统依赖 -通过GPIO安装人体传感器，做到人来屏亮，人走屏灭。 +```bash +# Debian/Ubuntu +sudo apt update +sudo apt install adb python3-curses -## 开发细节 +# 树莓派 OS +sudo apt install adb +``` -- 在windows上开发，在树莓派上执行 +### 2. 安装 Python 依赖 -- 需要安装adb命令行 +```bash +cd padsleep_adb +pip install -r requirements.txt +``` -- 启动后，能自动以服务方式守护运行 +依赖列表： +- `adbutils>=2.12.0` — Python ADB 库 -- 首次启动时，需要处理adb授权有关问题 \ No newline at end of file +## 使用方法 + +### 启动守护程序 + +```bash +python padsleep.py +``` + +守护程序会自动： +1. 加载配置文件 `config.json` +2. 检测 ADB 设备 +3. 启动 IPC 服务（`/tmp/padsleep.sock`） +4. 进入主循环，定期检查屏幕状态 + +### 启动 TUI 管理工具 + +```bash +# 新终端窗口 +python padsleep_tui.py +``` + +TUI 快捷键： + +| 按键 | 功能 | +|------|------| +| `↑/k` `↓/j` | 选择亮屏时间段 | +| `A` | 新增亮屏时间段 | +| `E` | 编辑选中时间段 | +| `D` | 删除选中时间段 | +| `S` | 选择 ADB 设备 | +| `W` | 无线连接设备 | +| `R` | 刷新状态 | +| `L` | 重载守护进程配置 | +| `F` | 强制息屏 | +| `N` | 强制亮屏 | +| `T` | 重新连接守护进程 | +| `H` | 显示帮助 | +| `Q` | 退出 | + +## 配置说明 + +配置文件 `config.json` 示例： + +```json +{ + "screen_on_periods": [ + {"start": "08:00", "end": "22:00"} + ], + "check_interval_seconds": 30, + "auto_sleep_delay_minutes": 5, + "adb_device_serial": "", + "adb_wireless_host": "", + "adb_wireless_port": 5555, + "adb_wireless_auto_connect": false +} +``` + +### 字段说明 + +| 字段 | 类型 | 默认值 | 说明 | +|------|------|--------|------| +| `screen_on_periods` | array | `[{"start":"08:00","end":"22:00"}]` | 亮屏时间段列表，支持跨天（如 22:00~08:00） | +| `check_interval_seconds` | int | `30` | 检查间隔（秒） | +| `auto_sleep_delay_minutes` | int | `5` | 非亮屏时段屏幕亮起多久后自动息屏（分钟） | +| `adb_device_serial` | string | `""` | 指定 USB 设备 serial，空表示自动选择第一个 | +| `adb_wireless_host` | string | `""` | 无线设备 IP 地址，空表示不使用无线连接 | +| `adb_wireless_port` | int | `5555` | 无线设备端口 | +| `adb_wireless_auto_connect` | bool | `false` | 启动时是否自动连接无线设备 | + +## 无线 ADB 连接 + +### 方式一：配置文件预设 + +在 `config.json` 中设置： + +```json +{ + "adb_wireless_host": "192.168.1.100", + "adb_wireless_port": 5555, + "adb_wireless_auto_connect": true +} +``` + +### 方式二：TUI 手动连接 + +1. 启动 TUI 工具 +2. 按 `W` 键 +3. 输入 IP 地址和端口 +4. 确认连接 + +### 开启安卓设备无线调试 + +1. 连接 USB，执行 `adb tcpip 5555` +2. 或在设备上：设置 → 开发者选项 → 无线调试 → 开启 + +### 连接优先级 + +程序采用 **USB 优先** 策略： +1. 首先尝试 USB 连接 +2. 如果 USB 不可用且配置了无线，尝试无线连接 +3. 无线连接断开后自动重连（最多 3 次） + +## 文件结构 + +``` +padsleep_adb/ +├── padsleep.py # 守护程序主程序 +├── padsleep_tui.py # TUI 管理工具 +├── config.json # 配置文件（首次运行自动生成） +├── requirements.txt # Python 依赖 +└── README.md # 本文档 +``` + +## 开发说明 + +- 在 Windows 上开发，在树莓派上运行 +- 使用 `adbutils` 库与 ADB Server 通信 +- IPC 使用 Unix Socket + JSON 行协议 +- TUI 使用 Python `curses` 库 + +## 许可证 + +MIT License