#!/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 = 0.05 # 轮询间隔 50ms 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._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) 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() # ═══════════════════════════════════════════════════════════════════ # 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.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, } # ── 辅助 ── 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 # ── 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) 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": ok = self.adb.screen_off() return {"ok": ok} if cmd == "screen_on": 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) 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() # 分段睡眠以快速响应退出信号 interval = self.config_mgr.get().get("check_interval_seconds", 30) for _ in range(max(1, interval)): if not self.running: break time.sleep(1) # 清理 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) # ═══════════════════════════════════════════════════════════════ # GPIO 触发检测（人来亮屏 / 人走息屏） # # 由于购买的模块有问题，无法稳定输出电平，但是人来人走时会发生 # 短暂电平切换，因此采用这个折中的法子：检测到 GPIO 电平变化 # 就亮屏并重置 gpio_countdown_seconds 秒倒计时，倒计时结束 # 且无新变化则息屏（除非当前在亮屏时段）。 # ═══════════════════════════════════════════════════════════════ gpio_triggered = False gpio_countdown_remain = 0 if self.gpio and self.gpio.available: gpio_pin = config.get("gpio_pin", 17) countdown = config.get("gpio_countdown_seconds", 60) elapsed = self.gpio.seconds_since_last_event gpio_triggered = elapsed < countdown gpio_countdown_remain = max(0, countdown - int(elapsed)) if gpio_triggered: # GPIO 刚刚触发过 → 确保屏幕亮起 if not screen_on: log.info("GPIO 检测到电平变化 → 亮屏 (%ds 倒计时)", gpio_countdown_remain) self.audit.log("GPIO触屏", f"BCM {gpio_pin} 电平变化，{countdown}s 倒计时") self.adb.screen_on() screen_on = True else: # 屏幕已亮，仅做调试日志 if gpio_countdown_remain >= countdown - 1: log.info("GPIO 电平变化 → 重置倒计时 %ds", countdown) else: # GPIO 倒计时已结束且屏幕不在亮屏时段 → 息屏 if screen_on and not within_period and elapsed > countdown: log.info("GPIO 倒计时结束（%.0fs 无变化）→ 息屏", elapsed) self.audit.log("GPIO息屏", f"BCM {gpio_pin} 在 {countdown}s 内无电平变化") self.adb.screen_off() screen_on = False # ── 更新状态（含 GPIO 信息）── device = self.adb.get_ready_device() 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, ) # ── 检测外部屏幕状态变化 ── if prev_screen_on is not None and prev_screen_on != screen_on: if screen_on: log.info("检测到屏幕被点亮 (外部操作)") self.audit.log("检测到屏幕被点亮", "外部操作或用户手动") else: log.info("检测到屏幕被熄灭 (外部操作)") self.audit.log("检测到屏幕被熄灭", "外部操作或用户手动") # ── 原有决策逻辑（仅在非 GPIO 触发时有效） ── delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60 period_desc = TimeChecker.nearest_period_description(periods) if within_period and not screen_on: # 在亮屏时段但屏幕灭着 → 立即亮屏（自动恢复） log.info("在亮屏时间段内但屏幕已熄灭 → 立即亮屏") self.audit.log("亮屏", f"在亮屏时间段 ({period_desc}) 内检测到屏幕熄灭，自动恢复") self.adb.screen_on() return True if not within_period and screen_on and not gpio_triggered: # 非亮屏时段、非 GPIO 触发、屏幕亮着 → 超时后息屏 on_duration = self.adb.get_screen_on_duration() if on_duration >= delay_sec: log.info("不在亮屏时段（已亮 %.0fs）→ 自动息屏", on_duration) self.audit.log("息屏", f"不在亮屏时间段 ({period_desc})，超时 {config.get('auto_sleep_delay_minutes')} 分钟") self.adb.screen_off() return False else: remaining = delay_sec - on_duration if remaining <= 10: log.info("不在亮屏时段，%.0f 秒后将息屏", remaining) elif within_period and screen_on: # 在亮屏时段且屏幕已亮 — 一切正常 if prev_screen_on is False: pass return screen_on # ═══════════════════════════════════════════════════════════════════ # 入口 # ═══════════════════════════════════════════════════════════════════ def main(): app = PadSleepApp() try: app.run() except KeyboardInterrupt: log.info("用户中断") finally: app.ipc.cleanup() if __name__ == "__main__": main()