padsleep_adb/padsleep.py

#!/usr/bin/env python3
"""padsleep_adb 守护程序

基于树莓派，通过 ADB 控制安卓平板自动亮屏和息屏。

阶段一功能：
- 根据配置的亮屏时间段自动亮屏/息屏
- 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏
- 在亮屏时间段内检测到屏幕熄灭 → 自动恢复亮屏
- 通过 Unix Socket 提供 IPC 接口，供 TUI 工具管理
- 审计日志记录所有事件，自动维护日志大小不超过 50MB
"""

import json
import logging
import os
import signal
import socket
import threading
import time
from datetime import datetime
from pathlib import Path
from socketserver import ThreadingUnixStreamServer, StreamRequestHandler

from adbutils import AdbClient, AdbDevice

# ── 常量 ────────────────────────────────────────────────────────

# 注意: .resolve().parent 确保即使通过 symlink 调用也能正确找到真实目录
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
LOG_PATH = SCRIPT_DIR / "padsleep.log"
SOCKET_PATH = "/tmp/padsleep.sock"

DEFAULT_CONFIG = {
    "screen_on_periods": [
        {"start": "08:00", "end": "22:00"}
    ],
    "check_interval_seconds": 30,
    "auto_sleep_delay_minutes": 5,
    "adb_device_serial": "",
    "adb_wireless_host": "",
    "adb_wireless_port": 5555,
    "adb_wireless_auto_connect": False,
    # ── GPIO 电平触发（人来亮屏 / 人走息屏） ──
    "gpio_enabled": True,
    "gpio_pin": 17,
    "gpio_countdown_seconds": 60,
}

# ── 日志 ────────────────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("padsleep")


# ═══════════════════════════════════════════════════════════════════
#  AuditLogger — 审计日志
# ═══════════════════════════════════════════════════════════════════

