Compare commits

..
2 Commits
Author SHA1 Message Date
kushidouandClaude Fable 5 e6fbe550bb 重构屏幕控制引擎 + 新增 Web 监控面板
- 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>
2026-07-20 00:15:07 +08:00
kushidouandClaude Fable 5 9940db451f feat: 添加 GPIO 电平检测触发亮屏/息屏
新增 GPIOEventDetector 后台线程,以 50ms 间隔轮询 GPIO 引脚。
检测到电平变化时唤醒屏幕并启动 60 秒倒计时,到期无新变化则息屏。

- 添加 GPIOEventDetector 类,支持 RPi.GPIO 硬读取和 Mock 回退
- 集成到 PadSleepApp 主循环,GPIO 触发优先级高于定时亮屏时段
- TUI 状态面板新增 GPIO 行,显示引脚号和倒计时
- 支持通过配置文件调整引脚号(gpio_pin)和倒计时长度
- install.sh 修复:补装 rpi-lgpio 依赖

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 16:10:42 +08:00
9 changed files with 1349 additions and 77 deletions
+93 -5
View File
@@ -6,6 +6,7 @@
- **定时控制**:根据配置的时间段自动亮屏/息屏 - **定时控制**:根据配置的时间段自动亮屏/息屏
- **超时保护**:屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏 - **超时保护**:屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏
- **GPIO 触发**:(新增)检测 GPIO 电平变化 → 亮屏 + 60 秒倒计时 → 倒计时结束息屏
- **无线 ADB**:支持通过 TCP/IP 连接安卓设备(无线调试) - **无线 ADB**:支持通过 TCP/IP 连接安卓设备(无线调试)
- **TUI 管理**:提供终端界面工具,方便查看状态和修改配置 - **TUI 管理**:提供终端界面工具,方便查看状态和修改配置
- **IPC 通信**:通过 Unix Socket 与 TUI 工具通信 - **IPC 通信**:通过 Unix Socket 与 TUI 工具通信
@@ -40,28 +41,63 @@ pip install -r requirements.txt
依赖列表: 依赖列表:
- `adbutils>=2.12.0` — Python ADB 库 - `adbutils>=2.12.0` — Python ADB 库
- `rpi-lgpio>=0.6` — 树莓派 GPIO 控制(仅在树莓派上需要)
### 3. 快速安装
```bash
sudo ./install.sh
```
安装脚本会将程序部署到 `/opt/apps/padsleep`,创建命令行入口和 systemd 服务。
## 使用方法 ## 使用方法
### 启动守护程序 ### 启动守护程序
```bash ```bash
# 直接启动
python padsleep.py python padsleep.py
# 或使用安装后的命令行入口
padsleep
``` ```
守护程序会自动: 守护程序会自动:
1. 加载配置文件 `config.json` 1. 加载配置文件 `config.json`
2. 检测 ADB 设备 2. 检测 ADB 设备
3. 启动 IPC 服务(`/tmp/padsleep.sock` 3. 启动 GPIO 电平检测(默认 BCM 17
4. 进入主循环,定期检查屏幕状态 4. 启动 IPC 服务(`/tmp/padsleep.sock`
5. 进入主循环,定期检查屏幕状态
### 启动 TUI 管理工具 ### 启动 TUI 管理工具
```bash ```bash
# 新终端窗口 # 新终端窗口
python padsleep_tui.py python padsleep_tui.py
# 或使用安装后的命令行入口
padsleep-config
``` ```
### 启动 Web 监控面板
```bash
# 新终端窗口(需先启动守护进程)
python padsleep_web.py
# 或使用安装后的命令行入口
padsleep-web
```
然后在浏览器打开 `http://树莓派IP:5000` 查看仪表盘。
Web 面板功能:
- **GPIO 电平实时监控**:独立线程 50ms 轮询 GPIO 引脚,毫秒级变化展示
- **守护进程状态**:屏幕状态、亮屏事件、ADB 连接、传感器触发等
- **电平事件日志**:实时记录所有 GPIO 电平变化
- **历史波形**:最近 40 次电平变化可视化
TUI 快捷键: TUI 快捷键:
| 按键 | 功能 | | 按键 | 功能 |
@@ -94,7 +130,10 @@ TUI 快捷键:
"adb_device_serial": "", "adb_device_serial": "",
"adb_wireless_host": "", "adb_wireless_host": "",
"adb_wireless_port": 5555, "adb_wireless_port": 5555,
"adb_wireless_auto_connect": false "adb_wireless_auto_connect": false,
"gpio_enabled": true,
"gpio_pin": 17,
"gpio_countdown_seconds": 60
} }
``` ```
@@ -109,6 +148,48 @@ TUI 快捷键:
| `adb_wireless_host` | string | `""` | 无线设备 IP 地址,空表示不使用无线连接 | | `adb_wireless_host` | string | `""` | 无线设备 IP 地址,空表示不使用无线连接 |
| `adb_wireless_port` | int | `5555` | 无线设备端口 | | `adb_wireless_port` | int | `5555` | 无线设备端口 |
| `adb_wireless_auto_connect` | bool | `false` | 启动时是否自动连接无线设备 | | `adb_wireless_auto_connect` | bool | `false` | 启动时是否自动连接无线设备 |
| `gpio_enabled` | bool | `true` | 是否启用 GPIO 电平检测 |
| `gpio_pin` | int | `17` | GPIO 引脚号(BCM 编号),如 GPIO17 = 物理 pin 11 |
| `gpio_countdown_seconds` | int | `60` | GPIO 触发后倒计时秒数,到期无新变化则息屏 |
## GPIO 电平检测
> 由于购买的模块有问题,无法稳定输出电平,但是人来人走时会发生短暂电平切换,
> 因此采用这个折中的法子:检测到 GPIO 电平变化就亮屏并重置倒计时,
> 倒计时结束且无新变化则息屏。
### 工作原理
1. **后台线程** 以 50ms 间隔高频轮询 GPIO 引脚
2. **检测到任意电平变化**(上升或下降沿)→ 立即亮屏,重置倒计时
3. **倒计时期间再次触发** → 倒计时重置,保持亮屏
4. **倒计时到期无新变化** → 息屏(除非当前在定时亮屏时段内)
### 决策优先级
```
GPIO 触发(60s 倒计时) > 定时亮屏时段 > 超时息屏
```
### 接线参考
GPIO 引脚号(BCM 编号)对照:
| BCM 编号 | 物理引脚 | 功能 |
|----------|----------|------|
| GPIO17 | Pin 11 | 默认 GPIO 检测引脚 |
| GND | Pin 6 / 9 / 14 / ... | 接地 |
将传感器信号线接 GPIO17,GND 接树莓派 GND 即可。
### 日志记录
所有 GPIO 事件记录在 `padsleep.log` 中:
```
[2026-07-19 15:59:25] GPIO触屏 — BCM 17 电平变化,60s 倒计时
[2026-07-19 15:59:28] GPIO息屏 — BCM 17 在 60s 内无电平变化
```
## 无线 ADB 连接 ## 无线 ADB 连接
@@ -147,10 +228,15 @@ TUI 快捷键:
``` ```
padsleep_adb/ padsleep_adb/
├── padsleep.py # 守护程序主程序 ├── padsleep.py # 守护程序主程序(含 GPIO 检测)
├── padsleep_tui.py # TUI 管理工具 ├── padsleep_tui.py # TUI 管理工具
├── config.json # 配置文件(首次运行自动生成 ├── padsleep_web.py # Web 监控面板(Flask + SocketIO
├── templates/
│ └── index.html # Web 面板前端
├── config.json # 配置文件
├── requirements.txt # Python 依赖 ├── requirements.txt # Python 依赖
├── install.sh # 安装脚本
├── padsleep.service # systemd 服务文件
└── README.md # 本文档 └── README.md # 本文档
``` ```
@@ -158,8 +244,10 @@ padsleep_adb/
- 在 Windows 上开发,在树莓派上运行 - 在 Windows 上开发,在树莓派上运行
- 使用 `adbutils` 库与 ADB Server 通信 - 使用 `adbutils` 库与 ADB Server 通信
- GPIO 使用 `RPi.GPIO`(通过 `rpi-lgpio` 驱动 `lgpio` 内核模块)
- IPC 使用 Unix Socket + JSON 行协议 - IPC 使用 Unix Socket + JSON 行协议
- TUI 使用 Python `curses` - TUI 使用 Python `curses`
- 非树莓派环境 GPIO 自动降级为模拟模式(不影响主功能)
## 许可证 ## 许可证
+4 -1
View File
@@ -7,5 +7,8 @@
"check_interval_seconds": 30, "check_interval_seconds": 30,
"auto_sleep_delay_minutes": 1, "auto_sleep_delay_minutes": 1,
"adb_device_serial": "", "adb_device_serial": "",
"adb_path": "adb" "adb_path": "adb",
"gpio_enabled": true,
"gpio_pin": 17,
"gpio_countdown_seconds": 60
} }
+50 -21
View File
@@ -63,6 +63,8 @@ info "已创建目录 $INSTALL_DIR"
# ── 复制文件 ── # ── 复制文件 ──
cp "$SRC_DIR/padsleep.py" "$INSTALL_DIR/" cp "$SRC_DIR/padsleep.py" "$INSTALL_DIR/"
cp "$SRC_DIR/padsleep_tui.py" "$INSTALL_DIR/" cp "$SRC_DIR/padsleep_tui.py" "$INSTALL_DIR/"
cp "$SRC_DIR/padsleep_web.py" "$INSTALL_DIR/" 2>/dev/null || true
cp -r "$SRC_DIR/templates" "$INSTALL_DIR/" 2>/dev/null || true
cp "$SRC_DIR/requirements.txt" "$INSTALL_DIR/" 2>/dev/null || true cp "$SRC_DIR/requirements.txt" "$INSTALL_DIR/" 2>/dev/null || true
# 配置文件:目标不存在则复制,存在则保留 # 配置文件:目标不存在则复制,存在则保留
@@ -77,25 +79,34 @@ fi
touch "$INSTALL_DIR/padsleep.log" touch "$INSTALL_DIR/padsleep.log"
chmod 644 "$INSTALL_DIR/padsleep.log" chmod 644 "$INSTALL_DIR/padsleep.log"
# ── 创建 Python 虚拟环境 ── # ── 创建 Python 虚拟环境并安装依赖 ──
# 进入目标目录确保 venv 上下文正确
cd "$INSTALL_DIR"
if [ ! -d "$VENV_DIR" ]; then if [ ! -d "$VENV_DIR" ]; then
info "正在创建 Python 虚拟环境..." info "正在创建 Python 虚拟环境..."
python3 -m venv "$VENV_DIR" python3 -m venv .venv
info "虚拟环境已创建: $VENV_DIR" info "虚拟环境已创建: $VENV_DIR"
else else
info "虚拟环境已存在,跳过创建" info "虚拟环境已存在,跳过创建"
fi fi
# ── 安装 Python 依赖 ── info "正在安装 Python 依赖 (requirements.txt)..."
info "正在安装 Python 依赖..." if [ -f "$INSTALL_DIR/requirements.txt" ]; then
"$VENV_DIR/bin/pip" install --quiet adbutils 2>&1 || { .venv/bin/pip install --quiet -r requirements.txt 2>&1 || {
warn "pip 安装失败,尝试离线安装..." warn "pip 安装失败,尝试逐个安装..."
if [ -f "$SRC_DIR/.venv" ]; then .venv/bin/pip install --quiet adbutils rpi-lgpio flask flask-socketio 2>&1 || {
cp -r "$SRC_DIR/.venv" "$VENV_DIR" warn "pip 安装仍然失败,请手动执行: cd $INSTALL_DIR && .venv/bin/pip install -r requirements.txt"
fi
} }
}
else
.venv/bin/pip install --quiet adbutils rpi-lgpio flask flask-socketio 2>&1 || \
warn "pip 安装失败,请手动安装依赖"
fi
info "Python 依赖安装完成" info "Python 依赖安装完成"
cd "$SRC_DIR" # 回到源目录
# 设置执行权限 # 设置执行权限
chmod 755 "$INSTALL_DIR/padsleep.py" chmod 755 "$INSTALL_DIR/padsleep.py"
chmod 755 "$INSTALL_DIR/padsleep_tui.py" chmod 755 "$INSTALL_DIR/padsleep_tui.py"
@@ -121,22 +132,38 @@ WRAPPER
create_wrapper "padsleep.py" "$BIN_DIR/padsleep" create_wrapper "padsleep.py" "$BIN_DIR/padsleep"
create_wrapper "padsleep_tui.py" "$BIN_DIR/padsleep-config" create_wrapper "padsleep_tui.py" "$BIN_DIR/padsleep-config"
if [ -f "$INSTALL_DIR/padsleep_web.py" ]; then
create_wrapper "padsleep_web.py" "$BIN_DIR/padsleep-web"
fi
info "已创建命令入口:" info "已创建命令入口:"
info " $BIN_DIR/padsleep → 启动守护进程 ($PYTHON_BIN)" info " $BIN_DIR/padsleep → 启动守护进程 ($PYTHON_BIN)"
info " $BIN_DIR/padsleep-config → 启动 TUI 配置工具 ($PYTHON_BIN)" info " $BIN_DIR/padsleep-config → 启动 TUI 配置工具 ($PYTHON_BIN)"
info " $BIN_DIR/padsleep-web → 启动 Web 监控面板 ($PYTHON_BIN)"
# ── 可选:复制 systemd 服务 ── # ── 可选:复制 systemd 服务 ──
if [ -f "$SRC_DIR/padsleep.service" ]; then install_service() {
cp "$SRC_DIR/padsleep.service" "$SERVICE_DIR/padsleep.service" local src="$1"
sed -i "s|WorkingDirectory=.*|WorkingDirectory=$INSTALL_DIR|" "$SERVICE_DIR/padsleep.service" local dst="$2"
sed -i "s|ExecStart=.*|ExecStart=$PYTHON_BIN $INSTALL_DIR/padsleep.py|" "$SERVICE_DIR/padsleep.service" local script="$3"
sed -i "s|^User=.*|User=$RUN_USER|" "$SERVICE_DIR/padsleep.service" if [ ! -f "$src" ]; then
systemctl daemon-reload warn "服务文件不存在: $src,跳过"
info "已复制 systemd 服务文件: $SERVICE_DIR/padsleep.service" return
info "服务用户已设为: $RUN_USER"
info "可执行以下命令启用:"
info " sudo systemctl enable --now padsleep"
fi fi
cp "$src" "$dst"
sed -i "s|WorkingDirectory=.*|WorkingDirectory=$INSTALL_DIR|" "$dst"
sed -i "s|ExecStart=.*|ExecStart=$PYTHON_BIN $INSTALL_DIR/$script|" "$dst"
sed -i "s|^User=.*|User=$RUN_USER|" "$dst"
info "已安装 $dst"
}
install_service "$SRC_DIR/padsleep.service" "$SERVICE_DIR/padsleep.service" "padsleep.py"
install_service "$SRC_DIR/padsleep-web.service" "$SERVICE_DIR/padsleep-web.service" "padsleep_web.py"
systemctl daemon-reload
info "systemd 已重载"
echo ""
info "启用服务(自启动 + 立即启动):"
info " sudo systemctl enable --now padsleep padsleep-web"
# ── 检查 ADB ── # ── 检查 ADB ──
if command -v adb &>/dev/null; then if command -v adb &>/dev/null; then
@@ -152,8 +179,10 @@ info "安装完成!"
echo "" echo ""
info "使用方法:" info "使用方法:"
echo " 1. 启动守护进程: padsleep" echo " 1. 启动守护进程: padsleep"
echo " 2. 打开配置界面: padsleep-config" echo " 2. 打开 TUI 配置界面: padsleep-config"
echo " 3. 注册系统服务: sudo systemctl enable --now padsleep" echo " 3. 打开 Web 监控面板: padsleep-web"
echo " 4. 注册系统服务: sudo systemctl enable --now padsleep padsleep-web"
echo " 5. Web 面板地址: http://树莓派IP:31400"
echo "" echo ""
info "配置文件: $INSTALL_DIR/config.json" info "配置文件: $INSTALL_DIR/config.json"
info "日志文件: $INSTALL_DIR/padsleep.log" info "日志文件: $INSTALL_DIR/padsleep.log"
+18
View File
@@ -0,0 +1,18 @@
[Unit]
Description=padsleep Web Monitor — 实时监控面板
Documentation=https://github.com/kushidou/padsleep_adb
Requires=padsleep.service
After=padsleep.service network.target
[Service]
Type=simple
User=_INSTALL_USER_
WorkingDirectory=/opt/apps/padsleep
ExecStart=/opt/apps/padsleep/.venv/bin/python3 /opt/apps/padsleep/padsleep_web.py
Restart=on-failure
RestartSec=5
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
+390 -42
View File
@@ -42,6 +42,10 @@ DEFAULT_CONFIG = {
"adb_wireless_host": "", "adb_wireless_host": "",
"adb_wireless_port": 5555, "adb_wireless_port": 5555,
"adb_wireless_auto_connect": False, "adb_wireless_auto_connect": False,
# ── GPIO 电平触发(人来亮屏 / 人走息屏) ──
"gpio_enabled": True,
"gpio_pin": 17,
"gpio_countdown_seconds": 60,
} }
# ── 日志 ──────────────────────────────────────────────────────── # ── 日志 ────────────────────────────────────────────────────────
@@ -257,6 +261,96 @@ class TimeChecker:
return "; ".join(descs) return "; ".join(descs)
# ═══════════════════════════════════════════════════════════════════
# GPIOEventDetector — GPIO 电平变化检测(人来亮屏 / 人走息屏)
# ═══════════════════════════════════════════════════════════════════
#
# 由于购买的模块有问题,无法稳定输出电平,但是人来人走时会发生短暂电平切换,
# 因此采用这个折中的法子:检测到 GPIO 电平变化就亮屏并重置 60 秒倒计时,
# 倒计时结束且无新变化则息屏。
class GPIOEventDetector:
"""在后台线程高频轮询 GPIO 引脚,检测电平变化并记录时间戳。
用法:
detector = GPIOEventDetector(pin=17)
detector.start()
...
elapsed = detector.seconds_since_last_event # 距上次变化秒数
if elapsed < 60:
... # 有人/有活动
detector.stop()
"""
POLL_INTERVAL = 1.0 # 轮询间隔 1s
def __init__(self, pin: int, callback=None):
self.pin = pin
self.callback = callback # 可选:每条变化时触发
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
@property
def seconds_since_last_event(self) -> float:
"""距上次 GPIO 电平变化的秒数。从未触发时返回 inf。"""
if self.last_event_time == 0:
return float("inf")
return time.time() - self.last_event_time
def start(self):
"""启动后台轮询线程。"""
# ── 初始化 GPIO ──
try:
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(self.pin, GPIO.IN)
self._gpio = GPIO
self.available = True
log.info("[GPIO] BCM %d — 电平检测已启动", self.pin)
except (ImportError, RuntimeError, OSError) as e:
log.warning("[GPIO] 无法初始化 GPIO%d: %s — 电平检测已禁用", self.pin, e)
self.available = False
return
self.last_value = self._gpio.input(self.pin)
self.last_event_time = time.time() # 启动瞬间作为一次"事件",防止开机立刻息屏
self._running = True
self._thread = threading.Thread(target=self._poll, daemon=True, name="gpio-poll")
self._thread.start()
def stop(self):
"""停止轮询并清理。"""
self._running = False
if self._thread:
self._thread.join(timeout=1)
if hasattr(self, "_gpio") and self._gpio:
self._gpio.cleanup(self.pin)
self.available = False
log.info("[GPIO] BCM %d — 电平检测已停止", self.pin)
def _poll(self):
"""后台轮询:50ms 间隔检测电平变化。"""
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()
edge = "上升" if val else "下降"
log.debug("[GPIO] 电平变化: %s%d", edge, val)
if self.callback:
self.callback(val)
except Exception as e:
log.error("[GPIO] 读取异常: %s", e)
time.sleep(self.POLL_INTERVAL)
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
# AdbManager — ADB 操作封装(使用 adbutils 库) # AdbManager — ADB 操作封装(使用 adbutils 库)
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
@@ -704,6 +798,130 @@ class IpcServer:
self.cleanup() 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 — 主应用 # PadSleepApp — 主应用
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
@@ -716,6 +934,8 @@ class PadSleepApp:
self.audit = AuditLogger() self.audit = AuditLogger()
self.config_mgr = ConfigManager() self.config_mgr = ConfigManager()
self.adb = AdbManager(self._get_config, self.audit) self.adb = AdbManager(self._get_config, self.audit)
self.screen_mgr = ScreenManager()
self.gpio: GPIOEventDetector | None = None
self.ipc = IpcServer(self) self.ipc = IpcServer(self)
self._lock = threading.Lock() self._lock = threading.Lock()
self._start_time = time.time() self._start_time = time.time()
@@ -728,6 +948,13 @@ class PadSleepApp:
"device_serial": "", "device_serial": "",
"uptime": 0, "uptime": 0,
"start_time": self._start_time, "start_time": self._start_time,
# ── GPIO 状态 ──
"gpio_available": False,
"gpio_triggered": False,
"gpio_countdown": 0,
"gpio_pin": 17,
# ── ScreenManager 状态 ──
"active_reason": None,
} }
# ── 辅助 ── # ── 辅助 ──
@@ -745,6 +972,32 @@ class PadSleepApp:
s["uptime"] = int(time.time() - self._start_time) s["uptime"] = int(time.time() - self._start_time)
return s 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 命令处理 ── # ── IPC 命令处理 ──
def handle_ipc(self, request: dict) -> dict: def handle_ipc(self, request: dict) -> dict:
@@ -761,6 +1014,9 @@ class PadSleepApp:
status["connection_type"] = self.adb.get_connection_type() status["connection_type"] = self.adb.get_connection_type()
status["wireless_host"] = cfg.get("adb_wireless_host", "") status["wireless_host"] = cfg.get("adb_wireless_host", "")
status["wireless_port"] = cfg.get("adb_wireless_port", 5555) 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} return {"ok": True, "data": status}
if cmd == "get_config": if cmd == "get_config":
@@ -795,10 +1051,12 @@ class PadSleepApp:
return {"ok": True} return {"ok": True}
if cmd == "screen_off": if cmd == "screen_off":
self.screen_mgr.release_manual()
ok = self.adb.screen_off() ok = self.adb.screen_off()
return {"ok": ok} return {"ok": ok}
if cmd == "screen_on": if cmd == "screen_on":
self.screen_mgr.trigger(ScreenManager.REASON_MANUAL)
ok = self.adb.screen_on() ok = self.adb.screen_on()
return {"ok": ok} return {"ok": ok}
@@ -859,6 +1117,20 @@ class PadSleepApp:
else: else:
log.warning("无线设备连接失败: %s", msg) log.warning("无线设备连接失败: %s", msg)
# ── 启动 GPIO 电平检测 ──
if config.get("gpio_enabled", True):
gpio_pin = config.get("gpio_pin", 17)
self.gpio = GPIOEventDetector(pin=gpio_pin, callback=self._on_gpio_event)
self.gpio.start()
if self.gpio.available:
log.info("GPIO 电平检测已启用 (BCM %d, 倒计时 %ds)",
gpio_pin, config.get("gpio_countdown_seconds", 60))
self.audit.log("GPIO检测", f"已启用 (BCM {gpio_pin})")
self._update_status(gpio_available=self.gpio.available, gpio_pin=gpio_pin)
else:
log.info("GPIO 电平检测已禁用 (配置 gpio_enabled=false)")
self._update_status(gpio_available=False)
# 启动 IPC # 启动 IPC
self.ipc.start() self.ipc.start()
@@ -884,14 +1156,17 @@ class PadSleepApp:
# 检测配置文件外部变更 # 检测配置文件外部变更
self.config_mgr.reload_if_changed() self.config_mgr.reload_if_changed()
# 分段睡眠以快速响应退出信号 # 分段睡眠,每秒做一次轻量决策检查(evaluate 纯内存运算)
interval = self.config_mgr.get().get("check_interval_seconds", 30) interval = self.config_mgr.get().get("check_interval_seconds", 30)
for _ in range(max(1, interval)): for _ in range(max(1, interval)):
if not self.running: if not self.running:
break break
time.sleep(1) time.sleep(1)
self._fast_check()
# 清理 # 清理
if self.gpio:
self.gpio.stop()
self.ipc.cleanup() self.ipc.cleanup()
self.audit.log("守护进程", "已退出") self.audit.log("守护进程", "已退出")
log.info("padsleep_adb 守护程序已退出") log.info("padsleep_adb 守护程序已退出")
@@ -909,12 +1184,82 @@ class PadSleepApp:
self._last_adb_warn = now self._last_adb_warn = now
return prev_screen_on if prev_screen_on is not None else False return prev_screen_on if prev_screen_on is not None else False
# ── 获取屏幕状态 ── # ── 获取屏幕实际状态 ──
screen_on, wakefulness = self.adb.get_screen_state() screen_on, wakefulness = self.adb.get_screen_state()
periods = config.get("screen_on_periods", []) periods = config.get("screen_on_periods", [])
within_period = TimeChecker.is_within_any_period(periods) within_period = TimeChecker.is_within_any_period(periods)
auto_sleep_delay = config.get("auto_sleep_delay_minutes", 5) * 60
# ── 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_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:
self.screen_mgr.trigger(ScreenManager.REASON_SENSOR)
# ── 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() device = self.adb.get_ready_device()
gpio_level = self.gpio.current_value if self.gpio and self.gpio.available else -1
self._update_status( self._update_status(
adb_ready=True, adb_ready=True,
device=device or "", device=device or "",
@@ -922,50 +1267,53 @@ class PadSleepApp:
screen_on=screen_on, screen_on=screen_on,
wakefulness=wakefulness, wakefulness=wakefulness,
within_period=within_period, 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("检测到屏幕被熄灭", "外部操作或用户手动")
# ── 决策逻辑 ──
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:
# 非亮屏时段但屏幕亮着 → 超时后息屏
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 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)
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
# 入口 # 入口
+40 -2
View File
@@ -29,7 +29,7 @@ CONFIG_PATH = SCRIPT_DIR / "config.json"
# ═══════════════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════════════
class IpcClient: class IpcClient:
"""Unix Socket JSON 行协议客户端。""" """Unix Socket JSON 行协议客户端(每次 send 新建连接)"""
def __init__(self, socket_path=SOCKET_PATH): def __init__(self, socket_path=SOCKET_PATH):
self.socket_path = socket_path self.socket_path = socket_path
@@ -58,7 +58,8 @@ class IpcClient:
self.connected = False self.connected = False
def send(self, cmd: str, **kwargs) -> dict: def send(self, cmd: str, **kwargs) -> dict:
if not self.sock: """发送命令,每次新建连接(守护进程每请求关闭连接)。"""
self.disconnect()
if not self.connect(): if not self.connect():
return {"ok": False, "error": "未连接守护进程"} return {"ok": False, "error": "未连接守护进程"}
req = {"cmd": cmd, **kwargs} req = {"cmd": cmd, **kwargs}
@@ -434,6 +435,43 @@ class TuiApp:
self.stdscr.addstr(f"{s.get('config_path', '?')}", curses.A_DIM) self.stdscr.addstr(f"{s.get('config_path', '?')}", curses.A_DIM)
except curses.error: except curses.error:
pass 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: else:
if y < h: if y < h:
self._clear_line(y) self._clear_line(y)
+201
View File
@@ -0,0 +1,201 @@
#!/usr/bin/env python3
"""
padsleep Web Monitor
====================
通过 Unix Socket 从 padsleep 守护进程读取全量状态(含 GPIO 实时电平),
通过 WebSocket 推送到浏览器实时展示。不与守护进程竞争硬件资源。
用法:
# 先启动 padsleep 守护进程
python padsleep.py
# 再启动 Web 监控(新终端)
python padsleep_web.py
# 浏览器打开 http://树莓派IP:31400
依赖:
pip install flask flask-socketio
配置项(环境变量):
PORT=31400 Web 端口
"""
import json
import os
import socket
import threading
import time
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
# ── 配置 ────────────────────────────────────────────
PORT = int(os.environ.get("PORT", "31400"))
SOCKET_PATH = "/tmp/padsleep.sock"
POLL_FAST_S = 0.2 # GPIO 电平轮询间隔(200ms,跟随守护进程 50ms 更新)
POLL_SLOW_S = 1 # 其他状态轮询间隔
# ═══════════════════════════════════════════════════════════════════
# IpcClient — 与 padsleep 守护进程通信(Unix Socket
# ═══════════════════════════════════════════════════════════════════
class IpcClient:
"""Unix Socket JSON 行协议客户端(每次 send 新建连接)。"""
def __init__(self, socket_path=SOCKET_PATH):
self.socket_path = socket_path
self.last_error = ""
def _connect(self) -> socket.socket | None:
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(2)
sock.connect(self.socket_path)
return sock
except (socket.error, OSError) as e:
self.last_error = str(e)
return None
def send(self, cmd: str, **kwargs) -> dict:
"""发送命令,每次新建连接(守护进程每请求关闭连接)。"""
sock = self._connect()
if sock is None:
return {"ok": False, "error": f"未连接: {self.last_error}"}
req = {"cmd": cmd, **kwargs}
try:
sock.sendall((json.dumps(req) + "\n").encode("utf-8"))
resp = sock.recv(65536)
return json.loads(resp.decode("utf-8").strip())
except (socket.error, json.JSONDecodeError) as e:
return {"ok": False, "error": str(e)}
finally:
try:
sock.close()
except Exception:
pass
# ═══════════════════════════════════════════════════════════════════
# Flask 应用
# ═══════════════════════════════════════════════════════════════════
app = Flask(__name__)
app.config["SECRET_KEY"] = os.urandom(16).hex()
socketio = SocketIO(app, cors_allowed_origins="*")
ipc = IpcClient()
# ── 全局缓存(供不同频率的轮询线程共享)──
_data_lock = threading.Lock()
_cached = {
"gpio_level": -1,
"gpio_time": "--",
"status": None,
}
# ── 路由 ─────────────────────────────────────────────
@app.route("/")
def index():
return render_template("index.html")
# ── 后台线程:高频轮询 GPIO 电平 ────────────────────
def poll_gpio():
"""通过 IPC 高频读取 GPIO 电平,变化时 WebSocket 广播。"""
last_val = -1
while True:
r = ipc.send("status")
if r.get("ok"):
level = r["data"].get("gpio_level", -1)
now = time.strftime("%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}"
if level != -1 and level != last_val:
socketio.emit("gpio_update", {
"value": level,
"time": now,
})
last_val = level
# 更新缓存(补上 connected 字段,前端据此判断连接状态)
r["data"]["connected"] = True
with _data_lock:
_cached["gpio_level"] = level
_cached["gpio_time"] = now
_cached["status"] = r["data"]
socketio.sleep(POLL_FAST_S)
# ── 后台线程:慢速轮询全量状态 ──────────────────────
def poll_status():
"""定期查询守护进程全量状态,通过 WebSocket 广播。"""
while True:
with _data_lock:
status = _cached.get("status")
if status:
socketio.emit("status_update", status)
socketio.sleep(POLL_SLOW_S)
# ── WebSocket 事件 ──────────────────────────────────
@socketio.on("connect")
def on_connect(auth=None):
print("[WS] 客户端已连接")
# 推送初始状态
r = ipc.send("status")
if r.get("ok"):
data = r["data"]
data["connected"] = True
level = data.get("gpio_level", -1)
now = time.strftime("%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}"
with _data_lock:
_cached["gpio_level"] = level
_cached["gpio_time"] = now
_cached["status"] = data
socketio.emit("gpio_update", {"value": level, "time": now})
socketio.emit("status_update", data)
else:
socketio.emit("status_update", {
"connected": False,
"error": r.get("error", "无法连接守护进程"),
})
@socketio.on("disconnect")
def on_disconnect():
print("[WS] 客户端已断开")
# ── 入口 ─────────────────────────────────────────────
def main():
print("╔══════════════════════════════════════════╗")
print("║ padsleep Web Monitor ║")
print("║ (通过 IPC 读取守护进程状态) ║")
print("╠══════════════════════════════════════════╣")
print(f"║ 监听地址: http://0.0.0.0:{PORT}")
print(f"║ 数据源: {SOCKET_PATH}")
print(f"║ GPIO 轮询: {POLL_FAST_S*1000:.0f}ms — 跟随守护进程 50ms 线程")
print(f"║ 状态轮询: {POLL_SLOW_S}s")
print(f"║ 请先启动 padsleep.py 守护进程")
print(f"║ 按 Ctrl+C 停止 ║")
print("╚══════════════════════════════════════════╝")
# 启动后台线程
socketio.start_background_task(poll_gpio)
socketio.start_background_task(poll_status)
try:
socketio.run(app, host="0.0.0.0", port=PORT, allow_unsafe_werkzeug=True)
except KeyboardInterrupt:
pass
finally:
print("👋 已退出")
if __name__ == "__main__":
main()
+3
View File
@@ -4,6 +4,9 @@
# .venv/bin/pip install -r requirements.txt # .venv/bin/pip install -r requirements.txt
adbutils>=2.12.0 adbutils>=2.12.0
rpi-lgpio>=0.6 # GPIO 电平检测(树莓派硬件传感器触发息屏/亮屏)
flask>=3.0 # Web 监控面板
flask-socketio>=5.0 # WebSocket 实时推送
# 系统依赖:adb (Android Debug Bridge) # 系统依赖:adb (Android Debug Bridge)
# 安装: sudo apt-get install adb # 安装: sudo apt-get install adb
+544
View File
@@ -0,0 +1,544 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>padsleep 监控面板</title>
<script src="https://cdn.socket.io/4.7.5/socket.io.min.js"></script>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
background: #0f172a;
color: #e2e8f0;
min-height: 100vh;
padding: 20px;
}
.container {
max-width: 900px;
margin: 0 auto;
}
h1 {
font-size: 1.4rem;
font-weight: 500;
color: #94a3b8;
margin-bottom: 24px;
letter-spacing: 0.02em;
text-align: center;
}
h1 small {
display: block;
font-size: 0.8rem;
color: #64748b;
margin-top: 4px;
font-weight: 400;
}
/* ── 网格布局 ── */
.dashboard {
display: grid;
grid-template-columns: 280px 1fr;
gap: 16px;
margin-bottom: 16px;
}
.card {
background: #1e293b;
border-radius: 12px;
padding: 20px;
border: 1px solid #334155;
}
.card-title {
font-size: 0.75rem;
color: #64748b;
text-transform: uppercase;
letter-spacing: 0.08em;
margin-bottom: 14px;
}
/* ── GPIO 指示球 ── */
.indicator {
width: 160px;
height: 160px;
border-radius: 50%;
margin: 0 auto 16px;
transition: all 0.15s ease;
display: flex;
align-items: center;
justify-content: center;
position: relative;
}
.indicator::after {
content: "";
position: absolute;
inset: 0;
border-radius: 50%;
box-shadow: 0 0 50px currentColor;
opacity: 0.3;
transition: opacity 0.15s ease;
}
.indicator.low {
background: radial-gradient(circle at 35% 35%, #1e40af, #1e3a5f);
color: #3b82f6;
box-shadow: inset 0 -6px 24px rgba(0,0,0,0.5), 0 4px 16px rgba(59,130,246,0.12);
}
.indicator.high {
background: radial-gradient(circle at 35% 35%, #dc2626, #7f1d1d);
color: #ef4444;
box-shadow: inset 0 -6px 24px rgba(0,0,0,0.5), 0 4px 24px rgba(239,68,68,0.3);
}
.indicator .gpio-value {
font-size: 3rem;
font-weight: 700;
z-index: 1;
text-shadow: 0 2px 8px rgba(0,0,0,0.4);
font-variant-numeric: tabular-nums;
}
.indicator.low .gpio-value { color: #60a5fa; }
.indicator.high .gpio-value { color: #fca5a5; }
.gpio-label {
text-align: center;
font-size: 1rem;
font-weight: 500;
}
.gpio-label.low { color: #60a5fa; }
.gpio-label.high { color: #fca5a5; }
.gpio-pin-info {
text-align: center;
color: #64748b;
font-size: 0.8rem;
margin-top: 4px;
}
.gpio-update-time {
text-align: center;
color: #475569;
font-size: 0.75rem;
margin-top: 2px;
}
/* ── 历史波形 ── */
.history {
margin-top: 12px;
}
.history-bar {
display: flex;
gap: 2px;
height: 28px;
align-items: flex-end;
justify-content: center;
}
.history-bar .bar {
width: 6px;
border-radius: 2px 2px 0 0;
transition: background 0.1s ease, height 0.2s ease;
min-height: 2px;
}
.history-bar .bar.high { background: #ef4444; }
.history-bar .bar.low { background: #3b82f6; }
/* ── 状态项 ── */
.status-grid {
display: grid;
grid-template-columns: 1fr 1fr;
gap: 10px;
}
.status-item {
padding: 8px 10px;
background: #0f172a;
border-radius: 8px;
display: flex;
justify-content: space-between;
align-items: center;
}
.status-item .label {
font-size: 0.78rem;
color: #94a3b8;
}
.status-item .value {
font-size: 0.85rem;
font-weight: 600;
font-variant-numeric: tabular-nums;
}
.status-item .value.ok { color: #22c55e; }
.status-item .value.warn { color: #eab308; }
.status-item .value.err { color: #ef4444; }
.status-item .value.info { color: #60a5fa; }
.status-item .value.neutral { color: #e2e8f0; }
/* ── 事件日志 ── */
.events-card {
border: 1px solid #334155;
}
.events-scroll {
max-height: 140px;
overflow-y: auto;
font-family: "SF Mono", "Fira Code", monospace;
font-size: 0.78rem;
line-height: 1.6;
}
.events-scroll::-webkit-scrollbar { width: 4px; }
.events-scroll::-webkit-scrollbar-thumb { background: #475569; border-radius: 2px; }
.event-entry {
display: flex;
gap: 8px;
padding: 2px 0;
border-bottom: 1px solid #1e293b;
}
.event-entry .event-time {
color: #475569;
white-space: nowrap;
min-width: 70px;
}
.event-entry .event-type {
color: #60a5fa;
white-space: nowrap;
}
.event-entry .event-detail {
color: #94a3b8;
overflow: hidden;
text-overflow: ellipsis;
}
.event-entry .event-detail.high { color: #fca5a5; }
.event-entry .event-detail.low { color: #60a5fa; }
.no-data {
color: #475569;
font-size: 0.8rem;
text-align: center;
padding: 20px;
}
/* ── 连接状态 ── */
.footer-status {
margin-top: 12px;
display: flex;
justify-content: space-between;
align-items: center;
font-size: 0.78rem;
color: #64748b;
}
.footer-status .dot {
display: inline-block;
width: 8px;
height: 8px;
border-radius: 50%;
margin-right: 6px;
}
.footer-status .dot.connected { background: #22c55e; }
.footer-status .dot.disconnected { background: #ef4444; }
.gpio-card { text-align: center; }
/* ── 响应式 ── */
@media (max-width: 680px) {
.dashboard { grid-template-columns: 1fr; }
.status-grid { grid-template-columns: 1fr; }
.indicator { width: 130px; height: 130px; }
.indicator .gpio-value { font-size: 2.4rem; }
}
</style>
</head>
<body>
<div class="container">
<h1>
📊 padsleep 监控面板
<small>padsleep_adb 守护进程 &middot; 通过 IPC 读取状态</small>
</h1>
<div class="dashboard">
<!-- 左侧:GPIO -->
<div class="card gpio-card">
<div class="card-title">GPIO 实时电平</div>
<div class="indicator low" id="gpioIndicator">
<span class="gpio-value" id="gpioValue">0</span>
</div>
<div class="gpio-label low" id="gpioLabel">LOW (0V)</div>
<div class="gpio-pin-info">BCM GPIO <strong id="gpioPin">--</strong></div>
<div class="gpio-update-time" id="gpioUpdateTime">--</div>
<div class="history">
<div class="history-bar" id="historyBar"></div>
</div>
</div>
<!-- 右侧:状态 -->
<div class="card">
<div class="card-title">守护进程状态</div>
<div class="status-grid" id="statusGrid">
<div class="status-item">
<span class="label">屏幕</span>
<span class="value neutral" id="sScreen">--</span>
</div>
<div class="status-item">
<span class="label">亮屏事件</span>
<span class="value neutral" id="sReason">--</span>
</div>
<div class="status-item">
<span class="label">在定时时段内</span>
<span class="value neutral" id="sPeriod">--</span>
</div>
<div class="status-item">
<span class="label">ADB</span>
<span class="value neutral" id="sAdb">--</span>
</div>
<div class="status-item">
<span class="label">连接类型</span>
<span class="value neutral" id="sConn">--</span>
</div>
<div class="status-item">
<span class="label">GPIO 传感器</span>
<span class="value neutral" id="sGpio">--</span>
</div>
<div class="status-item">
<span class="label">守护运行</span>
<span class="value neutral" id="sUptime">--</span>
</div>
<div class="status-item">
<span class="label">配置时间段</span>
<span class="value neutral" id="sPeriods">--</span>
</div>
</div>
</div>
</div>
<!-- 底部:事件日志 -->
<div class="card events-card">
<div class="card-title">GPIO 电平事件</div>
<div class="events-scroll" id="eventsList">
<div class="no-data">等待数据…</div>
</div>
</div>
<div class="footer-status">
<span>
<span class="dot disconnected" id="wsDot"></span>
<span id="wsStatus">⏳ 连接中…</span>
</span>
<span id="logUpdate">--</span>
</div>
</div>
<script>
const HISTORY_SIZE = 40;
const MAX_EVENTS = 50;
// ── 事件流(最近的在上面)──
let events = [];
// ── 格式化 ──
const REASON_LABELS = {
scheduled: "定时点亮",
sensor: "传感器触发",
manual: "手动控制",
external: "外部亮屏",
};
function formatUptime(sec) {
if (sec == null) return "--";
const h = Math.floor(sec / 3600);
const m = Math.floor((sec % 3600) / 60);
const s = sec % 60;
if (h > 0) return `${h}h${String(m).padStart(2, "0")}m`;
if (m > 0) return `${m}m${String(s).padStart(2, "0")}s`;
return `${s}s`;
}
// ── DOM refs ──
const indicator = document.getElementById("gpioIndicator");
const gpioValue = document.getElementById("gpioValue");
const gpioLabel = document.getElementById("gpioLabel");
const gpioUpdateTime = document.getElementById("gpioUpdateTime");
const historyBar = document.getElementById("historyBar");
const eventsList = document.getElementById("eventsList");
const wsDot = document.getElementById("wsDot");
const wsStatus = document.getElementById("wsStatus");
const gpioPin = document.getElementById("gpioPin");
// ── 状态 DOM refs ──
const statusFields = {
screen: document.getElementById("sScreen"),
reason: document.getElementById("sReason"),
period: document.getElementById("sPeriod"),
adb: document.getElementById("sAdb"),
conn: document.getElementById("sConn"),
gpio: document.getElementById("sGpio"),
uptime: document.getElementById("sUptime"),
periods: document.getElementById("sPeriods"),
};
// ── 初始化历史 ──
let history = new Array(HISTORY_SIZE).fill(0);
function initHistory() {
historyBar.innerHTML = "";
for (let i = 0; i < HISTORY_SIZE; i++) {
const bar = document.createElement("div");
bar.className = "bar low";
bar.style.height = "2px";
historyBar.appendChild(bar);
}
}
initHistory();
function updateHistory(value) {
history.push(value);
if (history.length > HISTORY_SIZE) history.shift();
const bars = historyBar.querySelectorAll(".bar");
for (let i = 0; i < bars.length; i++) {
const v = history[i];
bars[i].className = `bar ${v === 1 ? "high" : "low"}`;
bars[i].style.height = v === 1 ? "24px" : "2px";
}
}
// ── GPIO 更新 ──
function updateGPIO(data) {
const v = data.value;
if (v === 1) {
indicator.className = "indicator high";
gpioLabel.className = "gpio-label high";
gpioLabel.textContent = "HIGH (3.3V)";
} else {
indicator.className = "indicator low";
gpioLabel.className = "gpio-label low";
gpioLabel.textContent = "LOW (0V)";
}
gpioValue.textContent = v;
gpioUpdateTime.textContent = data.time || new Date().toLocaleTimeString();
updateHistory(v);
addEvent(v, data.time);
}
// ── 事件日志 ──
function addEvent(value, timeStr) {
const now = timeStr || new Date().toLocaleTimeString();
events.unshift({
time: now,
type: value === 1 ? "HIGH" : "LOW",
value: value,
});
if (events.length > MAX_EVENTS) events.pop();
renderEvents();
}
function renderEvents() {
if (events.length === 0) {
eventsList.innerHTML = '<div class="no-data">等待数据…</div>';
return;
}
let html = "";
for (const e of events) {
const cls = e.value === 1 ? "high" : "low";
html += `<div class="event-entry">`;
html += `<span class="event-time">${e.time}</span>`;
html += `<span class="event-type ${cls}">${e.type}</span>`;
html += `<span class="event-detail ${cls}">GPIO ${e.value === 1 ? "↑ 上升" : "↓ 下降"}</span>`;
html += `</div>`;
}
eventsList.innerHTML = html;
}
// ── 状态更新 ──
function updateStatus(data) {
if (!data.connected) {
for (const key of Object.keys(statusFields)) {
statusFields[key].className = "value err";
statusFields[key].textContent = "×";
}
statusFields.screen.textContent = "未连接";
return;
}
// 屏幕
const screenEl = statusFields.screen;
if (data.screen_on) {
screenEl.textContent = `亮 (${data.wakefulness || ""})`;
screenEl.className = "value ok";
} else {
screenEl.textContent = `灭 (${data.wakefulness || ""})`;
screenEl.className = "value err";
}
// 亮屏事件
const reasonEl = statusFields.reason;
if (data.active_reason) {
reasonEl.textContent = REASON_LABELS[data.active_reason] || data.active_reason;
reasonEl.className = "value info";
} else {
reasonEl.textContent = "无";
reasonEl.className = "value neutral";
}
// 在定时时段内
const periodEl = statusFields.period;
periodEl.textContent = data.within_period ? "是" : "否";
periodEl.className = data.within_period ? "value ok" : "value neutral";
// ADB
const adbEl = statusFields.adb;
adbEl.textContent = data.adb_ready ? "就绪" : "未就绪";
adbEl.className = data.adb_ready ? "value ok" : "value err";
// 连接类型
const connEl = statusFields.conn;
const connMap = { usb: "USB", wireless: "WiFi", none: "无" };
connEl.textContent = connMap[data.connection_type] || data.connection_type || "--";
connEl.className = data.connection_type !== "none" ? "value ok" : "value err";
// GPIO 传感器
const gpioEl = statusFields.gpio;
if (data.gpio_available) {
if (data.gpio_triggered) {
gpioEl.textContent = `触发中 ⏱ ${data.gpio_countdown}s`;
gpioEl.className = "value ok";
} else {
gpioEl.textContent = "等待中";
gpioEl.className = "value neutral";
}
} else {
gpioEl.textContent = "不可用";
gpioEl.className = "value err";
}
// 运行时间
statusFields.uptime.textContent = formatUptime(data.uptime);
statusFields.uptime.className = "value neutral";
// 时间段数
statusFields.periods.textContent = `${data.periods_count || 0}`;
statusFields.periods.className = "value neutral";
// GPIO 引脚号
gpioPin.textContent = data.gpio_pin || "--";
// 日志更新时间
document.getElementById("logUpdate").textContent = `更新: ${new Date().toLocaleTimeString()}`;
}
// ── SocketIO ──
const socket = io();
socket.on("connect", () => {
wsDot.className = "dot connected";
wsStatus.textContent = "🟢 已连接 (实时)";
});
socket.on("disconnect", () => {
wsDot.className = "dot disconnected";
wsStatus.textContent = "🔴 连接断开";
});
socket.on("connect_error", (err) => {
wsDot.className = "dot disconnected";
wsStatus.textContent = "⚠️ 连接失败: " + err.message;
});
socket.on("gpio_update", updateGPIO);
socket.on("status_update", updateStatus);
</script>
</body>
</html>