cbb
e91d53e0fc
- esp32_server: Popen 后外部 SetPriorityClass(RT/High)，native 自设会被钳成 High - 入库 skip_* 标定/实验脚本与 REPORT_ecat_skip_opt_20260917 - 前端 WIP：layout/app/polling/boot_ecat/index - 不含 lab/skip_meter 原始采样与 preview 截图
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""0.25ms 下 A/B：静默 vs 模拟 spi_test 前端轮询。"""
|
|
import json, time, urllib.request, os
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
LAB = os.path.join(ROOT, "lab", "skip_meter")
|
|
|
|
def post(p, b=None, timeout=5):
|
|
d = json.dumps(b or {}).encode()
|
|
r = urllib.request.Request("http://127.0.0.1:8090"+p, data=d, headers={"Content-Type":"application/json"})
|
|
with urllib.request.urlopen(r, timeout=timeout) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
def snap():
|
|
st = post("/api/motor/state")
|
|
dbg = post("/api/motor/dbg")
|
|
return {
|
|
"cycles": st.get("cycles") or 0,
|
|
"skip_total": dbg.get("skip_total") or 0,
|
|
"gseq": (st.get("head") or {}).get("global_seq") or 0,
|
|
"period": st.get("period_ms"),
|
|
"wkc": st.get("wkc"),
|
|
"pcore": st.get("pcore_mask"),
|
|
"cyc_us": st.get("cyc_us"),
|
|
}
|
|
|
|
def rate(a, b, dt):
|
|
dc = b["cycles"] - a["cycles"]
|
|
ds = (b["skip_total"] - a["skip_total"]) # may wrap? long long, no
|
|
return {
|
|
"dt": round(dt, 3),
|
|
"pdo_hz": round(dc / dt, 1) if dt else None,
|
|
"late_per_s": round(ds / dt, 2) if dt else None,
|
|
"skip_frac": round(100.0 * ds / dc, 2) if dc else None,
|
|
"gseq_per_s": round(((b["gseq"] - a["gseq"]) & 0xFFFFFFFF) / dt, 1) if dt else None,
|
|
"dcyc": dc, "dskip": ds,
|
|
}
|
|
|
|
def arm_period(ms):
|
|
try:
|
|
post("/api/motor/stop")
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.15)
|
|
post("/api/motor/period", {"period_ms": ms})
|
|
time.sleep(0.15)
|
|
post("/api/motor/start")
|
|
time.sleep(0.5)
|
|
|
|
print("ensure 0.25ms")
|
|
arm_period(0.25)
|
|
time.sleep(3)
|
|
|
|
# --- A: fully quiet ---
|
|
a0 = snap()
|
|
t0 = time.time()
|
|
# no HTTP
|
|
time.sleep(10)
|
|
a1 = snap()
|
|
A = rate(a0, a1, time.time() - t0)
|
|
print("QUIET", json.dumps(A, ensure_ascii=False))
|
|
print(" cyc_us", json.dumps(a1.get("cyc_us"), ensure_ascii=False))
|
|
|
|
# --- B: hammer like spi_test (state+load+dbg+rxraw) ---
|
|
time.sleep(1)
|
|
b0 = snap()
|
|
t0 = time.time()
|
|
n = 0
|
|
while time.time() - t0 < 10:
|
|
try:
|
|
post("/api/motor/state")
|
|
post("/api/app/load")
|
|
post("/api/motor/dbg")
|
|
post("/api/motor/rxraw")
|
|
n += 1
|
|
except Exception as e:
|
|
print("hammer err", e)
|
|
break
|
|
time.sleep(0.05) # ~20Hz full set, similar to busy panel
|
|
b1 = snap()
|
|
B = rate(b0, b1, time.time() - t0)
|
|
print("HAMMER", json.dumps(B, ensure_ascii=False), "polls", n)
|
|
print(" cyc_us", json.dumps(b1.get("cyc_us"), ensure_ascii=False))
|
|
|
|
# --- C: only motor/state at max (no dbg/rxraw) ---
|
|
time.sleep(1)
|
|
c0 = snap()
|
|
t0 = time.time()
|
|
n = 0
|
|
while time.time() - t0 < 10:
|
|
try:
|
|
post("/api/motor/state")
|
|
n += 1
|
|
except Exception:
|
|
break
|
|
time.sleep(0.05)
|
|
c1 = snap()
|
|
C = rate(c0, c1, time.time() - t0)
|
|
print("STATE_ONLY", json.dumps(C, ensure_ascii=False), "polls", n)
|
|
|
|
out = {"quiet": A, "hammer": B, "state_only": C,
|
|
"quiet_cyc": a1.get("cyc_us"), "hammer_cyc": b1.get("cyc_us")}
|
|
path = os.path.join(LAB, f"summary_fe_ab_{int(time.time())}.json")
|
|
json.dump(out, open(path, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
|
|
print("WROTE", path)
|