class AuditLogger:
    """审计日志，记录到 padsleep.log。

    日志格式: [YYYY-MM-DD HH:MM:SS] 事件 — 详情
    自动维护: 超过 50MB 时保留最后 10% 的内容。
    """

    MAX_BYTES = 50 * 1024 * 1024  # 50 MB
    _CHECK_INTERVAL = 100         # 每写入 100 次检查一次文件大小

    def __init__(self, path: Path = LOG_PATH):
        self.path = path
        self._write_count = 0
        # 启动时检查并裁剪
        self._maybe_trim()

    def log(self, event: str, detail: str = ""):
        """写入一条审计日志。"""
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        entry = f"[{now}] {event}"
        if detail:
            entry += f" — {detail}"
        try:
            with open(self.path, "a", encoding="utf-8") as f:
                f.write(entry + "\n")
        except OSError as e:
            log.error("审计日志写入失败: %s", e)

        self._write_count += 1
        if self._write_count >= self._CHECK_INTERVAL:
            self._write_count = 0
            self._maybe_trim()

    def _maybe_trim(self):
        """检查日志文件大小，超过上限则裁剪（保留最后 10% 内容）。"""
        try:
            if self.path.exists() and self.path.stat().st_size > self.MAX_BYTES:
                self._trim()
        except OSError:
            pass

    def _trim(self):
        """删除日志前 90% 行，只保留最后 10%。"""
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                lines = f.readlines()
            if not lines:
                return
            keep = lines[-max(1, len(lines) // 10):]
            old_size = self.path.stat().st_size
            with open(self.path, "w", encoding="utf-8") as f:
                f.writelines(keep)
            new_size = self.path.stat().st_size
            saved_mb = (old_size - new_size) / 1048576
            log.info("审计日志已裁剪: 释放 %.1f MB (保留 %.1f MB)",
                     saved_mb, new_size / 1048576)
        except OSError as e:
            log.error("审计日志裁剪失败: %s", e)

    def read_tail(self, n: int = 50) -> str:
        """读取日志末尾 N 行。"""
        try:
            if not self.path.exists():
                return "(日志文件不存在)"
            with open(self.path, "r", encoding="utf-8") as f:
                lines = f.readlines()
            return "".join(lines[-n:])
        except OSError as e:
            return f"(读取日志失败: {e})"


# ═══════════════════════════════════════════════════════════════════
#  ConfigManager — 配置加载/保存/监控
# ═══════════════════════════════════════════════════════════════════

class ConfigManager:
    """管理 JSON 配置文件的加载、保存和变更检测。"""

    def __init__(self, path: Path = CONFIG_PATH):
        self.path = path
        self.config = dict(DEFAULT_CONFIG)
        self._mtime: float = 0
        self._lock = threading.Lock()

    def load(self) -> bool:
        """从 JSON 文件加载配置，失败时使用默认值并不中断运行。"""
        try:
            with open(self.path, "r") as f:
                data = json.load(f)
            if "screen_on_periods" not in data:
                raise ValueError("缺少 screen_on_periods")
            for p in data["screen_on_periods"]:
                if "start" not in p or "end" not in p:
                    raise ValueError("时间段缺少 start 或 end")
            with self._lock:
                self.config.update(data)
                self._mtime = os.path.getmtime(self.path)
            log.info("配置已加载: %s", self.path)
            return True
        except FileNotFoundError:
            log.warning("配置文件不存在，写入默认配置: %s", self.path)
            self.save()
            return True
        except (json.JSONDecodeError, ValueError, OSError) as e:
            log.error("配置加载失败: %s，使用默认配置", e)
            return True  # 不崩溃，用默认配置继续运行

    def save(self) -> bool:
        """将当前配置写入 JSON 文件。"""
        with self._lock:
            data = dict(self.config)
        try:
            with open(self.path, "w") as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
            self._mtime = os.path.getmtime(self.path)
            log.info("配置已保存")
            return True
        except OSError as e:
            log.error("配置保存失败: %s", e)
            return False

    def get(self) -> dict:
        """线程安全地获取当前配置副本。"""
        with self._lock:
            return dict(self.config)

    def update(self, new_config: dict) -> bool:
        """合并新配置并写入文件。"""
        with self._lock:
            self.config.update(new_config)
        self.save()
        return True

    def has_changed(self) -> bool:
        """检查配置文件是否被外部修改。"""
        try:
            return os.path.getmtime(self.path) != self._mtime
        except OSError:
            return False

    def reload_if_changed(self):
        """检测到外部修改则自动重载。"""
        if self.has_changed():
            log.info("配置文件已变更，重新加载")
            self.load()


# ═══════════════════════════════════════════════════════════════════
#  TimeChecker — 时间判断工具
# ═══════════════════════════════════════════════════════════════════

class TimeChecker:
    """亮屏时间段判断，支持跨天（如 22:00 ~ 08:00）。"""

    @staticmethod
    def _to_minutes(tstr: str) -> int:
        h, m = tstr.strip().split(":")
        return int(h) * 60 + int(m)

    @staticmethod
    def now_minutes() -> int:
        now = datetime.now()
        return now.hour * 60 + now.minute

    @staticmethod
    def is_within_period(period: dict, now_m: int | None = None) -> bool:
        """判断 now_m 是否在单个时间段内（支持跨天）。"""
        if now_m is None:
            now_m = TimeChecker.now_minutes()
        start = TimeChecker._to_minutes(period["start"])
        end = TimeChecker._to_minutes(period["end"])

        if end > start:
            return start <= now_m < end
        elif end == start:
            return True          # 全天
        else:
            return now_m >= start or now_m < end  # 跨天

    @staticmethod
    def is_within_any_period(periods: list, now_m: int | None = None) -> bool:
        """判断当前时间是否在任意一个亮屏时间段内。"""
        if not periods:
            return False
        if now_m is None:
            now_m = TimeChecker.now_minutes()
        return any(TimeChecker.is_within_period(p, now_m) for p in periods)

    @staticmethod
    def nearest_period_description(periods: list) -> str:
        """返回时间段的文字描述，用于审计日志。"""
        if not periods:
            return "无配置"
        descs = []
        for p in periods:
            descs.append(f"{p['start']}-{p['end']}")
        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 库）
# ═══════════════════════════════════════════════════════════════════

class AdbManager:
    """封装所有 ADB 操作，使用 adbutils 库实现持久连接。"""

    def __init__(self, config_getter, audit_logger: AuditLogger | None = None):
        self._config_getter = config_getter
        self._audit = audit_logger
        self._client: AdbClient | None = None
        self._device: AdbDevice | None = None
        self._ready = False
        self._last_error = ""
        self._screen_on_since: float | None = None
        self._last_known_screen_on = False
        self._last_adb_check = 0  # 限频用
        self._prev_ready = False  # 前次 ADB 就绪状态，用于检测变化
        self._current_serial: str | None = None  # 自动检测到的设备串号
        self._wireless_addr: str = ""  # 当前无线连接地址 (IP:PORT)
        self._wireless_retries: int = 0  # 无线重连尝试次数

    # ── 连接管理 ──

    def _ensure_client(self) -> AdbClient:
        """懒初始化 ADB 客户端连接。"""
        if self._client is None:
            self._client = AdbClient(host="127.0.0.1", port=5037)
        return self._client

    def _ensure_device(self) -> AdbDevice | None:
        """获取目标设备，支持串号过滤。"""
        config = self._config_getter()
        target_serial = config.get("adb_device_serial", "")

        try:
            client = self._ensure_client()
            if target_serial:
                self._device = client.device(target_serial)
            else:
                devices = client.device_list()
                self._device = devices[0] if devices else None
            return self._device
        except Exception as e:
            self._device = None
            self._last_error = str(e)
            return None

    # ── 无线连接管理 ──

    @staticmethod
    def is_wireless_serial(serial: str) -> bool:
        """判断 serial 是否为无线格式（IP:PORT）。"""
        parts = serial.split(":")
        if len(parts) != 2:
            return False
        try:
            port = int(parts[1])
            ip_parts = parts[0].split(".")
            if len(ip_parts) == 4 and all(0 <= int(p) <= 255 for p in ip_parts):
                return True
        except (ValueError, IndexError):
            pass
        return False

    def connect_wireless(self, host: str, port: int = 5555, timeout: float = 5.0) -> tuple[bool, str]:
        """连接无线 ADB 设备。返回 (success, message)。"""
        addr = f"{host}:{port}"
        try:
            client = self._ensure_client()
            result = client.connect(addr, timeout=timeout)
            if "connected" in result.lower():
                self._wireless_addr = addr
                self._wireless_retries = 0
                log.info("无线 ADB 已连接: %s", addr)
                if self._audit:
                    self._audit.log("无线连接", f"已连接: {addr}")
                return True, result
            return False, result
        except Exception as e:
            self._last_error = str(e)
            log.error("无线连接失败: %s", e)
            if self._audit:
                self._audit.log("无线连接失败", f"{addr}: {e}")
            return False, str(e)

    def disconnect_wireless(self, addr: str = "") -> bool:
        """断开无线 ADB 连接。"""
        target = addr or self._wireless_addr
        if not target:
            return False
        try:
            client = self._ensure_client()
            client.disconnect(target)
            log.info("无线 ADB 已断开: %s", target)
            if self._audit:
                self._audit.log("无线断开", f"已断开: {target}")
            if target == self._wireless_addr:
                self._wireless_addr = ""
            return True
        except Exception as e:
            self._last_error = str(e)
            log.error("断开无线连接失败: %s", e)
            return False

    def get_connection_type(self) -> str:
        """返回当前连接类型：'usb' / 'wireless' / 'none'。"""
        if self._device is None:
            return "none"
        if self.is_wireless_serial(self._device.serial):
            return "wireless"
        return "usb"

    def ensure_wireless_connection(self) -> bool:
        """确保无线连接存活，失败时尝试重连。返回是否成功。"""
        config = self._config_getter()
        wireless_host = config.get("adb_wireless_host", "")
        wireless_port = config.get("adb_wireless_port", 5555)

        if not wireless_host:
            return False

        addr = f"{wireless_host}:{wireless_port}"

        # 检查是否已有连接
        if self._device and self._device.serial == addr:
            try:
                self._device.getprop("ro.build.version.sdk")
                return True
            except Exception:
                # 连接已断开，尝试重连
                pass

        # 尝试重连（最多 3 次）
        max_retries = 3
        for i in range(max_retries):
            self._wireless_retries += 1
            log.info("尝试重连无线设备 (%d/%d): %s", i + 1, max_retries, addr)
            success, msg = self.connect_wireless(wireless_host, wireless_port, timeout=3.0)
            if success:
                # 重新获取设备
                try:
                    client = self._ensure_client()
                    self._device = client.device(addr)
                    self._device.getprop("ro.build.version.sdk")
                    self._ready = True
                    self._last_error = ""
                    return True
                except Exception:
                    pass
            time.sleep(2)

        self._last_error = f"无线重连失败: {addr}"
        return False

    # ── 设备管理 ──

    def check_available(self) -> bool:
        """检查 ADB Server 是否可连接。"""
        try:
            client = self._ensure_client()
            client.device_list()
            return True
        except Exception as e:
            self._last_error = f"ADB Server 不可用: {e}"
            return False

    def list_devices(self) -> list[dict]:
        """返回 [{"serial": "...", "status": "device", "connection_type": "usb"/"wireless"}]。"""
        try:
            client = self._ensure_client()
            devices = client.device_list()
            result = []
            for d in devices:
                conn_type = "wireless" if self.is_wireless_serial(d.serial) else "usb"
                result.append({
                    "serial": d.serial,
                    "status": "device",
                    "connection_type": conn_type,
                })
            return result
        except Exception as e:
            self._last_error = str(e)
            return []

    def get_ready_device(self) -> str | None:
        """返回当前已连接设备的串号，无则返回 None。"""
        if self._device is not None:
            return self._device.serial
        return None

    def is_ready(self) -> bool:
        """检查是否有已授权的 ADB 设备可用。"""
        now = time.time()
        if now - self._last_adb_check < 1 and self._ready:
            return True
        self._last_adb_check = now

        config = self._config_getter()
        target_serial = config.get("adb_device_serial", "")
        wireless_host = config.get("adb_wireless_host", "")

        try:
            client = self._ensure_client()
            devices = client.device_list()

            # USB 连接优先
            for d in devices:
                if not self.is_wireless_serial(d.serial):
                    if not target_serial or d.serial == target_serial:
                        d.getprop("ro.build.version.sdk")
                        self._device = d
                        self._current_serial = d.serial
                        if not self._prev_ready and self._audit:
                            self._audit.log("ADB 设备已连接", f"串号: {d.serial}, 类型: USB")
                        self._prev_ready = True
                        self._ready = True
                        self._last_error = ""
                        return True

            # 尝试无线连接
            if wireless_host:
                wireless_port = config.get("adb_wireless_port", 5555)
                wireless_addr = f"{wireless_host}:{wireless_port}"

                # 检查是否已在设备列表中
                for d in devices:
                    if d.serial == wireless_addr:
                        d.getprop("ro.build.version.sdk")
                        self._device = d
                        if not self._prev_ready and self._audit:
                            self._audit.log("ADB 设备已连接", f"串号: {d.serial}, 类型: 无线")
                        self._prev_ready = True
                        self._ready = True
                        self._last_error = ""
                        self._wireless_addr = wireless_addr
                        return True

                # 不在列表中，尝试连接
                if self.ensure_wireless_connection():
                    if not self._prev_ready and self._audit:
                        self._audit.log("ADB 设备已连接", f"串号: {wireless_addr}, 类型: 无线")
                    self._prev_ready = True
                    return True

            if not devices and not wireless_host:
                self._last_error = "未检测到 ADB 设备，请连接平板"
            elif not devices:
                self._last_error = f"无线设备未连接: {wireless_host}:{config.get('adb_wireless_port', 5555)}"
            else:
                self._last_error = f"未找到指定设备: {target_serial}"
        except Exception as e:
            self._last_error = str(e)

        # 如果之前是就绪状态但检测失败 → 设备断开
        if self._prev_ready and self._audit:
            last_serial = self._current_serial or (self._device.serial if self._device else "未知")
            self._audit.log("ADB 设备已断开", f"上次串号: {last_serial}, 原因: {self._last_error}")
        self._prev_ready = False
        self._ready = False
        self._device = None
        return False

    # ── 屏幕状态 ──

    def get_screen_state(self) -> tuple[bool, str]:
        """返回 (is_on, wakefulness)，失败时返回上次已知状态。"""
        try:
            device = self._ensure_device()
            if device is None:
                return self._last_known_screen_on, "unknown"

            output = device.shell("dumpsys power")
            for line in output.split("\n"):
                line = line.strip()
                if "mWakefulness" in line:
                    state = line.split("=")[-1].strip()
                    is_on = state == "Awake"
                    self._last_known_screen_on = is_on
                    if is_on and self._screen_on_since is None:
                        self._screen_on_since = time.time()
                    elif not is_on:
                        self._screen_on_since = None
                    return is_on, state

            # fallback
            for line in output.split("\n"):
                if "Display Power" in line or "mScreenOn" in line:
                    is_on = "ON" in line.upper() or "=true" in line.lower()
                    return is_on, "unknown"
        except Exception as e:
            self._last_error = str(e)

        return self._last_known_screen_on, "unknown"

    def get_screen_on_duration(self) -> float:
        """返回屏幕已持续亮起的秒数，灭屏时返回 0。"""
        if self._screen_on_since is not None:
            return time.time() - self._screen_on_since
        return 0.0

    # ── 屏幕控制（含验证）──

    def screen_off(self) -> bool:
        """发送息屏命令，等待 2 秒后验证，失败时重试一次。"""
        log.info("执行息屏操作")
        try:
            device = self._ensure_device()
            if device is None:
                log.error("息屏失败: 无可用设备")
                return False
            if self._audit:
                self._audit.log("息屏", "发出息屏指令")
            device.shell("input keyevent 26")

            # 等待并验证
            time.sleep(2)
            is_on, _ = self.get_screen_state()
            if is_on:
                log.warning("息屏命令已发出但屏幕未熄灭，重试一次")
                device.shell("input keyevent 26")
                time.sleep(2)
                is_on, _ = self.get_screen_state()
                if is_on:
                    log.error("息屏重试后仍未熄灭")
                    self._last_error = "息屏验证失败"
                    if self._audit:
                        self._audit.log("息屏异常", "屏幕状态未变化")
                    return False

            self._screen_on_since = None
            log.info("息屏成功")
            if self._audit:
                self._audit.log("息屏成功", "屏幕已熄灭")
            return True
        except Exception as e:
            self._last_error = str(e)
            log.error("息屏失败: %s", e)
            if self._audit:
                self._audit.log("息屏失败", str(e))
            return False

    def screen_on(self) -> bool:
        """发送亮屏命令，等待 2 秒后验证，失败时重试一次。"""
        log.info("执行亮屏操作")
        try:
            device = self._ensure_device()
            if device is None:
                log.error("亮屏失败: 无可用设备")
                return False
            if self._audit:
                self._audit.log("亮屏", "发出亮屏指令")
            device.shell("input keyevent 26")

            # 等待并验证
            time.sleep(2)
            is_on, _ = self.get_screen_state()
            if not is_on:
                log.warning("亮屏命令已发出但屏幕未点亮，重试一次")
                device.shell("input keyevent 26")
                time.sleep(2)
                is_on, _ = self.get_screen_state()
                if not is_on:
                    log.error("亮屏重试后仍未点亮")
                    self._last_error = "亮屏验证失败"
                    if self._audit:
                        self._audit.log("亮屏异常", "屏幕状态未变化")
                    return False

            self._screen_on_since = time.time()
            # 解锁屏幕
            device.shell("input keyevent 82")
            log.info("亮屏成功")
            if self._audit:
                self._audit.log("亮屏成功", "屏幕已点亮")
            return True
        except Exception as e:
            self._last_error = str(e)
            log.error("亮屏失败: %s", e)
            if self._audit:
                self._audit.log("亮屏失败", str(e))
            return False


# ═══════════════════════════════════════════════════════════════════
#  IPC 服务端 — Unix Socket JSON 行协议
# ═══════════════════════════════════════════════════════════════════

class IpcHandler(StreamRequestHandler):
    """每个客户端连接由独立线程处理。"""

    def handle(self):
        try:
            raw = self.rfile.readline()
            if not raw:
                return
            req = json.loads(raw.decode("utf-8").strip())
            resp = self.server.app.handle_ipc(req)
            self.wfile.write((json.dumps(resp) + "\n").encode("utf-8"))
        except json.JSONDecodeError:
            self.wfile.write('{"ok":false,"error":"JSON 解析失败"}\n'.encode("utf-8"))
        except Exception as e:
            self.wfile.write(json.dumps({"ok": False, "error": str(e)}).encode() + b"\n")


class IpcServer:
    """管理 Unix Socket 生命周期。"""

    def __init__(self, app):
        self.app = app
        self.server: ThreadingUnixStreamServer | None = None
        self._thread: threading.Thread | None = None

    def start(self):
        try:
            os.unlink(SOCKET_PATH)
        except OSError:
            pass

        self.server = ThreadingUnixStreamServer(SOCKET_PATH, IpcHandler)
        self.server.app = self.app
        self.server.socket.settimeout(1.0)
        self._thread = threading.Thread(target=self._serve, daemon=True)
        self._thread.start()
        log.info("IPC 服务已启动: %s", SOCKET_PATH)

    def _serve(self):
        while self.app.running:
            try:
                self.server.handle_request()
            except socket.timeout:
                continue
            except OSError:
                break

    def cleanup(self):
        try:
            os.unlink(SOCKET_PATH)
        except OSError:
            pass

    def stop(self):
        if self.server:
            self.server.shutdown()
        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 — 主应用
# ═══════════════════════════════════════════════════════════════════

class PadSleepApp:
    """守护程序核心：加载配置 → 主循环检查屏幕状态 → IPC 通信。"""

    def __init__(self):
        self.running = True
        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()
        self._start_time = time.time()
        self._status = {
            "screen_on": False,
            "wakefulness": "unknown",
            "within_period": False,
            "adb_ready": False,
            "device": None,
            "device_serial": "",
            "uptime": 0,
            "start_time": self._start_time,
            # ── GPIO 状态 ──
            "gpio_available": False,
            "gpio_triggered": False,
            "gpio_countdown": 0,
            "gpio_pin": 17,
            # ── ScreenManager 状态 ──
            "active_reason": None,
        }

    # ── 辅助 ──

    def _get_config(self) -> dict:
        return self.config_mgr.get()

    def _update_status(self, **kwargs):
        with self._lock:
            self._status.update(kwargs)

    def get_status(self) -> dict:
        with self._lock:
            s = dict(self._status)
            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:
        cmd = request.get("cmd", "")
        log.debug("IPC 请求: %s", cmd)

        if cmd == "status":
            status = self.get_status()
            cfg = self.config_mgr.get()
            status["periods_count"] = len(cfg.get("screen_on_periods", []))
            status["check_interval"] = cfg.get("check_interval_seconds", 30)
            status["config_path"] = str(self.config_mgr.path)
            status["log_path"] = str(LOG_PATH)
            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":
            return {"ok": True, "data": self.config_mgr.get()}

        if cmd == "update_config":
            new = request.get("config", {})
            if "screen_on_periods" in new:
                for p in new["screen_on_periods"]:
                    if "start" not in p or "end" not in p:
                        return {"ok": False, "error": "时间段格式错误"}
            self.config_mgr.update(new)
            return {"ok": True, "data": {"reloaded": True}}

        if cmd == "reload":
            self.config_mgr.load()
            return {"ok": True, "data": {"reloaded": True}}

        if cmd == "list_devices":
            return {"ok": True, "data": {"devices": self.adb.list_devices()}}

        if cmd == "select_device":
            serial = request.get("serial", "")
            self.config_mgr.update({"adb_device_serial": serial})
            self.audit.log("设备选择", f"串号: {serial or '自动'}")
            return {"ok": True, "data": {"selected": serial}}

        if cmd == "shutdown":
            log.info("收到 IPC 关闭命令")
            self.audit.log("守护进程", "收到关闭命令")
            self.running = False
            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}

        if cmd == "log_tail":
            n = request.get("lines", 50)
            return {"ok": True, "data": {"log": self.audit.read_tail(n)}}

        if cmd == "connect_wireless":
            host = request.get("host", "")
            port = request.get("port", 5555)
            if not host:
                return {"ok": False, "error": "请提供 IP 地址"}
            success, msg = self.adb.connect_wireless(host, int(port))
            if success:
                # 更新配置
                self.config_mgr.update({
                    "adb_wireless_host": host,
                    "adb_wireless_port": int(port),
                })
            return {"ok": success, "data": {"message": msg}}

        if cmd == "disconnect_wireless":
            addr = request.get("addr", "")
            ok = self.adb.disconnect_wireless(addr)
            return {"ok": ok}

        return {"ok": False, "error": f"未知命令: {cmd}"}

    # ── 主循环 ──

    def run(self):
        log.info("=" * 50)
        log.info("padsleep_adb 守护程序启动")
        log.info("日志路径: %s", LOG_PATH)
        log.info("=" * 50)

        self.audit.log("守护进程启动", f"配置文件: {self.config_mgr.path}")

        # 加载配置
        self.config_mgr.load()

        # 检测 ADB
        if not self.adb.check_available():
            log.warning("ADB 未就绪，请安装: sudo apt install adb")
            log.warning("安装后自动检测到 ADB，无需重启本程序")
        else:
            log.info("ADB 可执行文件已就绪")

        # 自动连接无线设备
        config = self.config_mgr.get()
        if config.get("adb_wireless_auto_connect") and config.get("adb_wireless_host"):
            host = config["adb_wireless_host"]
            port = config.get("adb_wireless_port", 5555)
            log.info("尝试自动连接无线设备: %s:%d", host, port)
            success, msg = self.adb.connect_wireless(host, port, timeout=5.0)
            if success:
                log.info("无线设备已连接: %s", msg)
            else:
                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
        self.ipc.start()

        # 注册信号
        def sighandler(sig, _frame):
            log.info("收到信号 %s，正在退出...", sig)
            self.audit.log("守护进程", "收到退出信号")
            self.running = False

        signal.signal(signal.SIGTERM, sighandler)
        signal.signal(signal.SIGINT, sighandler)

        # 记录上一次屏幕状态用于外部变化检测
        prev_screen_on: bool | None = None

        # 主循环
        while self.running:
            try:
                prev_screen_on = self._tick(prev_screen_on)
            except Exception as e:
                log.error("主循环异常: %s", e, exc_info=True)

            # 检测配置文件外部变更
            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:
            self.gpio.stop()
        self.ipc.cleanup()
        self.audit.log("守护进程", "已退出")
        log.info("padsleep_adb 守护程序已退出")

    def _tick(self, prev_screen_on: bool | None) -> bool:
        """单次检查逻辑。返回当前屏幕亮灭状态。"""
        config = self.config_mgr.get()

        # ── 检查 ADB 就绪 ──
        if not self.adb.is_ready():
            self._update_status(adb_ready=False, device=None)
            now = time.time()
            if not hasattr(self, "_last_adb_warn") or now - self._last_adb_warn > 60:
                log.warning("ADB 未就绪: %s", self.adb._last_error)
                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

        # ── 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()
        gpio_level = self.gpio.current_value if self.gpio and self.gpio.available else -1
        self._update_status(
            adb_ready=True,
            device=device or "无",
            device_serial=config.get("adb_device_serial", ""),
            screen_on=screen_on,
            wakefulness=wakefulness,
            within_period=within_period,
            gpio_triggered=gpio_triggered,
            gpio_countdown=gpio_countdown_remain,
            gpio_level=gpio_level,
            **self.screen_mgr.get_status(),
        )

        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)


# ═══════════════════════════════════════════════════════════════════
#  入口
# ═══════════════════════════════════════════════════════════════════

def main():
    app = PadSleepApp()
    try:
        app.run()
    except KeyboardInterrupt:
        log.info("用户中断")
    finally:
        app.ipc.cleanup()


if __name__ == "__main__":
    main()