DCU_DesktopController/scripts/can_bridge_reconcile.py

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