butubb/auto_control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(设备): 电量显示（大屏 + 监控页）+ 低电量 webhook 告警

Browse Source 
需求：大屏和监控页都要能看到每台设备的电量，并且"低于多少电量就发通知"。

数据从哪来：`adb -s <serial> shell dumpsys battery`（**只读**，实测 0.2~0.35s/台，
11 台并发一轮 0.44s）。设备只要在 `adb devices` 里就说明 transport 现成，
**不需要 connect**——所以连空闲设备也能查。这是本模块敢每轮扫全量的前提，
也是这条只读路径与"空闲设备不主动 connect"红线（DEVELOPMENT.md §2 #3）的边界，
已写进文档免得后人误加 connect。

- core/device_battery.py（新）：后台 daemon 线程 `device-battery`（照 device_discovery
  的 init_app/shutdown 模式），默认 60s 一轮，采「设备池 ∩ 在线」（与 list_ready 同口径，
  adb 可见但不归平台管的设备不上屏也不告警）。
  · 结果**只放内存**（不落库 → 不建表、不动备份覆盖清单），离线设备保留最后读数；
  · 读失败不清缓存：list_devices 失败时若照清会把整份缓存抹掉、大屏全体变「—」；
  · 配置存 `app_meta.device_battery`（照 step_defaults：出厂值 + 白名单 + 范围钳制 +
    只落与出厂值不同的字段），改阈值不用重启（线程每轮重读配置）。
- 告警：事件 `device.battery.low` / `device.battery.recovered`（默认关，去通知页勾订阅）。
  **按档位做状态沿**（0 正常 / 1 低 / 2 严重）：掉档才报、严重再报一次、恢复报一次；
  迟滞 +5% 避免在阈值上下反复告警；同档位每 6h 重复提醒一次。
  "充电中"看的是 `status:` 行（2/5），**不是"插着电"**——所以"插着却没充电"
  （劣质线/温控/满电停充 status:4）仍会告警，那正是最该知道的情况（实测 fleet 里
  就有一台 8% 插着 AC 但 `Max charging current: 0`）。
- GET /api/status 每台设备多一个 `battery`：`{level,charging,at,tier}`。
  **tier 由后端算好**，前端只按它上色——阈值只在「工具 → 设备发现 → 电量监控」一处定义，
  免得 JS 里再判一遍、两边不一致。
- 前端：大屏卡片型号行右侧加电量徽标（绿/黄/红 + ⚡ 充电中，离线置灰），
  `cardHtml`/`renderGrid` 增量更新/脏检查白名单三处都改了（漏一处就不刷新）；
  监控页设备表新增可排序的「电量」列（空表 colspan 11 顺带对齐了）。
