From 4d0db4c3900bd1dfa8e573fbc7545b163b0e3641 Mon Sep 17 00:00:00 2001 From: kushidou Date: Thu, 9 Jul 2026 23:13:48 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9F=BA=E6=9C=AC=E5=8A=9F=E8=83=BD=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=E5=92=8C=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.json | 11 + padsleep.py | 589 ++++++++++++++++++++++++++++++++++++++++++ padsleep.service | 16 ++ padsleep.sh | 0 padsleep_tui.py | 659 +++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 7 + 6 files changed, 1282 insertions(+) create mode 100644 config.json create mode 100644 padsleep.py create mode 100644 padsleep.service delete mode 100644 padsleep.sh create mode 100644 padsleep_tui.py create mode 100644 requirements.txt diff --git a/config.json b/config.json new file mode 100644 index 0000000..766112a --- /dev/null +++ b/config.json @@ -0,0 +1,11 @@ +{ + "screen_on_periods": [ + { "start": "06:40", "end": "7:30" }, + { "start": "19:00", "end": "19:05" }, + { "start": "23:00", "end": "23:05" } + ], + "check_interval_seconds": 30, + "auto_sleep_delay_minutes": 1, + "adb_device_serial": "", + "adb_path": "adb" +} diff --git a/padsleep.py b/padsleep.py new file mode 100644 index 0000000..f496b58 --- /dev/null +++ b/padsleep.py @@ -0,0 +1,589 @@ +#!/usr/bin/env python3 +"""padsleep_adb 守护程序 + +基于树莓派，通过 ADB 控制安卓平板自动亮屏和息屏。 + +阶段一功能： +- 根据配置的亮屏时间段自动亮屏/息屏 +- 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏 +- 通过 Unix Socket 提供 IPC 接口，供 TUI 工具管理 +""" + +import json +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 + +# ── 常量 ──────────────────────────────────────────────────────── + +SCRIPT_DIR = Path(__file__).parent.resolve() +CONFIG_PATH = SCRIPT_DIR / "config.json" +SOCKET_PATH = "/tmp/padsleep.sock" + +DEFAULT_CONFIG = { + "screen_on_periods": [ + {"start": "08:00", "end": "22:00"} + ], + "check_interval_seconds": 30, + "auto_sleep_delay_minutes": 5, + "adb_device_serial": "", + "adb_path": "adb", +} + +# ── 日志 ──────────────────────────────────────────────────────── + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +log = logging.getLogger("padsleep") + + +# ═══════════════════════════════════════════════════════════════════ +# ConfigManager — 配置加载/保存/监控 +# ═══════════════════════════════════════════════════════════════════ + +class ConfigManager: + """管理 JSON 配置文件的加载、保存和变更检测。""" + + def __init__(self, path: Path = CONFIG_PATH): + self.path = path + self.config = dict(DEFAULT_CONFIG) + self._mtime: float = 0 + self._lock = threading.Lock() + + def load(self) -> bool: + """从 JSON 文件加载配置，失败时使用默认值并不中断运行。""" + try: + with open(self.path, "r") as f: + data = json.load(f) + if "screen_on_periods" not in data: + raise ValueError("缺少 screen_on_periods") + for p in data["screen_on_periods"]: + if "start" not in p or "end" not in p: + raise ValueError("时间段缺少 start 或 end") + with self._lock: + self.config.update(data) + self._mtime = os.path.getmtime(self.path) + log.info("配置已加载: %s", self.path) + return True + except FileNotFoundError: + log.warning("配置文件不存在，写入默认配置: %s", self.path) + self.save() + return True + except (json.JSONDecodeError, ValueError, OSError) as e: + log.error("配置加载失败: %s，使用默认配置", e) + return True # 不崩溃，用默认配置继续运行 + + def save(self) -> bool: + """将当前配置写入 JSON 文件。""" + with self._lock: + data = dict(self.config) + try: + with open(self.path, "w") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + self._mtime = os.path.getmtime(self.path) + log.info("配置已保存") + return True + except OSError as e: + log.error("配置保存失败: %s", e) + return False + + def get(self) -> dict: + """线程安全地获取当前配置副本。""" + with self._lock: + return dict(self.config) + + def update(self, new_config: dict) -> bool: + """合并新配置并写入文件。""" + with self._lock: + self.config.update(new_config) + self.save() + return True + + def has_changed(self) -> bool: + """检查配置文件是否被外部修改。""" + try: + return os.path.getmtime(self.path) != self._mtime + except OSError: + return False + + def reload_if_changed(self): + """检测到外部修改则自动重载。""" + if self.has_changed(): + log.info("配置文件已变更，重新加载") + self.load() + + +# ═══════════════════════════════════════════════════════════════════ +# TimeChecker — 时间判断工具 +# ═══════════════════════════════════════════════════════════════════ + +class TimeChecker: + """亮屏时间段判断，支持跨天（如 22:00 ~ 08:00）。""" + + @staticmethod + def _to_minutes(tstr: str) -> int: + h, m = tstr.strip().split(":") + return int(h) * 60 + int(m) + + @staticmethod + def now_minutes() -> int: + now = datetime.now() + return now.hour * 60 + now.minute + + @staticmethod + def is_within_period(period: dict, now_m: int | None = None) -> bool: + """判断 now_m 是否在单个时间段内（支持跨天）。""" + if now_m is None: + now_m = TimeChecker.now_minutes() + start = TimeChecker._to_minutes(period["start"]) + end = TimeChecker._to_minutes(period["end"]) + + if end > start: + return start <= now_m < end + elif end == start: + return True # 全天 + else: + return now_m >= start or now_m < end # 跨天 + + @staticmethod + def is_within_any_period(periods: list, now_m: int | None = None) -> bool: + """判断当前时间是否在任意一个亮屏时间段内。""" + if not periods: + return False + if now_m is None: + now_m = TimeChecker.now_minutes() + return any(TimeChecker.is_within_period(p, now_m) for p in periods) + + +# ═══════════════════════════════════════════════════════════════════ +# AdbManager — ADB 操作封装 +# ═══════════════════════════════════════════════════════════════════ + +class AdbManager: + """封装所有 ADB CLI 操作，所有异常都被捕获，不会导致进程退出。""" + + def __init__(self, config_getter): + self._config_getter = config_getter + self._ready = False + self._last_error = "" + self._screen_on_since: float | None = None + self._last_known_screen_on = False + self._last_adb_check = 0 # 限频用 + + # ── 底层命令执行 ── + + def _exec(self, args: list, timeout: int = 10) -> tuple: + """执行 ADB 命令，返回 (returncode, stdout, stderr)。永不抛出。""" + 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) + + 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 命令超时" + except Exception as e: + self._ready = False + self._last_error = str(e) + return -1, "", self._last_error + + # ── 设备管理 ── + + def check_available(self) -> bool: + """检查 ADB 是否可执行。""" + rc, _, _ = self._exec(["--version"], timeout=5) + return rc == 0 + + def list_devices(self) -> list[dict]: + """返回 [{"serial": "...", "status": "..."}]。""" + rc, out, _ = self._exec(["devices"]) + if rc != 0: + 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"] + return None + + 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", "") + + 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 + + # 诊断 + 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 + 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" + + 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 + + # 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: + """返回屏幕已持续亮起的秒数，灭屏时返回 0。""" + if self._screen_on_since is not None: + return time.time() - self._screen_on_since + return 0.0 + + def screen_off(self) -> bool: + """发送息屏命令。""" + log.info("执行息屏操作") + rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) + if rc == 0: + self._screen_on_since = None + return True + log.error("息屏失败: %s", err) + return False + + def screen_on(self) -> bool: + """发送亮屏并解锁命令。""" + log.info("执行亮屏操作") + rc, _, err = self._exec(["shell", "input", "keyevent", "26"]) + if rc == 0: + self._screen_on_since = time.time() + self._exec(["shell", "input", "keyevent", "82"]) # 解锁（MENU） + return True + log.error("亮屏失败: %s", err) + return False + + +# ═══════════════════════════════════════════════════════════════════ +# IPC 服务端 — Unix Socket JSON 行协议 +# ═══════════════════════════════════════════════════════════════════ + +class IpcHandler(StreamRequestHandler): + """每个客户端连接由独立线程处理。""" + + def handle(self): + try: + raw = self.rfile.readline() + if not raw: + return + req = json.loads(raw.decode("utf-8").strip()) + resp = self.server.app.handle_ipc(req) + self.wfile.write((json.dumps(resp) + "\n").encode("utf-8")) + except json.JSONDecodeError: + self.wfile.write('{"ok":false,"error":"JSON 解析失败"}\n'.encode("utf-8")) + except Exception as e: + self.wfile.write(json.dumps({"ok": False, "error": str(e)}).encode() + b"\n") + + +class IpcServer: + """管理 Unix Socket 生命周期。""" + + def __init__(self, app): + self.app = app + self.server: ThreadingUnixStreamServer | None = None + self._thread: threading.Thread | None = None + + def start(self): + try: + os.unlink(SOCKET_PATH) + except OSError: + pass + + self.server = ThreadingUnixStreamServer(SOCKET_PATH, IpcHandler) + self.server.app = self.app + self.server.socket.settimeout(1.0) + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + log.info("IPC 服务已启动: %s", SOCKET_PATH) + + def _serve(self): + while self.app.running: + try: + self.server.handle_request() + except socket.timeout: + continue + except OSError: + break + + def cleanup(self): + try: + os.unlink(SOCKET_PATH) + except OSError: + pass + + def stop(self): + if self.server: + self.server.shutdown() + self.cleanup() + + +# ═══════════════════════════════════════════════════════════════════ +# PadSleepApp — 主应用 +# ═══════════════════════════════════════════════════════════════════ + +class PadSleepApp: + """守护程序核心：加载配置 → 主循环检查屏幕状态 → IPC 通信。""" + + def __init__(self): + self.running = True + self.config_mgr = ConfigManager() + self.adb = AdbManager(self._get_config) + self.ipc = IpcServer(self) + self._lock = threading.Lock() + self._start_time = time.time() + self._status = { + "screen_on": False, + "wakefulness": "unknown", + "within_period": False, + "adb_ready": False, + "device": None, + "uptime": 0, + "start_time": self._start_time, + } + + # ── 辅助 ── + + def _get_config(self) -> dict: + return self.config_mgr.get() + + def _update_status(self, **kwargs): + with self._lock: + self._status.update(kwargs) + + def get_status(self) -> dict: + with self._lock: + s = dict(self._status) + s["uptime"] = int(time.time() - self._start_time) + return s + + # ── IPC 命令处理 ── + + def handle_ipc(self, request: dict) -> dict: + cmd = request.get("cmd", "") + log.debug("IPC 请求: %s", cmd) + + if cmd == "status": + status = self.get_status() + cfg = self.config_mgr.get() + status["periods_count"] = len(cfg.get("screen_on_periods", [])) + status["check_interval"] = cfg.get("check_interval_seconds", 30) + return {"ok": True, "data": status} + + if cmd == "get_config": + return {"ok": True, "data": self.config_mgr.get()} + + if cmd == "update_config": + new = request.get("config", {}) + if "screen_on_periods" in new: + for p in new["screen_on_periods"]: + if "start" not in p or "end" not in p: + return {"ok": False, "error": "时间段格式错误"} + self.config_mgr.update(new) + cfg = self.config_mgr.get() + # 重设串号以使 is_ready 立即用新串号检查 + self.config_mgr.config["adb_device_serial"] = cfg.get("adb_device_serial", "") + return {"ok": True, "data": {"reloaded": True}} + + if cmd == "reload": + self.config_mgr.load() + return {"ok": True, "data": {"reloaded": True}} + + if cmd == "list_devices": + return {"ok": True, "data": {"devices": self.adb.list_devices()}} + + if cmd == "select_device": + serial = request.get("serial", "") + self.config_mgr.update({"adb_device_serial": serial}) + return {"ok": True, "data": {"selected": serial}} + + if cmd == "shutdown": + log.info("收到 IPC 关闭命令") + self.running = False + return {"ok": True} + + if cmd == "screen_off": + ok = self.adb.screen_off() + return {"ok": ok} + + if cmd == "screen_on": + ok = self.adb.screen_on() + return {"ok": ok} + + return {"ok": False, "error": f"未知命令: {cmd}"} + + # ── 主循环 ── + + def run(self): + log.info("=" * 50) + log.info("padsleep_adb 守护程序启动") + log.info("=" * 50) + + # 加载配置 + self.config_mgr.load() + + # 检测 ADB + if not self.adb.check_available(): + log.warning("ADB 未就绪，请安装: sudo apt install adb") + log.warning("安装后将自动检测到 ADB，无需重启本程序") + else: + log.info("ADB 可执行文件已就绪") + + # 启动 IPC + self.ipc.start() + + # 注册信号 + def sighandler(sig, _frame): + log.info("收到信号 %s，正在退出...", sig) + self.running = False + + signal.signal(signal.SIGTERM, sighandler) + signal.signal(signal.SIGINT, sighandler) + + # 主循环 + while self.running: + try: + self._tick() + except Exception as e: + log.error("主循环异常: %s", e, exc_info=True) + + # 检测配置文件外部变更 + self.config_mgr.reload_if_changed() + + # 分段睡眠以快速响应退出信号 + interval = self.config_mgr.get().get("check_interval_seconds", 30) + for _ in range(max(1, interval)): + if not self.running: + break + time.sleep(1) + + # 清理 + self.ipc.cleanup() + log.info("padsleep_adb 守护程序已退出") + + def _tick(self): + """单次检查逻辑。""" + config = self.config_mgr.get() + + # ── 检查 ADB 就绪 ── + if not self.adb.is_ready(): + self._update_status(adb_ready=False, device=None) + now = time.time() + if not hasattr(self, "_last_adb_warn") or now - self._last_adb_warn > 60: + log.warning("ADB 未就绪: %s", self.adb._last_error) + self._last_adb_warn = now + return + + # ── 获取屏幕状态 ── + screen_on, wakefulness = self.adb.get_screen_state() + periods = config.get("screen_on_periods", []) + within_period = TimeChecker.is_within_any_period(periods) + + device = self.adb.get_ready_device() + self._update_status( + adb_ready=True, + device=device, + screen_on=screen_on, + wakefulness=wakefulness, + within_period=within_period, + ) + + # ── 决策逻辑 ── + delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60 + + if not within_period and screen_on: + # 非亮屏时段但屏幕亮着 → 超时后息屏 + on_duration = self.adb.get_screen_on_duration() + if on_duration >= delay_sec: + log.info("不在亮屏时段（已亮 %.0fs）→ 自动息屏", on_duration) + self.adb.screen_off() + else: + remaining = delay_sec - on_duration + if remaining <= 10: + log.info("不在亮屏时段，%.0f 秒后将息屏", remaining) + + elif within_period and not screen_on: + # 在亮屏时段但屏幕灭着 → 亮屏 + log.info("在亮屏时段内 → 自动亮屏") + self.adb.screen_on() + + +# ═══════════════════════════════════════════════════════════════════ +# 入口 +# ═══════════════════════════════════════════════════════════════════ + +def main(): + app = PadSleepApp() + try: + app.run() + except KeyboardInterrupt: + log.info("用户中断") + finally: + app.ipc.cleanup() + + +if __name__ == "__main__": + main() diff --git a/padsleep.service b/padsleep.service new file mode 100644 index 0000000..5693c96 --- /dev/null +++ b/padsleep.service @@ -0,0 +1,16 @@ +[Unit] +Description=padsleep_adb — 基于 ADB 的安卓平板亮屏/息屏控制 +After=network.target + +[Service] +Type=simple +User=kushidou +WorkingDirectory=/home/kushidou/apps/padsleep_adb +ExecStart=/usr/bin/python3 /home/kushidou/apps/padsleep_adb/padsleep.py +Restart=on-failure +RestartSec=10 +StandardOutput=journal +StandardError=journal + +[Install] +WantedBy=multi-user.target diff --git a/padsleep.sh b/padsleep.sh deleted file mode 100644 index e69de29..0000000 diff --git a/padsleep_tui.py b/padsleep_tui.py new file mode 100644 index 0000000..8e4f117 --- /dev/null +++ b/padsleep_tui.py @@ -0,0 +1,659 @@ +#!/usr/bin/env python3 +"""padsleep_tui — padsleep_adb 的 TUI 配置工具 + +通过 Unix Socket 与 padsleep.py 守护进程通信，提供： + - 查看/新增/编辑/删除亮屏时间段 + - 刷新 ADB 设备列表和选择设备 + - 查看守护进程运行状态 + - 重载配置、手动息屏/亮屏 + +依赖：Python 标准库（curses，需系统安装 python3-curses 或无额外操作） +""" + +import curses +import json +import os +import socket +import sys +import time +from pathlib import Path + +# ── 常量 ──────────────────────────────────────────────────────── + +SOCKET_PATH = "/tmp/padsleep.sock" +SCRIPT_DIR = Path(__file__).parent.resolve() +CONFIG_PATH = SCRIPT_DIR / "config.json" + +# ═══════════════════════════════════════════════════════════════════ +# IpcClient — 与守护进程通信 +# ═══════════════════════════════════════════════════════════════════ + +class IpcClient: + """Unix Socket JSON 行协议客户端。""" + + def __init__(self, socket_path=SOCKET_PATH): + self.socket_path = socket_path + self.sock: socket.socket | None = None + self.connected = False + + def connect(self) -> bool: + try: + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(3) + self.sock.connect(self.socket_path) + self.connected = True + return True + except (socket.error, OSError): + self.connected = False + self.sock = None + return False + + def disconnect(self): + if self.sock: + try: + self.sock.close() + except Exception: + pass + self.sock = None + self.connected = False + + def send(self, cmd: str, **kwargs) -> dict: + if not self.sock: + if not self.connect(): + return {"ok": False, "error": "未连接守护进程"} + req = {"cmd": cmd, **kwargs} + try: + self.sock.sendall((json.dumps(req) + "\n").encode("utf-8")) + resp = self.sock.recv(65536) + return json.loads(resp.decode("utf-8").strip()) + except (socket.error, json.JSONDecodeError) as e: + self.connected = False + return {"ok": False, "error": str(e)} + + +# ═══════════════════════════════════════════════════════════════════ +# TuiApp — curses 界面 +# ═══════════════════════════════════════════════════════════════════ + +class TuiApp: + """TUI 主应用，使用 curses 实现。""" + + def __init__(self, stdscr): + self.stdscr = stdscr + self.client = IpcClient() + # 数据缓存 + self.config: dict | None = None + self.status: dict | None = None + self.devices: list[dict] = [] + self.connected = False + self.selected = 0 # 列表选中索引 + self.message = "" + self.msg_time = 0.0 + self._last_refresh = 0.0 + + # ── 通信 ── + + def try_connect(self) -> bool: + self.connected = self.client.connect() + if self.connected: + self.refresh() + return self.connected + + def refresh(self): + """从守护进程拉取全量数据。""" + if not self.connected: + return + now = time.time() + if now - self._last_refresh < 0.5: + return + self._last_refresh = now + + r = self.client.send("get_config") + if r.get("ok"): + self.config = r["data"] + + r = self.client.send("status") + if r.get("ok"): + self.status = r["data"] + + r = self.client.send("list_devices") + if r.get("ok"): + self.devices = r["data"].get("devices", []) + + # ── 界面提示 ── + + def msg(self, text: str): + self.message = text + self.msg_time = time.time() + + # ── 主循环 ── + + def run(self): + self.try_connect() + if not self.connected: + self.msg("未连接守护进程 — 请先运行 padsleep.py") + + while True: + self.height, self.width = self.stdscr.getmaxyx() + self.stdscr.clear() + self._draw() + self.stdscr.refresh() + + curses.halfdelay(10) # 1 秒超时，用于自动刷新 + key = self.stdscr.getch() + + # ── 按键处理 ── + if key in (ord("q"), ord("Q")): + break + + elif key == ord("t") or key == ord("T"): + if self.try_connect(): + self.msg("已连接守护进程") + else: + self.msg("连接失败") + + elif key == ord("r") or key == ord("R"): + self.refresh() + self.msg("已刷新") + + elif key == ord("l") or key == ord("L"): + r = self.client.send("reload") + if r.get("ok"): + self.msg("配置已重载") + else: + self.msg(f"重载失败: {r.get('error', '')}") + self.refresh() + + elif key == ord("f") or key == ord("F"): + r = self.client.send("screen_off") + self.msg("已发送息屏指令" if r.get("ok") else f"息屏失败: {r.get('error','')}") + self.refresh() + + elif key == ord("n") or key == ord("N"): + r = self.client.send("screen_on") + self.msg("已发送亮屏指令" if r.get("ok") else f"亮屏失败: {r.get('error','')}") + self.refresh() + + elif key in (curses.KEY_UP, ord("k"), ord("K")): + if self.selected > 0: + self.selected -= 1 + + elif key in (curses.KEY_DOWN, ord("j"), ord("J")): + max_idx = len((self.config or {}).get("screen_on_periods", [])) - 1 + if self.selected < max_idx: + self.selected += 1 + + elif key == ord("a") or key == ord("A"): + if self.config is not None: + self._dlg_add_period() + + elif key == ord("e") or key == ord("E"): + if self.config is not None: + self._dlg_edit_period(self.selected) + + elif key == ord("d") or key == ord("D"): + if self.config is not None: + self._dlg_delete_period(self.selected) + + elif key == ord("s") or key == ord("S"): + self._dlg_select_device() + + elif key == ord("h") or key == ord("H"): + self._show_help() + + # 定时自动刷新 + if key == -1 and self.connected: + self.refresh() + + # ══════════════════════════════════════════════════════════════ + # 绘制 + # ══════════════════════════════════════════════════════════════ + + def _draw(self): + h, w = self.height, self.width + if h < 12 or w < 50: + self._center_text(0, "窗口太小，请放大终端 (至少 50×12)") + return + + self._draw_titlebar() + y = 2 + y = self._draw_periods_panel(y) + y += 1 + y = self._draw_adb_panel(y) + y += 1 + y = self._draw_status_panel(y) + self._draw_bottombar() + + def _draw_titlebar(self): + h, w = self.height, self.width + title = " padsleep 配置工具 " + st = "已连接" if self.connected else "未连接" + try: + self.stdscr.addstr(0, 0, "─" * w) + self.stdscr.addstr(0, 2, title, curses.A_BOLD) + sx = w - len(st) - 4 + if sx > len(title) + 4: + pair = 3 if self.connected else 2 + self.stdscr.addstr(0, sx, f"[{st}]", curses.color_pair(pair) | curses.A_BOLD) + except curses.error: + pass + + def _draw_separator(self, y: int, label: str = ""): + if y >= self.height: + return y + 1 + try: + if label: + self.stdscr.addstr(y, 0, "─" * self.width, curses.A_DIM) + self.stdscr.addstr(y, 2, label, curses.A_BOLD) + else: + self.stdscr.addstr(y, 0, "─" * self.width, curses.A_DIM) + except curses.error: + pass + return y + 1 + + def _draw_periods_panel(self, y: int) -> int: + h = self.height + y = self._draw_separator(y, "[1] 亮屏时间段") + + periods = (self.config or {}).get("screen_on_periods", []) + if not periods: + if y < h: + try: + self.stdscr.addstr(y, 4, "(暂无时间段 — 按 A 新增)") + except curses.error: + pass + y += 1 + else: + # 表头 + if y < h: + try: + self.stdscr.addstr(y, 4, " # │ 开始 │ 结束 │", curses.A_UNDERLINE) + except curses.error: + pass + y += 1 + for i, p in enumerate(periods): + if y >= h: + break + sel = "▸" if i == self.selected else " " + line = f"{sel}{i+1:>2} │ {p['start']} │ {p['end']} " + try: + if i == self.selected: + self.stdscr.addstr(y, 4, line, curses.A_REVERSE) + else: + self.stdscr.addstr(y, 4, line) + except curses.error: + pass + y += 1 + + if y < h: + try: + self.stdscr.addstr(y, 4, "[A]新增 [E]编辑 [D]删除 ↑↓选择", curses.A_DIM) + except curses.error: + pass + return y + 1 + + def _draw_adb_panel(self, y: int) -> int: + h = self.height + y = self._draw_separator(y, "[2] ADB 配置") + + if self.status: + dev = self.status.get("device") or "未选择" + try: + self.stdscr.addstr(y, 4, f"当前设备: {dev}") + except curses.error: + pass + y += 1 + + if self.devices: + for d in self.devices: + if y >= h: + break + icon = "✓" if d["status"] == "device" else "✗" + 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']})") + except curses.error: + pass + y += 1 + else: + if y < h: + try: + self.stdscr.addstr(y, 4, "(无设备 — 请连接平板)") + except curses.error: + pass + y += 1 + + if y < h: + try: + self.stdscr.addstr(y, 4, "[S]选择设备 [R]刷新", curses.A_DIM) + except curses.error: + pass + return y + 1 + + def _draw_status_panel(self, y: int) -> int: + h = self.height + y = self._draw_separator(y, "[3] 守护状态") + + if self.status: + s = self.status + sc = "亮" if s.get("screen_on") else "灭" + sc_color = 3 if s.get("screen_on") else 2 + wp = "是" if s.get("within_period") else "否" + wp_color = 3 if s.get("within_period") else 2 + upt = s.get("uptime", 0) + hh, mm = upt // 3600, (upt % 3600) // 60 + ar = "就绪" if s.get("adb_ready") else "未就绪" + ar_color = 3 if s.get("adb_ready") else 2 + wf = s.get("wakefulness", "?") + + if y < h: + try: + self.stdscr.addstr(y, 4, "屏幕: ") + self.stdscr.addstr(f"{sc}", curses.color_pair(sc_color) | curses.A_BOLD) + self.stdscr.addstr(f" ({wf})") + except curses.error: + pass + y += 1 + if y < h: + try: + self.stdscr.addstr(y, 4, "时段内: ") + self.stdscr.addstr(f"{wp}", curses.color_pair(wp_color) | curses.A_BOLD) + self.stdscr.addstr(f" │ 运行: {hh}h{mm:02d}m") + except curses.error: + pass + y += 1 + if y < h: + try: + self.stdscr.addstr(y, 4, "ADB: ") + self.stdscr.addstr(f"{ar}", curses.color_pair(ar_color) | curses.A_BOLD) + except curses.error: + pass + y += 1 + else: + if y < h: + try: + self.stdscr.addstr(y, 4, "(未连接守护进程 — 按 T 重连)") + except curses.error: + pass + y += 1 + + if y < h: + try: + self.stdscr.addstr(y, 4, "[L]重载 [F]息屏 [N]亮屏 [T]重连 [H]帮助 [Q]退出", + curses.A_DIM) + except curses.error: + pass + return y + 1 + + def _draw_bottombar(self): + h, w = self.height, self.width + bar_y = h - 2 + if bar_y <= 0: + return + try: + self.stdscr.addstr(bar_y, 0, "─" * w, curses.A_DIM) + except curses.error: + pass + bar_y += 1 + if bar_y >= h: + return + 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选设备 H帮助" + else: + msg = "未检测到守护进程 | T 尝试连接" + try: + self.stdscr.addstr(bar_y, 2, msg[: w - 4]) + except curses.error: + pass + + # ── 辅助 ── + + def _center_text(self, y: int, text: str): + try: + self.stdscr.addstr(y, max(0, (self.width - len(text)) // 2), text) + except curses.error: + pass + + # ══════════════════════════════════════════════════════════════ + # 弹窗 + # ══════════════════════════════════════════════════════════════ + + def _dlg_input(self, prompt: str, default: str = "") -> str | None: + """输入弹窗，返回字符串或 None（取消）。""" + h, w = self.height, self.width + bh, bw = 5, 52 + by = max(0, (h - bh) // 2) + bx = max(0, (w - bw) // 2) + + win = curses.newwin(bh, bw, by, bx) + win.box() + disp = prompt[: bw - 4] + win.addstr(1, 2, disp) + win.addstr(3, 2, "─" * (bw - 4), curses.A_DIM) + + # 在框内显示默认值 + if default: + win.addstr(3, 2, default, curses.A_DIM) + + win.refresh() + curses.echo() + curses.curs_set(1) + + raw = b"" + try: + raw = win.getstr(3, 2, 10) + except Exception: + pass + + curses.noecho() + curses.curs_set(0) + + result = raw.decode("utf-8") if isinstance(raw, bytes) else str(raw) if raw else "" + return result.strip() or None + + def _dlg_confirm(self, message: str) -> bool: + h, w = self.height, self.width + bh, bw = 5, 60 + by = max(0, (h - bh) // 2) + bx = max(0, (w - bw) // 2) + + win = curses.newwin(bh, bw, by, bx) + win.box() + win.addstr(1, 2, message[: bw - 4]) + win.addstr(3, 2, "[Y]确认 [N]取消") + win.refresh() + + while True: + k = win.getch() + if k in (ord("y"), ord("Y")): + return True + if k in (ord("n"), ord("N"), 27): + return False + + def _dlg_menu(self, title: str, options: list[str]) -> int | None: + """菜单选择弹窗，返回索引或 None。""" + h, w = self.height, self.width + bh = min(len(options) + 4, h - 4) + bw = 54 + by = max(0, (h - bh) // 2) + bx = max(0, (w - bw) // 2) + + win = curses.newwin(bh, bw, by, bx) + cur = 0 + while True: + win.box() + win.addstr(0, 2, title[: bw - 4]) + for i, opt in enumerate(options): + if i >= bh - 3: + break + d = opt[: bw - 6] + if i == cur: + win.addstr(i + 2, 2, f"▸ {d}", curses.A_REVERSE) + else: + win.addstr(i + 2, 2, f" {d}") + win.refresh() + + k = win.getch() + if k in (curses.KEY_UP, ord("k")): + cur = max(0, cur - 1) + elif k in (curses.KEY_DOWN, ord("j")): + cur = min(len(options) - 1, cur + 1) + elif k in (ord("\n"), ord(" ")): + return cur + elif k == 27: + return None + + # ══════════════════════════════════════════════════════════════ + # 对话框业务逻辑 + # ══════════════════════════════════════════════════════════════ + + def _get_periods(self) -> list: + return (self.config or {}).get("screen_on_periods", []) + + def _update_periods(self, periods: list): + r = self.client.send("update_config", config={"screen_on_periods": periods}) + if r.get("ok"): + self.msg(f"已更新亮屏时间段") + self.refresh() + else: + self.msg(f"更新失败: {r.get('error', '')}") + + def _dlg_add_period(self): + start = self._dlg_input("开始时间 (HH:MM):", "08:00") + if not start: + return + end = self._dlg_input("结束时间 (HH:MM):", "22:00") + if not end: + return + if not self._validate_time(start) or not self._validate_time(end): + self.msg("时间格式错误，请使用 HH:MM (00-23:00-59)") + return + periods = self._get_periods() + periods.append({"start": start, "end": end}) + self._update_periods(periods) + + def _dlg_edit_period(self, idx: int): + periods = self._get_periods() + if idx < 0 or idx >= len(periods): + return + p = periods[idx] + new_start = self._dlg_input("开始时间 (HH:MM):", p["start"]) + if not new_start: + return + new_end = self._dlg_input("结束时间 (HH:MM):", p["end"]) + if not new_end: + return + if not self._validate_time(new_start) or not self._validate_time(new_end): + self.msg("时间格式错误") + return + periods[idx] = {"start": new_start, "end": new_end} + self._update_periods(periods) + + def _dlg_delete_period(self, idx: int): + periods = self._get_periods() + if idx < 0 or idx >= len(periods): + return + p = periods[idx] + if not self._dlg_confirm(f"删除时间段 {p['start']}—{p['end']}？"): + return + periods.pop(idx) + if self.selected >= len(periods) and self.selected > 0: + self.selected -= 1 + self._update_periods(periods) + + def _dlg_select_device(self): + if not self.devices: + self.msg("未检测到设备，请先连接平板") + return + opts = [f"{d['serial']} ({d['status']})" for d in self.devices] + opts.append("(自动选择首个设备)") + idx = self._dlg_menu("选择 ADB 设备", opts) + if idx is None: + return + serial = self.devices[idx]["serial"] if idx < len(self.devices) else "" + r = self.client.send("select_device", serial=serial) + if r.get("ok"): + self.msg(f"设备已选择: {serial or '自动'}") + self.refresh() + else: + self.msg(f"选择失败: {r.get('error', '')}") + + @staticmethod + def _validate_time(tstr: str) -> bool: + try: + h, m = tstr.strip().split(":") + h, m = int(h), int(m) + return 0 <= h <= 23 and 0 <= m <= 59 + except (ValueError, IndexError): + return False + + # ── 帮助 ── + + def _show_help(self): + h, w = self.height, self.width + bh, bw = 20, 60 + by = max(1, (h - bh) // 2) + bx = max(1, (w - bw) // 2) + + win = curses.newwin(bh, bw, by, bx) + win.box() + + lines = [ + " padsleep 配置工具 — 快捷键", + "", + " ↑/k / ↓/j 选择亮屏时间段", + " A 新增亮屏时间段", + " E 编辑选中时间段", + " D 删除选中时间段", + " S 选择 ADB 设备", + " R 刷新状态", + " L 重载守护进程配置", + " F 强制息屏", + " N 强制亮屏", + " T 重新连接守护进程", + " H 显示此帮助", + " Q 退出", + "", + " 提示: 先运行 padsleep.py 启动守护进程", + ] + for i, line in enumerate(lines): + if i + 2 < bh - 2: + try: + win.addstr(i + 1, 2, line[: bw - 4]) + except curses.error: + pass + try: + win.addstr(bh - 2, 2, "按任意键返回") + except curses.error: + pass + win.refresh() + win.getch() + + +# ═══════════════════════════════════════════════════════════════════ +# 入口 +# ═══════════════════════════════════════════════════════════════════ + +def main(stdscr): + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_WHITE, -1) # default + curses.init_pair(2, curses.COLOR_RED, -1) # 未就绪 + curses.init_pair(3, curses.COLOR_GREEN, -1) # 已就绪 + curses.init_pair(4, curses.COLOR_YELLOW, -1) # 警告 + curses.curs_set(0) + + app = TuiApp(stdscr) + app.run() + + +if __name__ == "__main__": + try: + curses.wrapper(main) + except KeyboardInterrupt: + print("\n已退出") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8ba4822 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +# padsleep_adb 依赖 +# 本程序仅使用 Python 标准库，无需额外安装第三方包。 +# 系统依赖：adb (Android Debug Bridge) +# 安装: sudo apt-get install adb +# +# TUI 使用 curses（Python 标准库）， +# 某些系统需要: sudo apt-get install python3-curses