#!/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已退出")