#!/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__).resolve().parent CONFIG_PATH = SCRIPT_DIR / "config.json" # ═══════════════════════════════════════════════════════════════════ # IpcClient — 与守护进程通信 # ═══════════════════════════════════════════════════════════════════ class IpcClient: """Unix Socket JSON 行协议客户端（每次 send 新建连接）。""" 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: """发送命令，每次新建连接（守护进程每请求关闭连接）。""" self.disconnect() 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._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("w") or key == ord("W"): self._dlg_connect_wireless() elif key == ord("h") or key == ord("H"): self._show_help() elif key == ord("v") or key == ord("V"): if self.connected: self._dlg_view_log() # 定时自动刷新 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 _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 配置工具 " 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: self._clear_line(y) try: self.stdscr.addstr(y, 4, "(暂无时间段 — 按 A 新增)") except curses.error: pass y += 1 else: # 表头 if y < h: self._clear_line(y) 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 self._clear_line(y) 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: self._clear_line(y) 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: 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} (自动检测)" else: 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 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" [{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: pass y += 1 if y < h: self._clear_line(y) try: self.stdscr.addstr(y, 4, "[S]选择设备 [W]无线连接 [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: self._clear_line(y) 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: self._clear_line(y) 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: 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 try: self.stdscr.addstr(y, 4, "配置: ") self.stdscr.addstr(f"{s.get('config_path', '?')}", curses.A_DIM) except curses.error: pass # ── 亮屏事件状态 ── active_reason = s.get("active_reason") if active_reason: y += 1 reason_labels = { "scheduled": "定时点亮", "sensor": "传感器触发", "manual": "手动控制", "external": "外部亮屏", } label = reason_labels.get(active_reason, active_reason) if y < h: self._clear_line(y) try: self.stdscr.addstr(y, 4, "亮屏事件: ") self.stdscr.addstr(f"{label}", curses.color_pair(3) | curses.A_BOLD) except curses.error: pass # ── GPIO 状态 ── gpio_avail = s.get("gpio_available", False) if gpio_avail: y += 1 gpio_trig = s.get("gpio_triggered", False) gpio_cnt = s.get("gpio_countdown", 0) trig_text = f"触发中 ⏱ {gpio_cnt}s" if gpio_trig else "等待中" trig_color = 3 if gpio_trig else 2 gpio_pin = s.get("gpio_pin", "?") if y < h: self._clear_line(y) try: self.stdscr.addstr(y, 4, "GPIO: ") self.stdscr.addstr(f"BCM {gpio_pin} ", curses.A_DIM) self.stdscr.addstr(f"{trig_text}", curses.color_pair(trig_color) | curses.A_BOLD) except curses.error: pass else: if y < h: self._clear_line(y) 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]亮屏 [V]日志 [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 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选设备 W无线 V日志 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', '')}") 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: 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 _dlg_view_log(self): """查看审计日志""" r = self.client.send("log_tail", lines=100) if not r.get("ok"): self.msg(f"获取日志失败: {r.get('error', '')}") return log_text = r["data"].get("log", "") h, w = self.height, self.width bh, bw = min(h - 2, 30), min(w - 4, 80) by = max(1, (h - bh) // 2) bx = max(1, (w - bw) // 2) win = curses.newwin(bh, bw, by, bx) lines = log_text.split("\n") scroll = max(0, len(lines) - (bh - 3)) while True: win.box() win.addstr(0, 2, " 审计日志 (最近100行) ", curses.A_BOLD) visible = lines[scroll:scroll + bh - 3] for i, line in enumerate(visible): if i >= bh - 3: break try: win.addstr(i + 1, 2, line[:bw - 4]) except curses.error: pass footer = f" 行 {scroll+1}-{scroll+len(visible)}/{len(lines)} ↑↓滚动 ESC返回 " try: win.addstr(bh - 1, 2, footer[:bw - 4], curses.A_DIM) except curses.error: pass win.refresh() k = win.getch() if k == 27 or k == ord("q"): # ESC/q break elif k in (curses.KEY_UP, ord("k")) and scroll > 0: scroll -= 1 elif k in (curses.KEY_DOWN, ord("j")) and scroll < len(lines) - (bh - 3): scroll += 1 # ── 帮助 ── 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 设备", " W 无线连接设备", " R 刷新状态", " L 重载守护进程配置", " V 查看审计日志", " 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已退出")