615 lines
22 KiB
Python
615 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""padsleep_adb 守护程序
|
||
|
||
基于树莓派，通过 ADB 控制安卓平板自动亮屏和息屏。
|
||
|
||
阶段一功能：
|
||
- 根据配置的亮屏时间段自动亮屏/息屏
|
||
- 屏幕亮起超过设定时长且不在亮屏时间段内 → 自动息屏
|
||
- 通过 Unix Socket 提供 IPC 接口，供 TUI 工具管理
|
||
"""
|
||
|
||
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
|
||
|
||
# ── 常量 ────────────────────────────────────────────────────────
|
||
|
||
SCRIPT_DIR = Path(__file__).parent.resolve()
|
||
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
||
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_path": "adb",
|
||
}
|
||
|
||
# ── 日志 ────────────────────────────────────────────────────────
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
)
|
||
log = logging.getLogger("padsleep")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# 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)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# AdbManager — ADB 操作封装（使用 adbutils 库）
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class AdbManager:
|
||
"""封装所有 ADB 操作，使用 adbutils 库实现持久连接。"""
|
||
|
||
def __init__(self, config_getter):
|
||
self._config_getter = config_getter
|
||
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
|
||
|
||
# ── 连接管理 ──
|
||
|
||
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
|
||
|
||
# ── 设备管理 ──
|
||
|
||
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"}]。"""
|
||
try:
|
||
client = self._ensure_client()
|
||
devices = client.device_list()
|
||
return [{"serial": d.serial, "status": "device"} for d in devices]
|
||
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", "")
|
||
|
||
try:
|
||
client = self._ensure_client()
|
||
devices = client.device_list()
|
||
|
||
for d in devices:
|
||
if not target_serial or d.serial == target_serial:
|
||
# 尝试获取设备属性以验证连接
|
||
d.getprop("ro.build.version.sdk")
|
||
self._device = d
|
||
self._ready = True
|
||
self._last_error = ""
|
||
return True
|
||
|
||
if not devices:
|
||
self._last_error = "未检测到 ADB 设备，请连接平板"
|
||
else:
|
||
self._last_error = f"未找到指定设备: {target_serial}"
|
||
except Exception as e:
|
||
self._last_error = str(e)
|
||
|
||
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:
|
||
"""发送息屏命令。"""
|
||
log.info("执行息屏操作")
|
||
try:
|
||
device = self._ensure_device()
|
||
if device is None:
|
||
log.error("息屏失败: 无可用设备")
|
||
return False
|
||
device.shell("input keyevent 26")
|
||
self._screen_on_since = None
|
||
return True
|
||
except Exception as e:
|
||
self._last_error = str(e)
|
||
log.error("息屏失败: %s", e)
|
||
return False
|
||
|
||
def screen_on(self) -> bool:
|
||
"""发送亮屏并解锁命令。"""
|
||
log.info("执行亮屏操作")
|
||
try:
|
||
device = self._ensure_device()
|
||
if device is None:
|
||
log.error("亮屏失败: 无可用设备")
|
||
return False
|
||
device.shell("input keyevent 26")
|
||
self._screen_on_since = time.time()
|
||
device.shell("input keyevent 82") # 解锁（MENU）
|
||
return True
|
||
except Exception as e:
|
||
self._last_error = str(e)
|
||
log.error("亮屏失败: %s", 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.config_mgr = ConfigManager()
|
||
self.adb = AdbManager(self._get_config)
|
||
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,
|
||
"uptime": 0,
|
||
"start_time": self._start_time,
|
||
}
|
||
|
||
# ── 辅助 ──
|
||
|
||
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)
|
||
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)
|
||
cfg = self.config_mgr.get()
|
||
# 重设串号以使 is_ready 立即用新串号检查
|
||
self.config_mgr.config["adb_device_serial"] = cfg.get("adb_device_serial", "")
|
||
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})
|
||
return {"ok": True, "data": {"selected": serial}}
|
||
|
||
if cmd == "shutdown":
|
||
log.info("收到 IPC 关闭命令")
|
||
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}
|
||
|
||
return {"ok": False, "error": f"未知命令: {cmd}"}
|
||
|
||
# ── 主循环 ──
|
||
|
||
def run(self):
|
||
log.info("=" * 50)
|
||
log.info("padsleep_adb 守护程序启动")
|
||
log.info("=" * 50)
|
||
|
||
# 加载配置
|
||
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 可执行文件已就绪")
|
||
|
||
# 启动 IPC
|
||
self.ipc.start()
|
||
|
||
# 注册信号
|
||
def sighandler(sig, _frame):
|
||
log.info("收到信号 %s，正在退出...", sig)
|
||
self.running = False
|
||
|
||
signal.signal(signal.SIGTERM, sighandler)
|
||
signal.signal(signal.SIGINT, sighandler)
|
||
|
||
# 主循环
|
||
while self.running:
|
||
try:
|
||
self._tick()
|
||
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)
|
||
|
||
# 清理
|
||
self.ipc.cleanup()
|
||
log.info("padsleep_adb 守护程序已退出")
|
||
|
||
def _tick(self):
|
||
"""单次检查逻辑。"""
|
||
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
|
||
|
||
# ── 获取屏幕状态 ──
|
||
screen_on, wakefulness = self.adb.get_screen_state()
|
||
periods = config.get("screen_on_periods", [])
|
||
within_period = TimeChecker.is_within_any_period(periods)
|
||
|
||
device = self.adb.get_ready_device()
|
||
self._update_status(
|
||
adb_ready=True,
|
||
device=device,
|
||
screen_on=screen_on,
|
||
wakefulness=wakefulness,
|
||
within_period=within_period,
|
||
)
|
||
|
||
# ── 决策逻辑 ──
|
||
delay_sec = config.get("auto_sleep_delay_minutes", 5) * 60
|
||
|
||
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.adb.screen_off()
|
||
else:
|
||
remaining = delay_sec - on_duration
|
||
if remaining <= 10:
|
||
log.info("不在亮屏时段，%.0f 秒后将息屏", remaining)
|
||
|
||
elif within_period and not screen_on:
|
||
# 在亮屏时段但屏幕灭着 → 亮屏
|
||
log.info("在亮屏时段内 → 自动亮屏")
|
||
self.adb.screen_on()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# 入口
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
app = PadSleepApp()
|
||
try:
|
||
app.run()
|
||
except KeyboardInterrupt:
|
||
log.info("用户中断")
|
||
finally:
|
||
app.ipc.cleanup()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|