- 接口：GET/POST /api/devices/battery[/settings] + POST /scan（需设备权限，
  与 /api/devices/discovery/* 同权限档）。

自测：单元 30 项（解析真机输出/档位/状态机/钳制，含 scale=255、无 status 行、
垃圾输出等边界）、集成 27 项（临时库自建设备池：缓存/通知/离线保留/删设备清理/
配置往返/阈值改动即时生效）、真实 11 台设备一轮 0.44s 全读到、API 端到端
（登录/权限/钳制/越界/REST 往返/页面 DOM）、两份前端各自语法检查 + 渲染用例。
This commit is contained in:
butubb
2026-09-24 08:37:42 +08:00
parent 773cfbb763
commit beb51637dd
16 changed files with 651 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/device_battery.py
+404
View File
@@ -0,0 +1,404 @@
"""设备电量采集 + 低电量告警（大屏/监控展示 + webhook 通知）。
设计要点（都是被现有代码/实测约束逼出来的，改之前先读一遍）：
- **只读**：`adb -s <serial> shell dumpsys battery`（实测 0.2~0.35s/台）。
全模块不 connect / 不 disconnect / 不 kill-server（技术红线，见 doc/DEVELOPMENT.md）。
能查到的前提是设备已在 adb server 里（= device_pool.list_online()），
所以连"空闲设备"也能查——**不需要 connect**，这是本模块敢每轮扫全量的原因。
- **后台独立线程**采集（照 core/device_discovery.py 的 daemon 循环）：`/api/status`
只读内存缓存。绝不能把 dumpsys 放进 /api/status——前端 5 秒轮询一次，
14 台设备同步查会把 Flask 拖死（`get_status` 的 5s 缓存注释也写了这条）。
- **不落库**：电量是易变值，重启重新采一轮即可 → 不建表、**不动备份覆盖清单**。
- 告警走 notifier（`device.battery.low` / `device.battery.recovered`），
**按档位做状态沿检测**（正常/低/严重），掉档才报，绝不满屏刷。
- 配置存 `app_meta.device_battery` 一个键（照 core/step_defaults.py 的做法：
出厂默认 + 字段白名单 + 只落"与出厂值不同"的字段）。
"""
import json
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from config import USB_ADB_HOST, USB_ADB_PORT
from core import notifier, device_pool
from core.adb_helper import _adb, _adb_remote
from core.logger import get_logger
from core.models import db
_log = get_logger("core.battery")
# 出厂默认（前端「工具 → 设备发现 → 电量监控」的初始内容，也是没配过时的回退）
FACTORY = {
"enabled": True,
"low": 20, # 低电量阈值（%）
"critical": 10, # 严重低电量阈值（%）——通知等级升到 error
"skip_charging": True, # 充电中不告警（插着充电器说明正在补电）
"interval": 60, # 采集间隔（秒）
}
# 字段规格：("num", 最小, 最大) / ("bool",)。白名单之外的一律丢弃。
SPEC = {
"enabled": ("bool",),
"low": ("num", 1, 100),
"critical": ("num", 1, 100),
"skip_charging": ("bool",),
"interval": ("num", 30, 3600),
}
META_KEY = "device_battery"
# 恢复迟滞：掉到阈值下之后，要回升到 low+HYSTERESIS 才算"恢复"。
# 没有它的话，电量在阈值上下浮动（充电器接触不良）会反复告警。
HYSTERESIS = 5
# 持续低电量的重复提醒间隔（秒）。0 = 只提醒一次（等恢复后才重新武装）。
RE_NOTIFY_S = 6 * 3600
# 单台设备读取超时（秒）。正常 0.2~0.35s，8s 已经非常宽松。
_READ_TIMEOUT = 8
_app = None
_cache = {} # serial -> {"level": int, "charging": bool, "at": ts}
_tiers = {} # serial -> {"tier": int, "at": ts} 告警档位状态（0 正常/1 低/2 严重）
_lock = threading.Lock()
_stop_event = threading.Event()
_scanning = False
_last_scan = None # (时间串, 采集成功数, 在线数)
def _ctx():
"""后台线程访问 db 时自行推 app context（仿 device_pool._ctx）。"""
return _app.app_context() if _app else None
# ================== 配置读写 ==================
def factory():
"""出厂默认的拷贝（别让调用方改到常量）。"""
return dict(FACTORY)
def _clean(raw):
"""按 SPEC 校验清洗；非法字段**丢弃**（不让脏值进库）。"""
if not isinstance(raw, dict):
return {}
out = {}
for k, rule in SPEC.items():
if k not in raw:
continue
v = raw[k]
try:
if rule[0] == "bool":
out[k] = bool(v)
elif rule[0] == "num":
out[k] = int(round(max(rule[1], min(rule[2], float(v)))))
except (TypeError, ValueError):
continue
return out
def get_settings():
"""当前生效配置（读库；模块未初始化时回退出厂值，别让调用方炸）。"""
ctx = _ctx()
if ctx is None:
return factory()
try:
from core.db_config import meta_get
with ctx:
raw = meta_get(META_KEY) or ""
if not raw:
return factory()
return _merge(json.loads(raw))
except Exception:
return factory()
def _merge(user):
"""出厂默认 + 用户覆盖（逐字段合并 + 钳制）。"""
vals = dict(FACTORY)
vals.update(_clean(user))
# 严重阈值不能高于低电量阈值，否则"严重"永远先触发，语义反了
if vals["critical"] > vals["low"]:
vals["critical"] = vals["low"]
return vals
def save_settings(payload):
"""保存配置。返回 (ok, msg, saved)。
与出厂默认**相同的字段不入库**（照 step_defaults）：以后调出厂值时，
没配过的字段能跟着更新，库里只留"用户真的改过"的部分。
"""
vals = _merge(payload)
diff = {k: v for k, v in vals.items() if FACTORY.get(k) != v}
ctx = _ctx()
if ctx is None:
return False, "电量模块未初始化", {}
try:
from core.db_config import meta_set
with ctx:
meta_set(META_KEY, json.dumps(diff, ensure_ascii=False))
db.session.commit()
except Exception as e:
return False, f"保存失败: {e}", {}
_log.info(f"电量监控配置已保存: {vals}")
return True, "已保存", vals
# ================== 单台读取与解析 ==================
_BAT_LEVEL_RE = re.compile(r"^\s*level:\s*(\d+)", re.M)
_BAT_SCALE_RE = re.compile(r"^\s*scale:\s*(\d+)", re.M)
_BAT_STATUS_RE = re.compile(r"^\s*status:\s*(\d+)", re.M)
# 部分 ROM 没有 status 行时的兜底：插着电就算充电中
_POWERED_RE = re.compile(r"^\s*(?:AC|USB|Wireless) powered:\s*true", re.M)
def _parse_battery(text):
"""解析 `dumpsys battery` 输出，返回 (level, charging) 或 None。
status: 2=充电中 5=已充满 → charging；3=放电 4=未充电（插着但不充，
例如劣质线/温控/满电停充）→ 不算充电中。所以"插着却掉电"仍然会告警，
这正是最该让人知道的情况。
"""
if not text or "level:" not in text:
return None
m = _BAT_LEVEL_RE.search(text)
if not m:
return None
level = int(m.group(1))
ms = _BAT_SCALE_RE.search(text)
scale = int(ms.group(1)) if ms else 100
if scale > 0 and scale != 100:
level = int(round(level * 100.0 / scale))
st = _BAT_STATUS_RE.search(text)
if st:
charging = int(st.group(1)) in (2, 5)
else:
charging = bool(_POWERED_RE.search(text))
return max(0, min(100, level)), charging
def _read_battery(serial):
"""读一台设备的电量，返回 (level, charging) 或 None。
与 device_pool.refresh_model 同思路：IP:5555 设备走本机 adb server；
USB 设备本机看不到，再走 220 远程 adb server 兜底。
"""
out = ""
try:
out = _adb("-s", serial, "shell", "dumpsys", "battery", timeout=_READ_TIMEOUT)
except Exception:
out = ""
if not out and ":" not in serial:
try:
out = _adb_remote(USB_ADB_HOST, USB_ADB_PORT, "-s", serial,
"shell", "dumpsys", "battery")
except Exception:
out = ""
return _parse_battery(out)
# ================== 告警（档位状态沿） ==================
def _tier_of(level, charging, spec, prev):
"""电量档位：0 正常 / 1 低 / 2 严重。
充电中（且配了「充电中不告警」）直接算 0——正在补电就不用管它。
回升要越过 low+HYSTERESIS 才回 0，避免在阈值上下反复告警。
"""
if spec["skip_charging"] and charging:
return 0
if level <= spec["critical"]:
return 2
if level <= spec["low"]:
return 1
if level >= spec["low"] + HYSTERESIS:
return 0
return prev # 迟滞带内：维持原档位（不报也不恢复）
def _check_alerts(serial, level, charging, spec, dname, model):
"""比上一轮档位，决定是否发通知。只在档位变化/到重复提醒点时发。"""
now = time.time()
with _lock:
st = _tiers.setdefault(serial, {"tier": 0, "at": 0.0})
prev = st["tier"]
tier = _tier_of(level, charging, spec, prev)
# 掉档（正常→低→严重）立刻报；同一档位停留太久到重复提醒间隔再报一次
fire = tier > prev or (tier > 0 and now - st["at"] >= RE_NOTIFY_S)
if fire:
st["tier"], st["at"] = tier, now
recovered = prev > 0 and tier == 0
if recovered:
st["tier"], st["at"] = 0, now
if fire:
notifier.notify(
"device.battery.low", serial=serial, device_name=dname, model=model,
battery=level, charging=charging,
threshold=spec["critical"] if tier == 2 else spec["low"],
level="error" if tier == 2 else "warning")
if recovered:
notifier.notify(
"device.battery.recovered", serial=serial, device_name=dname,
model=model, battery=level, charging=charging, threshold=spec["low"])
# ================== 一轮全量采集 ==================
def _scan_all():
"""采集池内所有在线设备的电量（后台线程执行）。
读失败的设备（离线/超时）**保持上一次的读数不动**——大屏上显示的是
"最后一次知道的电量"，比直接变成空白有用。
"""
global _scanning, _last_scan
if _scanning:
return
_scanning = True
try:
spec = get_settings()
try:
online = list(device_pool.list_online())
except Exception as e:
_log.warning(f"电量采集: 获取在线设备失败: {e}")
return
if not online:
_last_scan = (time.strftime("%H:%M:%S"), 0, 0)
return
# 设备名/型号：通知里要显示「A08（Redmi K40）」而不是一串 IP
names, models, known = {}, {}, set()
try:
for d in device_pool.list_devices():
s = d.get("serial") or ""
if not s:
continue
known.add(s)
names[s] = d.get("name") or ""
models[s] = d.get("model") or ""
except Exception:
pass
# 只采集「设备池 ∩ 在线」：平台管的是设备池（与 list_ready 同口径），
# 池外的设备不上大屏、也不告警——adb 里能看到但不归平台管的设备不该吵人。
# known 为空 = 设备表没读到（不能因此整轮不采），退回按在线设备采。
targets = [s for s in online if s in known] if known else list(online)
if not targets:
_last_scan = (time.strftime("%H:%M:%S"), 0, 0)
return
results = {}
try:
with ThreadPoolExecutor(max_workers=min(10, len(targets))) as pool:
futures = {pool.submit(_read_battery, s): s for s in targets}
for fut in as_completed(futures, timeout=_READ_TIMEOUT + 15):
s = futures[fut]
try:
results[s] = fut.result()
except Exception:
results[s] = None
except Exception as e:
_log.warning(f"电量采集: 部分设备读取超时（保留已取到的）: {e}")
now = time.time()
with _lock:
for s, r in results.items():
if r:
_cache[s] = {"level": r[0], "charging": r[1], "at": now}
# 设备已从池里删除的，连缓存一起清掉（避免大屏上冒出幽灵设备）。
# ⚠️ 只在真的读到设备清单时才清理：list_devices 失败时 known 是空的，
# 照清下去会把整份缓存抹掉、大屏全体变「—」。
if known:
for s in [k for k in _cache if k not in known]:
_cache.pop(s, None)
_tiers.pop(s, None)
for s, r in results.items():
if r:
_check_alerts(s, r[0], r[1], spec, names.get(s, ""), models.get(s, ""))
ok = sum(1 for r in results.values() if r)
_last_scan = (time.strftime("%H:%M:%S"), ok, len(targets))
_log.info(f"电量采集完成: {ok}/{len(targets)} 台（池内在线；adb 可见 {len(online)} 台）")
except Exception as e:
_log.warning(f"电量采集异常: {e}")
finally:
_scanning = False
def scan_once():
"""手动触发一轮采集（后台线程执行，不阻塞调用者）。返回是否已启动。
不看 enabled 开关：用户点了「立即采集」就应该采，开关只管定时循环。
"""
if _scanning:
return False
threading.Thread(target=_scan_all, name="battery-scan-once", daemon=True).start()
return True
# ================== 对外读取 ==================
def get(serial):
"""设备电量快照（供 /api/status 组装）或 None。
结构 {"level": int, "charging": bool, "at": ts, "tier": 0|1|2}。
**带上 tier 是刻意的**：前端只按档位上色，不在 JS 里重算阈值——
否则「工具 → 设备发现」改了阈值，大屏/监控页的判定就和告警对不上了。
"""
with _lock:
v = _cache.get(serial)
if not v:
return None
out = dict(v)
st = _tiers.get(serial)
out["tier"] = st["tier"] if st else 0
return out
def get_status():
"""采集器状态（面板显示用）。"""
with _lock:
cached = len(_cache)
return {
"settings": get_settings(),
"scanning": _scanning,
"last_scan": _last_scan,
"cached": cached,
}
# ================== 定时采集线程 ==================
def _battery_loop():
"""定时采集 daemon 线程。每轮重读配置（开关/间隔即时生效）。
启动先等 20s：让开 web_server 的预连接与型号采集（都抢 adb）。
等待用 15s 切片，关停/改间隔 ≤15s 生效。
"""
try:
_stop_event.wait(20)
while not _stop_event.is_set():
interval = 60
try:
spec = get_settings()
interval = max(30, int(spec.get("interval", 60)))
if spec.get("enabled"):
_scan_all()
except Exception as e:
_log.warning(f"电量定时采集异常: {e}")
waited = 0
while waited < interval and not _stop_event.is_set():
_stop_event.wait(min(15, interval - waited))
waited += min(15, interval - waited)
except Exception:
pass
def init_app(app):
"""web_server 启动时调用：绑 app + 起定时采集 daemon 线程。"""
global _app
_app = app
_stop_event.clear()
t = threading.Thread(target=_battery_loop, name="device-battery", daemon=True)
t.start()
_log.info("设备电量采集线程已启动（默认 60s 一次）")
def shutdown():
"""优雅退出：置 stop_event，等待中的循环在切片边界退出。"""
_stop_event.set()