基本功能实现和验证

This commit is contained in:
kushidou
2026-07-09 23:13:48 +08:00
parent 0d14a5de4e
commit 4d0db4c390
6 changed files with 1282 additions and 0 deletions
+659
View File
@@ -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已退出")