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 #