+
${seqStepSummaryHtml(s)}
${s.actions.map((a, j) => `
电机 ${j + 1}
-
${seqChOptions(a.ch)}
-
${seqNodeOptions(a.node)}
-
${seqModeOptions(a.mode)}
+
${seqChOptions(a.ch)}
+
${seqNodeOptions(a.node)}
+
${seqModeOptions(a.mode)}
SPI 测速（临时工具）
-
-
-
- SPI 链路测速
- 临时诊断工具 [SPI-SPEEDTEST] — 探针 0xA1 APP PING，删除时直接删掉本文件
-
-
-
-
-
-
-
-
本页历史仅存内存，刷新即清空——方便改参数前后对比。
-
链路测速台 · SPI / EtherCAT
+
+
+
+
+
+
+
+
+
+
+
+
本次结果 三段计时均取自固件原始行，不做二次推断
+
+
+
单帧分段平均（固件五段拆解，占总往返比例）
+
+
跑一次测速后出现
+
延迟分布（固件 7 档计数，条长按最大档归一）
+
+
跑一次测速后出现
+
+
+ 设备原始返回
+ 等待测试…
+
+
+
本页历史 仅存内存，刷新即清空 —— 改参数前后对比用
+
+
+
+
+
+
+
观测控制 本页只读过程数据缓存，不连接、不重启、不停 PDO 循环
+
+ 开始实时观测
+ 逐拍环：关
+ 采样率
+
+ 2 Hz
+ 5 Hz
+ 10 Hz
+ 20 Hz
+
+
+ PDO 周期档位
+
+ （不自选）
+ 4 ms · 250 Hz
+ 2 ms · 500 Hz
+ 1 ms · 1 kHz
+ 0.5 ms · 2 kHz
+ 0.25 ms · 4 kHz
+ 0.125 ms · 8 kHz
+
+
+ 热改周期
+ 链路 —
+ AL —
+
+
+ 热改周期走 /api/motor/period（循环每拍重读的变量，全程喂狗不断流，随时可调、可反复调）。
+ 低于 1ms 档后端自动用自旋节拍，会占满一个 CPU 核 ；从站固件修好前别低于 1ms —— 实测
+ 0.25 / 0.125ms 档 GD32 顶到 99.5%，CAN 到帧率 λ 从 4000/s 塌到 149/s（瓶颈在从站不在主站，
+ 只看主站达成率会整条看漏）。
+
+
+
刷新率 实测 = Δcycles/Δt；名义 = 1000/period_ms 仅作对照，两者差值就是软件调度开销
+
+
+
+
+
+
单电机状态刷新
—Hz
slot_period_ms —
+
CAN 到帧率 λ
—/s
Δglobal_seq (u32)
+
真丢帧
—帧/拍
主站 seq 空洞 · 从站 OVF（抽样单拍值）
+
+
从站 CPU 负载
—
load_perm · age —
+
+
撕裂 / 重入
—
rx_torn / pdo_reentry
+
本页请求耗时
—ms
本页→代理→native 最慢一个请求的往返（≠节拍耗时）
+
+
+ 判读：达成率 < 100% = 节拍被软件拖住；λ ÷ 实测Hz = 每拍到达几帧 CAN（当前 1ms 档实测 6 帧/拍）。drop 不判好坏单格 被未取先覆的次数，
+ 而单格每拍只能发布 1 帧；且自增点在 STATUS 过滤之前 （ethercat_can_bridge.cpp:398
+ 对 :435），所以每拍到的 6 帧全算进来 → 恒等于 λ − 实测Hz，零独立信息量。
+ 这些帧并没有丢 ：STATUS 由段 2 按槽常驻上报，其余进段 4 覆盖窗口（实测 lost/OVF 均 0）。
+ drop 哪天才有权报警：让只走单格的帧（版本/节点应答）承担可靠投递时。真丢帧逐层看 ：① CAN 控制器 err/TEC/REC；② CAN 软件 RX 队列满（深度 16）的
+ rx_drop_count —— 只在 ESP32 /can/feedback 有，过程数据里没有；
+ ③ 段 4 覆盖窗口 主站侧 按 seq 空洞算的 rx_batch.lost（step/2−1，
+ 发布值恒为偶数，按 step−1 会把每帧虚报成丢 1 个）；④ 段 4 从站侧 未取先覆
+ rx_batch.pending（= OVF，与 ③ 是两个独立仪表，正常应当同步增长）；
+ ⑤ CAN 取帧撕裂 rx_torn。③④ 是本页 10Hz 抽样看到的单拍值 ，只能当下界。
+ 撕裂 / 重入是饱和 计数（到 65535 停），封顶时只能报下界。
+
+
+
实时曲线 滚动 90s · 紫 = 实测 PDO Hz（左轴） · 青 = 单拍 p50 µs（右轴）
+
+
+
总线与从站 /api/status · /api/native/health · state.head
+
+
窗口自适应：win_bases 是当前收缩出的接收窗口基址数，
+ 单槽刷新 = period_ms × win_bases.length；开机首圈先扫全表（6 拍）。
+
+
+
桥接计数 state.bridge · 桥全局，无按通道拆分 / 无 TEC·REC
+
+
过程数据里没有逐通道 TEC/REC —— 没有的就显示 —，
+ 不编 0 让"总线正常"和"没这个数据"长得一模一样。
+
+
+
收到的东西 ① 电机状态槽 state.motors[] · 48B STATUS 解码（DBC Motor_CAN_FD v0.74）
+
+
逐槽原始 hex（48B 全长，16B 一行、多列排）
+ —
+
+
「速度参考」= RT_Velocity，是内部斜坡参考值 而非实测转速，
+ 不能拿去回答"三轴实际是否同步"。InitStatus 显示 — 表示这条链路没送
+ B32-35 —— 补 0 会把"无数据"显示成"初始化正常"。
+
+
收到的东西 ② 批量接收区 rxraw.parsed（与下面槽表同一拍快照）· 段 4 覆盖窗口去重后的事件流
+
+ 本拍新帧 / seq 空洞 / 从站 OVF —
+ 本页帧样本 —
+ 清零计数
+
+
+
+ 覆盖窗口 5 槽原样（从站每拍整块拷出，写游标绕圈覆盖 —— 槽下标 ≠ 到达顺序 ，seq 才是顺序）。
+ 末列「本拍新上报」：是 = 这一拍第一次见到这帧，主站把它交给上层；否 = 它是上一拍就交过的旧帧
+ （窗口只留"最近 5 帧"，一帧会连续几拍都躺在里面，但不会被重复上报）。
+
+
按子码汇总（每 sub 一行的采样次数 / 最近载荷）
+
+
+
+ 同一 ID + 同一载荷出现两次不是重复显示 ：主站已按每槽 seq 去重，同一帧不会报第二次。
+ 实测 120 个窗口快照里「同 ID 且 seq 相同」0 次、「同 ID 但 seq 不同」240 次 —— 那是两帧真的各到达一次 ，
+ 只是电机空闲时 CONFIG 回报的 24B 逐字节相同，肉眼看一模一样。看下面槽表的 seq 列即可分辨。
+ "前面像少了一帧"也不是丢帧 ：① 主站每拍只把 seq 比上次大的槽交给上层，上一拍已经交过的帧
+ 仍躺在窗口里（就是末列的"否"），但不会再进上面的表；② 页面 10Hz 采样、PDO 1kHz，两次采样之间约
+ 100 拍的窗口内容已被覆盖，看不见（所以"样本数"只占全量约 0.8%，全量看 λ 那张卡）。
+ 本表能看到的丢帧判据只有两个：seq 空洞（主站侧）与 OVF（从站侧），
+ 上面 pill 里实时显示，实测同为 0；本表看不到的还有三层 —— CAN 控制器 err/TEC/REC、
+ GD32 软件 RX 队列满丢（深度 16，只在 ESP32 /can/feedback 有）、CAN 取帧撕裂
+ rx_torn。逐层对账跑 scripts/can_bridge_reconcile.py。
+ state.xrx 长帧槽，
+ 逐字节的真凭实据在下面第 ③ 区的原始窗口里。
+
+
收到的东西 ③ 原始接收窗口 + 长帧 + 逐拍环 /api/motor/rxraw · state.xrx · /api/motor/dbg（全为只读缓存拷贝）
+
+
+
+
+ layout：win=1.6.15 定长覆盖窗口 / slim=变长 / old=旧布局，解析器必须按 layout 选，
+ 否则拿变长游标解定长窗口会整片错位。
+
+
xrx 长帧完整载荷（kv 里那行是截断的，这里不截）
+
—
+
+
+
+
+
逐拍环未开启。开启后实测每次约 160KB（后端 RING=2048 拍），
+ 只在需要看抖动分布时开，别常年挂着 —— 仪表本身也是负载。
+
+
+
+
覆盖窗口原始字节 —— 按 44B/槽分组：头 12B（ch dlc flags pad ‖ id LE ‖ seq LE）+ 数据 32B（两行）
+
—
+
+
+
电机控制台 — Motor CAN FD Console
@@ -401,7 +402,7 @@
-
+
diff --git a/scripts/build_ecat_report_html.py b/scripts/build_ecat_report_html.py
new file mode 100644
index 0000000..301e4b0
--- /dev/null
+++ b/scripts/build_ecat_report_html.py
@@ -0,0 +1,677 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""把扫描数据渲染成自包含 HTML 调研报告（图表为内联 SVG）。
+
+为什么手写 SVG 而不用 matplotlib：产出的 HTML 要**单文件自包含**，不引外部
+JS/CSS/图片依赖，也不能要求阅读者装 matplotlib；SVG 直接内嵌，浏览器打开即看，
+且数值标注可以精确控制位置（matplotlib 的 annotate 在这类密集小图上很容易压线）。
+
+纪律同 plot_ecat_sweep.py：
+ * 图里画的**只有实测点**。模型/参考一律用虚线并写明"外推"，不与实测混同。
+ * 无功效的轴（整段恒零）不画成"平坦的零线"，改成功效声明块。
+ * 从站停报的档位（wkc 不达预期 / load_perm 为残值）一律排除出图，并在页面上
+ 显式列出被排除的档位与原因 —— 空图比假图诚实。
+
+用法：
+ python scripts/build_ecat_report_html.py
+ python scripts/build_ecat_report_html.py --out ECAT_RATE_TRADEOFF.html
+"""
+import argparse
+import html
+import json
+import math
+import os
+import sys
+from collections import OrderedDict
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+sys.path.insert(0, HERE)
+from ecat_sweep_common import post # noqa: E402 （仅用于探测 native 是否在跑）
+
+import csv as _csv
+
+
+# ───────────────────────────────────────────────────────── 数据读取 ─────────────────────────────────────────────────────────
+def read_csv(path):
+ if not os.path.isfile(path):
+ return []
+ rows = []
+ for r in _csv.DictReader(open(path, encoding="utf-8")):
+ def num(k):
+ v = r.get(k, "")
+ try:
+ return float(v) if v not in ("", "None", None) else None
+ except ValueError:
+ return None
+ rows.append({
+ "period_ms": float(r["period_ms"]),
+ "hz_set": 1000.0 / float(r["period_ms"]),
+ "hz_meas": num("measured_hz"),
+ "achieve": num("achieve_pct"),
+ "load": num("gd32_load_pct"),
+ "load_samples": num("gd32_load_samples"),
+ "load_stale": num("gd32_load_stale"),
+ "host": num("host_cpu_1core_pct"),
+ "lambda_": num("can_lambda_per_s"),
+ "drop": num("drop_per_s"),
+ "drop_pct": num("drop_pct_of_lambda"),
+ "skip": num("skip_per_s"),
+ "wkc": r.get("wkc"), "exp_wkc": r.get("expected_wkc"),
+ "wr": r.get("wr"), "hijacked": r.get("hijacked"),
+ "pid_stable": r.get("pid_stable"),
+ "online_frac": num("online_frac"),
+ "untrusted": num("drop_untrusted"),
+ "path": path})
+ rows[-1]["hz"] = rows[-1]["hz_meas"] or rows[-1]["hz_set"]
+ rows[-1]["wkc_ok"] = rows[-1]["wkc"] == rows[-1]["exp_wkc"]
+ rows.sort(key=lambda r: r["hz"])
+ return rows
+
+
+# ───────────────────────────────────────────────────────── SVG 图元 ─────────────────────────────────────────────────────────
+def _ticks(lo, hi, n=5):
+ """线性刻度：给一组整齐步进的刻度值。"""
+ if hi <= lo:
+ return [lo]
+ raw = (hi - lo) / max(1, n - 1)
+ mag = 10 ** math.floor(math.log10(raw)) if raw else 1
+ step = mag * next((k for k in (1, 2, 2.5, 5, 10) if raw / mag <= k), 10)
+ v = math.ceil(lo / step) * step
+ out = []
+ while v <= hi + 1e-9:
+ out.append(round(v, 6))
+ v += step
+ return out
+
+
+def _log_ticks(lo, hi):
+ out = []
+ for e in range(int(math.floor(math.log10(lo))), int(math.ceil(math.log10(hi))) + 1):
+ for m in (1, 2, 5):
+ v = m * (10 ** e)
+ if lo - 1e-9 <= v <= hi + 1e-9:
+ out.append(v)
+ return out or [lo]
+
+
+def _lab(v):
+ if v is None:
+ return "n/a"
+ a = abs(v)
+ if a >= 1000:
+ return f"{v:,.0f}"
+ if a >= 100:
+ return f"{v:.0f}"
+ if a >= 10:
+ return f"{v:.1f}"
+ if a >= 1:
+ return f"{v:.1f}" if v != int(v) else str(int(v))
+ return f"{v:.2f}" if a else "0"
+
+
+class Svg:
+ """极简 SVG 累加器。"""
+
+ def __init__(self, w, h):
+ self.w, self.h, self.b = w, h, []
+ self.L, self.R = 64, w - 16
+ self.T, self.B = 26, h - 42
+
+ def scale(self, xlog=False, ylog=False, xr=None, yr=None):
+ self.xlog, self.ylog = xlog, ylog
+ self.xr, self.yr = xr, yr
+
+ def X(self, v):
+ if self.xlog:
+ lo, hi = self.xr
+ return self.L + (math.log10(max(v, 1e-9)) - math.log10(lo)) / \
+ (math.log10(hi) - math.log10(lo)) * (self.R - self.L)
+ lo, hi = self.xr
+ return self.L + (v - lo) / (hi - lo or 1) * (self.R - self.L)
+
+ def Y(self, v):
+ if self.ylog:
+ lo, hi = self.yr
+ return self.B - (math.log10(max(v, 1e-9)) - math.log10(lo)) / \
+ (math.log10(hi) - math.log10(lo)) * (self.B - self.T)
+ lo, hi = self.yr
+ return self.B - (v - lo) / (hi - lo or 1) * (self.B - self.T)
+
+ def grid(self, xt, yt, xf=_lab, yf=_lab):
+ for v in yt:
+ try:
+ y = self.Y(v)
+ except ValueError:
+ continue
+ if not (self.T - 1 <= y <= self.B + 1):
+ continue
+ self.b.append(f'
')
+ self.b.append(f'
{html.escape(yf(v))} ')
+ for v in xt:
+ x = self.X(v)
+ if not (self.L - 1 <= x <= self.R + 1):
+ continue
+ self.b.append(f'
')
+ self.b.append(f'
{html.escape(xf(v))} ')
+
+ def axes(self, xl, yl):
+ self.b.append(f'
')
+ self.b.append(f'
')
+ self.b.append(f'
{html.escape(xl)} ')
+ self.b.append(f'
{html.escape(yl)} ')
+
+ def line(self, pts, color, dash=None, wid=2.4, marker="circle", r=4.2,
+ label=None, lab_dxy=(9, -9)):
+ live = [(x, y) for x, y in pts if x is not None and y is not None]
+ if not live:
+ return
+ if len(live) > 1:
+ d = " ".join(f"{'M' if i == 0 else 'L'}{self.X(x):.1f},{self.Y(y):.1f}"
+ for i, (x, y) in enumerate(live))
+ da = f' stroke-dasharray="{dash}"' if dash else ""
+ self.b.append(f'
')
+ for i, (x, y) in enumerate(live):
+ cx, cy = self.X(x), self.Y(y)
+ if marker == "circle":
+ self.b.append(f'
')
+ elif marker == "square":
+ self.b.append(f'
')
+ elif marker == "tri":
+ p = f"{cx:.1f},{cy-r-1:.1f} {cx-r-1:.1f},{cy+r:.1f} {cx+r+1:.1f},{cy+r:.1f}"
+ self.b.append(f'
')
+ if label and i == len(live) - 1:
+ dx, dy = lab_dxy
+ self.b.append(f'
{html.escape(label)} ')
+
+ def vtext(self, x, y, txt, color="#374151", anchor="middle", size=10.5, weight="400"):
+ self.b.append(f'
{html.escape(txt)} ')
+
+ def legend(self, entries, x=None, y=None):
+ x = self.L + 10 if x is None else x
+ y = self.T + 12 if y is None else y
+ for i, (c, t, dash) in enumerate(entries):
+ yy = y + i * 17
+ da = f' stroke-dasharray="{dash}"' if dash else ""
+ self.b.append(f'
')
+ self.b.append(f'
')
+ self.b.append(f'
{html.escape(t)} ')
+
+ def hline(self, v, color="#9ca3af", dash="3,3", wid=1.2):
+ y = self.Y(v)
+ if not (self.T - 1 <= y <= self.B + 1):
+ return
+ self.b.append(f'
')
+
+ def band(self, x0, x1, color="#ef4444", alpha=0.07):
+ a, b = self.X(x0), self.X(x1)
+ self.b.append(f'
')
+
+ def note(self, txt, x=None, y=None, color="#6b7280", anchor="start", size=10.5):
+ x = self.L + 6 if x is None else x
+ y = self.B - 8 if y is None else y
+ self.b.append(f'
{html.escape(txt)} ')
+
+ def finish(self, title, sub=""):
+ head = ""
+ if sub:
+ head = f'
' \
+ f'{html.escape(sub)} '
+ return (f'
'
+ f'{head}{"".join(self.b)} ')
+
+
+# ───────────────────────────────────────────────────────── 具体图 ─────────────────────────────────────────────────────────
+C_RED, C_BLUE, C_GREEN, C_PINK, C_ORANGE, C_GREY, C_PURPLE = (
+ "#dc2626", "#2563eb", "#16a34a", "#db2777", "#ea580c", "#6b7280", "#7c3aed")
+
+
+def chart_achieve(rows, rows_no=()):
+ s = Svg(560, 320)
+ allr = list(rows) + list(rows_no)
+ lo = min(r["hz_set"] for r in allr) * 0.7
+ hi = max(r["hz_set"] for r in allr) * 1.5
+ s.scale(True, True, (lo, hi), (lo, hi))
+ s.grid(_log_ticks(lo, hi), _log_ticks(lo, hi))
+ s.line([(lo, lo), (hi, hi)], C_GREY, dash="5,4", label=None)
+ s.line([(r["hz_set"], r["hz_meas"]) for r in rows], C_RED, marker="circle")
+ s.line([(r["hz_set"], r["hz_meas"]) for r in rows_no], C_BLUE, marker="square")
+ for r in allr: # 只标偏离点，免得满图都是 99.x%
+ if (r["achieve"] or 100) < 99.5:
+ s.vtext(r["hz_set"], r["hz_meas"] * 0.72,
+ _lab(r["achieve"]) + "%", color="#1e40af", size=9.5, anchor="middle")
+ s.legend([(C_GREY, "理想 1:1（设定率）", "5,4"),
+ (C_RED, "实测循环率 · 带 CAN", None),
+ (C_BLUE, "实测循环率 · 无 CAN", None)], x=s.L + 12, y=s.T + 46)
+ s.note("标注为达成率%（<99.5% 才标）。无 CAN 设定 >8kHz 后实测全顶在 ≈8.2kHz —— 硬上限",
+ y=s.B - 8)
+ s.axes("设定循环率 (Hz，对数)", "实测循环率 (Hz，对数)")
+ return s.finish("循环率：设定 vs 实测")
+
+
+def chart_load(rows_no, rows_wc):
+ s = Svg(620, 350)
+ allr = rows_no + rows_wc
+ lo = min(r["hz"] for r in allr) * 0.75
+ hi = max(r["hz"] for r in allr) * 1.35
+ s.scale(True, False, (lo, hi), (0, 112))
+ s.grid(_log_ticks(lo, hi), _ticks(0, 110, 6))
+ s.hline(100, color="#6b7280", dash="2,4")
+ s.b.append(f'
100% ')
+ pts_wc = [(r["hz"], r["load"]) for r in rows_wc if r["load"] is not None]
+ pts_no = [(r["hz"], r["load"]) for r in rows_no if r["load"] is not None]
+ s.line(pts_wc, C_RED, marker="circle")
+ s.line(pts_no, C_BLUE, marker="square")
+ wc_by_p = {round(r["period_ms"], 4): r for r in rows_wc}
+ for r in [x for x in rows_no if x["load"] is not None]:
+ w = wc_by_p.get(round(r["period_ms"], 4))
+ if w and w["load"] is not None:
+ # 任一侧饱和（≥95%）时 Δ 只反映夹紧，不标注
+ if (r["load"] or 0) >= 95 or (w["load"] or 0) >= 95:
+ continue
+ s.vtext(r["hz"], (r["load"] + w["load"]) / 2, f"Δ{w['load']-r['load']:.1f}pp",
+ color=C_PURPLE, size=10, weight="600")
+ s.legend([(C_RED, "带 CAN（电机在报 STATUS/CONFIG）", None),
+ (C_BLUE, "拔掉全部 CAN 设备", None)], x=s.L + 12, y=s.T + 12)
+ s.note("纵轴是固件 load_perm（只计 ISR+非延迟段，排除 SSC 轮询）—— 不是整机 CPU 占用率",
+ y=s.B - 8)
+ s.note("无 CAN 补测：0.25/0.125ms 满速；设定再快实测全顶在 ≈8.2kHz。"
+ "≥4kHz 的负载读数全是饱和值，不能定位拐点", y=s.B - 22, color="#b45309")
+ s.note("带 CAN 的 0.25/0.125ms 两点同理停在 ~99.x% —— 饱和读数不等于\"已测到上限\"",
+ y=s.B - 36, color="#b45309")
+ s.axes("实测循环率 (Hz，对数)", "从站 load_perm (%)")
+ return s.finish("从站每拍负担")
+
+
+def chart_traffic(rows):
+ """CAN 流量分层：STATUS 刷新（≈CAN 总接收）与 CONFIG 事件流。
+
+ drop 不再入图：它与 rx_count/gseq 同速增长（各约 4000/s），拿它做分子分母
+ 只会得到无意义的 100%（本次实际踩过的坑，详见第四节仪表陷阱）。
+ """
+ s = Svg(620, 350)
+ ok = [r for r in rows if (r["lambda_"] or 0) > 0]
+ if not ok:
+ s.note("无可用的 CAN 流量档位（本轮 CAN 已拔除）", y=(s.T + s.B) / 2, anchor="middle")
+ return s.finish("CAN 流量分层", "（无数据）")
+ lo = min(r["hz"] for r in ok) * 0.75
+ hi = max(r["hz"] for r in ok) * 1.35
+ s.scale(True, True, (lo, hi), (10, 20000))
+ s.grid(_log_ticks(lo, hi), _log_ticks(10, 20000))
+ s.line([(r["hz"], r["lambda_"]) for r in ok], C_GREEN, marker="circle")
+ s.band(2500, 9000)
+ s.vtext(1400, 4000 * 1.35, "带 CAN 轮：STATUS 全流量", color="#166534", size=10)
+ s.vtext(4200, 150, "塌陷", color="#b91c1c", size=11, weight="700")
+ s.legend([(C_GREEN, "CAN 总接收 rx_count ≡ STATUS 刷新 gseq（实测同为 ~4000/s）", None)],
+ x=s.L + 12, y=s.T + 12)
+ s.note("CONFIG 事件流独立实测仅约 92 帧/s（去重后），不随周期放大", y=s.B - 8)
+ s.note(">2kHz 处 STATUS 与 CONFIG 一同塌到 ~150/s", y=s.B - 22, color="#b45309")
+ s.axes("实测循环率 (Hz，对数)", "帧/s（对数）")
+ return s.finish("CAN 流量分层")
+
+
+def chart_skip(rows_no, rows_wc):
+ """skip 有真 0 档 → 对数轴画不了。做法：>0 的点画在对数区，真 0/近 0 档
+ 落到轴下方的"≈0"标记轨道，两种点含义不同、不混在同一条线上。"""
+ s = Svg(560, 320)
+ pos = [r for r in rows_no + rows_wc if (r["skip"] or 0) > 0.05]
+ zero = [r for r in rows_no + rows_wc if (r["skip"] or 0) <= 0.05]
+ if not pos:
+ s.note("本轮无 skip>0 的档位", y=(s.T + s.B) / 2, anchor="middle")
+ return s.finish("主站错拍率", "（无数据）")
+ lo = min(r["hz"] for r in pos) * 0.75
+ hi = max(r["hz"] for r in pos) * 1.35
+ vlo, vhi = max(min(r["skip"] for r in pos), 0.05), max(r["skip"] for r in pos) * 3
+ s.T -= 6
+ s.scale(True, True, (lo, hi), (vlo, vhi))
+ s.grid(_log_ticks(lo, hi), _log_ticks(vlo, vhi))
+ s.line([(r["hz"], r["skip"]) for r in rows_wc if (r["skip"] or 0) > 0.05],
+ C_RED, marker="circle")
+ s.line([(r["hz"], r["skip"]) for r in rows_no if (r["skip"] or 0) > 0.05],
+ C_BLUE, marker="square")
+ for r in zero: # ≈0 轨道：贴在轴下方，明确不属于对数刻度
+ y = s.B + 4
+ mk = ("circle" if r in rows_wc else "square")
+ col = C_RED if r in rows_wc else C_BLUE
+ cx = s.X(r["hz"])
+ s.b.append(f'
')
+ ent = [(C_RED, "带 CAN", None), (C_BLUE, "无 CAN", None)]
+ if zero: # 没有真 0 档时不要挂一条对不上号的图例
+ ent.append(("#9ca3af", "▽ = skip ≤0.05 次/s（真 0 档，不在对数刻度上）", None))
+ s.legend(ent, x=s.L + 12, y=s.T + 12)
+ s.note("skip=错过的网格拍。0.0625ms 设定档 15720 次/s：实测率被钉在 8.2kHz，"
+ "错拍数随设定率线性涨", y=s.B - 10)
+ s.axes("实测循环率 (Hz，对数)", "主站错拍 skip_total (次/s，对数)")
+ return s.finish("主站错拍率")
+
+
+# ───────────────────────────────────────────────────────── 页面组装 ─────────────────────────────────────────────────────────
+PAGE = """
+
+
+
EtherCAT 周期速度 × 故障率 × CPU 占用 调研
+
+{body}
+
+"""
+
+
+def tbl(rows, cols, cap):
+ h = [f'
{html.escape(cap)} ' +
+ "".join(f"{html.escape(t)} " for t, _ in cols) + " "]
+ for r in rows:
+ cls = ""
+ if r.get("wkc_ok") is False:
+ cls = ' class="dead"'
+ elif r["period_ms"] == 0.5:
+ cls = ' class="best"'
+ h.append(f"" + "".join(f"{f(r)} " for _, f in cols) + "")
+ h.append("
")
+ return "".join(h)
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--nocan", default=os.path.join(ROOT, "lab", "ecat_rate_sweep_nocan.csv"))
+ ap.add_argument("--withcan", default=os.path.join(ROOT, "lab", "ecat_rate_sweep_withcan.csv"))
+ ap.add_argument("--calib", default=os.path.join(ROOT, "lab", "ecat_calib.json"))
+ ap.add_argument("--out", default=os.path.join(ROOT, "ECAT_RATE_TRADEOFF.html"))
+ a = ap.parse_args()
+
+ no, wc = read_csv(a.nocan), read_csv(a.withcan)
+ if not wc:
+ raise SystemExit(f"缺带 CAN 数据：{a.withcan}")
+ cal = json.load(open(a.calib, encoding="utf-8")) if os.path.isfile(a.calib) else {}
+ no_ok = [r for r in no if r["wkc_ok"] and r["load"] is not None]
+ wc_ok = [r for r in wc if r["wkc_ok"] and r["load"] is not None]
+ excl = [r for r in no if not r["wkc_ok"] or r["load"] is None]
+
+ # 电机刷新率：带 CAN 那轮的 8s/档定点实测。(周期文本, 实测Hz, 刷新率帧/s)
+ # 周期直接带文本 —— 由 hz 反算会浮点漂移（1000/1999 → 显示成 0.50025ms）。
+ motor = [("1", 1000.0, 4000.2), ("0.5", 1999.0, 4000.2),
+ ("0.25", 3989.0, 87.9), ("0.125", 7762.0, 147.5)]
+ # 跨轮配对必须用**设定周期**做键：两轮的实测 hz 会差零点几（1999.3 vs 1999.6），
+ # 拿四舍五入的 hz 当键会把 0.5ms 档静默丢掉，KPI 均值就跟着算错。
+ wc_by_period = {round(r["period_ms"], 4): r for r in wc_ok}
+ deltas, unm, sat = [], [], []
+ for r in no_ok:
+ w = wc_by_period.get(round(r["period_ms"], 4))
+ if w:
+ # 任一侧 load 已饱和（≥95%）时 Δ 只反映夹紧，不代表 CAN 负担
+ if (r["load"] or 0) >= 95 or (w["load"] or 0) >= 95:
+ sat.append(r["period_ms"])
+ else:
+ deltas.append(w["load"] - r["load"])
+ else:
+ unm.append(r["period_ms"])
+ if unm:
+ print("!! 无 CAN 档位在带 CAN 轮中找不到配对，已从 Δ 中排除：", unm)
+ if sat:
+ print(" 饱和档（Δ 不计入）：", sat)
+
+ b = [f'
EtherCAT 周期速度 × 故障率 × CPU 占用 调研 ',
+ f'
调研日期 {html.escape(str(cal.get("started_at","2026-09-08"))[:10])}'
+ f' · 本机 I210 + 1 个 GD32E508 从站 · 两轮扫描：带 CAN / 拔掉全部 CAN'
+ f' · 数据 lab/ecat_rate_sweep_*.csv · 生成器 '
+ f'scripts/build_ecat_report_html.py
']
+
+ b.append('
'
+ '
0.5ms 实测可用上限（2kHz，带 CAN 工况）
'
+ f'
≈{(sum(deltas)/len(deltas) if deltas else 0):.1f}pp '
+ f'CAN 每拍恒定负担（{len(deltas)} 档配对，饱和档除外）
'
+ '
4000 帧/s 电机 STATUS 实测刷新（1/0.5ms）
'
+ '
≈8.2kHz 纯 ECAT 实测硬上限（设 10/12.5/16kHz 实测均不涨）
')
+
+ b.append('
结论 '
+ '带 CAN 工况的可用上限在 0.5ms，瓶颈在从站 GD32 不在主站。 '
+ '实测可用到 0.5ms；0.25ms 起电机数据刷新率从 4000 帧/s 掉到 88 帧/s'
+ '（带 CAN 那轮当场确认 online=True，是真失效）。'
+ '可用上限只夹在 (0.25ms, 0.5ms] 内，具体位置没测到 。纯 ECAT 硬上限 ≈8.2kHz，这次测到了墙本身。 断电重启后无 CAN 补测：'
+ '0.25/0.125ms 满速通过（99.7% / 98.0%）；再往下设 0.1 / 0.08 / 0.0625ms，'
+ '实测循环率全部顶在 8196–8238Hz（达成 82.3% / 65.9% / 51.2%），'
+ 'wkc 全程 3/3、从站始终应答。墙的来历：主站单拍硬耗时 cyc p50 '
+ '各档恒 117–126µs，1/121µs ≈ 8.2kHz —— 是"每拍必须花的往返+处理时间"，'
+ '不是主机 CPU 不够（顶墙时主机 CPU 反而降到 ~85%）。原「3.5kHz 外推」'
+ '被证伪 ；首轮无 CAN 0.25ms 停报为残状态偶发，不复现。'
+ '注意"跑得快"≠"能用"：带 CAN 的可用上限仍由电机数据刷新决定（下一条）。拔 CAN 对照解开了混淆 ：未饱和三档 Δ 稳定 8.7–9.3pp，'
+ '不随频率张开 → 快档塌陷不是 CAN 挤出来的；饱和两档 Δ 收窄到 ≤0.5pp，'
+ '正是读数被夹紧的特征。主机 native CPU 恒约 100% 单核、与周期无关 （自旋恒转，'
+ '/api/motor/spin 已是空操作）。唯一的例外是带 CAN 轮的 '
+ '0.125ms 档降到 87.5% —— 因为从站已停答。无 CAN 补测里从站活着，'
+ '同档就回到 99.7%。主机 CPU 掉下来那一刻，正是从站死掉那一刻。 建议：要吃到余量就停在 0.5ms 。0.25ms 不是"更快"，是功能损坏。 '
+ ' ')
+
+ b.append('
一、曲线 ')
+ b.append('
')
+ b.append(f'
① 循环率：设定 vs 实测（主站侧）
'
+ f'{chart_achieve(wc_ok, no_ok)}
')
+ b.append(f'
② 从站每拍负担：带 CAN vs 无 CAN
'
+ f'{chart_load(no_ok, wc_ok)}
')
+ b.append(f'
③ CAN 流量分层：STATUS 刷新与 CONFIG 事件流
'
+ f'{chart_traffic(wc_ok)}
')
+ b.append(f'
④ 主站错拍率
'
+ f'{chart_skip(no_ok, wc_ok)}
')
+ b.append('
')
+
+ if excl:
+ b.append('
已从图中排除的档位：' + "；".join(
+ f"{r['period_ms']}ms（{'wkc 不达预期，从站停报' if not r['wkc_ok'] else '无有效负载读数'}"
+ f"，load_perm 为残值）" for r in excl) + '。该档已由断电重启后的补测覆盖：'
+ '无 CAN 的 0.25 / 0.125ms 两点均已满速测得（见无 CAN 表）。空图比假图诚实。
')
+
+ b.append('
名词解释（表里这几个计数器到底是什么） '
+ 'WKC（working counter，工作计数器） ：EtherCAT 报文自带一个计数器，'
+ '每有一个从站硬件处理了该报文就 +1。主站知道自己发出了几个报文，'
+ '这个应得值就是 expected_wkc（本台子 1 个从站、每拍 3 个报文 '
+ '→ 期望 3）。所以 wkc 3/3 = 从站每拍都完整处理了；'
+ 'wkc -1 = 整帧没回来（从站停答/掉 OP）；'
+ 'wr（watchdog reset）0→1 = 看门狗把总线重拉过一次。主站错拍 skip ：主站按周期给自己排了一个"节拍网格"'
+ '（ecat_native.c:776-791）。每拍做完 EtherCAT 收发后，'
+ '若发现已经错过了下一个节拍点（甚至错过多个），就把错过的个数累进 '
+ 'skip_total。它量的是"主机软件迟到"，不是从站故障 —— '
+ '1ms 档 1.1 次/s 只是偶发 OS 调度抖动；0.125ms 档 5107 次/s 说明'
+ '单拍耗时已超过两个周期，主机也在勉强。它不区分"从站慢"和"主机慢"。在线 ：电机槽位的 online 标志（槽位数据在超时窗内'
+ '持续刷新才为 true）。带 CAN 轮显示"未记录"是因为当时脚本还没有这列，'
+ '不是电机掉线 ；该轮的电机在线状况另有 8s/档定点实测佐证。 ')
+
+ b.append('
二、数据全表 ')
+ fmt = lambda v, p=1, suf="": ("n/a" if v is None else f"{v:,.{p}f}{suf}")
+ base = [("周期", lambda r: f"{r['period_ms']:g}ms"),
+ ("实测Hz", lambda r: fmt(r["hz_meas"], 0)),
+ ("达成率", lambda r: fmt(r["achieve"], 1, "%")),
+ ("从站load", lambda r: fmt(r["load"], 1, "%")),
+ ("CAN接收/s", lambda r: fmt(r["lambda_"], 0)),
+ ("skip/s", lambda r: fmt(r["skip"], 2)),
+ ("wkc", lambda r: f'{html.escape(str(r["wkc"]))}/{html.escape(str(r["exp_wkc"]))}')]
+ cols_wc = base + [("在线", lambda r: "未记录*")]
+ cols_no = base + [("在线", lambda r: fmt(r["online_frac"], 2))]
+ b.append(tbl(wc, cols_wc, "带 CAN（主结论轮）"))
+ b.append(tbl(no, cols_no, "拔掉全部 CAN 设备（补测轮）"))
+ b.append('
* 「在线」列是扫描脚本后来才加的（'
+ 'motors_online/online_frac），带 CAN 那轮跑在加列'
+ '之前 ，逐档数据没采 —— 不是全部掉线 。该轮电机在线另有'
+ ' 8s/档定点实测（「电机数据刷新率」表）佐证。
')
+ mrows = "".join(
+ f"
{lab}ms {v:,.1f} "
+ f"{'正常' if v > 3000 else '塌陷'} "
+ for lab, _hz, v in motor)
+ b.append('
电机数据刷新率（带 CAN 轮，8s/档定点实测） '
+ '周期 刷新率 帧/s 判定
'
+ '
这是唯一直接回答"用户还能不能用"的指标 —— λ 与 drop 都是中间量。'
+ '塌陷两档当场确认 online=True，故为真失效而非掉线。
')
+
+ b.append('
三、机理与上限推算 ')
+ b.append('
上限已实测：≈8.2kHz，墙是单拍 ≈121µs 的硬耗时 '
+ '
证据链 ：无 CAN 设定 0.25 → 0.0625ms 五档，实测循环率在 ≤0.125ms '
+ '（8kHz）前跟随设定，再快就被钉死在 8196–8238Hz；与此同时主站单拍耗时 '
+ 'cyc p50 在所有档位 （含 2ms 慢档）恒为 117–126µs，'
+ '1/121µs ≈ 8.2kHz，与顶墙实测严丝合缝。即：每拍有一段不可压缩的'
+ '「发送→从站处理→回读+主站处理」往返时间，周期设定小于它时就堵在这里。
'
+ '
排除项 ：不是主机 CPU 不够 —— 顶墙档主机 CPU 只有 ~85%'
+ '（自旋等待占比变小）；不是从站拒答 —— wkc 全程 3/3。'
+ 'load_perm 在 ≥4kHz 全饱和 99.x%，既定位不了拐点也喂不准模型，'
+ '早先「两项模型 load=F/T+R → 2.7/3.5kHz」的外推已被满速补测证伪，仅留档。
'
+ '
未闭合的一处 ：121µs 里主站协议栈与从站 ESC/固件各占多少，'
+ '本轮数据分不开 —— 需给固件 fillRxWindow 加直接计时'
+ '（固件改动，需批准）或抓总线帧时序才能归因。
')
+
+ b.append('
四、判据口径与踩过的坑 ')
+ b.append('
仪表陷阱（每条都实际踩过） '
+ '【本次真正翻车处】把两个同速计数器相除当成比率 —— '
+ 'drop/rx_count 都约 4000/s，相除恒得 100%，'
+ '看着像"全丢"，其实只是同速量之比 。判丢帧必须有独立 的分子与分母：'
+ '本次真正的独立证据是「STATUS 实测 4001 帧/s 正常刷新」+「CONFIG 事件流去重后'
+ '仅约 92 帧/s」，两者一起否掉了 100% 丢帧。任何"A/B 比率"指标，先问 A 与 B '
+ '是否独立计数。 从站停报后 load_perm 保留残值 —— 实测 '
+ 'age_ms=214338 时仍报 99.2%。任何取 load_perm 的脚本必须先过 '
+ 'age_ms 闸。扫描脚本已加该闸 并新增 '
+ 'gd32_load_samples/gd32_load_stale 两列。load_perm 不是整机 CPU 占用率sys/cpu_load.h 定义收窄为"中断+非延迟处理段"，'
+ '排除 SSC MainLoop 轮询；当前 free-run（dc=0）下 PDO '
+ '应用走主循环轮询路径、不计入 。所以不能把"load→100%"读成"CPU 耗尽"。OVF 每拍被 s_rxWinOvf=0 清零不能 当"无丢帧"证据'
+ '（我据此误判 bridge.drop 是死字段，结论已收回）。/api/can/cadence 只存最近 512 拍count/lost '
+ '是行数 不是每拍帧数。对应 CSV 列已从结论排除。u16 回绕 ：drop 在 4000/s 下 16 秒绕一圈 → 10Hz 逐次累加 '
+ '+ 取模，单次增量>32768 记为作废。本次作废点数=0 。「seq 值域 − distinct」不是丢帧率 ：报 99.686%，方向与 drop≈λ 一致'
+ '但分母含 STATUS 全流量，口径不干净 → 只作旁证。周期可被 localStorage 劫持 （transport-ecat.js:227-228）'
+ '→ 每档下发后回读校验。本次六档 hijacked=no、native PID 全程 29284 未变。cyc_us.late/lost 恒 0 ')
+
+ b.append('
五、可信度边界 ')
+ fw = cal.get("firmware", {})
+ b.append('
这份数据不能证明什么 '
+ f'板载固件版本无法确认 ：烧写台账 '
+ f'{html.escape(str(fw.get("ledger",""))[:16])}… 与在盘 .hex 展平哈希'
+ f'（{html.escape("、".join(str(c.get("sha256",""))[:8]+"…" for c in fw.get("candidates",[])))}）'
+ f'均不匹配 ，且产物时间戳晚于台账。→ 本报告不复用任何历史固件结论。主站在测量期间又重编过（源码 02:39 / exe 02:40，PID 起于 02:32），'
+ '无法证明运行中的进程就是盘上那份。 '
+ 'rx_torn 与 pdo_reentry 整段恒零 = 无检出功效λ≈3993/s 是本次 CAN 工况。λ 变则负担变 ，0.5ms 这个数不可外搬到别流量。 '
+ '单台单次，未做重复性与温漂验证。 '
+ ' ')
+
+ b.append('
六、下一步 ')
+ b.append('
按优先级 '
+ '查明 drop 口径（当前最高优先，唯一未决的承重项） 。'
+ '必须用一对独立可比 的计数，不能再拿同速量相除：从 WiFi 侧读固件 u32 '
+ 'rx_drop_count（app_spi_cmd_can_feedback.cpp:99）'
+ '对速率，同时用 CAN 监控页统计实收 CONFIG 帧数作分母。'
+ '只有「独立到达计数 − 独立交付计数」才是丢帧数。台子现状 ：断电重启后重新 open+start 成功（OP、wkc=3），'
+ '无 CAN 的 0.25→0.0625ms 五档补测完成，周期已自动恢复 1ms。'
+ 'CAN 设备当前处于拔除状态 ，电机数据相关指标要插回 CAN 才能再测。插回 CAN 后补测 0.35 / 0.4 / 0.5ms 三档的电机数据刷新率 ，'
+ '把带 CAN 工况的可用上限定位出来。load_perm 已被证明在快档饱和，'
+ '不能再用它定位拐点 ；扫描脚本的 '
+ 'motors_online/online_frac/'
+ 'motor_refresh_per_s 三列才是可用性判据。 '
+ '给 fillRxWindow 加直接计时，把 121µs 硬耗时里的主/从占比'
+ '拆开（属固件改动，需批准）。 '
+ '0.5ms 档实跑电机页完整命令序列 —— 本调研只测了计数器，没测界面实际可用性。 '
+ ' ')
+
+ live = ""
+ try:
+ h = post("http://127.0.0.1:8090", "/api/native/health", timeout=1.5)
+ st = post("http://127.0.0.1:8090", "/api/motor/state", timeout=1.5)
+ live = (f'native 实时：op={h.get("op")} loop={h.get("loop")} wkc={h.get("wkc")} '
+ f'period={st.get("period_ms")}ms')
+ except Exception:
+ live = "native 当前不可达（生成本报告时未连上 8090）"
+ b.append(f'')
+
+ out = a.out
+ with open(out, "w", encoding="utf-8") as f:
+ f.write(PAGE.format(body="\n".join(b)))
+ print("HTML 报告 ->", out)
+ print(f" 参与绘图：带CAN {len(wc_ok)} 档、无CAN {len(no_ok)} 档；排除 {len(excl)} 档")
+ if deltas:
+ print(f" CAN 负担 Δ：{'、'.join(f'{d:.1f}pp' for d in deltas)}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/calibrate_ecat_metrics.py b/scripts/calibrate_ecat_metrics.py
new file mode 100644
index 0000000..3a11289
--- /dev/null
+++ b/scripts/calibrate_ecat_metrics.py
@@ -0,0 +1,418 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Phase 0：ECAT 指标现场标定 —— 跑速度扫描之前，先确认每个计数器还有没有意义。
+
+为什么必须有这一步：固件与主站都在被反复改（当前 ecat_native.c 有 239 行未提交改动，
+批量区已从"变长紧排 3 帧/拍"换成 1.6.15 定长覆盖窗口），任何沿用历史结论的判据都可能
+已经失效。本脚本只认现场读数，不认文档和历史。
+
+标定八件事：
+ 1 总线能否正常起（open + motor/start）
+ 2 板上固件身份（台账 sha256 对 Release/*.hex 展平后的哈希）
+ 3 批量区实际布局（rxraw.layout，不假设）
+ 4 每拍入窗帧数分布（取 cadence 环，按拍采样，不受 HTTP 轮询速率影响）
+ 5 丢帧率的主判据：**seq 值域 − 实测 distinct seq**（对槽位顺序不敏感）
+ 对账判据：固件 OVF / bridge.drop —— 二者与主判据是否一致
+ 6 撕裂类计数器是否还有功效（rx_torn / pdo_reentry，恒零即"测不到"≠"没坏"）
+ 7 主站错拍 skip_total 与达成率、cyc_us 占空
+ 8 CPU 两口径（GD32 load_perm / 主机 native）+ 进程 PID 稳定性
+
+用法：
+ python scripts/calibrate_ecat_metrics.py
+ python scripts/calibrate_ecat_metrics.py --seconds 60 --json lab/ecat_calib.json
+ python scripts/calibrate_ecat_metrics.py --no-start # 总线没起时不自动拉起
+"""
+import argparse
+import collections
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from ecat_sweep_common import delta_saturate, delta_wrap, percentile, post
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DEFAULT_URL = "http://127.0.0.1:8090"
+
+# 1.6.15 定长覆盖窗口槽布局（对应 ecat_native.c RXB_WIN_OFF_* / 固件 ECAT_RXW_OFF_*）
+WIN_SLOT_SIZE, WIN_SLOTS = 44, 5
+WIN_OFF_FLAGS, WIN_OFF_ID, WIN_OFF_SEQ, WIN_OFF_DLC = 2, 4, 8, 1
+FLAG_WRITTEN = 0x01
+
+
+def parse_win_slots(rawhex):
+ """从 rxraw.raw 解出每个槽的 (id, seq)（跳过 bit0=未写过的空槽）。
+
+ parsed 字段不带 seq，而 seq 是这个布局的丢帧判据，必须自己解。
+ """
+ try:
+ buf = bytes.fromhex(rawhex or "")
+ except ValueError:
+ return []
+ out = []
+ for k in range(WIN_SLOTS):
+ o = k * WIN_SLOT_SIZE
+ if o + WIN_SLOT_SIZE > len(buf):
+ break
+ if not buf[o + WIN_OFF_FLAGS] & FLAG_WRITTEN:
+ continue
+ out.append((int.from_bytes(buf[o + WIN_OFF_ID:o + WIN_OFF_ID + 4], "little"),
+ int.from_bytes(buf[o + WIN_OFF_SEQ:o + WIN_OFF_SEQ + 4], "little")))
+ return out
+
+
+def window_coverage(seqmap, slot_slots=WIN_SLOTS):
+ """覆盖窗口观测到的 seq 值域 / distinct seq 数（**描述性指标，不是丢帧判据**）。
+
+ 曾用「值域 − distinct」当丢帧主判据，已被现场证伪：窗口只有 5 槽、HTTP 又只几
+ Hz，值域内根本不可能观测到全部 seq，实测报出 99.686% "丢帧"而固件 OVF=0，自相
+ 矛盾 —— 它量的是**采样覆盖率**。真丢帧主判据是 bridge.drop（固件
+ canfd.h:53 rx_drop_count，已实测确认为回绕计数器且增量与 rx_count 增量相等）。
+ """
+ out = {}
+ for fid, obs in seqmap.items():
+ if not obs:
+ continue
+ lo, hi = min(obs), max(obs)
+ out[fid] = {"n_obs": len(obs), "span": hi - lo + 1,
+ "coverage_pct": round(len(obs) / (hi - lo + 1) * 100, 3)}
+ return {"per_id": out, "ids": len(out),
+ "note": "覆盖率仅说明观测密度，不可当丢帧率引用"}
+
+
+def cadence_rings(url):
+ """按拍采到的环形缓冲：每拍 count/lost/pending/wkc 的真实分布。
+
+ HTTP 轮询只有几 Hz，用它看"每拍几帧"是欠采样；cadence 环由 native 自己在
+ 每拍写，512 拍一段，才是每拍口径的真值。
+ """
+ try:
+ c = post(url, "/api/can/cadence")
+ except Exception as e:
+ return {"error": str(e)[:120]}
+ cols, rows = c.get("cols") or [], c.get("rows") or []
+ if not rows:
+ return {"error": "环为空"}
+ ix = {k: i for i, k in enumerate(cols)}
+ def col(name):
+ return [r[ix[name]] for r in rows] if name in ix else []
+ out = {"n": len(rows), "period_ms": c.get("period_ms")}
+ for name in ("count", "lost", "pending", "backlog", "wkc"):
+ v = col(name)
+ if v:
+ out[name + "_hist"] = {str(k): n for k, n in sorted(collections.Counter(v).items())}
+ out[name + "_max"] = max(v)
+ span_ms = (col("t")[-1] - col("t")[0]) if len(rows) > 1 else 0
+ out["span_ms"] = round(span_ms, 1)
+ if span_ms > 0 and "rxcnt" in ix:
+ out["lambda_per_s"] = round((col("rxcnt")[-1] - col("rxcnt")[0]) / (span_ms / 1000.0), 1)
+ out["gseq_per_s"] = round((col("gseq")[-1] - col("gseq")[0]) / (span_ms / 1000.0), 1) \
+ if "gseq" in ix else None
+ return out
+
+
+def find_native_pid(url):
+ """反查占用该端口的进程 PID。PID 变了 = 换了实例 = 之前采的全部作废。"""
+ try:
+ out = subprocess.check_output(["netstat", "-ano", "-p", "TCP"],
+ text=True, errors="replace")
+ except Exception as e:
+ return None, f"netstat 失败: {e}"
+ port = url.rsplit(":", 1)[-1].strip("/")
+ for line in out.splitlines():
+ cols = line.split()
+ if len(cols) >= 5 and cols[0] == "TCP" and cols[3] == "LISTENING" \
+ and cols[1].endswith(":" + port):
+ return int(cols[-1]), None
+ return None, f"没人监听 :{port}"
+
+
+def firmware_identity():
+ """台账 sha256 对固件产物。
+
+ 口径必须与烧写脚本一致：`lab/_last_gd32_sha.txt` 记的是 Intel-Hex **展平成
+ APP 基址连续 blob 后**的 sha256（flash_gd32.py:91-92），不是 .hex 文件本身的哈希，
+ 产物也是 `.hex` 而非 `.bin`。按原始字节算或按 .bin 过滤会一个都扫不到，
+ 把"本可判定"误报成"无法判定"。
+
+ 即便命中也只能说明"板上镜像与当前产物同源"；产物比台账新时必然不等。
+ **不能据此推断板上正在运行的版本** —— 那要设备回报版本字。
+ """
+ shafile = os.path.join(REPO, "lab", "_last_gd32_sha.txt")
+ rec = {"ledger": None, "ledger_path": shafile, "match": None,
+ "candidates": [], "method": "hex_to_flat + sha256"}
+ if not os.path.isfile(shafile):
+ rec["error"] = "台账不存在"
+ return rec
+ with open(shafile, encoding="utf-8", errors="replace") as f:
+ rec["ledger"] = f.read().strip()
+ try:
+ from flash_gd32 import hex_to_flat
+ except Exception as e:
+ rec["error"] = f"无法复用烧写脚本的展平算法({e})"
+ return rec
+ app = os.path.abspath(os.path.join(REPO, "..", "..", "DCU_APP", "out"))
+ if not os.path.isdir(app):
+ rec["error"] = f"固件产物目录不存在: {app}"
+ return rec
+ hexes = []
+ for dirpath, _d, files in os.walk(app):
+ hexes += [os.path.join(dirpath, fn) for fn in files if fn.endswith(".hex")]
+ if not hexes:
+ rec["error"] = "DCU_APP/out 下没有 .hex 产物"
+ return rec
+ for p in sorted(hexes):
+ try:
+ d = hashlib.sha256(hex_to_flat(p)).hexdigest()
+ except (OSError, SystemExit, ValueError) as e:
+ rec["candidates"].append({"path": p, "error": str(e)[:120]})
+ continue
+ rec["candidates"].append({"path": p, "sha256": d[:16], "mtime": int(os.path.getmtime(p))})
+ if d == rec["ledger"]:
+ rec["match"] = p
+ rec["candidates"].sort(key=lambda c: -c.get("mtime", 0))
+ rec["candidates"] = rec["candidates"][:5]
+ return rec
+
+
+def ensure_loop(url, autostart, substr):
+ st = post(url, "/api/motor/state")
+ if st.get("ok") and st.get("cycles"):
+ return st, "已在跑"
+ if not autostart:
+ return st, "循环未跑（--no-start）"
+ op = post(url, "/api/open", {"substr": substr}, timeout=25.0)
+ if not op.get("ok"):
+ return op, "open 失败: %s" % (op.get("log") or [])
+ ms = post(url, "/api/motor/start", {"period_ms": st.get("period_ms") or 1.0}, timeout=40.0)
+ if not ms.get("ok"):
+ return ms, "motor/start 失败(典型=SSC 邮箱楔死): %s" % (ms.get("log") or [])
+ return post(url, "/api/motor/state"), "已自动 open+start"
+
+
+def main():
+ ap = argparse.ArgumentParser(description="ECAT 指标现场标定")
+ ap.add_argument("--url", default=DEFAULT_URL)
+ ap.add_argument("--seconds", type=float, default=30.0)
+ ap.add_argument("--hz", type=float, default=10.0,
+ help="HTTP 采样频率。u16 的 drop 在 4000/s 下 16s 就绕一圈，"
+ "采样间隔须远小于回绕周期")
+ ap.add_argument("--no-start", action="store_true")
+ ap.add_argument("--substr", default="I210")
+ ap.add_argument("--json", default=None)
+ a = ap.parse_args()
+
+ rep = {"url": a.url, "started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+ "seconds": a.seconds, "sample_hz": a.hz, "logical_cpus": os.cpu_count()}
+ pid, perr = find_native_pid(a.url)
+ rep["native_pid"], rep["native_pid_error"] = pid, perr
+ rep["firmware"] = firmware_identity()
+ rep["cadence"] = cadence_rings(a.url)
+
+ st, how = ensure_loop(a.url, not a.no_start, a.substr)
+ rep["loop_start"] = {"how": how, "ok": bool(st.get("ok"))}
+ if not st.get("ok"):
+ print("总线起不来，标定终止：", how)
+ return 2
+ rep["period_ms_set"] = st.get("period_ms")
+ rep["pcore_mask"] = st.get("pcore_mask")
+
+ proc = None
+ try:
+ import psutil
+ for p in psutil.process_iter(["name", "pid"]):
+ if p.info["pid"] == pid:
+ proc = psutil.Process(pid)
+ proc.cpu_percent(None)
+ break
+ rep["psutil"] = proc is not None
+ except Exception as e:
+ rep["psutil"] = f"不可用({e})"
+
+ interval = 1.0 / a.hz
+ seqmap = collections.defaultdict(set)
+ counts, loads, hc1, hctot, cyc_hz = collections.Counter(), [], [], [], []
+ drop_rate, rx_rate, cyc_samples = [], [], []
+ drop_untrusted = 0
+ skip_first = skip_last = None
+ torn_c = reentry_c = 0
+ torn_cap = reentry_cap = False
+ torn_prev = reentry_prev = None
+ prev = None
+ prev_t = None
+ layout = collections.Counter()
+ err = iters = 0
+ t_end = time.perf_counter() + a.seconds
+ while time.perf_counter() < t_end:
+ tc = time.perf_counter()
+ try:
+ s = post(a.url, "/api/motor/state")
+ L = post(a.url, "/api/app/load")
+ R = post(a.url, "/api/motor/rxraw")
+ D = post(a.url, "/api/motor/dbg")
+ except Exception:
+ err += 1
+ time.sleep(interval)
+ continue
+ iters += 1
+ b = s.get("bridge") or {}
+ rb = s.get("rx_batch") or {}
+ cur = {"cycles": s.get("cycles"), "drop": b.get("drop"), "rx": b.get("rx_count")}
+ if prev is not None and prev_t is not None and tc > prev_t:
+ dt = tc - prev_t
+ if cur["cycles"] is not None and prev["cycles"] is not None:
+ hz = (cur["cycles"] - prev["cycles"]) / dt
+ if 0 < hz < 1e6:
+ cyc_hz.append(hz)
+ if cur["drop"] is not None and prev["drop"] is not None:
+ d = delta_wrap(prev["drop"], cur["drop"], 16)
+ # u16 回绕计数器：单次增量逼近模数一半即说明两次采样间已绕圈，
+ # 该次不可信，只能丢弃并计数暴露采样率不足（真值下降同理）。
+ if d < 32768:
+ drop_rate.append(d / dt)
+ else:
+ drop_untrusted += 1
+ if cur["rx"] is not None and prev["rx"] is not None:
+ dr = delta_wrap(prev["rx"], cur["rx"], 32)
+ if dr < (1 << 31):
+ rx_rate.append(dr / dt)
+ for fid, sq in parse_win_slots(R.get("raw")):
+ seqmap[fid].add(sq)
+ counts[rb.get("count")] += 1
+ layout[R.get("layout")] += 1
+ if L.get("valid") and L.get("load_perm") is not None:
+ loads.append(L["load_perm"] / 10.0)
+ if L.get("rx_torn") is not None:
+ if torn_prev is not None:
+ d, capped = delta_saturate(torn_prev, L["rx_torn"])
+ torn_c += d
+ torn_cap |= capped or L["rx_torn"] >= 65535
+ torn_prev = L["rx_torn"]
+ if L.get("pdo_reentry") is not None:
+ if reentry_prev is not None:
+ d, capped = delta_saturate(reentry_prev, L["pdo_reentry"])
+ reentry_c += d
+ reentry_cap |= capped or L["pdo_reentry"] >= 65535
+ reentry_prev = L["pdo_reentry"]
+ sk = D.get("skip_total")
+ if sk is not None:
+ if skip_first is None:
+ skip_first = sk
+ skip_last = sk
+ if s.get("cyc_us"):
+ cyc_samples.append(s["cyc_us"])
+ if proc is not None:
+ try:
+ c = proc.cpu_percent(None)
+ hc1.append(c)
+ hctot.append(c / max(1, rep["logical_cpus"] or 1))
+ except Exception:
+ pass
+ prev, prev_t = cur, tc
+ time.sleep(max(0.0, interval - (time.perf_counter() - tc)))
+
+ dur = time.perf_counter() - (t_end - a.seconds)
+ stf = post(a.url, "/api/motor/state")
+ pid2, _ = find_native_pid(a.url)
+ cad = rep["cadence"]
+
+ lam = percentile(rx_rate, 0.5) or cad.get("lambda_per_s") or 0
+ dropr = percentile(drop_rate, 0.5) or 0
+ rep["dur_s"] = round(dur, 2)
+ rep["iters"] = iters
+ rep["errors"] = err
+ rep["drop_untrusted_samples"] = drop_untrusted
+ rep["period_ms_effective"] = stf.get("period_ms")
+ rep["wkc"] = [stf.get("wkc"), stf.get("expected_wkc")]
+ rep["wr_end"] = stf.get("wr")
+ rep["native_pid_end"] = pid2
+ rep["pid_stable"] = (pid == pid2)
+ rep["layout_seen"] = dict(layout)
+ rep["rx_batch_count_hist_http"] = {str(k): v for k, v in sorted(counts.items(), key=lambda x: (x[0] is None, x[0]))}
+ rep["window_coverage"] = window_coverage(seqmap)
+ rep["can_lambda_per_s"] = round(lam, 1)
+ rep["drop_per_s"] = round(dropr, 1)
+ rep["drop_pct_of_lambda"] = round(dropr / lam * 100, 2) if lam else None
+ rep["achieved_hz_p50"] = round(percentile(cyc_hz, 0.5) or 0, 1) if cyc_hz else None
+ pm = rep["period_ms_effective"] or 1
+ rep["achieve_pct"] = round((rep["achieved_hz_p50"] or 0) / (1000.0 / pm) * 100, 1)
+ rep["skip_delta"] = (skip_last - skip_first) if (skip_first is not None and skip_last is not None) else None
+ rep["skip_per_s"] = round((rep["skip_delta"] or 0) / dur, 2) if rep["skip_delta"] is not None else None
+ rep["skip_total_cum"] = skip_last
+ rep["rx_torn_delta"] = torn_c
+ rep["rx_torn_capped"] = torn_cap
+ rep["pdo_reentry_delta"] = reentry_c
+ rep["pdo_reentry_capped"] = reentry_cap
+ tot = [(c.get("total") or {}) for c in cyc_samples if c.get("total")]
+ rep["cyc_us_total"] = {"p50": percentile([t.get("p50") for t in tot if t.get("p50") is not None], 0.5),
+ "p90": percentile([t.get("p90") for t in tot if t.get("p90") is not None], 0.5),
+ "max": max([t.get("max") or 0 for t in tot], default=None)}
+ rep["cyc_us_late_lost_sample"] = tot[0] if tot else None
+ rep["gd32_load_pct_p50"] = round(percentile(loads, 0.5) or 0, 1)
+ rep["host_cpu_1core_p50"] = round(percentile(hc1, 0.5) or 0, 1)
+ rep["host_cpu_total_p50"] = round(percentile(hctot, 0.5) or 0, 2)
+
+ cap = cad.get("count_max") or 0
+ rep["per_cycle_cap_observed"] = cap
+ rep["capacity_model"] = {
+ "lambda_per_s": round(lam, 1), "per_cycle_cap": cap,
+ "drain_per_s_at_current_T": round(cap / pm * 1000, 0) if pm else None,
+ "knee_ms_if_linear": round(cap / lam * 1000, 3) if lam and cap else None,
+ "note": "cap 取自 cadence 环实测 count 峰值；拐点以扫描实测为准，此值仅供事后对照"}
+ rep["power"] = {
+ "drop_rx_queue": dropr > 0,
+ "skip_total": (rep["skip_delta"] or 0) > 0,
+ "achieve_deficit": (rep["achieve_pct"] or 100) < 99.0,
+ "rx_torn": torn_c > 0,
+ "pdo_reentry": reentry_c > 0,
+ "cadence_lost_or_pending": bool((cad.get("lost_max") or 0) or (cad.get("pending_max") or 0)),
+ "cyc_us_late_lost": bool((rep["cyc_us_late_lost_sample"] or {}).get("late")
+ or (rep["cyc_us_late_lost_sample"] or {}).get("lost"))}
+
+ print(f"=== Phase 0 标定 · {dur:.0f}s · HTTP 采样 {iters} 轮(错 {err}) @ {a.hz:g}Hz ===")
+ print(f"native PID {pid} -> {pid2} 稳定={rep['pid_stable']} 钉核 mask={rep['pcore_mask']}")
+ fw = rep["firmware"]
+ print(f"固件台账 {str(fw.get('ledger'))[:16]}… 匹配={os.path.basename(fw['match']) if fw.get('match') else fw.get('error') or '无'}")
+ print(f"周期 设定 {rep['period_ms_set']}ms -> 生效 {pm}ms 实测 {rep['achieved_hz_p50']}Hz 达成率 {rep['achieve_pct']}%")
+ print(f"批量区布局 {dict(layout)}")
+ print(f"每拍入窗帧数(cadence 环 {cad.get('n')}拍): count={cad.get('count_hist')} "
+ f"lost={cad.get('lost_hist')} pending={cad.get('pending_hist')}")
+ print(f" ↑ 每拍口径必须用环，HTTP 轮询看 count 是欠采样")
+ print(f"【主判据】λ={lam:.0f}/s drop={dropr:.0f}/s 丢帧占 λ = {rep['drop_pct_of_lambda']}%"
+ f" （drop=固件 canfd.h rx_drop_count，CAN RX 队列(kRxQueueDepth=16)满即丢）")
+ if drop_untrusted:
+ print(f" !! {drop_untrusted} 个采样点单次增量 >32768，说明已绕圈 -> 采样率不足，提高 --hz")
+ print(f" 对账: cadence lost_max={cad.get('lost_max')} pending_max={cad.get('pending_max')}"
+ f"（覆盖窗口下这两个旧计数器语义已改，仅供对账）")
+ print(f" 窗口 seq 覆盖率(非丢帧率！): {rep['window_coverage']['per_id']}")
+ print(f"rx_torn +{torn_c}(封顶={torn_cap}) pdo_reentry +{reentry_c}(封顶={reentry_cap})")
+ print(f"主站错拍 skip_total Δ={rep['skip_delta']} = {rep['skip_per_s']}/s（累计 {rep['skip_total_cum']}）")
+ print(f"cyc_us.total p50={rep['cyc_us_total']['p50']} p90={rep['cyc_us_total']['p90']} "
+ f"max={rep['cyc_us_total']['max']} -> 占空 {round((rep['cyc_us_total']['p50'] or 0)/(pm*1000)*100,1)}%")
+ print(f"CPU: GD32 load_perm {rep['gd32_load_pct_p50']}% 主机 native 单核 {rep['host_cpu_1core_p50']}% "
+ f"/ 整机 {rep['host_cpu_total_p50']}%（逻辑核 {rep['logical_cpus']}）")
+ print(f"wkc {rep['wkc']} wr={rep['wr_end']}")
+ km = rep["capacity_model"]
+ print(f"\n容量对照: cap={km['per_cycle_cap']} λ={km['lambda_per_s']:.0f} "
+ f"当前档排空 {km['drain_per_s_at_current_T']} 线性外推拐点 {km['knee_ms_if_linear']}ms"
+ f"（仅事后对照，拐点以扫描实测为准）")
+ dead = [k for k, v in rep["power"].items() if not v]
+ live = [k for k, v in rep["power"].items() if v]
+ print("有功效:", live or "无")
+ if dead:
+ print(f"无功效: {dead}")
+ print(" → 这些轴在报告里只能写「本次测不到」，**禁止**写成「无故障/已修好」")
+
+ if a.json:
+ os.makedirs(os.path.dirname(os.path.abspath(a.json)), exist_ok=True)
+ with open(a.json, "w", encoding="utf-8") as f:
+ json.dump(rep, f, ensure_ascii=False, indent=2)
+ print(f"\n标定结果: {a.json}")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/can_bridge_reconcile.py b/scripts/can_bridge_reconcile.py
new file mode 100644
index 0000000..aba4aff
--- /dev/null
+++ b/scripts/can_bridge_reconcile.py
@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""逐层对账 CAN→ECAT：谁在丢帧、丢多少、哪一层丢的。
+
+动机：面板上的 `drop`（段1/段3"最近一帧"单格被未取先覆的帧数）恒等于 λ − 实测Hz，
+是个会自证的数，拿它说"丢帧"没有说服力。真丢帧要逐层各数一遍：
+
+ L1 CAN 控制器 err/TEC/REC —— ESP32 /can/feedback
+ L2 CAN 软件队列 rx_drop_count（深度 16 满则丢） —— 只在 ESP32 侧有，过程数据里没有
+ L3 ECAT 桥 bridge.rx_count（进桥帧数）
+ L4 段4 覆盖窗口 s_rxWinSeq 每入窗帧 +2 → Δmax_seq/2 是**精确**的入窗帧数
+ L5 主站取帧 rx_batch.lost(主站 seq 空洞) / pending(从站 OVF) —— 单拍值，
+ 由本脚本按 cycle 去重后统计，仍是**下界**（轮询看不到每一拍）
+
+L3−L4 就是 STATUS 帧数（STATUS 在入窗之前就 return，见固件
+ethercat_can_bridge.cpp:435），每节点 STATUS 率另用段2 槽 seq 步长反推
+（率 = λ/step，与轮询率无关）。
+
+只读：全部走只读诊断路由，不发报、不改周期、不碰生命周期。
+用法: python scripts/can_bridge_reconcile.py [秒数=20] [esp32_ip=192.168.31.127]
+"""
+import json
+import socket
+import sys
+import time
+import urllib.request
+
+# 控制台默认 GBK：一个字符编不出来就整轮数据白采，所以先把输出流换成 UTF-8
+try:
+ sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+except Exception:
+ pass
+
+HOST = "127.0.0.1"
+ECAT_PORTS = range(8090, 8096)
+DUR_S = float(sys.argv[1]) if len(sys.argv) > 1 else 20.0
+ESP_IP = sys.argv[2] if len(sys.argv) > 2 else "192.168.31.127"
+
+# 段4 覆盖窗口（固件 1.6.15 定长布局）：raw 从 PDO 偏移 364 起，5 槽 × 44B
+WIN_SLOT_SIZE = 44
+WIN_SLOTS = 5
+W_DLC, W_FLAGS, W_ID, W_SEQ = 1, 2, 4, 8
+
+
+def post(path, port, timeout=6.0):
+ """裸 socket + 连接阶段不设 timeout：给 urllib 传 timeout 会撞上 Windows
+ 非阻塞 connect 的 ~10ms 调度粒度，把仪表自己变成扰动源。"""
+ body = b"{}"
+ req = (f"POST {path} HTTP/1.0\r\nHost: {HOST}:{port}\r\n"
+ "Content-Type: application/json\r\n"
+ f"Content-Length: {len(body)}\r\n\r\n").encode("ascii") + body
+ s = socket.create_connection((HOST, port))
+ try:
+ s.settimeout(timeout)
+ s.sendall(req)
+ buf = bytearray()
+ while True:
+ c = s.recv(65536)
+ if not c:
+ break
+ buf.extend(c)
+ finally:
+ s.close()
+ return json.loads(bytes(buf).partition(b"\r\n\r\n")[2].decode("utf-8", "replace"))
+
+
+def find_port():
+ for p in ECAT_PORTS:
+ try:
+ if post("/api/motor/state", p, 2.0).get("ok"):
+ return p
+ except Exception:
+ continue
+ return None
+
+
+def wrap(a, b, bits):
+ return (b - a) % (1 << bits)
+
+
+def le32(bs, off):
+ return int.from_bytes(bs[off:off + 4], "little")
+
+
+def can_ch(ip):
+ """一次读全部 CAN 通道，返回 {通道号: 通道 dict}；读不到返回 None（绝不填 0）。
+
+ 不能在这里挑"rx 最大的通道"：ESP32 的 ID 快照是分页轮转的，开机初期某通道还没
+ 轮到 valid 时 rx=0，两次采样各挑各的就会挑到不同通道，差值直接变成整个累计值
+ （实测虚报成 3.0M 帧/s）。通道由调用方跨两次采样锁定。"""
+ try:
+ with urllib.request.urlopen(f"http://{ip}/can/feedback", timeout=3) as r:
+ d = json.loads(r.read().decode("utf-8"))
+ except Exception:
+ return None
+ chs = {c.get("ch"): c for c in (d.get("channels") or []) if c.get("valid")}
+ return chs or None
+
+
+def can_pick(a, b):
+ """两次采样都 valid 且 rx 最大的那个通道 —— 保证差分是同一计数器的两端。"""
+ if not a or not b:
+ return None
+ common = [k for k in a if k in b]
+ if not common:
+ return None
+ return max(common, key=lambda k: max(a[k].get("rx") or 0, b[k].get("rx") or 0))
+
+
+def win_slots(raw_hex):
+ """原始字节 → [(seq, id, dlc)]，只收 flags.bit0=1 且 seq 为偶数（整槽一致）的槽。"""
+ bs = bytes.fromhex(raw_hex)
+ out = []
+ for i in range(WIN_SLOTS):
+ p = bs[i * WIN_SLOT_SIZE:(i + 1) * WIN_SLOT_SIZE]
+ if len(p) < W_ID + 4 or (p[W_FLAGS] & 1) == 0:
+ continue
+ sq = le32(p, W_SEQ)
+ if sq == 0 or sq & 1:
+ continue
+ out.append((sq, le32(p, W_ID), p[W_DLC]))
+ return out
+
+
+def main():
+ port = find_port()
+ if not port:
+ print("找不到在监听的 ECAT native（8090..8095）—— 主站没起，先连一次总线")
+ return 1
+ print(f"native :{port} 窗口 {DUR_S:.0f}s ESP32 {ESP_IP}")
+
+ seen_cycle = None
+ seq_first = seq_last = None
+ per_id = {}
+ lost_max = pend_max = cnt_max = online = 0
+ n_state = n_raw = 0
+ t0 = time.perf_counter()
+ c0 = can_ch(ESP_IP)
+ s0 = post("/api/motor/state", port)
+ while time.perf_counter() - t0 < DUR_S:
+ st = post("/api/motor/state", port)
+ n_state += 1
+ rb = st.get("rx_batch") or {}
+ online = max(online, sum(1 for m in (st.get("motors") or []) if m.get("online")))
+ # 每拍回填清零的单拍值，只能取抽样最大值当下界
+ lost_max = max(lost_max, rb.get("lost") or 0)
+ pend_max = max(pend_max, rb.get("pending") or 0)
+ cnt_max = max(cnt_max, rb.get("count") or 0)
+ try:
+ rr = post("/api/motor/rxraw", port)
+ except Exception:
+ rr = None
+ if rr and rr.get("layout") != "win":
+ print(f"段4 布局是 {rr.get('layout')!r}，不是 1.6.15 覆盖窗口(win) —— 本脚本的解析器不适用")
+ return 2
+ # 注意：rxraw 的 count/lost/pending 是**旧布局偏移**(360/361/362)的原始字节，
+ # 在 win 布局下分别是"窗口槽数常量""环形写游标 0..4""从站 OVF"。只有第三个
+ # 有意义，前两个拿来当丢帧数会得出"lost 最大 4、84% 的拍非 0"这种假结论。
+ if rr and rr.get("cycle") != seen_cycle: # 按拍去重：同一拍重复读会重复计数
+ seen_cycle = rr.get("cycle")
+ n_raw += 1
+ for sq, _id, _dlc in win_slots(rr.get("raw") or ""):
+ seq_first = sq if seq_first is None else min(seq_first, sq)
+ seq_last = sq if seq_last is None else max(seq_last, sq)
+ per_id.setdefault(_id, set()).add(sq)
+ t1 = time.perf_counter()
+ time.sleep(0.05) # 约 20Hz：开环全速轮询会让仪表自己变成扰动源
+ t1 = time.perf_counter()
+ c1 = can_ch(ESP_IP)
+ s1 = post("/api/motor/state", port)
+ dt = t1 - t0
+
+ hz = wrap(s0.get("cycles", 0), s1.get("cycles", 0), 64) / dt
+ lam = wrap((s0.get("head") or {}).get("global_seq", 0),
+ (s1.get("head") or {}).get("global_seq", 0), 32) / dt
+ bridge = wrap((s0.get("bridge") or {}).get("rx_count", 0),
+ (s1.get("bridge") or {}).get("rx_count", 0), 32) / dt
+ win_rate = ((seq_last - seq_first) / 2 / dt) if seq_last is not None else None
+ got = sum(len(v) for v in per_id.values()) / dt
+ drop_single = wrap((s0.get("bridge") or {}).get("drop", 0),
+ (s1.get("bridge") or {}).get("drop", 0), 16) / dt
+
+ print(f"\n窗口 {dt:.2f}s 档位 {s1.get('period_ms')}ms PDO {hz:.0f}Hz "
+ f"轮询 state {n_state} 次 / rxraw 新拍 {n_raw} 次"
+ f"（{n_raw / dt:.1f} 拍/s → 抽样比 {n_raw / max(1e-9, hz * dt):.1%}）")
+ print("\n层 计数 帧/s 判读")
+ ck = can_pick(c0, c1)
+ can_rate = None
+ if ck is not None:
+ rx0 = c0[ck].get("rx") or 0
+ dr0 = c0[ck].get("drop") or 0
+ rx1 = c1[ck].get("rx") or 0
+ dr1 = c1[ck].get("drop") or 0
+ err, tec, rec, bus = (c1[ck].get(k) for k in ("err", "tec", "rec", "bus_state"))
+ can_rate = wrap(rx0, rx1, 32) / dt # rx / drop 都是 u32
+ can_drop = wrap(dr0, dr1, 32) / dt
+ print(f"L1 CAN err/TEC/REC/bus —— {err}/{tec}/{rec}/{bus}"
+ f" {'干净' if not (err or tec or rec) else '有错'} (通道 {ck})")
+ print(f"L2 CAN 软件队列满丢 {can_drop:10.0f} "
+ f"{'0 = 队列(深度16)没满过' if not can_drop else '在丢帧，查从站取帧速率'}")
+ print(f"L3 CAN 控制器收帧 {can_rate:10.0f}")
+ else:
+ print("L1/L2/L3 ESP32 /can/feedback 不可读 —— 不填 0，这三层本轮无数据")
+ print(f"L3 ECAT桥 进桥 rx_count {bridge:10.0f}")
+ if can_rate is not None:
+ gap = can_rate - bridge
+ print(f" CAN - 桥 {gap:10.0f} "
+ f"{'≈0 两层同域' if abs(gap) < bridge * 0.01 else '差额：到了 CAN 但没进桥计数'}")
+ print(f"L4 段4窗 入窗帧 Δseq/2 {win_rate if win_rate else float('nan'):10.0f}"
+ f" 精确值（窗口 seq 是累计计数器，与抽样率无关）")
+ print(f" STATUS（=桥−入窗） {bridge - (win_rate or 0):10.0f} "
+ f"STATUS 在入窗前就 return，固件 ethercat_can_bridge.cpp:435")
+ print(f" 本脚本见到的唯一入窗帧 {got:10.0f} 下界：轮询只覆盖 {n_state / max(1e-9, hz * dt):.1%} 的拍")
+ print(f"L5 单拍值 主站空洞 lost 最大 {lost_max:10.0f} 每拍回填清零，只能当下界")
+ print(f" 从站 OVF pending 最大 {pend_max:10.0f} 同上")
+ print(f" 主站上报 count 最大 {cnt_max:10.0f} 单拍上报帧数（窗口 5 槽）")
+ print(f"-- 段1/段3 单格 drop {drop_single:10.0f} "
+ f"恒等于 λ−Hz = {lam - hz:.0f}，不参与判丢帧")
+
+ if online:
+ print(f"\n按节点拆（在线 {online} 个）: STATUS ≈ {(bridge - (win_rate or 0)) / online:6.0f} 帧/s/节点，"
+ f"入窗 ≈ {(win_rate or 0) / online:6.0f} 帧/s/节点")
+ if per_id:
+ print("\n入窗帧按 ID（唯一 seq 计数，下界）:")
+ for _id in sorted(per_id):
+ n = len(per_id[_id])
+ print(f" 0x{_id:08X}: {n / dt:8.0f} 帧/s 子码 0x{_id & 0xFFFF:04X}")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/ecat_sweep_common.py b/scripts/ecat_sweep_common.py
new file mode 100644
index 0000000..960b921
--- /dev/null
+++ b/scripts/ecat_sweep_common.py
@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""ECAT native 主站的采样公共件：HTTP 调用 + 计数器差分。
+
+固件 1.6.15 覆盖窗口布局下，各计数器的口径并不一致，混用会静默算错：
+ * bridge.rx_count / head.global_seq —— u32，**回绕**
+ * bridge.drop —— u16，**回绕**；段 1/段 3 单格"未取先覆"数，恒等于 λ − 实测Hz
+ * rx_batch.lost(主站按 seq 空洞算的单拍丢帧) / pending(从站窗口 OVF) ——
+ **每拍回填清零的单拍值**（ecat_native.c:461），不是累计计数：拿它做差得速率恒为 0，
+ 拿轮询见过的最大值当总丢帧也只是下界
+ * app/load.rx_torn / pdo_reentry —— u16，**饱和**（到 65535 就停，不是回绕）
+饱和量按回绕算会把"已封顶"误报成"发生了 6 万多次"，故二者分开处理。
+"""
+import json
+import urllib.request
+
+SATURATE = 65535 # u16 饱和计数的顶
+
+
+def post(url, path, body=None, timeout=6.0):
+ req = urllib.request.Request(
+ url + path, data=json.dumps(body or {}).encode("utf-8"),
+ headers={"Content-Type": "application/json"}, method="POST")
+ with urllib.request.urlopen(req, timeout=timeout) as r:
+ return json.loads(r.read().decode("utf-8"))
+
+
+def delta_wrap(a, b, bits=16):
+ """回绕计数器的增量。真值下降（重启清零）会表现为一个巨大的正数，
+ 调用方须用 sanity 上界剔除 —— 这里只负责把模数算对。"""
+ m = 1 << bits
+ return (b - a) % m
+
+
+def delta_saturate(a, b):
+ """饱和计数的增量。返回 (增量, 是否已封顶)。
+
+ 封顶后增量恒为 0 —— 读数 0 既可能是"真没发生"也可能是"早发生了但计不动"，
+ 报告里必须写成下界而不是精确值。
+ """
+ if b >= SATURATE:
+ return 0, True
+ if b < a:
+ return 0, False # 计数器被回填清零：该窗口作废，不臆造增量
+ return b - a, False
+
+
+def percentile(values, q):
+ """小样本分位数（最近邻）。不引 numpy，采样窗口只有百来个点。"""
+ if not values:
+ return None
+ s = sorted(values)
+ i = min(len(s) - 1, max(0, int(round(q * (len(s) - 1)))))
+ return s[i]
diff --git a/scripts/plot_ecat_nocan_vs_can.py b/scripts/plot_ecat_nocan_vs_can.py
new file mode 100644
index 0000000..00b70fd
--- /dev/null
+++ b/scripts/plot_ecat_nocan_vs_can.py
@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""对比「带 CAN」与「拔掉 CAN」两轮扫描，分离每拍负担的组成。
+
+用途：判断 ECAT 速度上限究竟由什么决定 —— 是每拍都要跑的过程数据交换
+（PDI_Isr / PDO 填充，与 CAN 无关），还是叠加在上面的 CAN 收发处理。
+
+判读：两条 load 曲线的**垂直间距**就是 CAN 在每个周期档上贡献的负担。
+间距近似恒定 → CAN 负担与周期无关，上限由纯 ECAT 那条线单独决定；
+间距随频率张开 → CAN 负担本身也按拍放大，提速会被 CAN 先拖垮。
+
+用法：
+ python scripts/plot_ecat_nocan_vs_can.py \
+ --nocan lab/ecat_rate_sweep_nocan.csv --withcan lab/ecat_rate_sweep_withcan.csv
+"""
+import argparse
+import csv
+import os
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+from matplotlib import rcParams
+
+rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"]
+rcParams["axes.unicode_minus"] = False
+rcParams["figure.dpi"] = 130
+
+
+def load(path):
+ out = {}
+ for r in csv.DictReader(open(path, encoding="utf-8")):
+ def num(k):
+ v = r.get(k, "")
+ try:
+ return float(v) if v not in ("", "None", None) else None
+ except ValueError:
+ return None
+ out[float(r["period_ms"])] = {
+ "load": num("gd32_load_pct"), "hz": num("measured_hz"),
+ "achieve": num("achieve_pct"), "skip": num("skip_per_s"),
+ "wkc_ok": r.get("wkc") == r.get("expected_wkc"),
+ "stale": num("gd32_load_stale")}
+ return out
+
+
+def main():
+ ap = argparse.ArgumentParser(description="带 CAN vs 无 CAN 每拍负担分离")
+ d = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+ ap.add_argument("--nocan", default=os.path.join(d, "lab", "ecat_rate_sweep_nocan.csv"))
+ ap.add_argument("--withcan", default=os.path.join(d, "lab", "ecat_rate_sweep_withcan.csv"))
+ ap.add_argument("--out", default=None)
+ a = ap.parse_args()
+
+ no, wc = load(a.nocan), load(a.withcan)
+ common = sorted(set(no) & set(wc), reverse=True)
+ out = a.out or os.path.splitext(a.nocan)[0] + "_vs.png"
+
+ fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(13.0, 5.2))
+ fig.suptitle("每拍负担分离：拔掉 CAN 之后，速度上限由谁决定", fontsize=13)
+
+ # 只取两侧 wkc 都达预期的档位：从站停报后 load_perm 会保留**残值**
+ # （实测楔死后仍报 99.2%），留在图里会被当成真数据、把 Δ 趋势带歪。
+ healthy = [t for t in common if no[t]["wkc_ok"] and wc[t]["wkc_ok"]]
+ dropped = [t for t in common if t not in healthy]
+ T = [t for t in healthy if no[t]["load"] is not None and wc[t]["load"] is not None]
+ if not T:
+ raise SystemExit("两文件没有可对比的公共档位（且均需有效负载读数）")
+ Hz = [1000.0 / t for t in T]
+ ax1.plot(Hz, [wc[t]["load"] for t in T], "o-", color="#d62728", lw=2,
+ label="带 CAN（电机在报 STATUS/CONFIG）")
+ ax1.plot(Hz, [no[t]["load"] for t in T], "s-", color="#1f77b4", lw=2,
+ label="拔掉全部 CAN 设备")
+ for t, h in zip(T, Hz):
+ ax1.annotate(f"Δ={wc[t]['load'] - no[t]['load']:.1f}pp", (h, no[t]["load"]),
+ textcoords="offset points", xytext=(0, -14), ha="center", fontsize=8)
+ ax1.axhline(100, ls=":", color="k", lw=1)
+ ax1.set_xscale("log"); ax1.set_ylim(0, 112)
+ ax1.set_xlabel("循环率 (Hz)"); ax1.set_ylabel("load_perm (%)")
+ ax1.set_title("从站每拍负担（load_perm 口径，非整机 CPU）")
+ ax1.legend(fontsize=8, loc="upper left"); ax1.grid(alpha=0.3)
+
+ ax2.plot(Hz, [wc[t]["skip"] for t in T], "o-", color="#d62728", lw=2, label="带 CAN")
+ ax2.plot(Hz, [no[t]["skip"] for t in T], "s-", color="#1f77b4", lw=2, label="无 CAN")
+ ax2.set_xscale("log"); ax2.set_yscale("symlog", linthresh=1)
+ ax2.set_xlabel("循环率 (Hz)"); ax2.set_ylabel("skip_total (次/s)")
+ ax2.set_title("主站错拍率：CAN 有无对节拍精度的影响")
+ ax2.legend(fontsize=8, loc="upper left"); ax2.grid(alpha=0.3)
+
+ fig.tight_layout(rect=(0, 0, 1, 0.94))
+ fig.savefig(out, bbox_inches="tight")
+ print(f"对比图 -> {out}\n")
+ print(f"{'周期':>7} {'循环率':>8} {'带CAN':>8} {'无CAN':>8} {'Δpp':>7} {'带CAN skip':>11} {'无CAN skip':>11}")
+ for t in T:
+ print(f"{t:>6}ms {1000.0/t:>7.0f}Hz {wc[t]['load']:>7.1f}% {no[t]['load']:>7.1f}% "
+ f"{wc[t]['load']-no[t]['load']:>6.1f} {wc[t]['skip']:>10.2f} {no[t]['skip']:>10.2f}")
+ skipped = [t for t in common if t not in T]
+ if skipped:
+ print(f"\n排除的档位：{skipped}")
+ for t in skipped:
+ why = []
+ if not no[t]["wkc_ok"]:
+ why.append("无CAN 侧 wkc 不达预期（从站已停报，load_perm 为残值）")
+ if not wc[t]["wkc_ok"]:
+ why.append("带CAN 侧 wkc 不达预期")
+ if no[t]["load"] is None or wc[t]["load"] is None:
+ why.append("缺有效负载读数")
+ print(f" {t}ms: " + "; ".join(why or ["未知"]))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/plot_ecat_sweep.py b/scripts/plot_ecat_sweep.py
new file mode 100644
index 0000000..2cb5e96
--- /dev/null
+++ b/scripts/plot_ecat_sweep.py
@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""把 sweep_ecat_rate.py 的 CSV 画成「速度-故障率-CPU」四联图。
+
+画法纪律（对应调研报告的判据要求）：
+ * 先画实测点/折线，参考模型只在事后叠加，且线型与实测明显区分 —— 避免拿模型
+ 冒充数据。容量参考线用的是**本次实测**的每拍入窗帧数，不是源码常量。
+ * 无功效的轴（整段恒零）不画成"平坦的零线" —— 那会被读成"没问题"。改为在子图
+ 里写明"本次测不到"，并保留轴刻度供对比。
+ * 横轴用频率（=1000/周期）而不是周期：物理上限在时间轴上是反比双曲线，
+ 在频率轴上是直线，读者不会看错方向。
+"""
+import argparse
+import csv
+import os
+import sys
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+from matplotlib import rcParams
+
+rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"]
+rcParams["axes.unicode_minus"] = False
+rcParams["figure.dpi"] = 130
+
+MX = MY = []
+DEFAULT_CSV = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ "lab", "ecat_rate_sweep.csv")
+
+
+def load(path):
+ rows = list(csv.DictReader(open(path, encoding="utf-8")))
+ if not rows:
+ raise SystemExit(f"{path} 没有数据行")
+ for r in rows:
+ r["hz_set"] = 1000.0 / float(r["period_ms"])
+ for k in ("measured_hz", "achieve_pct", "can_lambda_per_s", "drop_per_s",
+ "drop_pct_of_lambda", "skip_per_s", "torn_per_s", "reentry_per_s",
+ "gd32_load_pct", "host_cpu_1core_pct", "cyc_p50_us", "cyc_p90_us",
+ "cyc_duty_pct", "lambda_u16_crosscheck", "rxb_count_mode_per_cycle"):
+ v = r.get(k, "") # 旧 CSV 可能没有后加的列
+ try:
+ r[k] = float(v) if v not in ("", "None", None) else None
+ except ValueError:
+ r[k] = None
+ # CPU / 丢帧曲线的真正自变量是**实测**循环率：快档未必设得到设定值，
+ # 拿设定周期当横轴会把点画到错的位置上。
+ r["hz"] = r["measured_hz"] or r["hz_set"]
+ rows.sort(key=lambda r: r["hz"])
+ return rows
+
+
+def ys(rows, key):
+ """取该列的 y 值序列；x 一律用 xs()。col() 曾按 (y, x) 顺序返回，调用处按
+ (x, y) 解包会把 y 当横轴画（实测把 CPU% 画成了 2–100Hz），故改名并各自独立。"""
+ return [r[key] for r in rows]
+
+
+def xs(rows):
+ return [r["hz"] for r in rows]
+
+
+def no_power(rows, key):
+ """该轴整段恒零 → 无检出功效，不能画成"平坦的零线"。"""
+ vals = [r[key] for r in rows if r[key] is not None]
+ return (not vals) or max(vals) == 0
+
+
+def annotate_flags(ax, rows):
+ """把采样作废/劫持/PID 易主这些不可信标记写在图边上，避免读者信了坏数据。"""
+ bad = []
+ for r in rows:
+ if r.get("hijacked") not in ("no", None):
+ bad.append(f"{r['period_ms']}ms 周期被劫持")
+ if r.get("pid_stable") != "yes":
+ bad.append(f"{r['period_ms']}ms native PID 易主")
+ if int(r.get("drop_untrusted") or 0) > 0:
+ bad.append(f"{r['period_ms']}ms drop 有 {r['drop_untrusted']} 点绕圈作废")
+ if bad:
+ ax.set_xlabel(ax.get_xlabel() + " ⚠ " + "; ".join(bad[:3]), color="darkred")
+
+
+def main():
+ ap = argparse.ArgumentParser(description="画 ECAT 速度-故障率-CPU 曲线")
+ ap.add_argument("--csv", default=DEFAULT_CSV)
+ ap.add_argument("--out", default=None, help="输出 PNG（默认与 CSV 同名 .png）")
+ a = ap.parse_args()
+
+ rows = load(a.csv)
+ out = a.out or os.path.splitext(a.csv)[0] + ".png"
+ # 电机槽 seq 前进速率：扫描后另用 8s/档定点实测（判据见报告"方法"节）。
+ # 单独测是因为它要看的是 u32 槽 seq 的秒增量，与扫描的 60s 平均口径不同源。
+ global MX, MY
+ MX = [1000.0, 1999.0, 3989.0, 7762.0]
+ MY = [4000.2, 4000.2, 87.9, 147.5]
+ fig, axes = plt.subplots(2, 2, figsize=(13.5, 9.2))
+ fig.suptitle("EtherCAT 周期速度 × 故障率 × CPU 占用（实测）", fontsize=13, y=0.985)
+
+ # ---- 1 速度达成：设定 vs 实测 ----
+ ax = axes[0][0]
+ xset = [r["hz_set"] for r in rows]
+ ax.plot([min(xset), max(xset)], [min(xset), max(xset)], "k--", lw=1,
+ label="理想 1:1（设定率）")
+ ax.plot(xset, [r["measured_hz"] for r in rows], "o-", color="#1f77b4", label="实测达成")
+ ax.set_xscale("log"); ax.set_yscale("log")
+ ax.set_xlabel("设定循环率 (Hz)"); ax.set_ylabel("实测循环率 (Hz)")
+ ax.set_title("① 循环率：设定 vs 实测（主站侧）"); ax.legend(fontsize=8, loc="upper left")
+ ax.grid(alpha=0.3, which="both")
+ txt = "达成率 " + "\n".join(
+ f"{r['period_ms']}ms: {r['achieve_pct']:.1f}%" for r in rows if r["achieve_pct"])
+ ax.annotate(txt, (0.98, 0.5), xycoords="axes fraction", fontsize=7.5,
+ ha="right", va="center")
+
+ # ---- 2 CPU 占用（两路：从站负载 vs 主机自旋）----
+ ax = axes[0][1]
+ X = xs(rows)
+ gy, hy = ys(rows, "gd32_load_pct"), ys(rows, "host_cpu_1core_pct")
+ ax.plot(X, gy, "s-", color="#d62728", lw=2, label="GD32 负载 load_perm")
+ ax.plot(X, hy, "^-", color="#7f7f7f", lw=1.4, label="主机 native（自旋，单核%）")
+ ax.axhline(95, ls=":", color="darkred", lw=1)
+ ax.annotate("95% 逼近饱和", (1.02, 95), fontsize=7.5, color="darkred", va="bottom")
+ ax.set_xscale("log"); ax.set_ylim(0, 112)
+ ax.set_xlabel("循环率 (Hz)"); ax.set_ylabel("CPU 占用 (%)")
+ ax.set_title("② CPU 占用：GD32 与主机 native 两路")
+ ax.legend(fontsize=8, loc="center left"); ax.grid(alpha=0.3)
+ sat = [r for r in rows if r["gd32_load_pct"] is not None and r["gd32_load_pct"] >= 95]
+ if sat:
+ ax.annotate(f"≥95% 的档位: {', '.join(str(r['period_ms'])+'ms' for r in sat)}",
+ (0.02, 0.04), xycoords="axes fraction", fontsize=7.5, color="darkred")
+
+ # ---- 3 故障率主曲线：功能是否还活着（电机数据刷新率）vs 从站负载 ----
+ # 横轴用对数；λ 与 drop 两条线**刻意画在一起**，用来展示"加速对丢帧无改善"
+ ax = axes[1][0]
+ ax.plot(X, ys(rows, "can_lambda_per_s"), "o-", color="#2ca02c", lw=2,
+ label="λ CAN 到帧（u32 gseq）")
+ ax.plot(X, ys(rows, "drop_per_s"), "v--", color="#e377c2", lw=2,
+ label="drop 队列满丢帧（与 λ 重合）")
+ ax.plot(MX, MY, "^-", color="#d62728", lw=2.2,
+ label="电机数据刷新率（真故障轴）")
+ ax.axvspan(2000, 4000, color="red", alpha=0.07)
+ ax.annotate("GD32 饱和 → 功能塌陷", (2550, 40), fontsize=8, color="darkred")
+ ax.set_xscale("log"); ax.set_yscale("log") # 帧/s 恒为正，用 symlog 会把轴拖到负值
+ ax.set_ylim(10, 20000)
+ ax.set_xlabel("实测循环率 (Hz)"); ax.set_ylabel("帧/s")
+ ax.set_title("③ 故障率：加速对丢帧毫无改善，且 >2kHz 直接打断功能")
+ ax.grid(alpha=0.3, which="both")
+ ax.legend(fontsize=7.5, loc="lower left")
+
+ # ---- 4 错拍率 + 撕裂/重入功效声明 ----
+ ax = axes[1][1]
+ ax.plot(X, ys(rows, "skip_per_s"), "D-", color="#ff7f0e", lw=2,
+ label="主站错拍 skip_total（次/s）")
+ ax.set_xscale("log"); ax.set_yscale("symlog", linthresh=1) # skip 有真 0 档，需 symlog
+ ax.set_xlabel("循环率 (Hz)"); ax.set_ylabel("次/s")
+ ax.set_ylim(-10, None)
+ ax.set_title("④ 主站错拍率，及撕裂/重入轴功效声明"); ax.legend(fontsize=8, loc="upper left")
+ ax.grid(alpha=0.3)
+ dead = [n for n, k in (("rx_torn", "torn_per_s"), ("pdo_reentry", "reentry_per_s"))
+ if no_power(rows, k)]
+ if dead:
+ ax.annotate("恒零轴（无检出功效）：" + ", ".join(dead) +
+ "\n→ 本次测不到 ≠ 无故障，不得据此判\"已修好\"\n"
+ " （且固件版本无法与在盘产物对账）",
+ (0.03, 0.06), xycoords="axes fraction", fontsize=8,
+ color="darkred", bbox=dict(fc="#fff3f3", ec="darkred", alpha=0.9))
+ annotate_flags(ax, rows)
+
+ fig.tight_layout(rect=(0, 0, 1, 0.975))
+ fig.savefig(out, bbox_inches="tight")
+ print(f"曲线图 -> {out}")
+ for r in rows:
+ def fmt(v, w=7, p=0, unit=""):
+ return f"{v:>{w}.{p}f}{unit}" if v is not None else f"{'n/a':>{w}}"
+ ld = r["gd32_load_pct"]
+ extra = (f"{ld:.1f}%" if ld is not None else "n/a")
+ if r.get("gd32_load_stale") is not None:
+ extra += f"(样{r.get('gd32_load_samples')}/废{r['gd32_load_stale']})"
+ print(f" {r['period_ms']:>6}ms {fmt(r['hz']):>7}Hz 实测{fmt(r['measured_hz'])}Hz "
+ f"GD32={extra:<18} λ={fmt(r['can_lambda_per_s'])}/s "
+ f"drop={fmt(r['drop_per_s'])}/s skip={fmt(r['skip_per_s'], 6, 1)}/s "
+ f"在线={r.get('online_frac', 'n/a')}")
+ if dead:
+ print(f" 无功效轴: {dead} —— 报告按「测不到」写")
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/repro_cmdload.py b/scripts/repro_cmdload.py
index a1dbcf1..dc96a64 100644
--- a/scripts/repro_cmdload.py
+++ b/scripts/repro_cmdload.py
@@ -62,6 +62,14 @@ lock = threading.Lock()
stats = {"sends": 0, "send_err": 0, "t0": time.time()}
torn_events = [] # (t, node, dlc, data48hex, cfg_hex)
batch_events = []
+fp_events = [] # 过滤后仍残余的指纹违规
+fp_total = [0] # （弃用，保留兼容）
+cfg_total = [0] # 监控实际消费的 CONFIG 帧总数（活性自证，非 0 才有资格谈违规=0）
+fp_after = [0] # 过滤器放行掉的违规（应为 0）
+fp_dropped = [0] # 过滤器正确丢弃的违规
+fp_falsekill = [0] # 过滤器误杀的真帧（应为 0）
+cfg_seen = {} # node -> 最近 4 个已接受 CONFIG hex（页面过滤器模拟）
+cfg_susp = {} # node -> 连续疑似计数
torn_base = [0]
@@ -86,12 +94,14 @@ def sender():
except Exception:
time.sleep(0.05)
continue
- if len(cfgs) >= 2:
- # 判别实验：两电机反相正弦 + 不同速度限，给帧内容打节点指纹。
- # node2: ang=1000+200sin, vel=200 node3: ang=1000-200sin, vel=150
+ if len(cfgs) >= 2 or time.time() - stats["t0"] > 3:
+ # 电机偶尔暂停 CONFIG 自报（如正弦轰炸后）：拿不到回声就用页面默认
+ # 限压 24.0V（u16 LE 0x5DC0 → bytes c0 5d）兜底，保证指令压力不断。
+ fallback = bytes([0x03, 0xc0, 0x5d]) + bytes(21)
s = math.sin(2.0 * math.pi * (time.time() - stats["t0"]) / 10.0)
frames = []
- for n, c in sorted(cfgs.items()):
+ for n in (2, 3):
+ c = cfgs.get(n, fallback)
ang = 1000.0 + (200.0 * s if n == 2 else -200.0 * s)
vel = 200.0 if n == 2 else 150.0
frames.append({"ch": 2, "id": hex((n << 16) | 0x0200), "dlc": 32,
@@ -162,11 +172,42 @@ def monitor():
node = (fid >> 16) & 0xFF
if (fid & 0xFFFF) == 0x0101:
cfgs[node] = d
+ cfg_total[0] += 1 # 活性自证：本拍监控实际看到的 CONFIG 帧数
if any(d[15:24]):
with lock:
if len(batch_events) < 40:
batch_events.append((round(time.time() - stats["t0"], 2),
node, d.hex()))
+ # CONFIG↔CONFIG 拼接：结构校验抓不到，用速度指纹暴露。
+ # 预热 3s：sender 尚未生效时电机回显的是页面/上一轮的 vel，
+ # 不是拼接，不能算违规。
+ if len(d) >= 15 and time.time() - stats["t0"] > 3.0:
+ vel = struct.unpack_from("
1.0
+ with lock:
+ # 模拟页面 CONFIG 跨节点拼接过滤器（transport-ecat.js
+ # dispatchFrame 同逻辑），统计过滤后残余违规与误杀。
+ hx = d.hex()
+ seen = cfg_seen.setdefault(node, [])
+ own = hx in seen
+ foreign = (not own) and any(
+ hx in s2 for n2, s2 in cfg_seen.items() if n2 != node)
+ if own or not foreign or cfg_susp.get(node, 0) >= 5:
+ cfg_susp[node] = 0
+ seen.insert(0, hx)
+ del seen[4:]
+ if viol:
+ fp_after[0] += 1
+ if len(fp_events) < 40:
+ fp_events.append((round(time.time() - stats["t0"], 2),
+ node, vel, exp_vel, hx))
+ else:
+ cfg_susp[node] = cfg_susp.get(node, 0) + 1
+ if viol:
+ fp_dropped[0] += 1
+ else:
+ fp_falsekill[0] += 1
off += 8 + w
@@ -198,6 +239,12 @@ def main():
print(f"段4 污染事件: {len(batch_events)}（前 8）")
for e in batch_events[:8]:
print(f" t={e[0]}s node={e[1]} data={e[2]}")
+ print(f"CONFIG↔CONFIG 指纹(预热后): 违规={fp_after[0]+fp_dropped[0]} "
+ f"过滤器正确丢弃={fp_dropped[0]} 过滤后残余={fp_after[0]} "
+ f"误杀真帧={fp_falsekill[0]}（样本 {len(fp_events)}）")
+ print(f"监控消费的 CONFIG 帧总数: {cfg_total[0]}（活性自证：>0 才说明违规=0 是真结果）")
+ for e in fp_events[:8]:
+ print(f" t={e[0]}s node={e[1]} vel={e[2]:.1f} 应为{e[3]:.0f} data={e[4]}")
if __name__ == "__main__":
diff --git a/scripts/sweep_ecat_rate.py b/scripts/sweep_ecat_rate.py
new file mode 100644
index 0000000..f0f5904
--- /dev/null
+++ b/scripts/sweep_ecat_rate.py
@@ -0,0 +1,335 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""PDO 周期扫描：一次跑完多档，同时记「速度达成 / 故障率 / CPU 占用」三条曲线。
+
+与 tools/ecat/rate_probe.py 的区别：rate_probe 只量速度达成率，本脚本把故障率与
+CPU 一并采齐并落成机读 CSV，供 plot_ecat_sweep.py 画曲线。速率达成部分的
+"cycles 与时刻同取同舍"纪律直接沿用 rate_probe（它注释里记过一次把 99.2% 误报
+成 96% 的实踩教训：分子取末次 cycles、分母却在循环退出后才算，dt 多出一次 sleep）。
+
+必须绕开的四件事（均由 Phase 0 现场标定确认，不是照抄文档）：
+ 1 u16 回绕：bridge.drop 在 4000/s 下 16s 绕一圈，首尾相减必错 → 高频逐次累加，
+ 且单次增量 >32768 视为绕圈、该点作废并计数暴露采样率不足。
+ 2 rx_torn / pdo_reentry 是**饱和**计数（到 65535 停），不能按回绕算 → 走
+ delta_saturate，封顶即标注"下界"而不是报精确值。
+ 3 周期能被 localStorage 劫持：任何开着的电机页标签在 ensureStarted 时会把总线
+ 周期拽回存量值（transport-ecat.js:227-228）→ 每档下发后回读 period_ms 确认
+ 真的生效，不一致就把该档标记为被劫持。
+ 4 λ（CAN 到帧率）取 u32 的 head.global_seq，不取 u16 的 bridge.rx_count。
+ 本次实测到关键分界：1ms/0.5ms 档 λ=4000/s、GD32 负载 44%/77%；0.25ms/0.125ms
+ 档 GD32 顶到 99.5%/99.3%，λ 塌到 149/s —— 从站主循环被 PDO 填充吃光、CAN 接收
+ 停摆。而主站达成率反而仍很高（0.125ms 实测 7780Hz），所以瓶颈在从站不在主站：
+ 只看达成率会把这条线整个看漏。
+
+用法：
+ python scripts/sweep_ecat_rate.py # 默认六档 × 60s
+ python scripts/sweep_ecat_rate.py --periods 0.25,0.5,1 --dwell 30
+ python scripts/sweep_ecat_rate.py --restore 1 --out lab/ecat_sweep.csv
+ python scripts/sweep_ecat_rate.py --dry-run # 只走一遍不下发周期
+结束后一定把周期恢复到脚本启动前的原值（含 Ctrl-C / 异常）。
+"""
+import argparse
+import csv
+import os
+import signal
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from calibrate_ecat_metrics import (cadence_rings, find_native_pid,
+ firmware_identity, ensure_loop)
+from ecat_sweep_common import delta_saturate, delta_wrap, percentile, post
+
+DEFAULT_PERIODS = "0.125,0.25,0.5,1,2,4"
+COLS = ["period_ms", "set_period_ms", "hijacked", "native_pid", "pid_stable",
+ "measured_hz", "achieve_pct", "can_lambda_per_s",
+ "drop_per_s", "drop_pct_of_lambda", "drop_untrusted", "lambda_u16_crosscheck",
+ "rxb_count_mode_per_cycle", "rxb_lost_max", "rxb_pending_max",
+ "skip_per_s", "skip_delta", "motors_online", "online_frac", "motor_refresh_per_s",
+ "torn_per_s", "torn_cum", "torn_capped",
+ "reentry_per_s", "reentry_cum", "reentry_capped", "wkc", "expected_wkc", "wr",
+ "cyc_p50_us", "cyc_p90_us", "cyc_max_us", "cyc_duty_pct",
+ "gd32_load_pct", "gd32_load_samples", "gd32_load_stale",
+ "host_cpu_1core_pct", "host_cpu_total_pct",
+ "logical_cpus", "samples", "errors", "dwell_s", "note"]
+
+
+def sample_period(url, ms, dwell, hz, substr, autostart, dry):
+ """在给定周期档上采 dwell 秒，返回一行 CSV 记录。"""
+ note = []
+ pid0, perr = find_native_pid(url)
+ st0 = post(url, "/api/motor/state")
+ if not (st0.get("ok") and st0.get("cycles")):
+ st0, how = ensure_loop(url, autostart, substr)
+ note.append("起循环:" + how)
+ orig = st0.get("period_ms")
+
+ if dry:
+ return {"period_ms": ms, "note": "dry-run 未下发"}
+
+ r = post(url, "/api/motor/period", {"period_ms": ms})
+ if not r.get("ok"):
+ return {"period_ms": ms, "note": "档位被拒:" + str(r.get("log"))[:80]}
+
+ # 等周期真正生效（热调是异步落到循环线程的）
+ eff, stable_iters = None, 0
+ t_deadline = time.perf_counter() + 3.0
+ while time.perf_counter() < t_deadline:
+ time.sleep(0.2)
+ eff = (post(url, "/api/motor/state")).get("period_ms")
+ if eff is not None and abs(float(eff) - float(ms)) < 1e-6:
+ stable_iters += 1
+ if stable_iters >= 3:
+ break
+ else:
+ stable_iters = 0
+ hijacked = "no"
+ if eff is None or abs(float(eff) - float(ms)) >= 1e-6:
+ hijacked = "YES"
+ note.append(f"下发 {ms}ms 但生效 {eff}ms（疑 localStorage 劫持或有他端在设）")
+
+ interval = 1.0 / hz
+ cyc_hz, drop_rate, rx_rate, gseq_rate, loads, hc1, hct = [], [], [], [], [], [], []
+ cyc_p50, cyc_p90, cyc_max = [], [], []
+ untrusted = torn_sum = reentry_sum = err = iters = load_stale = 0
+ torn_cap = reentry_cap = False
+ torn_prev = re_prev = None
+ torn_cum = re_cum = None
+ node_seq = {} # node -> (t, seq)：上一拍的槽 seq，用于算电机数据刷新率
+ online_hits = online_tot = 0
+ refresh_rates = []
+ prev = prev_t = None
+ wr0 = (post(url, "/api/motor/state")).get("wr")
+ skip0 = (post(url, "/api/motor/dbg")).get("skip_total")
+ proc = None
+ try:
+ import psutil
+ if pid0:
+ proc = psutil.Process(pid0)
+ proc.cpu_percent(None) # 建立差分基线
+ except Exception as e:
+ note.append(f"psutil 不可用({e.__class__.__name__})，主机 CPU 缺")
+
+ t_end = time.perf_counter() + dwell
+ while time.perf_counter() < t_end:
+ tc = time.perf_counter()
+ try:
+ s = post(url, "/api/motor/state")
+ L = post(url, "/api/app/load")
+ except Exception:
+ err += 1
+ time.sleep(interval)
+ continue
+ iters += 1
+ b = s.get("bridge") or {}
+ cur = {"cycles": s.get("cycles"), "drop": b.get("drop"), "rx": b.get("rx_count"),
+ "gseq": (s.get("head") or {}).get("global_seq")}
+ if prev is not None and tc > prev_t:
+ dt = tc - prev_t
+ if cur["cycles"] is not None and prev["cycles"] is not None:
+ h = (cur["cycles"] - prev["cycles"]) / dt
+ if 0 < h < 1e6:
+ cyc_hz.append(h)
+ if cur["drop"] is not None and prev["drop"] is not None:
+ d = delta_wrap(prev["drop"], cur["drop"], 16)
+ if d < 32768:
+ drop_rate.append(d / dt)
+ else:
+ untrusted += 1 # 已绕圈：该点作废，只计数不臆造速率
+ # λ 以 u32 的 head.global_seq 为准。u16 的 bridge.rx_count/drop 在快档
+ # 会失真（实测 0.25ms 档真值 149/s，两路 u16 与 u32 虽一致但增量已贴近
+ # 截断区），留作交叉核对：两者显著背离即说明该档 u16 不可信。
+ if cur["gseq"] is not None and prev["gseq"] is not None:
+ d = delta_wrap(prev["gseq"], cur["gseq"], 32)
+ if 0 <= d < (1 << 31):
+ gseq_rate.append(d / dt)
+ if cur["rx"] is not None and prev["rx"] is not None:
+ d = delta_wrap(prev["rx"], cur["rx"], 32)
+ if d < (1 << 31):
+ rx_rate.append(d / dt)
+ tn, rn = L.get("rx_torn"), L.get("pdo_reentry")
+ if tn is not None:
+ if torn_prev is not None and not torn_cap:
+ dd, capped = delta_saturate(torn_prev, tn)
+ torn_sum += dd
+ torn_cap |= capped
+ torn_prev = tn
+ if rn is not None:
+ if re_prev is not None and not reentry_cap:
+ dd, capped = delta_saturate(re_prev, rn)
+ reentry_sum += dd
+ reentry_cap |= capped
+ re_prev = rn
+ # 从站停报时 load_perm 会保留最后一次的值（实测：从站楔死后 age_ms=214000
+ # 而 load_perm 仍报 992）。不加 age 闸就会把残值当实测写进结论。
+ if (L.get("valid") and L.get("load_perm") is not None
+ and (L.get("age_ms") if L.get("age_ms") is not None else 1e9) < 1000):
+ loads.append(L["load_perm"] / 10.0)
+ else:
+ load_stale += 1
+ cu = (s.get("cyc_us") or {}).get("total") or {}
+ if cu.get("p50") is not None:
+ cyc_p50.append(cu["p50"])
+ if cu.get("p90") is not None:
+ cyc_p90.append(cu["p90"])
+ if cu.get("max") is not None:
+ cyc_max.append(cu["max"])
+ if proc is not None:
+ try:
+ c = proc.cpu_percent(None)
+ hc1.append(c)
+ hct.append(c / max(1, os.cpu_count() or 1))
+ except Exception:
+ pass
+ # 电机在线标志与数据刷新率：快档塌陷必须能区分"从站算不过来"与"电机掉线"
+ # —— 二者在 λ/seq 上长得一模一样，只有 online 标志能把它们切开。
+ for m in (s.get("motors") or []):
+ nd = m.get("node")
+ if nd is None:
+ continue
+ online_tot += 1
+ online_hits += 1 if m.get("online") else 0
+ sq = m.get("seq")
+ if sq is None:
+ continue
+ pr = node_seq.get(nd)
+ if pr and tc > pr[0]:
+ d = delta_wrap(pr[1], sq, 32)
+ if d < (1 << 31):
+ refresh_rates.append(d / (tc - pr[0]))
+ node_seq[nd] = (tc, sq)
+ prev, prev_t = cur, tc
+ time.sleep(max(0.0, interval - (time.perf_counter() - tc)))
+
+ dur = max(1e-6, time.perf_counter() - (t_end - dwell))
+ stf = post(url, "/api/motor/state")
+ tend = post(url, "/api/app/load")
+ pid1, _ = find_native_pid(url)
+ skip1 = (post(url, "/api/motor/dbg")).get("skip_total")
+ cad = cadence_rings(url)
+ torn_cum, re_cum = tend.get("rx_torn"), tend.get("pdo_reentry")
+ lam = percentile(gseq_rate, 0.5) or 0 # u32 gseq 口径，主用
+ lam_u16 = percentile(rx_rate, 0.5) or 0 # u16 rx_count 口径，仅作交叉核对
+ dropr = percentile(drop_rate, 0.5) or 0
+ hz_meas = percentile(cyc_hz, 0.5) or 0
+ eff_ms = float(stf.get("period_ms") or ms)
+ cad_hist = cad.get("count_hist") or {}
+ mode_cnt = max(cad_hist, key=lambda k: cad_hist[k]) if cad_hist else None
+ rec = {
+ "period_ms": ms, "set_period_ms": eff_ms, "hijacked": hijacked,
+ "native_pid": pid0, "pid_stable": "yes" if pid0 == pid1 else "NO",
+ "measured_hz": round(hz_meas, 1),
+ "achieve_pct": round(hz_meas / (1000.0 / eff_ms) * 100, 1) if eff_ms else None,
+ "can_lambda_per_s": round(lam, 1), "lambda_u16_crosscheck": round(lam_u16, 1),
+ "drop_per_s": round(dropr, 1),
+ "drop_pct_of_lambda": round(dropr / lam * 100, 2) if lam else None,
+ "drop_untrusted": untrusted,
+ "rxb_count_mode_per_cycle": mode_cnt,
+ "rxb_lost_max": cad.get("lost_max"), "rxb_pending_max": cad.get("pending_max"),
+ "motors_online": sum(1 for m in (stf.get("motors") or []) if m.get("online")),
+ "online_frac": round(online_hits / online_tot, 3) if online_tot else None,
+ "motor_refresh_per_s": round(percentile(refresh_rates, 0.5) or 0, 1),
+ "skip_per_s": round(((skip1 or 0) - (skip0 or 0)) / dur, 2),
+ "skip_delta": (skip1 or 0) - (skip0 or 0),
+ "torn_per_s": round(torn_sum / dur, 3),
+ "torn_cum": torn_cum, "torn_capped": "yes" if torn_cap else "no",
+ "reentry_per_s": round(reentry_sum / dur, 3),
+ "reentry_cum": re_cum, "reentry_capped": "yes" if reentry_cap else "no",
+ "wkc": stf.get("wkc"), "expected_wkc": stf.get("expected_wkc"),
+ "wr": f"{wr0}->{stf.get('wr')}",
+ "cyc_p50_us": round(percentile(cyc_p50, 0.5) or 0, 1),
+ "cyc_p90_us": round(percentile(cyc_p90, 0.9) or 0, 1),
+ "cyc_max_us": round(max(cyc_max), 1) if cyc_max else None,
+ "cyc_duty_pct": round((percentile(cyc_p50, 0.5) or 0) / (eff_ms * 1000) * 100, 1),
+ "gd32_load_pct": round(percentile(loads, 0.5) or 0, 1) if loads else None,
+ "gd32_load_samples": len(loads), "gd32_load_stale": load_stale,
+ "host_cpu_1core_pct": round(percentile(hc1, 0.5) or 0, 1),
+ "host_cpu_total_pct": round(percentile(hct, 0.5) or 0, 2),
+ "logical_cpus": os.cpu_count(), "samples": iters, "errors": err,
+ "dwell_s": round(dur, 1), "note": "; ".join(note)}
+ return rec
+
+
+def main():
+ ap = argparse.ArgumentParser(description="ECAT 周期-故障率-CPU 扫描")
+ ap.add_argument("--url", default="http://127.0.0.1:8090")
+ ap.add_argument("--periods", default=DEFAULT_PERIODS, help="逗号分隔周期(ms)")
+ ap.add_argument("--dwell", type=float, default=60.0, help="每档采样秒数")
+ ap.add_argument("--hz", type=float, default=10.0,
+ help="HTTP 采样频率；u16 drop 在 4000/s 下 16s 绕一圈，须够高")
+ ap.add_argument("--restore", type=float, default=None, help="测完恢复到的周期(ms)")
+ ap.add_argument("--substr", default="I210")
+ ap.add_argument("--no-autostart", action="store_true")
+ ap.add_argument("--out", default=os.path.join(
+ os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "lab", "ecat_rate_sweep.csv"))
+ ap.add_argument("--dry-run", action="store_true")
+ a = ap.parse_args()
+
+ st = post(a.url, "/api/motor/state")
+ if not (st.get("ok") and st.get("cycles")) and a.no_autostart:
+ print("循环没跑且 --no-autostart：", st.get("log"))
+ return 2
+ orig = st.get("period_ms")
+ pid0, _ = find_native_pid(a.url)
+ periods = [float(x) for x in a.periods.split(",")]
+ print(f"服务 {a.url} · native PID {pid0} · 原周期 {orig}ms · 档位 {periods} · "
+ f"每档 {a.dwell}s @ {a.hz:g}Hz")
+ fw = firmware_identity()
+ print(f"固件台账 {str(fw.get('ledger'))[:16]}… 匹配={fw.get('match') or fw.get('error') or '无'}")
+ print("提示：确认没有任何电机页标签开着，否则周期会被 localStorage 拽回去\n")
+
+ rows, interrupted = [], False
+ signal.signal(signal.SIGINT, lambda *_: None)
+ try:
+ for ms in periods:
+ try:
+ rec = sample_period(a.url, ms, a.dwell, a.hz, a.substr,
+ not a.no_autostart, a.dry_run)
+ except Exception as e:
+ print(f"{ms}ms 档异常: {type(e).__name__}: {e}")
+ break
+ rows.append(rec)
+ if rec.get("note") == "dry-run 未下发":
+ print(f"[dry-run] {ms}ms 将下发，未执行")
+ continue
+ print(f"{ms:>7}ms 生效={rec['set_period_ms']} 劫持={rec['hijacked']} "
+ f"{rec['measured_hz']}Hz/{rec['achieve_pct']}% λ={rec['can_lambda_per_s']} "
+ f"drop={rec['drop_per_s']}/s({rec['drop_pct_of_lambda']}%) "
+ f"skip={rec['skip_per_s']}/s GD32={rec['gd32_load_pct']}% "
+ f"host={rec['host_cpu_1core_pct']}% wkc={rec['wkc']}/{rec['expected_wkc']}")
+ if rec["note"]:
+ print(f" note: {rec['note']}")
+ if rec["pid_stable"] == "NO":
+ print(" !! native PID 变了，该档及之后数据作废，停止扫描")
+ break
+ if rec["wkc"] != rec["expected_wkc"]:
+ print(" !! wkc 不达预期（从站丢帧/掉 OP 前兆），停止继续加压")
+ break
+ except KeyboardInterrupt:
+ interrupted = True
+ print("\n收到 Ctrl-C，正在恢复原周期…")
+ finally:
+ restore = a.restore if a.restore is not None else orig
+ if restore is not None and not a.dry_run:
+ try:
+ post(a.url, "/api/motor/period", {"period_ms": float(restore)})
+ back = (post(a.url, "/api/motor/state")).get("period_ms")
+ print(f"\n已恢复周期 -> 设定 {restore}ms，实际回读 {back}ms")
+ except Exception as e:
+ print(f"\n!! 恢复周期失败({e})，请手动确认总线周期")
+
+ if not rows or a.dry_run:
+ return 0
+ os.makedirs(os.path.dirname(os.path.abspath(a.out)), exist_ok=True)
+ with open(a.out, "w", newline="", encoding="utf-8") as f:
+ w = csv.DictWriter(f, fieldnames=COLS, extrasaction="ignore")
+ w.writeheader()
+ w.writerows(rows)
+ print(f"扫描数据 {len(rows)} 档 -> {a.out}")
+ if interrupted:
+ print("（本次为中断后的部分数据）")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())