合并远程变更: adbutils 改造 + 无线 ADB 支持 + TUI 优化
合并内容: - 远程: adbutils 库替换 subprocess、无线 ADB 连接、TUI 闪烁修复 - 本地: 审计日志、命令执行后验证重试、安装脚本 - 合并后保留双方特性: adbutils + 审计日志 + 无线 ADB + 命令验证
This commit is contained in:
kushidou
+25
@@ -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
|
||||
|
||||
|
||||
@@ -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授权有关问题
|
||||
## 使用方法
|
||||
|
||||
### 启动守护程序
|
||||
|
||||
```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
|
||||
|
||||
+319
-124
@@ -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()
|
||||
|
||||
|
||||
+95
-13
@@ -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 查看审计日志",
|
||||
|
||||
+2
-1
@@ -1,5 +1,6 @@
|
||||
# padsleep_adb 依赖
|
||||
# 本程序仅使用 Python 标准库，无需额外安装第三方包。
|
||||
adbutils>=2.12.0
|
||||
|
||||
# 系统依赖：adb (Android Debug Bridge)
|
||||
# 安装: sudo apt-get install adb
|
||||
#
|
||||
|
||||
Reference in New Issue
Block a user