137 lines
6.4 KiB
Python
137 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""ECAT 段2 撕裂 A/B 判据（修复前后的唯一对照指标）。
|
||
|
||
主判据（零假阳性，直接来自写侧契约）：
|
||
固件 MotorSlot 的 seq 与 data 属同一次更新，seq 先自增、data 后 memcpy。
|
||
⇒ 同一 seq 在总线上只能对应唯一一份 48B payload。
|
||
若同一 seq 先后被读到两份不同 payload，只可能是消费者（PDI_Isr 的 buildInput）
|
||
在生产者 memcpy 中途取走了半成品 —— 即撕裂（含"seq 新、data 整体旧"的错拍型）。
|
||
1.6.5 的 seqlock 把整个更新段（seq 自增 + data 拷贝）都盖进奇数区间，
|
||
故修复后该不变量必须零例外。
|
||
|
||
功效条件：只有当该节点的 seq 在相邻两拍之间不前进时，同一 seq 才可能被读两次。
|
||
所以脚本同时报"同 seq 重复读"的覆盖率——重复率低则本次判定功效不足，需换工况重跑。
|
||
|
||
旁判据（与速率无关）：字段值的"跳一下又回来"（median 偏离 >8× 局部抖动带且下一帧
|
||
回到带内）——真撕裂必然表现为这种形态。
|
||
"""
|
||
import json
|
||
import os
|
||
import socket
|
||
import statistics
|
||
import struct
|
||
import sys
|
||
import time
|
||
|
||
HOST = "127.0.0.1"
|
||
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8090
|
||
DUR_S = float(sys.argv[2]) if len(sys.argv) > 2 else 150.0
|
||
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ecat_torn_hits.jsonl")
|
||
|
||
FLOATS = (("rtAngle", 13), ("rtVelocity", 17), ("rtCurrent", 21), ("testFloat", 25))
|
||
I16 = (("mcuTemp", 0), ("motTemp", 2), ("uCur", 4), ("vCur", 6), ("wCur", 8))
|
||
|
||
|
||
def once():
|
||
s = socket.create_connection((HOST, PORT))
|
||
try:
|
||
s.sendall(REQ)
|
||
buf = bytearray()
|
||
while True:
|
||
c = s.recv(65536)
|
||
if not c:
|
||
break
|
||
buf.extend(c)
|
||
finally:
|
||
s.close()
|
||
_, _, d = bytes(buf).partition(b"\r\n\r\n")
|
||
return json.loads(d.decode("utf-8", "replace"))
|
||
|
||
|
||
REQ = (f"POST /api/motor/state HTTP/1.0\r\nHost: {HOST}:{PORT}\r\n"
|
||
f"Content-Type: application/json\r\nContent-Length: 2\r\n"
|
||
"Connection: close\r\n\r\n{{}}").encode("ascii")
|
||
|
||
|
||
def main():
|
||
# per node: 当前 seq 的首见/末见 payload、上一 seq 的确定 payload、字段历史
|
||
st = {}
|
||
viol, spikes, polls, err = [], [], 0, 0
|
||
cov = 0
|
||
t0 = time.time()
|
||
lf = open(LOG, "w", encoding="utf-8")
|
||
while time.time() - t0 < DUR_S:
|
||
try:
|
||
snap = once()
|
||
except Exception:
|
||
err += 1
|
||
continue
|
||
polls += 1
|
||
for m in snap.get("motors") or []:
|
||
if not m.get("online") or not m.get("data"):
|
||
continue
|
||
nd, seq = m["node"], m["seq"]
|
||
d = bytes.fromhex(m["data"])
|
||
e = st.setdefault(nd, {"cur": None, "first": None, "last": None,
|
||
"done": None, "hist": {}, "seen": 0})
|
||
if e["cur"] == seq:
|
||
cov += 1 # 一次"同一 seq 被读第二遍"的判定机会
|
||
if e["first"] != d:
|
||
rec = {"t": round(time.time() - t0, 2), "node": nd, "seq": seq,
|
||
"first": e["first"].hex(), "later": d.hex(),
|
||
"prev_seq_data": (e["done"].hex() if e["done"] else None)}
|
||
# 分类：整体等于上一帧 = 错拍型；前缀/后缀混合 = 真混合
|
||
a, b, p = e["first"], d, e["done"]
|
||
rec["cls"] = ("STALE(错拍)" if (p is not None and a == p) else
|
||
"MATCH" if a == b else "VIOLATION")
|
||
if p is not None and a != p:
|
||
n = min(len(a), len(b), len(p))
|
||
rec["mix_j"] = next((j for j in range(n + 1)
|
||
if a[:j] == b[:j] and a[j:n] == p[j:n]), None)
|
||
viol.append(rec)
|
||
lf.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||
lf.flush()
|
||
e["last"] = d
|
||
else:
|
||
# seq 前进：上一 seq 的末见即其确定内容
|
||
if e["cur"] is not None and e["last"] is not None:
|
||
e["done"] = e["last"]
|
||
if len(e["last"]) >= 29: # 覆盖到 testFloat 末字节才做形态判据
|
||
for name, off in FLOATS:
|
||
e["hist"].setdefault(name, []).append(
|
||
struct.unpack_from("<f", e["last"], off)[0])
|
||
if len(e["hist"][name]) > 400:
|
||
e["hist"][name].pop(0)
|
||
# 形态判据：本帧各字段是否"跳出带又回来"需要下一帧，留到下次比较
|
||
for name, off in FLOATS:
|
||
h = e["hist"].get(name) or []
|
||
if len(h) >= 4:
|
||
a, b2 = h[-3], h[-2] # 上上帧 / 上一帧(刚确定)
|
||
# 检查上一帧相对其邻居是否孤立尖峰（其前其后由 hist 提供）
|
||
if len(h) >= 4:
|
||
pre, post = h[-4], h[-1]
|
||
band = max(abs(b2 - pre), abs(post - b2), 1e-6)
|
||
if abs(b2 - (pre + post) / 2) > 20 * band:
|
||
spikes.append({"t": round(time.time() - t0, 2),
|
||
"node": nd, "field": name,
|
||
"pre": pre, "spike": b2, "post": post})
|
||
e["cur"], e["first"], e["last"], e["seen"] = seq, d, d, 0
|
||
lf.close()
|
||
|
||
print(f"轮询 {polls} 次 / {DUR_S:.0f}s（{polls/DUR_S:.0f} 次/s，连接错误 {err}）")
|
||
print(f"判定功效：同一 seq 被读第二遍的机会 = {cov} 次")
|
||
print(f"主判据违规（同一 seq 出现两份不同 48B）= {len(viol)}")
|
||
for r in viol[:6]:
|
||
print(" ", r["cls"], f"t={r['t']}s node={r['node']} seq={r['seq']} mix_j={r.get('mix_j')}")
|
||
print(f" first={r['first'][:64]}")
|
||
print(f" later={r['later'][:64]}")
|
||
print(f"旁判据（孤立尖峰，形态学，可能有真突变混入）= {len(spikes)}")
|
||
for s in spikes[:4]:
|
||
print(f" t={s['t']}s node={s['node']} {s['field']}: {s['pre']:.6g} -> {s['spike']:.6g} -> {s['post']:.6g}")
|
||
print(f"明细: {LOG}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|