Files
DCU_DesktopController/scripts/sweep_ecat_rate.py
T

336 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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())