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