auto_control/core/device_battery.py

"""设备电量采集 + 低电量告警（大屏/监控展示 + 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()