DCU_DesktopController/scripts/check_web_send_latency.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""模拟网页全链路自测（改 native/前端后必跑）。

覆盖浏览器实际路径（同源 :2625 /ecat/api/* + 静态资源），
以及 GD32 出口（bridge.tx_count/ack）与 WiFi 对照。

用法:
  python scripts/e2e_web_send_latency.py
  python scripts/e2e_web_send_latency.py --n 8
退出码: 0=PASS, 1=FAIL
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625")
ECAT = PAGE + "/ecat/api"
ESP = os.environ.get("DCU_ESP_IP", "192.168.31.127")

PASS: list[str] = []
FAIL: list[str] = []


def ok(msg: str) -> None:
    PASS.append(msg)
    print(f"  ✓ {msg}")


def bad(msg: str) -> None:
    FAIL.append(msg)
    print(f"  ✗ {msg}")


def http_json(url: str, obj=None, timeout=8.0):
    data = b"{}" if obj is None else json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    req = urllib.request.Request(
        url, data=data,
        headers={"Content-Type": "application/json", "Accept": "application/json"},
        method="POST" if obj is not None else "GET",
    )
    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, r.status, json.loads(body.decode("utf-8", "replace") or "{}")
        except Exception:
            return ms, r.status, {"_raw_len": len(body)}
    except Exception as e:
        ms = (time.perf_counter() - t0) * 1000.0
        return ms, None, {"ok": False, "error": str(e)}


def http_get(url: str, timeout=5.0):
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r:
            body = r.read()
        return (time.perf_counter() - t0) * 1000.0, r.status, body
    except Exception as e:
        return (time.perf_counter() - t0) * 1000.0, None, str(e).encode()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--n", type=int, default=6)
    args = ap.parse_args()

    print("=== 1) 页面与静态资源（模拟浏览器打开电机页） ===")
    ms, st, body = http_get(PAGE + "/pages/motor.html")
    html = body.decode("utf-8", "replace") if isinstance(body, (bytes, bytearray)) else str(body)
    if st == 200 and "motor.html" or st == 200:
        ok(f"motor.html HTTP {st} ({ms:.1f}ms)")
    else:
        bad(f"motor.html 失败 st={st} {body[:80]!r}")
    css_ref = re.search(r'motor\.css\?v=([0-9a-z]+)', html)
    js_ref = re.search(r'main\.js\?v=([0-9a-z]+)', html)
    if css_ref:
        ok(f"CSS 引用 motor.css?v={css_ref.group(1)}")
    else:
        bad("motor.html 无 motor.css 引用")
    if js_ref:
        ok(f"JS 引用 main.js?v={js_ref.group(1)}")
    else:
        bad("motor.html 无 main.js 引用")
    if "sendLatency" in html or "send-trace" in html:
        bad("页面仍含延迟埋点（应已删除）")
    else:
        ok("页面无延迟埋点残留")

    for rel in (
        "/css/motor.css",
        "/js/motor/main.js",
        "/js/motor/transport.js",
        "/js/motor/transport-ecat.js",
        "/js/motor/transport-wifi.js",
        "/js/common/desktop-bridge.js",
        "/js/common/device-link.js",
        "/js/common/app-lease.js",
    ):
        ms, st, b = http_get(PAGE + rel)
        if st == 200 and isinstance(b, (bytes, bytearray)) and len(b) > 100:
            ok(f"{rel} {st} {len(b)}B ({ms:.1f}ms)")
        else:
            bad(f"{rel} 资源异常 st={st} {b[:60]!r}")

    print("\n=== 2) EtherCAT 控制面（网页同源 /ecat/api/*） ===")
    ms, st, h = http_json(ECAT + "/native/health", {})
    if h.get("ok") and h.get("bus_open") == 1:
        ok(f"native/health {ms:.1f}ms dc={h.get('dc')} dc_us={h.get('dc_cycle_us')} loop={h.get('loop')} wkc={h.get('wkc')}")
    else:
        bad(f"native/health 异常 {h}")
    if h.get("dc") == 1 and h.get("dc_cycle_us") == 250:
        ok("DC/周期为 250µs（改动生效）")
    else:
        bad(f"DC/周期未钉死 250µs: dc={h.get('dc')} us={h.get('dc_cycle_us')}")

    ms, st, r = http_json(ECAT + "/motor/start",
                          {"period_ms": 0.25, "dc_cycle_us": 250, "dc_shift_us": 0,
                           "force_reopen": True}, 30)
    log = r.get("log") if isinstance(r, dict) else None
    if r.get("ok"):
        ok(f"motor/start ok period={r.get('period_ms')} ({ms:.0f}ms) log={log}")
    else:
        bad(f"motor/start 失败 {r}")
    ms, st, h = http_json(ECAT + "/native/health", {})
    if h.get("dc") == 1 and h.get("dc_cycle_us") == 250 and h.get("loop") == 1:
        ok(f"start 后 health: dc=250 loop=1 wkc={h.get('wkc')} ({ms:.1f}ms)")
    else:
        bad(f"start 后 health 不健康: {h}")

    ms, st, snap = http_json(ECAT + "/motor/state", {})
    if snap.get("ok"):
        br = snap.get("bridge") or {}
        ok(f"motor/state {ms:.1f}ms period={snap.get('period_ms')} wkc={snap.get('wkc')} "
           f"tx={br.get('tx_count')} ack={br.get('ack')} online="
           f"{[m.get('node') for m in (snap.get('motors') or []) if m.get('online')]}")
    else:
        bad(f"motor/state 失败 {snap}")

    print("\n=== 3) 模拟网页点击「发送 CMD」×N（POST motor/send + 看 GD32 出口） ===")
    frame = {"frames": [{"ch": 2, "id": 0x00030200, "dlc": 32, "data": "03" * 32}]}

    def bridge_snap():
        _, _, s = http_json(ECAT + "/motor/state", {})
        if not isinstance(s, dict) or not s.get("ok"):
            return None
        br = s.get("bridge") or {}
        return {
            "ms_tx": br.get("tx_count"),
            "ack": br.get("ack"),
            "sent_seq": s.get("sent_seq"),
            "tx_pack": (s.get("tx_pack") or {}).get("seq"),
            "txres": br.get("tx_result"),
        }

    send_lats = []
    outlet_lats = []
    for i in range(args.n):
        pre = bridge_snap()
        ms, st, r = http_json(ECAT + "/motor/send", frame, 6.0)
        send_lats.append(ms)
        send_ok = isinstance(r, dict) and r.get("ok") is True
        outlet = None
        t0 = time.perf_counter()
        while (time.perf_counter() - t0) < 2.0:
            post = bridge_snap()
            if pre and post and (
                post["ms_tx"] != pre["ms_tx"]
                or post["ack"] != pre["ack"]
            ):
                outlet = (time.perf_counter() - t0) * 1000.0
                outlet_lats.append(outlet)
                break
            time.sleep(0.001)
        if send_ok and outlet is not None:
            ok(f"send[{i}] HTTP {ms:.1f}ms → GD32 出口 {outlet:.1f}ms "
               f"tx {pre['ms_tx']}→{bridge_snap() and bridge_snap()['ms_tx']} "
               f"ack {pre['ack']}→{'?'}")
        elif send_ok and outlet is None:
            post = bridge_snap()
            bad(f"send[{i}] HTTP ok {ms:.1f}ms，但 2s 内 GD32 出口无变化 "
                f"pre={pre} post={post} resp={r}")
        else:
            bad(f"send[{i}] HTTP 失败 {ms:.1f}ms {r}")
        time.sleep(0.05)

    def pct(xs, q):
        if not xs:
            return None
        xs = sorted(xs)
        return xs[min(len(xs) - 1, int(q * (len(xs) - 1)))]

    if send_lats:
        s_p50 = pct(send_lats, 0.5)
        if s_p50 is not None and s_p50 < 50:
            ok(f"控制面 HTTP send p50={s_p50:.1f}ms（网页路径）")
        elif s_p50 is not None:
            bad(f"控制面 HTTP send 过慢 p50={s_p50:.1f}ms")
    if outlet_lats:
        o_p50 = pct(outlet_lats, 0.5)
        o_max = max(outlet_lats)
        if o_p50 is not None and o_p50 < 50 and o_max < 200:
            ok(f"GD32 CAN 出口 p50={o_p50:.1f}ms max={o_max:.1f}ms")
        else:
            bad(f"GD32 CAN 出口过慢 p50={o_p50} max={o_max:.1f}ms（或仅有部分命中）")
    else:
        bad("没有任何 send 观测到 GD32 bridge.tx/ack 变化 — 电机侧未消费输出 PDO")

    print("\n=== 4) WiFi 对照（ESP32 /can/send_batch） ===")
    try:
        body = {"ch": 2, "frames": [{"id": 0x00030200, "dlc": 32, "data": "03" * 32}]}
        data = json.dumps(body).encode()
        req = urllib.request.Request(
            f"http://{ESP}/can/send_batch", data=data,
            headers={"Content-Type": "application/json"}, method="POST")
        t0 = time.perf_counter()
        with urllib.request.urlopen(req, timeout=5) as resp:
            raw = resp.read()
        wms = (time.perf_counter() - t0) * 1000.0
        j = json.loads(raw.decode("utf-8", "replace") or "{}")
        if j.get("status") == "ok":
            ok(f"WiFi send_batch {wms:.0f}ms ok（链路可用）")
        else:
            bad(f"WiFi send_batch 回执异常 {wms:.0f}ms {j}")
    except Exception as e:
        bad(f"WiFi send_batch 失败: {e}")

    print("\n=== 5) 点击→电机 STATUS 开始动（软件端到端） ===")
    print("  调用 check_click_to_motion.py（会发一小段速度指令后自动 Disable）")
    import subprocess
    py = sys.executable
    motion = os.path.join(ROOT, "scripts", "check_click_to_motion.py")
    rc = subprocess.call([py, motion, "--node", "3", "--ch", "2", "--vel", "40", "--timeout", "3"], cwd=ROOT)
    if rc == 0:
        ok("点击→STATUS 开始动 已测通（详见上表）")
    else:
        bad("点击→STATUS 开始动 测量失败/超时")

    print("\n=== 6) tx 冻结守卫可观测面（改 native 后必查） ===")
    # 这一段只回答"守卫还活着吗"，不测触发（触发要 8s+60s 冷却，见
    # scripts/check_guard_idle_silence.py 与 tools/ecat/check_guard_regression.py）。
    # 本脚本每次发送只等 2s，结构上看不到 8s 判据，所以以前"守卫被改死"与
    # "守卫健康"在这里完全同像——必须显式读 guard 块。
    try:
        _, h = post(PAGE + "/ecat/api/native/health", {}, 5)
        g = h.get("guard")
        if not isinstance(g, dict):
            bad("health 无 guard 块 ⇒ 跑的是旧 exe（不含守卫可观测面）；"
                "跑 tools/ecat/build_native.bat 换二进制")
        else:
            t0 = g.get("alive_ticks", 0)
            time.sleep(1.0)
            _, h2 = post(PAGE + "/ecat/api/native/health", {}, 5)
            t1 = (h2.get("guard") or {}).get("alive_ticks", 0)
            if t1 - t0 >= 3:
                ok(f"守卫心跳 +{t1 - t0}/1s，armed={g.get('armed')} fires={g.get('fires')} "
                   f"wr={h2.get('wr')} build={h2.get('guard_build')}")
            else:
                bad(f"守卫心跳仅 +{t1 - t0}/1s ⇒ wedge_worker 没在跑（守卫死了而页面仍全绿）")
            if h2.get("recovering"):
                ok("health.recovering=1：此刻正在重拉，绿灯不作数（可见面工作正常）")
    except Exception as e:
        bad(f"guard 可见面读取失败: {e}")

    print("\n=== 汇总 ===")
    print(f"PASS {len(PASS)}  FAIL {len(FAIL)}")
    if FAIL:
        print("失败项:")
        for x in FAIL:
            print("  -", x)
        return 1
    print("全链路自测通过：网页资源 + ECAT 控制面 + 发送出口 + 点击→STATUS 动作延迟")
    return 0


if __name__ == "__main__":
    sys.exit(main())