kushidou/padsleep_adb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构屏幕控制引擎 + 新增 Web 监控面板

Browse Source 
- ScreenManager: 基于优先级(9/5/0)的统一屏幕控制引擎，替代 _tick 中的
  散乱 if-else 逻辑
- 四个亮屏事件: scheduled(p9), sensor(p9), manual(p5), external(p0)
- GPIO 回调: 电平变化立即触发亮屏（异步线程，不阻塞轮询）
- _fast_check: 每秒轻量评估，提高息屏超时精度（±1s 替代 ±30s）
- padsleep_web: 通过 IPC 读取守护进程状态的实时 Web 面板
  （GPIO 电平/屏幕状态/ADB 连接，端口 31400）
- padsleep-web.service: 关联 padsleep.service 的自启动服务
- install.sh: 修复 venv 初始化目录上下文，使用 requirements.txt
- IpcClient: 修复 socket 复用导致交替失败的问题
- padsleep_tui: 适配 ScreenManager 状态显示

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
kushidou
2026-07-20 00:15:07 +08:00
co-authored by Claude Fable 5
parent 9940db451f
commit e6fbe550bb
8 changed files with 1134 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
padsleep.py
+271 -75
View File
@@ -282,7 +282,7 @@ class GPIOEventDetector:
detector.stop()
"""
POLL_INTERVAL = 0.05 # 轮询间隔 50ms
POLL_INTERVAL = 1.0 # 轮询间隔 1s
def __init__(self, pin: int, callback=None):
self.pin = pin
@@ -290,6 +290,7 @@ class GPIOEventDetector:
self.last_event_time: float = 0.0 # 0 表示从未触发
self.last_value: int | None = None
self.available = False # 硬件是否可用
self.current_value: int = 0 # 当前引脚电平（供外部读取）
self._running = False
self._thread: threading.Thread | None = None
self._reader = None
@@ -337,6 +338,7 @@ class GPIOEventDetector:
while self._running:
try:
val = self._gpio.input(self.pin)
self.current_value = val # 对外暴露当前电平
if val != self.last_value:
self.last_value = val
self.last_event_time = time.time()
@@ -796,6 +798,130 @@ class IpcServer:
self.cleanup()
# ═══════════════════════════════════════════════════════════════════
# ScreenManager — 统一屏幕状态管理（flag 组 + 优先级决策）
# ═══════════════════════════════════════════════════════════════════
class ScreenManager:
"""统一管理四个亮屏事件，基于优先级决定息屏时机。
四个事件（优先级从高到低）:
1. 定时点亮 (scheduled) — p9: 在亮屏时间段内 → 亮屏
2. 传感器点亮 (sensor) — p9: GPIO 检测到活动 → 亮屏 + 倒计时
3. IPC 控制 (manual) — p5: 收到 IPC screen_on → 亮屏
4. 检测到外部亮屏 (external) — p0: 检测到物理按键亮屏
决策规则:
- 每个事件的 flag 记录触发时间戳
- 每次 tick 检查最高优先级事件的息屏条件
- 同优先级需 ALL 满足息屏条件才息屏
- 息屏时清除所有 flag
"""
REASON_SCHEDULED = "scheduled" # 定时点亮, p9
REASON_SENSOR = "sensor" # 传感器点亮, p9
REASON_MANUAL = "manual" # IPC 控制, p5
REASON_EXTERNAL = "external" # 检测到外部亮屏, p0
PRIORITIES = {
REASON_SCHEDULED: 9,
REASON_SENSOR: 9,
REASON_MANUAL: 5,
REASON_EXTERNAL: 0,
}
def __init__(self):
self._flags: dict[str, float] = {} # reason → trigger timestamp
self._manual_released = False # manual 是否已要求释放
# ── 接口 ──
def trigger(self, reason: str):
"""设置亮屏事件 flag（记录当前时间戳）。多次调用刷新时间戳。"""
self._flags[reason] = time.time()
if reason == self.REASON_MANUAL:
self._manual_released = False
def release_manual(self):
"""标记手动息屏请求（收到 IPC screen_off）。"""
self._manual_released = True
def clear(self):
"""清除所有 flag（息屏时调用）。"""
self._flags.clear()
self._manual_released = False
# ── 核心决策 ──
def evaluate(self, *, actual_screen_on: bool, within_period: bool,
gpio_elapsed: float, gpio_countdown: int,
screen_on_duration: float, auto_sleep_delay: float) -> dict:
"""综合各事件 flag 判断应执行的操作。
返回:
{"action": "on", "reason": "..."} → 需要亮屏
{"action": "off", "reason": "..."} → 需要息屏（调用者负责 clear()）
{"action": "none", "reason": "..."} → 保持当前状态
"""
active = list(self._flags.keys())
if not active:
return {"action": "none", "reason": "no_flags"}
# 按优先级分组
by_pri: dict[int, list[str]] = {}
for r in active:
p = self.PRIORITIES.get(r, 0)
by_pri.setdefault(p, []).append(r)
max_pri = max(by_pri)
top = by_pri[max_pri]
# 检查最高优先级组：是否 ALL 满足息屏条件
all_ok = all(
self._can_off(r, within_period, gpio_elapsed, gpio_countdown,
screen_on_duration, auto_sleep_delay)
for r in top
)
if all_ok:
if actual_screen_on:
return {"action": "off", "reason": f"p{max_pri}_all_ok"}
return {"action": "none", "reason": "already_off"}
# 有事件仍要求亮屏
if not actual_screen_on:
return {"action": "on", "reason": f"p{max_pri}_wants_on"}
return {"action": "none", "reason": f"p{max_pri}_active"}
def _can_off(self, reason: str, within_period: bool,
gpio_elapsed: float, gpio_countdown: int,
screen_on_duration: float, auto_sleep_delay: float) -> bool:
"""判断单个事件是否满足息屏条件。"""
if reason == self.REASON_SCHEDULED:
return not within_period
if reason == self.REASON_SENSOR:
return gpio_elapsed >= gpio_countdown
if reason == self.REASON_MANUAL:
return self._manual_released or screen_on_duration >= auto_sleep_delay
if reason == self.REASON_EXTERNAL:
return screen_on_duration >= auto_sleep_delay
return True # 未知事件默认可息屏
# ── 状态查询 ──
def get_active_reason(self) -> str | None:
"""返回当前最高优先级的 active 事件名（用于显示）。"""
if not self._flags:
return None
return max(self._flags, key=lambda r: self.PRIORITIES.get(r, 0))
def get_status(self) -> dict:
return {
"active_reason": self.get_active_reason(),
"flags": dict(self._flags),
}
# ═══════════════════════════════════════════════════════════════════
# PadSleepApp — 主应用
# ═══════════════════════════════════════════════════════════════════
@@ -808,6 +934,7 @@ class PadSleepApp:
self.audit = AuditLogger()
self.config_mgr = ConfigManager()
self.adb = AdbManager(self._get_config, self.audit)
self.screen_mgr = ScreenManager()
self.gpio: GPIOEventDetector | None = None
self.ipc = IpcServer(self)
self._lock = threading.Lock()
@@ -826,6 +953,8 @@ class PadSleepApp:
"gpio_triggered": False,
"gpio_countdown": 0,
"gpio_pin": 17,
# ── ScreenManager 状态 ──
"active_reason": None,
}
# ── 辅助 ──
@@ -843,6 +972,32 @@ class PadSleepApp:
s["uptime"] = int(time.time() - self._start_time)
return s
# ── GPIO 回调（在 50ms 轮询线程中触发）──
def _on_gpio_event(self, value):
"""GPIO 电平变化回调：设置传感器 flag，异步处理亮屏。
运行在 GPIOEventDetector 的 50ms 轮询线程中，
必须尽快返回以避免阻塞电平检测。
"""
self.screen_mgr.trigger(ScreenManager.REASON_SENSOR)
# 独立线程处理亮屏（screen_on 有 2s ADB 验证延迟，不阻塞 GPIO 轮询）
threading.Thread(target=self._gpio_screen_on, daemon=True).start()
def _gpio_screen_on(self):
"""在独立线程中检查屏幕状态，灭屏则立即亮屏。"""
try:
screen_on, _ = self.adb.get_screen_state()
if not screen_on:
config = self.config_mgr.get()
gpio_pin = config.get("gpio_pin", 17)
gpio_countdown = config.get("gpio_countdown_seconds", 60)
log.info("GPIO 回调 → 亮屏")
self.audit.log("GPIO触屏", f"BCM {gpio_pin} 电平变化，{gpio_countdown}s 倒计时")
self.adb.screen_on()
except Exception as e:
log.error("GPIO 亮屏异常: %s", e)
# ── IPC 命令处理 ──
def handle_ipc(self, request: dict) -> dict:
@@ -859,6 +1014,9 @@ class PadSleepApp:
status["connection_type"] = self.adb.get_connection_type()
status["wireless_host"] = cfg.get("adb_wireless_host", "")
status["wireless_port"] = cfg.get("adb_wireless_port", 5555)
# 实时 GPIO 电平（由 GPIOEventDetector 50ms 轮询线程更新）
if self.gpio and self.gpio.available:
status["gpio_level"] = self.gpio.current_value
return {"ok": True, "data": status}
if cmd == "get_config":
@@ -893,10 +1051,12 @@ class PadSleepApp:
return {"ok": True}
if cmd == "screen_off":
self.screen_mgr.release_manual()
ok = self.adb.screen_off()
return {"ok": ok}
if cmd == "screen_on":
self.screen_mgr.trigger(ScreenManager.REASON_MANUAL)
ok = self.adb.screen_on()
return {"ok": ok}
@@ -960,7 +1120,7 @@ class PadSleepApp:
# ── 启动 GPIO 电平检测 ──
if config.get("gpio_enabled", True):
gpio_pin = config.get("gpio_pin", 17)
self.gpio = GPIOEventDetector(pin=gpio_pin)
self.gpio = GPIOEventDetector(pin=gpio_pin, callback=self._on_gpio_event)
self.gpio.start()
if self.gpio.available:
log.info("GPIO 电平检测已启用 (BCM %d, 倒计时 %ds)",
@@ -996,12 +1156,13 @@ class PadSleepApp:
# 检测配置文件外部变更
self.config_mgr.reload_if_changed()
# 分段睡眠以快速响应退出信号
# 分段睡眠，每秒做一次轻量决策检查（evaluate 纯内存运算）
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._fast_check()
# 清理
if self.gpio:
@@ -1023,49 +1184,82 @@ class PadSleepApp:
self._last_adb_warn = now
return prev_screen_on if prev_screen_on is not None else False
# ── 获取屏幕状态 ──
# ── 获取屏幕实际状态 ──
screen_on, wakefulness = self.adb.get_screen_state()
periods = config.get("screen_on_periods", [])
within_period = TimeChecker.is_within_any_period(periods)
auto_sleep_delay = config.get("auto_sleep_delay_minutes", 5) * 60
# ═══════════════════════════════════════════════════════════════
# GPIO 触发检测（人来亮屏 / 人走息屏）
#
# 由于购买的模块有问题，无法稳定输出电平，但是人来人走时会发生
# 短暂电平切换，因此采用这个折中的法子：检测到 GPIO 电平变化
# 就亮屏并重置 gpio_countdown_seconds 秒倒计时，倒计时结束
# 且无新变化则息屏（除非当前在亮屏时段）。
# ═══════════════════════════════════════════════════════════════
# ── 1. 设置定时 flag ──
if within_period:
self.screen_mgr.trigger(ScreenManager.REASON_SCHEDULED)
# ── 2. 设置传感器 flag ──
gpio_triggered = False
gpio_countdown_remain = 0
gpio_elapsed = float("inf")
gpio_countdown = config.get("gpio_countdown_seconds", 60)
if self.gpio and self.gpio.available:
gpio_pin = config.get("gpio_pin", 17)
countdown = config.get("gpio_countdown_seconds", 60)
elapsed = self.gpio.seconds_since_last_event
gpio_triggered = elapsed < countdown
gpio_countdown_remain = max(0, countdown - int(elapsed))
gpio_elapsed = self.gpio.seconds_since_last_event
gpio_triggered = gpio_elapsed < gpio_countdown
gpio_countdown_remain = max(0, gpio_countdown - int(gpio_elapsed))
if gpio_triggered:
# GPIO 刚刚触发过 → 确保屏幕亮起
if not screen_on:
log.info("GPIO 检测到电平变化 → 亮屏 (%ds 倒计时)", gpio_countdown_remain)
self.audit.log("GPIO触屏", f"BCM {gpio_pin} 电平变化，{countdown}s 倒计时")
self.adb.screen_on()
screen_on = True
else:
# 屏幕已亮，仅做调试日志
if gpio_countdown_remain >= countdown - 1:
log.info("GPIO 电平变化 → 重置倒计时 %ds", countdown)
else:
# GPIO 倒计时已结束且屏幕不在亮屏时段 → 息屏
if screen_on and not within_period and elapsed > countdown:
log.info("GPIO 倒计时结束（%.0fs 无变化）→ 息屏", elapsed)
self.audit.log("GPIO息屏", f"BCM {gpio_pin} 在 {countdown}s 内无电平变化")
self.adb.screen_off()
screen_on = False
self.screen_mgr.trigger(ScreenManager.REASON_SENSOR)
# ── 更新状态（含 GPIO 信息）──
# ── 3. 检测外部状态变化（用户按电源键）──
if prev_screen_on is not None and prev_screen_on != screen_on:
if screen_on:
# 屏幕被外部点亮——设置 external flag 让 ScreenManager 追踪
self.screen_mgr.trigger(ScreenManager.REASON_EXTERNAL)
log.info("检测到屏幕被点亮 (外部操作)")
self.audit.log("检测到屏幕被点亮", "外部操作或用户手动")
else:
log.info("检测到屏幕被熄灭 (外部操作)")
self.audit.log("检测到屏幕被熄灭", "外部操作或用户手动")
# ── 4. ScreenManager 统一决策 ──
decision = self.screen_mgr.evaluate(
actual_screen_on=screen_on,
within_period=within_period,
gpio_elapsed=gpio_elapsed,
gpio_countdown=gpio_countdown,
screen_on_duration=self.adb.get_screen_on_duration(),
auto_sleep_delay=auto_sleep_delay,
)
# ── 5. 执行决策 ──
if decision["action"] == "on" and not screen_on:
log.info("ScreenManager 决策 → 亮屏 (%s)", decision["reason"])
# 区分亮屏原因
if ScreenManager.REASON_SENSOR in self.screen_mgr._flags:
gpio_pin = config.get("gpio_pin", 17)
countdown = config.get("gpio_countdown_seconds", 60)
self.audit.log("GPIO触屏", f"BCM {gpio_pin} 电平变化，{countdown}s 倒计时")
elif ScreenManager.REASON_SCHEDULED in self.screen_mgr._flags:
period_desc = TimeChecker.nearest_period_description(periods)
self.audit.log("亮屏", f"在亮屏时间段 ({period_desc}) 内检测到屏幕熄灭，自动恢复")
elif ScreenManager.REASON_MANUAL in self.screen_mgr._flags:
self.audit.log("亮屏", "手动 IPC 控制")
self.adb.screen_on()
screen_on = True
elif decision["action"] == "off" and screen_on:
reason = decision["reason"]
log.info("ScreenManager 决策 → 息屏 (%s)", reason)
if "sensor" in str(self.screen_mgr._flags):
self.audit.log("GPIO息屏",
f"BCM {config.get('gpio_pin', 17)} 在 {gpio_countdown}s 内无电平变化")
elif reason == "p5_all_ok" or reason == "p0_all_ok":
period_desc = TimeChecker.nearest_period_description(periods)
self.audit.log("息屏",
f"不在亮屏时间段 ({period_desc})，超时 {config.get('auto_sleep_delay_minutes')} 分钟")
self.screen_mgr.clear()
self.adb.screen_off()
screen_on = False
# ── 6. 更新状态 ──
device = self.adb.get_ready_device()
gpio_level = self.gpio.current_value if self.gpio and self.gpio.available else -1
self._update_status(
adb_ready=True,
device=device or "无",
@@ -1075,49 +1269,51 @@ class PadSleepApp:
within_period=within_period,
gpio_triggered=gpio_triggered,
gpio_countdown=gpio_countdown_remain,
gpio_level=gpio_level,
**self.screen_mgr.get_status(),
)
# ── 检测外部屏幕状态变化 ──
if prev_screen_on is not None and prev_screen_on != screen_on:
if screen_on:
log.info("检测到屏幕被点亮 (外部操作)")
self.audit.log("检测到屏幕被点亮", "外部操作或用户手动")
else:
log.info("检测到屏幕被熄灭 (外部操作)")
self.audit.log("检测到屏幕被熄灭", "外部操作或用户手动")
# ── 原有决策逻辑（仅在非 GPIO 触发时有效） ──
delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60
period_desc = TimeChecker.nearest_period_description(periods)
if within_period and not screen_on:
# 在亮屏时段但屏幕灭着 → 立即亮屏（自动恢复）
log.info("在亮屏时间段内但屏幕已熄灭 → 立即亮屏")
self.audit.log("亮屏", f"在亮屏时间段 ({period_desc}) 内检测到屏幕熄灭，自动恢复")
self.adb.screen_on()
return True
if not within_period and screen_on and not gpio_triggered:
# 非亮屏时段、非 GPIO 触发、屏幕亮着 → 超时后息屏
on_duration = self.adb.get_screen_on_duration()
if on_duration >= delay_sec:
log.info("不在亮屏时段（已亮 %.0fs）→ 自动息屏", on_duration)
self.audit.log("息屏", f"不在亮屏时间段 ({period_desc})，超时 {config.get('auto_sleep_delay_minutes')} 分钟")
self.adb.screen_off()
return False
else:
remaining = delay_sec - on_duration
if remaining <= 10:
log.info("不在亮屏时段，%.0f 秒后将息屏", remaining)
elif within_period and screen_on:
# 在亮屏时段且屏幕已亮 — 一切正常
if prev_screen_on is False:
pass
return screen_on
def _fast_check(self):
"""每秒轻量决策（不查 ADB，纯内存 evaluate）。
仅在缓存状态显示屏幕亮、且时间条件接近超时时，
才做一次 ADB 验证并执行息屏，避免 30 秒 tick 的超时误差。
"""
config = self.config_mgr.get()
delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60
on_duration = self.adb.get_screen_on_duration() # 纯内存
if on_duration < delay_sec:
return # 远未超时，跳过
# 缓存状态显示屏幕灭 → 无需操作
if not self._status.get("screen_on", False):
return
within_period = TimeChecker.is_within_any_period(
config.get("screen_on_periods", []))
gpio_elapsed = self.gpio.seconds_since_last_event if self.gpio else 999
gpio_countdown = config.get("gpio_countdown_seconds", 60)
decision = self.screen_mgr.evaluate(
actual_screen_on=True,
within_period=within_period,
gpio_elapsed=gpio_elapsed,
gpio_countdown=gpio_countdown,
screen_on_duration=on_duration,
auto_sleep_delay=delay_sec,
)
if decision["action"] == "off":
# 验证屏幕是否真的亮着（一次 ADB 调用）
is_on, _ = self.adb.get_screen_state()
if is_on:
log.info("快速决策 → 息屏 (%s)", decision["reason"])
self.screen_mgr.clear()
self.adb.screen_off()
self._update_status(screen_on=False)
# ═══════════════════════════════════════════════════════════════════
# 入口