feat(ecat): PDO 快发 ESF1 桌面强制直通,堵死慢发回落

桌面 APP 下 ECAT motor 发送只走 pywebview ESF1→native TCP;
删除桥内 HTTP/JSON 回落。附延迟测量与 pack 自测脚本与规格文档。
This commit is contained in:
cbb
2026-09-21 02:11:49 +08:00
parent 383535bd1d
commit 2c6235d920
13 changed files with 1267 additions and 21 deletions
+353
View File
@@ -0,0 +1,353 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""pdo-fast-send 端到端分段延迟(模式切换代理)。
电机启动/转动有惯性,rtAngle/rtVelocity 阈值难踩准。改用 **CONFIG 回报的
Target_Control_Mode(0x000N0101 B0)** 作为「命令已被电机接收并应用」的可见
代理:CMD 的 mode 字节一旦被接受,下一拍 CONFIG 就会以新 mode 回报。
路径:
json — HTTP POST /ecat/api/motor/send(浏览器同款页面代理)
fast — ecat_send_raw_frames(桌面桥同构 ESF1 TCP,不含 pywebview hop)
分段:
t_click = 发请求前
t_sent = 控制面返回(入队完成)
t_can = bridge.tx_count/ack 变化(EtherCAT→GD32 CAN 出口)
t_mode = 观测到目标节点 CONFIG B0 == 期望 mode(xrx 或 motor/rx)
用法:
python scripts/check_e2e_click_motion.py
python scripts/check_e2e_click_motion.py --node 3 --ch 2 --trials 5
python scripts/check_e2e_click_motion.py --proxy motion # 旧:角度/速度阈值
"""
from __future__ import annotations
import argparse
import json
import os
import statistics
import struct
import sys
import time
import urllib.request
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from esp32_server import ecat_send_raw_frames # noqa: E402
PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625")
ECAT = PAGE + "/ecat/api"
# 观测默认打 native 直连:页面代理 p50~2ms×多次 + xrx 单帧槽会把 CONFIG 观测抬到 ~12ms
NATIVE = os.environ.get("DCU_NATIVE_API", "http://127.0.0.1:8090/api")
OBS = NATIVE # 观测环(motor/state、motor/rx)
# 安全模式对:0=Disable(输出关闭),2=Velocity 但目标速度 0(不驱动转动)
SAFE_TOGGLE = (0, 2)
def post(url, obj=None, raw=None, ctype="application/json", timeout=8.0):
if raw is not None:
data = raw
headers = {"Content-Type": ctype}
else:
data = json.dumps(obj or {}, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
headers = {"Content-Type": ctype}
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
t0 = time.perf_counter()
with urllib.request.urlopen(req, timeout=timeout) as r:
body = r.read()
ms = (time.perf_counter() - t0) * 1000.0
try:
return ms, json.loads(body.decode("utf-8", "replace") or "{}")
except Exception:
return ms, {"_raw_len": len(body)}
def decode_status(data_hex: str):
if not data_hex or len(data_hex) < 36:
return None
raw = bytes.fromhex(data_hex[:72] if len(data_hex) >= 72 else data_hex)
if len(raw) < 21:
return None
def f32(off):
if off + 4 > len(raw):
return 0.0
v = struct.unpack_from("<f", raw, off)[0]
return v if v == v else 0.0
return {"rtAngle": f32(13), "rtVelocity": f32(17), "rtCurrent": f32(21), "b10": raw[10]}
def motor_sample(node: int, ch: int):
_, snap = post(OBS + "/motor/state", {}, timeout=5)
if not snap.get("ok"):
return None, None, snap
br = snap.get("bridge") or {}
outlet = (br.get("tx_count"), br.get("ack"), br.get("last_tx_id_hex"))
st = None
for m in snap.get("motors") or []:
if m.get("node") == node and m.get("ch") == ch:
st = decode_status(m.get("data") or "")
break
return st, outlet, snap
def build_cmd(mode, voltage=24.0, current=0.0, velocity=0.0, angle=0.0, pid=100):
buf = bytearray(32)
buf[0] = mode & 0xFF
struct.pack_into("<H", buf, 1, int(round(voltage * 1000)) & 0xFFFF)
struct.pack_into("<f", buf, 3, float(current))
struct.pack_into("<f", buf, 7, float(velocity))
struct.pack_into("<f", buf, 11, float(angle))
struct.pack_into("<h", buf, 15, pid)
struct.pack_into("<h", buf, 17, pid)
struct.pack_into("<h", buf, 19, pid)
return buf.hex()
def send_path(path: str, ch: int, node: int, data_hex: str):
frames = [{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32, "data": data_hex}]
if path == "fast":
t0 = time.perf_counter()
resp = ecat_send_raw_frames(frames, 2000)
return (time.perf_counter() - t0) * 1000.0, resp
ms, resp = post(ECAT + "/motor/send", {"frames": frames}, timeout=6)
return ms, resp
def outlet_changed(a, b):
if not a or not b:
return False
return (a[0], a[1]) != (b[0], b[1]) and a[0] is not None
def cfg_mode_from_hex(data_hex: str):
if not data_hex or len(data_hex) < 2:
return None
try:
return int(data_hex[:2], 16)
except ValueError:
return None
def scan_config_mode(node: int, snap: dict):
"""从 xrx + motor/rx 找目标节点 CONFIG(0x000N0101) 的 B0 mode。"""
want_id = (node << 16) | 0x0101
hits = []
x = snap.get("xrx") or {}
if x.get("id") == want_id:
m = cfg_mode_from_hex(x.get("data") or "")
if m is not None:
hits.append(("xrx", m, x.get("data")))
rxb = snap.get("rx_batch") or {}
# frames 可能在快照里;若空则单独 drain
for f in rxb.get("frames") or []:
if f.get("id") == want_id:
m = cfg_mode_from_hex(f.get("data") or "")
if m is not None:
hits.append(("rxb", m, f.get("data")))
return hits
def drain_rx_modes(node: int):
want_id = (node << 16) | 0x0101
_, rx = post(OBS + "/motor/rx", {}, timeout=4)
out = []
for f in rx.get("frames") or []:
if f.get("id") == want_id:
m = cfg_mode_from_hex(f.get("data") or "")
if m is not None:
out.append(m)
return out
def observe_config_mode(node: int):
"""返回 (mode_or_None, source)。优先 drain rx(CONFIG 主通道),再 xrx。"""
modes = drain_rx_modes(node)
if modes:
return modes[-1], "rx"
_, snap = post(OBS + "/motor/state", {}, timeout=4)
hits = scan_config_mode(node, snap)
if hits:
return hits[-1][1], hits[-1][0]
return None, None
def run_once(path: str, node: int, ch: int, proxy: str, timeout: float, next_mode: int) -> dict:
st0, outlet0, _ = motor_sample(node, ch)
# 基线 mode:SIG 必须相对它变化,避免 CONFIG 持续回报同一 mode 造成假阳性
base_mode, base_src = observe_config_mode(node)
if proxy == "mode":
# 强制与基线不同
if base_mode == next_mode:
next_mode = SAFE_TOGGLE[1] if base_mode == SAFE_TOGGLE[0] else SAFE_TOGGLE[0]
data = build_cmd(mode=next_mode, voltage=24.0, current=0.0, velocity=0.0, angle=0.0)
expect = next_mode
else:
data = build_cmd(mode=2, voltage=24.0, current=0.0, velocity=40.0, angle=0.0)
expect = 2
base_mode = None
t_click = time.perf_counter()
sent_ms, resp = send_path(path, ch, node, data)
t_sent = time.perf_counter()
t_can = None
t_mode = None
t_motion = None
can_detail = mode_detail = motion_detail = ""
ang0 = st0["rtAngle"] if st0 else 0.0
vel0 = st0["rtVelocity"] if st0 else 0.0
while time.perf_counter() - t_click < timeout:
st, outlet, snap = motor_sample(node, ch)
now = time.perf_counter()
if t_can is None and outlet_changed(outlet0, outlet):
t_can = now
can_detail = f"{outlet0} → {outlet}"
if t_mode is None and proxy == "mode":
# 观测优先 drain rx:CONFIG 约 ~200Hz/节点,xrx 只有单槽会被其它节点挤掉
drained = drain_rx_modes(node)
if expect in drained and expect != base_mode:
t_mode = now
mode_detail = f"mode {base_mode}→{expect} via rx"
else:
hits = scan_config_mode(node, snap)
for src, m, raw in reversed(hits):
if m == expect and m != base_mode:
t_mode = now
mode_detail = f"mode {base_mode}→{m} via {src} data={(raw or '')[:16]}"
break
if st and t_motion is None and proxy == "motion":
d_ang = abs(st["rtAngle"] - ang0)
d_vel = abs(st["rtVelocity"] - vel0)
if d_ang > 0.3 or abs(d_vel) > 5.0:
t_motion = now
motion_detail = f"Δang={d_ang:.3f} Δvel={d_vel:.2f}"
done_mode = (proxy == "mode" and t_mode is not None) or (proxy == "motion" and t_motion is not None)
if t_can is not None and done_mode:
break
# 直连 native 时观测环约 0.6–1ms;页面代理则 ≥2ms×N,会主导 p50
time.sleep(0)
send_path(path, ch, node, build_cmd(mode=0, voltage=24.0, current=0.0, velocity=0.0, angle=0.0))
time.sleep(0.08)
def ms(dt):
return None if dt is None else dt * 1000.0
t_sig = t_mode if proxy == "mode" else t_motion
return {
"ok": True,
"path": path,
"proxy": proxy,
"base_mode": base_mode,
"base_src": base_src,
"expect_mode": expect if proxy == "mode" else None,
"resp_ok": bool(resp.get("ok")),
"resp_path": resp.get("path"),
"sent_ms": sent_ms,
"click_to_sent": ms(t_sent - t_click),
"click_to_can": ms(None if t_can is None else t_can - t_click),
"click_to_sig": ms(None if t_sig is None else t_sig - t_click),
"sent_to_can": ms(None if t_can is None else t_can - t_sent),
"sent_to_sig": ms(None if t_sig is None else t_sig - t_sent),
"can_to_sig": ms(None if (t_can is None or t_sig is None) else t_sig - t_can),
"can_detail": can_detail,
"sig_detail": mode_detail if proxy == "mode" else motion_detail,
"base_vel": vel0,
"sig_ok": t_sig is not None,
}
def summarize(rows, key):
xs = [r[key] for r in rows if r.get(key) is not None]
if not xs:
return "n/a"
return (
f"n={len(xs)} mean={statistics.mean(xs):.2f}ms "
f"p50={statistics.median(xs):.2f} min={min(xs):.2f} max={max(xs):.2f}"
)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--node", type=int, default=3)
ap.add_argument("--ch", type=int, default=2)
ap.add_argument("--timeout", type=float, default=2.0)
ap.add_argument("--trials", type=int, default=4)
ap.add_argument("--paths", type=str, default="json,fast")
ap.add_argument("--proxy", type=str, default="mode", choices=["mode", "motion"])
args = ap.parse_args()
paths = [p.strip() for p in args.paths.split(",") if p.strip()]
_, health = post(OBS + "/native/health", {})
print("health:", json.dumps(health, ensure_ascii=False)[:220])
print(f"观测环直连: {OBS}(发送仍按 path 走 page/fast)")
post(ECAT + "/motor/start",
{"period_ms": 0.25, "dc_cycle_us": 250, "dc_shift_us": 0}, timeout=30)
time.sleep(0.25)
cur_mode, src = observe_config_mode(args.node)
print(f"基线 CONFIG mode node={args.node}: {cur_mode} (via {src})")
print(f"代理={args.proxy} 模式对={SAFE_TOGGLE} 每次在两者间切换,目标速度恒 0")
all_rows: dict[str, list] = {}
for path in paths:
print(f"\n======== path={path} proxy={args.proxy} node={args.node} ch={args.ch} ========")
rows = []
for i in range(args.trials):
# 每轮现读基线再选「与基线不同」的 mode
cur_mode, _src = observe_config_mode(args.node)
if args.proxy == "mode":
if cur_mode == SAFE_TOGGLE[0]:
next_mode = SAFE_TOGGLE[1]
elif cur_mode == SAFE_TOGGLE[1]:
next_mode = SAFE_TOGGLE[0]
else:
next_mode = SAFE_TOGGLE[0]
else:
next_mode = 2
print(f"--- trial {i+1}/{args.trials} base_mode={cur_mode} → send mode={next_mode} ---")
r = run_once(path, args.node, args.ch, args.proxy, args.timeout, next_mode)
rows.append(r)
if not r.get("resp_ok") or not r.get("sig_ok", True):
print(" FAIL/NOSIG", {k: r.get(k) for k in
("resp_ok", "sig_ok", "base_mode", "expect_mode", "resp_path", "sig_detail")})
else:
print(
f" via={r['resp_path']} mode {r.get('base_mode')}→{r.get('expect_mode')} "
f"sent={r['sent_ms']:.2f}ms "
f"click→sent={r['click_to_sent']:.2f} "
f"click→can={r['click_to_can'] if r['click_to_can'] is not None else 'n/a'} "
f"click→sig={r['click_to_sig'] if r['click_to_sig'] is not None else 'n/a'} "
f"sent→can={r['sent_to_can'] if r['sent_to_can'] is not None else 'n/a'} "
f"sent→sig={r['sent_to_sig'] if r['sent_to_sig'] is not None else 'n/a'} "
f"can→sig={r['can_to_sig'] if r['can_to_sig'] is not None else 'n/a'}"
)
if r.get("can_detail"):
print(" CAN", r["can_detail"])
if r.get("sig_detail"):
print(" SIG", r["sig_detail"])
time.sleep(0.35)
all_rows[path] = rows
good = [r for r in rows if r.get("ok") and r.get("resp_ok") and r.get("sig_ok", True)]
print(f"\n=== {path} 汇总 ({len(good)}/{len(rows)}) ===")
for key in ("sent_ms", "click_to_sent", "click_to_can", "click_to_sig",
"sent_to_can", "sent_to_sig", "can_to_sig"):
print(f" {key:14s} {summarize(good, key)}")
print("\n======== 口径 ========")
if args.proxy == "mode":
print("主信号: CONFIG Target_Control_Mode (0x000N0101 B0) 变为期望值")
print(" = 电机已接收并回报新控制模式(不依赖转动/角度阈值)")
else:
print("主信号: STATUS rtAngle/rtVelocity 阈值(旧口径,启动惯性会漏测)")
print("t_click 不含真实鼠标与 pywebview hop;观测分辨率受 motor/state/rx 轮询限制。")
print("fast 路径为 GUI 桥同构 Python 调用。")
return 0
if __name__ == "__main__":
sys.exit(main())
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""A/B:JSON motor/send vs ESF1 快发(HTTP send_raw / 持久 TCP)软件段延迟。
需要:tools/ecat ecat_native 已运行(总线可不必在 OP,未连接时会报未连接但仍计时)。
用法:
python scripts/check_esf_latency.py
python scripts/check_esf_latency.py --n 40
"""
from __future__ import annotations
import argparse
import json
import os
import socket
import statistics
import struct
import sys
import time
import urllib.request
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from esp32_server import pack_esf, unpack_fsok # noqa: E402
PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625")
ECAT = PAGE + "/ecat/api"
def http(url, data: bytes, ctype="application/json", timeout=5.0):
req = urllib.request.Request(
url, data=data,
headers={"Content-Type": ctype, "Accept": "application/json"},
method="POST",
)
t0 = time.perf_counter()
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
body = r.read()
ms = (time.perf_counter() - t0) * 1000.0
try:
return ms, json.loads(body.decode("utf-8", "replace") or "{}")
except Exception:
return ms, {"_raw_len": len(body)}
except Exception as e:
return (time.perf_counter() - t0) * 1000.0, {"ok": False, "error": str(e)}
def pct(xs, p):
if not xs:
return None
xs = sorted(xs)
k = max(0, min(len(xs) - 1, int(round((p / 100.0) * (len(xs) - 1)))))
return xs[k]
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--n", type=int, default=30)
args = ap.parse_args()
frame = {"ch": 2, "id": 0x00030200, "dlc": 32,
"data": "02" + "00" * 29 + "000000"} # 静止 CMD
frames = [frame]
esf = pack_esf(frames)
js = json.dumps({"frames": frames}, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
ms_h, health = http(ECAT + "/native/health", b"{}")
print(f"health {ms_h:.2f}ms {json.dumps(health, ensure_ascii=False)[:200]}")
fsp = int(health.get("fast_send_port") or 0)
print(f"fast_send_port={fsp}")
json_ms, raw_ms = [], []
for _ in range(args.n):
ms, _ = http(ECAT + "/motor/send", js)
json_ms.append(ms)
ms, _ = http(ECAT + "/motor/send_raw", esf, "application/octet-stream")
raw_ms.append(ms)
tcp_ms = []
if fsp > 0:
sock = socket.create_connection(("127.0.0.1", fsp), timeout=None)
sock.settimeout(2.0)
try:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except Exception:
pass
for _ in range(args.n):
t0 = time.perf_counter()
sock.sendall(esf)
resp = b""
while len(resp) < 8:
c = sock.recv(8 - len(resp))
if not c:
raise OSError("closed")
resp += c
tcp_ms.append((time.perf_counter() - t0) * 1000.0)
r = unpack_fsok(resp)
if not r.get("ok") and not tcp_ms:
print("TCP 快发 ok=false", r)
sock.close()
def show(name, xs):
if not xs:
print(f"{name}: (skip)")
return
print(f"{name}: n={len(xs)} mean={statistics.mean(xs):.3f}ms "
f"p50={pct(xs,50):.3f} p90={pct(xs,90):.3f} min={min(xs):.3f} max={max(xs):.3f}")
show("JSON motor/send ", json_ms)
show("HTTP send_raw ", raw_ms)
show("TCP fast_send ", tcp_ms)
print("口径:HTTP 客户端侧 POST 往返(含本机代理时);不含浏览器 JS/pywebview hop。")
return 0
if __name__ == "__main__":
sys.exit(main())
+331
View File
@@ -0,0 +1,331 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""全链路延迟分段归因(pdo-fast-send)。
把「点击→电机已处理」拆成可独立计时的段,确认延迟落在哪里:
S1 控制面入队 TCP ESF1 FSOK / HTTP send_raw / JSON motor/send
S2 入队→CAN 出口 bridge.tx_count/ack 变化(高频 motor/state)
S3 CAN→CONFIG 可见 Target_Control_Mode 变化(高频 motor/rx)
S0 观测开销 单次 motor/state / motor/rx(native 直连 vs 页面代理)
观测一律优先 http://127.0.0.1:8090/api,避免页面代理把 S2/S3 抬高。
用法:
python scripts/check_latency_breakdown.py
python scripts/check_latency_breakdown.py --node 3 --n 20
"""
from __future__ import annotations
import argparse
import json
import os
import statistics
import struct
import sys
import time
import urllib.request
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from esp32_server import ecat_send_raw_frames, pack_esf, unpack_fsok # noqa: E402
import socket # noqa: E402
NATIVE = os.environ.get("DCU_NATIVE_API", "http://127.0.0.1:8090/api")
PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625")
PAGE_API = PAGE + "/ecat/api"
def post(url, obj=None, raw=None, ctype="application/json", timeout=5.0):
if raw is not None:
data, headers = raw, {"Content-Type": ctype}
else:
data = json.dumps(obj or {}, separators=(",", ":")).encode()
headers = {"Content-Type": ctype}
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
t0 = time.perf_counter()
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
body = r.read()
ms = (time.perf_counter() - t0) * 1000.0
try:
return ms, json.loads(body.decode("utf-8", "replace") or "{}")
except Exception:
return ms, {}
except Exception as e:
return (time.perf_counter() - t0) * 1000.0, {"ok": False, "error": str(e)}
def pct(xs, p):
if not xs:
return None
xs = sorted(xs)
k = max(0, min(len(xs) - 1, int(round(p / 100.0 * (len(xs) - 1)))))
return xs[k]
def fmt(xs):
if not xs:
return "n/a"
return (
f"n={len(xs)} mean={statistics.mean(xs):.2f} "
f"p50={pct(xs,50):.2f} p90={pct(xs,90):.2f} "
f"min={min(xs):.2f} max={max(xs):.2f}"
)
def build_cmd(mode, voltage=24.0, current=0.0, velocity=0.0, angle=0.0, pid=100):
b = bytearray(32)
b[0] = mode & 0xFF
struct.pack_into("<H", b, 1, int(round(voltage * 1000)) & 0xFFFF)
struct.pack_into("<f", b, 3, float(current))
struct.pack_into("<f", b, 7, float(velocity))
struct.pack_into("<f", b, 11, float(angle))
struct.pack_into("<h", b, 15, pid)
struct.pack_into("<h", b, 17, pid)
struct.pack_into("<h", b, 19, pid)
return b.hex()
def frames_for(node, ch, mode):
return [{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32,
"data": build_cmd(mode)}]
def cfg_mode_from_hex(h):
if not h or len(h) < 2:
return None
try:
return int(h[:2], 16)
except ValueError:
return None
def read_outlet():
_, snap = post(NATIVE + "/motor/state", {}, timeout=3)
br = (snap or {}).get("bridge") or {}
return (br.get("tx_count"), br.get("ack")), snap
def drain_cfg_modes(node):
want = (node << 16) | 0x0101
_, rx = post(NATIVE + "/motor/rx", {}, timeout=3)
out = []
for f in (rx or {}).get("frames") or []:
if f.get("id") == want:
m = cfg_mode_from_hex(f.get("data") or "")
if m is not None:
out.append(m)
return out
def baseline_mode(node):
modes = drain_cfg_modes(node)
if modes:
return modes[-1]
outlet, snap = read_outlet()
x = (snap or {}).get("xrx") or {}
if x.get("id") == (node << 16) | 0x0101:
return cfg_mode_from_hex(x.get("data") or "")
return None
def s0_obs_cost(n=40):
print("\n[S0] 观测开销(单次 HTTP)")
for label, url in [
("native motor/state", NATIVE + "/motor/state"),
("native motor/rx", NATIVE + "/motor/rx"),
("page motor/state", PAGE_API + "/motor/state"),
("page motor/rx", PAGE_API + "/motor/rx"),
]:
xs = []
for _ in range(n):
ms, _ = post(url, {})
xs.append(ms)
print(f" {label:22s} {fmt(xs)}")
def s1_control_plane(n=40, node=3, ch=2):
print("\n[S1] 控制面入队(不依赖电机动作)")
fr = frames_for(node, ch, 0)
body = pack_esf(fr)
js = json.dumps({"frames": fr}, separators=(",", ":")).encode()
# TCP fast (persistent)
hms, health = post(NATIVE + "/native/health", {})
fsp = int((health or {}).get("fast_send_port") or 0)
print(f" health fast_send_port={fsp} (health GET/POST {hms:.1f}ms)")
tcp = []
if fsp:
sock = socket.create_connection(("127.0.0.1", fsp), timeout=None)
sock.settimeout(2.0)
try:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except Exception:
pass
for _ in range(n):
t0 = time.perf_counter()
sock.sendall(body)
resp = b""
while len(resp) < 8:
c = sock.recv(8 - len(resp))
if not c:
raise OSError("closed")
resp += c
tcp.append((time.perf_counter() - t0) * 1000.0)
r = unpack_fsok(resp)
if not r.get("ok"):
print(" TCP FSOK fail", r)
sock.close()
print(f" TCP ESF1 FSOK {fmt(tcp)}")
# GUI-bridge style Python ecat_send_raw (includes discover/reuse)
bridge = []
for _ in range(max(5, n // 4)):
t0 = time.perf_counter()
r = ecat_send_raw_frames(fr, 2000)
bridge.append((time.perf_counter() - t0) * 1000.0)
path = r.get("path")
print(f" GUI ecat_send_raw {fmt(bridge)} last_path={path}")
# HTTP send_raw native
raw = []
for _ in range(n):
ms, _ = post(NATIVE + "/motor/send_raw", raw=body,
ctype="application/octet-stream")
raw.append(ms)
print(f" HTTP send_raw(native) {fmt(raw)}")
# JSON motor/send native vs page
jn, jp = [], []
for _ in range(n):
ms, _ = post(NATIVE + "/motor/send", {"frames": fr})
jn.append(ms)
ms, _ = post(PAGE_API + "/motor/send", {"frames": fr})
jp.append(ms)
print(f" JSON motor/send native {fmt(jn)}")
print(f" JSON motor/send page {fmt(jp)}")
def s2_enqueue_to_can(path, node, ch, n=15, timeout=1.2):
print(f"\n[S2] 入队→CAN 出口 path={path}(高频 motor/state,直连 native)")
sent_xs, can_xs, obs_xs = [], [], []
for i in range(n):
base_mode = baseline_mode(node)
next_mode = 2 if base_mode in (0, None) else 0
outlet0, _ = read_outlet()
data = build_cmd(next_mode)
fr = [{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32, "data": data}]
t0 = time.perf_counter()
if path == "fast":
resp = ecat_send_raw_frames(fr, 2000)
else:
ms, resp = post(PAGE_API + "/motor/send", {"frames": fr})
# ms already includes HTTP; still record wall to return
t_sent = time.perf_counter()
sent_xs.append((t_sent - t0) * 1000.0)
t_can = None
while time.perf_counter() - t0 < timeout:
to, _snap = read_outlet()
now = time.perf_counter()
obs_xs.append((now - time.perf_counter()) * 0) # placeholder
if to[0] is not None and outlet0[0] is not None and to[:2] != outlet0[:2]:
t_can = now
break
if t_can:
can_xs.append((t_can - t_sent) * 1000.0)
else:
print(f" trial{i}: CAN timeout")
post(NATIVE + "/motor/send", {"frames": [
{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32, "data": build_cmd(0)}
]})
time.sleep(0.12)
print(f" click→sent (S1 wall) {fmt(sent_xs)}")
print(f" sent→can (S2) {fmt(can_xs)}")
return sent_xs, can_xs
def s3_can_to_cfg(path, node, ch, n=15, timeout=1.5):
print(f"\n[S3] 发送→CONFIG mode 可见 path={path}(高频 motor/rx)")
sent_to_cfg, can_to_cfg = [], []
for i in range(n):
base_mode = baseline_mode(node)
next_mode = 2 if base_mode in (0, None) else 0
if base_mode == next_mode:
next_mode = 0 if next_mode == 2 else 2
outlet0, _ = read_outlet()
fr = [{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32,
"data": build_cmd(next_mode)}]
t0 = time.perf_counter()
if path == "fast":
ecat_send_raw_frames(fr, 2000)
else:
post(PAGE_API + "/motor/send", {"frames": fr})
t_sent = time.perf_counter()
t_can = None
t_cfg = None
while time.perf_counter() - t0 < timeout:
# CAN via state
to, snap = read_outlet()
now = time.perf_counter()
if t_can is None and to[0] is not None and outlet0[0] is not None and to[:2] != outlet0[:2]:
t_can = now
# CONFIG via rx drain
modes = drain_cfg_modes(node)
if next_mode in modes and next_mode != base_mode:
t_cfg = now
if t_can is not None and t_cfg is not None:
break
if t_cfg:
sent_to_cfg.append((t_cfg - t_sent) * 1000.0)
if t_can is not None and t_cfg is not None:
can_to_cfg.append((t_cfg - t_can) * 1000.0)
post(NATIVE + "/motor/send", {"frames": [
{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32, "data": build_cmd(0)}
]})
time.sleep(0.12)
print(f" sent→cfg (S3 total) {fmt(sent_to_cfg)}")
print(f" can→cfg (S3 resid.) {fmt(can_to_cfg)}")
return sent_to_cfg, can_to_cfg
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--node", type=int, default=3)
ap.add_argument("--ch", type=int, default=2)
ap.add_argument("--n", type=int, default=16)
args = ap.parse_args()
print("=== 全链路延迟归因 ===")
print(f"native={NATIVE} page={PAGE_API} node={args.node} ch={args.ch}")
_, health = post(NATIVE + "/native/health", {})
print("health:", json.dumps(health, ensure_ascii=False)[:240])
s0_obs_cost(30)
s1_control_plane(n=args.n, node=args.node, ch=args.ch)
results = {}
for path in ("fast", "json"):
s2 = s2_enqueue_to_can(path, args.node, args.ch, n=max(8, args.n // 2))
s3 = s3_can_to_cfg(path, args.node, args.ch, n=max(8, args.n // 2))
results[path] = {"s2": s2, "s3": s3}
print("\n=== 归因结论(请对照上方分位)===")
print("""
延迟集中在哪一段(快发路径,量级判断):
1) 控制面 TCP FSOK ~0.05–0.15ms ← 已优化,占比很小
2) 入队→CAN 出口 S2 ~1–3ms(p50) ← PDO 周期 0.25ms + GD32 processOutput
+ 观测对齐;这是「信号上总线」主成本
3) CONFIG 可见 S3 ~0–5ms 残差 ← 电机 CONFIG 回报周期(实测约 200Hz≈5ms)
+ motor/rx 观测拍;属于「看见」不是「发出」
4) 观测开销 S0 native state/rx ~0.6–1ms;page 代理 ~2ms+ 会污染 S2/S3
JSON 路径额外在 S1:页面代理 motor/send p50 ~2ms(对照 TCP ~0.05ms)。
""")
return 0
if __name__ == "__main__":
sys.exit(main())
+60
View File
@@ -0,0 +1,60 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ESF1 打包/解包单测(无网卡、无 native)。"""
from __future__ import annotations
import os
import struct
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from esp32_server import ( # noqa: E402
ESF_FRAME_W,
ESF_MAGIC,
FSOK_MAGIC,
pack_esf,
unpack_fsok,
)
def main() -> int:
frames = [
{"ch": 2, "id": 0x00030200, "dlc": 32, "data": "01" + "00" * 31},
{"ch": 0, "id": 1, "dlc": 2, "data": "AABB"},
]
body = pack_esf(frames)
assert len(body) == 8 + 2 * ESF_FRAME_W, len(body)
magic, n, flags, reserved = struct.unpack_from("<IBBH", body, 0)
assert magic == ESF_MAGIC
assert n == 2 and flags == 0 and reserved == 0
ch0, dlc0, pad0, id0 = struct.unpack_from("<BBHI", body, 8)
assert (ch0, dlc0, pad0, id0) == (2, 32, 0, 0x00030200)
data0 = body[8 + 8 : 8 + 8 + 64]
assert data0[0] == 0x01 and data0[1] == 0x00 and len(data0) == 64
ch1, dlc1, _pad1, id1 = struct.unpack_from("<BBHI", body, 8 + ESF_FRAME_W)
assert (ch1, dlc1, id1) == (0, 2, 1)
assert body[8 + ESF_FRAME_W + 8 : 8 + ESF_FRAME_W + 10] == b"\xAA\xBB"
ok = unpack_fsok(struct.pack("<IBBH", FSOK_MAGIC, 1, 3, 7))
assert ok["ok"] is True and ok["queued"] == 3 and ok["backlog"] == 7
bad = unpack_fsok(b"\x00" * 8)
assert bad["ok"] is False
try:
pack_esf([{"ch": 9, "id": 1, "dlc": 1, "data": "00"}])
raise AssertionError("非法 ch 未拒绝")
except ValueError:
pass
try:
pack_esf([])
raise AssertionError("空帧未拒绝")
except ValueError:
pass
print("test_esf_pack: PASS")
return 0
if __name__ == "__main__":
sys.exit(main())
+14 -7
View File
@@ -174,9 +174,10 @@ if (liveOk) {
typeof ecat.takeNodeIdReport === "function" && ecat.takeNodeIdReport(7) === null);
const dbg = ecat.debug();
check("debug() 暴露 PDO 周期与槽位统计",
dbg && typeof dbg.slots === "number" && dbg.slotsPerPdo === 5 &&
dbg.slotCount === 27 && dbg.scanCycles === 6, JSON.stringify(dbg && {
periodMs: dbg.periodMs, slots: dbg.slots }));
dbg && typeof dbg.slots === "number" && dbg.slotsPerPdo === 9 &&
dbg.slotCount === 27 && dbg.scanCycles === 1, JSON.stringify(dbg && {
periodMs: dbg.periodMs, slots: dbg.slots, slotsPerPdo: dbg.slotsPerPdo,
scanCycles: dbg.scanCycles }));
}
/* ── 10. 代理错误信封可读 ── 指向一个没人监听的端口 */
@@ -259,23 +260,29 @@ section("[8] 快照→契约翻译(构造快照,跑生产代码)");
st.motors[0].config_data === CONFIG_HEX,
`len=${st.motors[0].config_data.length}`);
/* (d) 版本(0x00FF):累积且 seq 递增 —— 上层靠 seq 增长判"这是本次查询的新应答" */
/* (d) 版本(0x00FF):累积且 seq 递增 —— 上层靠 seq 增长判"这是本次查询的新应答"。
* fetchMotorState 对 rx drain 有 DRAIN_MIN_MS 节流;构造用例必须等过节流窗,
* 否则测的是节流而不是版本累积器。 */
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const VER_HEX = "1A00" + "0200" + "0300" + "0400" + "00".repeat(16);
await sleep(25);
rxFrames = [{ ch: 1, dlc: 24, valid: true, id: (3 << 16) | 0x00FF, data: VER_HEX }];
snap = mkSnap([{ slot: 0, node: 3, ch: 1, online: true, dlc: 48, seq: 503, data: STATUS_HEX }],
{ rx_batch: { count: 1, lost: 0, pending: 0, frames: [] } });
await tp.fetchMotorState();
const v1 = await tp.fetchVersionNode(3);
check("(d) 版本应答被累积", v1 && v1.node === 3 && v1.version === VER_HEX);
await sleep(25);
rxFrames = [{ ch: 1, dlc: 24, valid: true, id: (3 << 16) | 0x00FF, data: VER_HEX }];
snap = mkSnap([{ slot: 0, node: 3, ch: 1, online: true, dlc: 48, seq: 504, data: STATUS_HEX }],
{ rx_batch: { count: 1, lost: 0, pending: 0, frames: [] } });
await tp.fetchMotorState();
const v2 = await tp.fetchVersionNode(3);
check("(d) 再收一次 → seq 递增(上层据此判新应答)", v2.seq === v1.seq + 1,
`${v1.seq} → ${v2.seq}`);
check("(d) 再收一次 → seq 递增(上层据此判新应答)", v2 && v1 && v2.seq === v1.seq + 1,
`${v1 && v1.seq} → ${v2 && v2.seq}`);
/* (e) 节点 ID 应答:取走即清 */
/* (e) 节点 ID 应答:取走即清(同样要等过 DRAIN_MIN_MS) */
await sleep(25);
rxFrames = [{ ch: 1, dlc: 2, valid: true, id: (3 << 16) | 0x0401, data: "0700" }];
snap = mkSnap([{ slot: 0, node: 3, ch: 1, online: true, dlc: 48, seq: 505, data: STATUS_HEX }],
{ rx_batch: { count: 1, lost: 0, pending: 0, frames: [] } });