Files
DCU_ECAT_Master/_probe_stuck.py
T
cbb fc9e8f34e5 diag: 加邮箱楔死探针三件套, 移除 _repro_hang.py
_probe_wedge.py   逐环节定位死点: A 写 SM0 的 WKC → B 0x220 bit0 是否置位
                  → C SSC 是否清事件并置 SM1 就绪 → D SM1 是否滞留旧响应,
                  用于区分 ESC 硬件楔死与 SSC 软件死循环。
_probe_wedge2.py  按真实邮箱协议走: 查写邮箱空 → 写满 256B → 验事件/响应。
_probe_stuck.py   验"SM1 响应滞留 → 邮箱死锁"假说: 复现失败 → 手动读走 SM1
                  清 bit3 → 再读是否恢复。若恢复则主站侧可自愈。

_repro_hang.py 的闪退成因已定位(见 a1ec486, config_map 撞楔死邮箱), 且规避
已内建到 op/start 入口, 该复现脚本失去用途, 删除。

README 同步脚本清单, 并记下 SM = 0x0800 + 8n 这个易错点。
.gitignore 排除 _analyze_log.py(硬编码已失效的临时日志路径)。
2026-09-03 02:36:07 +08:00

78 lines
2.4 KiB
Python

"""_probe_stuck.py — 验证"SM1 响应滞留 → 邮箱死锁"假说。
板上现状: SM1 挂着一个未读走的响应(上一探针注入)。实验:
1. SOEM SDO 读 → 预期失败(复现楔死现象)
2. 手动读走 SM1 全长(清 bit3) → 模拟"主站把滞留响应读掉"
3. SOEM SDO 再读 → 是否恢复?
若恢复 = 根治方向明确:邮箱死锁的物理本质是响应滞留,主站侧可自愈,从站侧可加超时复位。
"""
import struct
import sys
import ecat_server as es
import pysoem
r = es.api_open({"substr": "I226"})
if not r.get("ok"):
sys.exit(1)
m = es.g_master
s = m.slaves[0]
def fprd(addr, size):
try:
return bytes(s._fprd(addr, size, 20000))
except Exception:
return None
def sm1_state(tag):
d = fprd(0x0808, 8)
if d is None:
print(f"[{tag}] SM1 读失败", flush=True)
return None
start, length = struct.unpack("<HH", d[:4])
st = d[5]
print(f"[{tag}] SM1 st=0x{st:02X}(bit3={st >> 3 & 1} 读邮箱满)", flush=True)
return {"start": start, "len": length, "st": st}
m.state = pysoem.INIT_STATE | pysoem.STATE_ACK
m.write_state()
m.state_check(pysoem.INIT_STATE, 1_000_000)
m.config_init()
m.state = pysoem.PREOP_STATE | pysoem.STATE_ACK
m.write_state()
print("PREOP:", "OK" if m.state_check(pysoem.PREOP_STATE, 1_000_000) == pysoem.PREOP_STATE else "FAIL", flush=True)
# --- 1. 现状: SM1 挂着滞留响应 ---
sm1_state("实验前")
print("\n--- SOEM SDO #1(预期失败) ---", flush=True)
try:
d = bytes(s.sdo_read(0x1000, 0, 8))
print(f" 成功?! = {d.hex().upper()}", flush=True)
except Exception as e:
print(f" 失败: {type(e).__name__}: {e}", flush=True)
sm1_state("SDO#1 后")
# --- 2. 读走 SM1 滞留响应(全长度读取,读到末字节才清 bit3) ---
sm1 = sm1_state("读滞留前")
if sm1 and (sm1["st"] & 0x08):
d = fprd(sm1["start"], sm1["len"])
print(f" 读走 SM1 {sm1['len']}B: {d[:16].hex().upper()}...", flush=True)
sm1_state("读走后")
# --- 3. SOEM SDO 再试 ---
print("\n--- SOEM SDO #2(读走滞留后) ---", flush=True)
try:
d = bytes(s.sdo_read(0x1000, 0, 8))
print(f" 成功! = {d.hex().upper()}", flush=True)
print(">>> 假说成立:响应滞留是死锁根源,读走即恢复!", flush=True)
except Exception as e:
print(f" 仍失败: {type(e).__name__}: {e}", flush=True)
sm1_state("SDO#2 后")
es.api_close({})
print("DONE", flush=True)