#!/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())