From 2a267967b4981a13b1e3168e10b1745e799256a7 Mon Sep 17 00:00:00 2001 From: wangfq Date: Tue, 25 Aug 2026 10:09:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(vd960Loop):=20=E7=8E=AF=E5=A2=83=E5=81=A5?= =?UTF-8?q?=E5=BA=B7=E5=BA=A6=E7=AE=97=E6=B3=95=20=C2=A710=20+=20drift=5Fh?= =?UTF-8?q?ealth.py=20=E5=8F=82=E8=80=83=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - variation-analysis.md §10: 无车漂移监测 (零成本, 不改协议) - 口径修正: 绝对峰=1.5×窗口漂移(Origin 滞后半窗) → 用峰谷差 - 600ms 采样系统性低估 ~12% (峰在窗口末采不到) - 冻结态失效(>4×dlt_ORG) → 双指标: 漂移速率 + 冻结占比 - 冻结判定双判据: 跨窗跳变消失 + 峰谷差突变(>4×前4窗中位数) - tools/drift_health.py: CSV 逐窗统计/冻结检测/JSON/自测 - devlog 2026-08-25 条目追加环境健康度小节 - 自测: 慢漂移 0.02%/s 识别准确, 冻结窗正确排除 --- vd960Loop/docs/devlog.md | 16 +++ vd960Loop/docs/variation-analysis.md | 65 ++++++++++ vd960Loop/tools/drift_health.py | 177 +++++++++++++++++++++++++++ 3 files changed, 258 insertions(+) create mode 100644 vd960Loop/tools/drift_health.py diff --git a/vd960Loop/docs/devlog.md b/vd960Loop/docs/devlog.md index 64307bf..eccf13e 100644 --- a/vd960Loop/docs/devlog.md +++ b/vd960Loop/docs/devlog.md @@ -57,6 +57,22 @@ variation/Origin = 1 − √(1 + ΔL/L₀) 精确 代码无改动（`main.c` 饱和边界本就是 −8388608 ~ +8388607）。 +### 环境健康度算法（§10） + +用户提出零成本环境健康度方案：无车时 variation 锯齿的峰 = 漂移信号，后台逐窗统计画健康度曲线。模拟验证后修正两处口径并落地： + +1. **绝对峰 ≠ 窗口漂移**：Origin 更新到窗口均值（滞后半窗 2.5s），稳态绝对峰 = **1.5×** 窗口漂移 → 改用**峰谷差**（max−min，误差 <1%） +2. **600ms 采样系统性低估 ~12%**：峰恒在窗口末（500 tick），末采样点 480 tick 永远错过 → 单边偏差，趋势可用，绝对标定 ×1.14 +3. **冻结态失效**：漂移 > 4×dlt_ORG（中档 0.13%/s）→ 锯齿消失，速率口径失效 → 双指标：`漂移速率 = 峰谷差/Origin/5s` + `冻结占比` +4. 冻结判定双判据：跨窗跳变消失（|jump|≈0）+ 峰谷差突变（>4×前 4 窗中位数），最后一窗也可判 + +| 文件 | 内容 | +|------|------| +| `docs/variation-analysis.md` | 新增 §10（口径修正表/双指标/冻结判定/工具用法） | +| `tools/drift_health.py` | 新增参考实现：CSV 输入 + 逐窗统计 + 冻结检测 + JSON 输出 + 自测 | + +自测：前 6 窗慢漂移 0.02%/s 全正常，后 2 窗快漂移正确标冻结，平均速率/冻结占比准确。 + --- ## 2026-07-14 — variation 上报量 2B→3B 有符号 (协议 V1.05) diff --git a/vd960Loop/docs/variation-analysis.md b/vd960Loop/docs/variation-analysis.md index 9d1c493..50d6247 100644 --- a/vd960Loop/docs/variation-analysis.md +++ b/vd960Loop/docs/variation-analysis.md @@ -327,3 +327,68 @@ python3 variation_calc.py --parse-var F6FFFF python3 variation_calc.py --sens-table # 四档对照 python3 variation_calc.py --interactive # 交互 ``` + +--- + +## 10. 环境健康度算法：无车漂移监测（2026-08-25 补充） + +> 配套工具：`tools/drift_health.py`。零成本方案——不改协议、不改固件，直接用现有 `variation` 锯齿。 + +### 10.1 思路 + +无车时 Origin 每 5s 阶跃更新（窗口均值），CAPVD 缓慢漂移 → variation 呈锯齿。**锯齿就是漂移信号**：峰谷差 = 窗口内漂移量，后台逐窗口统计即可画环境健康度曲线（湿度突变、变频器干扰加剧提前可见）。 + +### 10.2 正确口径（踩坑修正） + +| 口径 | 关系 | 结论 | +|------|------|------| +| 绝对峰 \|variation\| | = **1.5×** 窗口漂移量 | ❌ 高估 50%——Origin 更新到**窗口均值**（滞后半窗 2.5s），峰含半窗累积 | +| **峰谷差**（窗口内 max−min） | ≈ 窗口漂移量（误差 <1%） | ✅ 正确 | +| 600ms 采样（8 点/窗） | 峰谷差**系统性低估 ~12%** | ⚠ 峰恒在窗口末（500 tick），末采样点 480 tick 永远错过它；实测 −11%~−14% | + +**推荐公式：** + +``` +漂移速率(%/s) = (窗口内 max − min) / Origin / 5s × 100 +``` + +- 方向保留：V1.05 带符号，正向锯齿（CAPVD 下降/频率上升）= 正速率，反向为负 +- 采样低估 ~12% 为单边已知偏差：趋势曲线看相对变化可直接接受；要绝对标定乘 1.14 + +### 10.3 失效场景：冻结态（必须双指标） + +`dev ≥ 4×dlt_ORG` 基线冻结 → **锯齿消失**，variation 跨窗口单调累积不归零，"每窗口取峰"变成取历史累积，速率口径失效： + +| 灵敏度档 | 冻结门槛（窗口漂移） | 对应速率 | +|---------|-------------------|---------| +| 0 低 | 1.32% / 5s | 0.26%/s | +| 1 中 | 0.66% / 5s | **0.13%/s** | +| 3 最高 | 0.06% / 5s | 0.012%/s | + +变频器启动、湿度突变轻松超中档 0.13%/s → 冻结。**冻结本身 + 累积 variation 就是更强的恶化信号**，健康度曲线用双指标： + +``` +① 漂移速率 = 峰谷差 / Origin / 5s ← 慢漂移场景（温/湿度），慢且稳 +② 冻结占比 = 冻结 tick / 总 tick ← 快干扰场景（变频器），触发即预警 +``` + +**冻结判定**（后台可用跨窗回落检测）：正常锯齿窗口边界 variation **跳回谷值**，跳变幅度 ≈ 峰谷差；冻结时 Origin 不动，窗口边界无跳变、variation 继续同向爬升。 + +``` +jump = variation[下窗起点] − variation[本窗终点] +正常: |jump| ≈ 峰谷差（窗口边界跳回谷值） +冻结: |jump| ≈ 0（Origin 不动, 连续爬升） +``` + +### 10.4 工具用法 + +```bash +cd tools +python3 drift_health.py --selftest # 模拟数据自测 +python3 drift_health.py --csv drift.csv --window 500 # CSV: tick,origin,variation +python3 drift_health.py --csv drift.csv --window 500 --json # JSON 输出 +``` + +### 10.5 与 §8"改上报斜率"的关系 + +当前零成本方案精度受上报窗限制（600ms 采样 vs 固件 10ms tick，斜率分辨率差 60 倍）。§8 的固件侧斜率字段是**远期升级**：先跑通后台算法验证指标价值，确认值得后，再由固件直接上报 dCAPVD/dt 消除采样混叠与相位滞后。 diff --git a/vd960Loop/tools/drift_health.py b/vd960Loop/tools/drift_health.py new file mode 100644 index 0000000..e27ff33 --- /dev/null +++ b/vd960Loop/tools/drift_health.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +drift_health.py — vd960Loop 无车漂移环境健康度分析（零成本） + +原理（docs/variation-analysis.md §10）: + 无车时 Origin 每 5s 阶跃更新（窗口均值）→ variation 呈锯齿 + 锯齿峰谷差 = 窗口内漂移量 （绝对峰 = 1.5×窗口漂移, Origin 滞后半窗） + 漂移速率(%/s) = 峰谷差 / Origin / 窗口秒数 × 100 + +输入: CSV (tick,origin,variation 三列) 或 --selftest 模拟数据 +输出: 逐窗口统计 + 汇总（平均速率/方向/冻结占比） + +注意: + - 600ms 上报采样会使峰谷差系统性低估 ~12%（峰恒在窗口末, 末采样点错过） + - 漂移 > 4×dlt_ORG 时基线冻结, 锯齿消失 → 冻结占比是第二指标 + - 冻结判定: 跨窗回落检测 (正常窗口边界回落≈峰谷差; 冻结继续同向爬升) + +用法: + python3 drift_health.py --selftest + python3 drift_health.py --csv drift.csv --window 500 + python3 drift_health.py --csv drift.csv --window 500 --json +""" + +import argparse +import csv +import json +import math +import random +import sys + +WINDOW_DEFAULT = 500 # tick, 5s @10ms +TICK_S_DEFAULT = 0.01 # 10ms + + +def analyze(variation, origins, window, tick_s): + """逐窗口统计。返回 (窗口列表, 汇总字典)""" + import statistics + n = len(variation) + nw = n // window + rows = [] + pkpk_hist = [] + for k in range(nw): + a = k * window + b = a + window + seg = variation[a:b] + oseg = origins[a:b] + vmax = max(seg) + vmin = min(seg) + pkpk = vmax - vmin # 峰谷差 = 窗口漂移量 + origin = sum(oseg) / len(oseg) + # 跨窗回落: jump = 下窗起点 - 本窗终点 + jump = None + if k + 1 < nw: + jump = variation[b] - variation[b - 1] + # 冻结判定 (双判据): + # ① 跨窗跳变消失: 正常 |jump| ≈ 峰谷差, 冻结 |jump| ≈ 0 + # ② 峰谷差突变: 当前窗 pkpk > 4× 前 4 窗中位数 (冻结=漂移持续累积) + # 判据②让最后一窗(无 jump)也能判冻结, 实时滚动同样适用 + baseline = statistics.median(pkpk_hist[-4:]) if pkpk_hist else 0.0 + frozen = False + if pkpk > 0: + if jump is not None and abs(jump) < 0.3 * pkpk: + frozen = True + if baseline > 0 and pkpk > 4 * baseline: + frozen = True + pkpk_hist.append(pkpk) + # 速率 %/s (按峰谷差) + rate = pkpk / origin / (window * tick_s) * 100.0 if origin > 0 else 0.0 + rows.append({ + "window": k, + "max": round(vmax, 4), + "min": round(vmin, 4), + "pkpk": round(pkpk, 4), + "origin": round(origin, 1), + "rate_pct_per_s": round(rate, 4), + "jump": round(jump, 4) if jump is not None else None, + "frozen": frozen, + }) + if not rows: + return [], {} + rates = [r["rate_pct_per_s"] for r in rows if not r["frozen"]] + frozen_cnt = sum(1 for r in rows if r["frozen"]) + summary = { + "windows": len(rows), + "frozen_ratio_pct": round(frozen_cnt / len(rows) * 100, 1), + "avg_rate_pct_per_s": round(sum(rates) / len(rates), 4) if rates else None, + "max_rate_pct_per_s": round(max(rates), 4) if rates else None, + "last_direction": "正(频率上升/CAPVD下降)" if rows[-1]["pkpk"] > 0 and rows[-1]["max"] >= abs(rows[-1]["min"]) else "负(频率下降/CAPVD上升)", + } + return rows, summary + + +def gen_selftest(window=WINDOW_DEFAULT, tick_s=TICK_S_DEFAULT): + """生成模拟数据: 前 6 窗慢漂移(0.02%/s, 正常锯齿), 后 2 窗快漂移(0.5%/s, 冻结) + 归一化坐标: Origin≈1.0, 漂移为相对比例""" + r_slow = 2e-6 # 0.02%/s → 窗口漂移 0.1% + r_fast = 5e-5 # 0.5%/s → 窗口漂移 2.5%, 远超 4×dlt_ORG → 冻结 + n = 8 * window + capvd = [] + for i in range(n): + if i < 6 * window: + capvd.append(1.0 + i * r_slow) + else: + capvd.append(1.0 + 6 * window * r_slow + (i - 6 * window) * r_fast) + # Origin: 窗口 k 期间 = 窗口 k-1 均值 (k>=1); 冻结窗(6,7) 锁在窗口 5 均值 + origin = [0.0] * n + base6 = sum(capvd[5 * window:6 * window]) / window + for i in range(n): + k = i // window + if k == 0: + origin[i] = capvd[0] + elif k < 6: + origin[i] = sum(capvd[(k - 1) * window:k * window]) / window + else: + origin[i] = base6 + variation = [origin[i] - capvd[i] for i in range(n)] + return variation, origin + + +def fmt_table(rows, summary): + lines = [] + lines.append(f"{'窗口':<5}{'max':>10}{'min':>10}{'峰谷差':>10}{'Origin':>12}{'速率%/s':>10}{'跨窗jump':>10} 状态") + for r in rows: + state = "冻结" if r["frozen"] else "正常" + jump = f"{r['jump']:+.4f}" if r["jump"] is not None else " -" + lines.append(f"{r['window']:<5}{r['max']:>10.4f}{r['min']:>10.4f}{r['pkpk']:>10.4f}" + f"{r['origin']:>12.1f}{r['rate_pct_per_s']:>10.4f}{jump:>10} {state}") + lines.append("-" * 66) + lines.append(f"平均速率: {summary['avg_rate_pct_per_s']} %/s " + f"最大速率: {summary['max_rate_pct_per_s']} %/s " + f"冻结占比: {summary['frozen_ratio_pct']}% " + f"末窗方向: {summary['last_direction']}") + return "\n".join(lines) + + +def main(): + p = argparse.ArgumentParser( + description="vd960Loop 无车漂移环境健康度分析（零成本, 用现有 variation 锯齿）", + formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__) + p.add_argument("--csv", help="CSV 文件, 表头含 tick,origin,variation") + p.add_argument("--window", type=int, default=WINDOW_DEFAULT, help=f"窗口 tick (默认 {WINDOW_DEFAULT}=5s)") + p.add_argument("--tick-s", type=float, default=TICK_S_DEFAULT, help=f"tick 秒数 (默认 {TICK_S_DEFAULT}=10ms)") + p.add_argument("--json", action="store_true", help="JSON 输出") + p.add_argument("--selftest", action="store_true", help="模拟数据自测") + args = p.parse_args() + + if args.selftest: + variation, origin = gen_selftest(args.window, args.tick_s) + rows, summary = analyze(variation, origin, args.window, args.tick_s) + if args.json: + print(json.dumps({"rows": rows, "summary": summary}, ensure_ascii=False, indent=2)) + else: + print("模拟: 前 6 窗慢漂移 r=0.0002/tick (0.1%/窗 → 0.02%/s), 后 2 窗快漂移 r=0.002 (冻结)") + print(fmt_table(rows, summary)) + return + + if not args.csv: + p.error("需要 --csv 或 --selftest") + + variations, origins = [], [] + with open(args.csv, newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + variations.append(float(row["variation"])) + origins.append(float(row["origin"])) + if len(variations) < args.window: + p.error(f"数据不足一个窗口 ({len(variations)} < {args.window})") + + rows, summary = analyze(variations, origins, args.window, args.tick_s) + if args.json: + print(json.dumps({"rows": rows, "summary": summary}, ensure_ascii=False, indent=2)) + else: + print(fmt_table(rows, summary)) + + +if __name__ == "__main__": + sys.exit(main())