feat(vd960Loop): 环境健康度算法 §10 + drift_health.py 参考实现
- 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 识别准确, 冻结窗正确排除
This commit is contained in:
@@ -57,6 +57,22 @@ variation/Origin = 1 − √(1 + ΔL/L₀) 精确
|
|||||||
|
|
||||||
代码无改动(`main.c` 饱和边界本就是 −8388608 ~ +8388607)。
|
代码无改动(`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)
|
## 2026-07-14 — variation 上报量 2B→3B 有符号 (协议 V1.05)
|
||||||
|
|||||||
@@ -327,3 +327,68 @@ python3 variation_calc.py --parse-var F6FFFF
|
|||||||
python3 variation_calc.py --sens-table # 四档对照
|
python3 variation_calc.py --sens-table # 四档对照
|
||||||
python3 variation_calc.py --interactive # 交互
|
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 消除采样混叠与相位滞后。
|
||||||
|
|||||||
@@ -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())
|
||||||
Reference in New Issue
Block a user