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/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 diff --git a/padsleep.py b/padsleep.py index 77d1550..f06a0af 100644 --- a/padsleep.py +++ b/padsleep.py @@ -16,14 +16,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 + # ── 常量 ──────────────────────────────────────────────────────── # 注意: .resolve().parent 确保即使通过 symlink 调用也能正确找到真实目录 @@ -39,7 +39,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, } # ── 日志 ──────────────────────────────────────────────────────── @@ -256,74 +258,187 @@ class TimeChecker: # ═══════════════════════════════════════════════════════════════════ -# AdbManager — ADB 操作封装 +# AdbManager — ADB 操作封装（使用 adbutils 库） # ═══════════════════════════════════════════════════════════════════ class AdbManager: - """封装所有 ADB CLI 操作，所有异常都被捕获，不会导致进程退出。""" + """封装所有 ADB 操作，使用 adbutils 库实现持久连接。""" def __init__(self, config_getter, audit_logger: AuditLogger | None = None): self._config_getter = config_getter self._audit = audit_logger + 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._current_serial: str | None = None # 自动检测到的设备串号 + self._wireless_addr: str = "" # 当前无线连接地址 (IP:PORT) + self._wireless_retries: int = 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", "") or self._current_serial or "" - - cmd = [adb_path] - # "devices" / "--version" 命令不加 -s - if serial and args and args[0] not in ("devices", "--version"): - 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 + + # ── 无线连接管理 ── + + @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: - """检查 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", "connection_type": "usb"/"wireless"}]。""" + try: + client = self._ensure_client() + devices = client.device_list() + 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 [] - 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。""" - 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: @@ -334,53 +449,86 @@ class AdbManager: self._last_adb_check = now config = self._config_getter() - target = config.get("adb_device_serial", "") + target_serial = config.get("adb_device_serial", "") + wireless_host = config.get("adb_wireless_host", "") - devices = self.list_devices() - for d in devices: - if d["status"] == "device": - if not target or d["serial"] == target: - self._ready = True - self._current_serial = d["serial"] - self._last_error = f"已就绪: {d['serial']}" + try: + client = self._ensure_client() + devices = client.device_list() + + # USB 连接优先 + for d in devices: + 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._current_serial = d.serial + 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 - # 诊断 - 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 设备" + 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: + self._last_error = str(e) + 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: @@ -394,69 +542,82 @@ class AdbManager: def screen_off(self) -> bool: """发送息屏命令，等待 2 秒后验证，失败时重试一次。""" log.info("执行息屏操作") - if self._audit: - self._audit.log("息屏", "发出息屏指令") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) - if rc != 0: - log.error("息屏命令失败: %s", err) + try: + device = self._ensure_device() + if device is None: + log.error("息屏失败: 无可用设备") + return False if self._audit: - self._audit.log("息屏失败", err) - return False + self._audit.log("息屏", "发出息屏指令") + device.shell("input keyevent 26") - # 等待并验证 - time.sleep(2) - is_on, _ = self.get_screen_state() - if is_on: - log.warning("息屏命令已发出但屏幕未熄灭，重试一次") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) + # 等待并验证 time.sleep(2) is_on, _ = self.get_screen_state() if is_on: - log.error("息屏重试后仍未熄灭") - if self._audit: - self._audit.log("息屏异常", "屏幕状态未变化") - # 不更新 _screen_on_since - return False + log.warning("息屏命令已发出但屏幕未熄灭，重试一次") + device.shell("input keyevent 26") + time.sleep(2) + is_on, _ = self.get_screen_state() + if is_on: + log.error("息屏重试后仍未熄灭") + self._last_error = "息屏验证失败" + if self._audit: + self._audit.log("息屏异常", "屏幕状态未变化") + return False - self._screen_on_since = None - log.info("息屏成功") - if self._audit: - self._audit.log("息屏成功", "屏幕已熄灭") - return True + self._screen_on_since = None + log.info("息屏成功") + if self._audit: + self._audit.log("息屏成功", "屏幕已熄灭") + return True + except Exception as e: + self._last_error = str(e) + log.error("息屏失败: %s", e) + if self._audit: + self._audit.log("息屏失败", str(e)) + return False def screen_on(self) -> bool: """发送亮屏命令，等待 2 秒后验证，失败时重试一次。""" log.info("执行亮屏操作") - if self._audit: - self._audit.log("亮屏", "发出亮屏指令") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) - if rc != 0: - log.error("亮屏命令失败: %s", err) + try: + device = self._ensure_device() + if device is None: + log.error("亮屏失败: 无可用设备") + return False if self._audit: - self._audit.log("亮屏失败", err) - return False + self._audit.log("亮屏", "发出亮屏指令") + device.shell("input keyevent 26") - # 等待并验证 - time.sleep(2) - is_on, _ = self.get_screen_state() - if not is_on: - log.warning("亮屏命令已发出但屏幕未点亮，重试一次") - rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) + # 等待并验证 time.sleep(2) is_on, _ = self.get_screen_state() if not is_on: - log.error("亮屏重试后仍未点亮") - if self._audit: - self._audit.log("亮屏异常", "屏幕状态未变化") - return False + log.warning("亮屏命令已发出但屏幕未点亮，重试一次") + device.shell("input keyevent 26") + time.sleep(2) + is_on, _ = self.get_screen_state() + if not is_on: + log.error("亮屏重试后仍未点亮") + self._last_error = "亮屏验证失败" + if self._audit: + self._audit.log("亮屏异常", "屏幕状态未变化") + return False - self._screen_on_since = time.time() - # 解锁屏幕 - self._exec(["shell", "input", "keyevent", "82"]) - log.info("亮屏成功") - if self._audit: - self._audit.log("亮屏成功", "屏幕已点亮") - return True + self._screen_on_since = time.time() + # 解锁屏幕 + device.shell("input keyevent 82") + log.info("亮屏成功") + if self._audit: + self._audit.log("亮屏成功", "屏幕已点亮") + return True + except Exception as e: + self._last_error = str(e) + log.error("亮屏失败: %s", e) + if self._audit: + self._audit.log("亮屏失败", str(e)) + return False # ═══════════════════════════════════════════════════════════════════ @@ -576,6 +737,9 @@ class PadSleepApp: status["check_interval"] = cfg.get("check_interval_seconds", 30) status["config_path"] = str(self.config_mgr.path) status["log_path"] = str(LOG_PATH) + 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": @@ -621,6 +785,25 @@ class PadSleepApp: n = request.get("lines", 50) return {"ok": True, "data": {"log": self.audit.read_tail(n)}} + 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}"} # ── 主循环 ── @@ -643,6 +826,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 bd22d8b..042fd7c 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() @@ -198,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() @@ -228,6 +230,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 配置工具 " @@ -262,6 +272,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: @@ -270,6 +281,7 @@ class TuiApp: else: # 表头 if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, " # │ 开始 │ 结束 │", curses.A_UNDERLINE) except curses.error: @@ -278,6 +290,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: @@ -290,6 +303,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: @@ -303,38 +317,59 @@ class TuiApp: if self.status: serial_cfg = self.status.get("device_serial", "") detected = self.status.get("device", "") + conn_type = self.status.get("connection_type", "none") + conn_icon = "USB" if conn_type == "usb" else "WiFi" if conn_type == "wireless" else "---" + if serial_cfg: dev_str = f"{serial_cfg} (已指定)" elif detected and detected != "无": dev_str = f"{detected} (自动检测)" - - if detected and detected != "无": - try: - self.stdscr.addstr(y, 4, f"当前设备: {dev_str or detected}") - self.stdscr.addstr(y, 4 + len(f"当前设备: {dev_str or detected}"), " ✓", curses.color_pair(3)) - except curses.error: - pass else: - try: + dev_str = detected or "未连接" + + has_device = conn_type != "none" or (detected and detected != "无") + self._clear_line(y) + try: + label = f"当前设备: {dev_str} [{conn_icon}]" + if has_device: + self.stdscr.addstr(y, 4, label) + self.stdscr.addstr(y, 4 + len(label), " ✓", curses.color_pair(3)) + else: self.stdscr.addstr(y, 4, "当前设备: 未连接", curses.color_pair(2)) + except curses.error: + pass + y += 1 + 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 + 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 else: if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "(无设备 — 请连接平板)", curses.color_pair(2)) except curses.error: @@ -342,8 +377,9 @@ class TuiApp: y += 1 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 @@ -365,6 +401,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) @@ -373,6 +410,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) @@ -381,9 +419,13 @@ 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) + conn_type = s.get("connection_type", "none") + if conn_type != "none": + self.stdscr.addstr(f" ({conn_type})") except curses.error: pass y += 1 @@ -394,6 +436,7 @@ class TuiApp: pass else: if y < h: + self._clear_line(y) try: self.stdscr.addstr(y, 4, "(未连接守护进程 — 按 T 重连)") except curses.error: @@ -420,12 +463,13 @@ 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 else: if self.connected: - msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 V日志 H帮助" + msg = "已连接 | ↑↓/jk 选择 A新增 E编辑 D删除 S选设备 W无线 V日志 H帮助" else: msg = "未检测到守护进程 | T 尝试连接" try: @@ -605,6 +649,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: @@ -678,6 +759,7 @@ class TuiApp: " E 编辑选中时间段", " D 删除选中时间段", " S 选择 ADB 设备", + " W 无线连接设备", " R 刷新状态", " L 重载守护进程配置", " V 查看审计日志", 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 #