refactor(server): P2 拆出 ecat_client/ecat_proc 并归一化桥路径

This commit is contained in:
cbb
2026-09-21 02:22:32 +08:00
parent 67dd1702f0
commit b70f5ed0d6
4 changed files with 943 additions and 794 deletions
+497
View File
@@ -0,0 +1,497 @@
#!/usr/bin/env python3
"""ecat_client.py — EtherCAT native 客户端(端口探测 / 裸 socket HTTP / ESF1 快发)。
从 esp32_server.py 外提(R8)。本模块只做「怎么跟 ecat_native 说话」:
- 路径归一化(R9:桥与代理共用 normalize_ecat_api_path)
- 端口探测(8090..8095 漂移)与裸 socket HTTP/1.0 客户端
- ESF1 二进制快发打包/解包 + 持久 TCP
- 代理重发策略(R1/P0:非幂等不重发、先复用健康实例由 proc 负责)
双路径说明(R10)——两条入口,同一 ecat_client:
- 浏览器 = HTTP 代理:
页面 fetch POST /ecat/api/* → esp32_server.Handler.do_POST
→ ecat_client.ecat_proxy → _ecat_post
- 桌面 APP = GuiApi 桥:
JS window.pywebview.api.ecat_api → esp32_server._GuiApi.ecat_api
→ ecat_client.api_call → ecat_proxy → _ecat_post
两者最终都走本模块的 normalize_ecat_api_path + _ecat_post/_ecat_proxy。
不删桥的理由:桌面桥少一跳 HTTP(GUI 进程直连 native);浏览器没有
pywebview.api,只能走同源代理。外部 API 语义不变。
进程治理(autostart / kill / RealTime / 跨进程锁)在 ecat_proc.py。
本模块通过 set_autostart_hook 接收回调,避免 ecat_client ↔ ecat_proc
循环 import。BASE_DIR 只在 esp32_server 定义一次并传给 ecat_proc.init。
"""
from __future__ import annotations
import json
import socket
import struct
import threading
# ── EtherCAT 后端反向代理常量 ──────────────────────────────────────────
# 电机页的 EtherCAT 传输要打 tools/ecat 的 ecat_native.exe 的 /api/motor/*,
# 但那个服务**只绑 127.0.0.1 且完全没有 CORS**。本服务做同源转发。
# 端口自动探测: EtherCAT 服务(含 native)会在 8090..8095 间漂移。
ECAT_PROXY_PREFIX = "/ecat/api/"
ECAT_PORTS = (8090, 8091, 8092, 8093, 8094, 8095)
g_ecat_port = None # 记住上次探到的端口, 避免每次请求都从 8090 试起
# ── PDO 快发(ESF1)── tools/ecat ecat_native 的二进制直通 ──────────
# 布局与 docs/compose/spec/pdo-fast-send.md 一致。**仅** native 专用 TCP 口
# (health.fast_send_port);2026-09-23 堵死慢发:已删除 HTTP/JSON 回落。
# 持久连接:连接阶段不设 timeout(Windows 非阻塞 connect 会撞
# ~10ms 调度粒度,见 _ecat_post 注释);连上后再 settimeout。
ESF_MAGIC = 0x31465345
FSOK_MAGIC = 0x4B4F5346
ESF_FRAME_W = 72
g_fast_send_port = 0 # 从 native health 发现的快发 TCP 口;0=未知/不可用
g_fast_sock = None
g_fast_sock_lock = threading.Lock()
# ── R1/P0: 代理重发策略 ────────────────────────────────────────────────
# 连接失败后 autostart 再打一次同一 POST,对写路径 = 可能二次执行。
# 仅对明确只读/可安全重试的路由允许 autostart 后自动重发一次。
_ECAT_IDEMPOTENT_TAILS = frozenset({
"status", "native/health", "app/load", "motor/state", "motor/rx",
"motor/version", "motor/dbg", "load",
})
# ecat_proc.init 注入的进程拉起回调: () -> (port|None, log_lines)
_autostart_hook = None
def set_autostart_hook(fn):
"""由 ecat_proc.init 调用:注入 autostart,避免循环 import。"""
global _autostart_hook
_autostart_hook = fn
def _autostart():
if _autostart_hook is None:
return None, ["EtherCAT 进程管理未初始化(ecat_proc.init 未调用)"]
return _autostart_hook()
# ── R9: 桥与代理共用的路径归一化 ──────────────────────────────────────
def normalize_ecat_api_path(path):
"""任意形态 ECAT API 路径 → (proxy_path, native_target)。
输入可为:
"motor/send" | "api/motor/send" | "/api/motor/send"
| "/ecat/api/motor/send" | "ecat/api/motor/send"
返回:
proxy_path = "/ecat/api/<tail>" # 浏览器 HTTP 代理路径
native_target = "/api/<tail>" # 直连 native 的路径
_GuiApi.ecat_api(桌面桥)与 _ecat_proxy(浏览器代理)共用本函数,
删除两边各自实现的剥前缀逻辑(R9)。
"""
p = str(path or "").strip()
p = p.split("?", 1)[0]
while p.startswith("/"):
p = p[1:]
if p.startswith("ecat/api/"):
p = p[len("ecat/"):] # → api/<tail>
elif p.startswith("ecat/"):
p = p[len("ecat/"):]
if p.startswith("api/"):
tail = p[len("api/"):]
else:
tail = p
proxy_path = ECAT_PROXY_PREFIX + tail
native_target = "/api/" + tail
return proxy_path, native_target
# ── 端口探测 ──────────────────────────────────────────────────────────
def _port_taken(port):
"""Windows 的 SO_REUSEADDR 允许同端口重复 bind(实测会双监听、连接随机落),
故先用一次真实连接探测占用, 不能只靠 bind 是否抛错。"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(0.3)
return s.connect_ex(("127.0.0.1", port)) == 0
def _ecat_find_port():
"""返回正在监听的 ecat 服务端口, 没有则 None。"""
ports = ([g_ecat_port] if g_ecat_port else []) + \
[p for p in ECAT_PORTS if p != g_ecat_port]
for p in ports:
if _port_taken(p):
return p
return None
# ── ESF1 打包/解包 ────────────────────────────────────────────────────
def _hex_to_bytes64(data) -> bytes:
"""帧 data → 64 字节(短则右侧补 0)。"""
if data is None:
return b"\x00" * 64
if isinstance(data, (bytes, bytearray)):
b = bytes(data)[:64]
return b + b"\x00" * (64 - len(b))
s = str(data).strip().replace(" ", "")
if s.lower().startswith("0x"):
s = s[2:]
if len(s) % 2:
s = "0" + s
b = bytes.fromhex(s)[:64] if s else b""
return b + b"\x00" * (64 - len(b))
def pack_esf(frames) -> bytes:
"""frames=[{ch,id,dlc,data(hex)}] → ESF1 请求体。"""
items = list(frames or [])
n = len(items)
if n < 1 or n > 64:
raise ValueError(f"frames 数量必须 1..64,收到 {n}")
out = bytearray(struct.pack("<IBBH", ESF_MAGIC, n, 0, 0))
for fr in items:
if not isinstance(fr, dict):
raise ValueError(f"帧必须是对象,收到 {type(fr).__name__}")
ch = int(fr.get("ch", 0))
dlc = int(fr.get("dlc", 0) or 0)
if ch < 0 or ch > 2:
raise ValueError(f"通道必须是 0/1/2,收到 {ch}")
if dlc < 0 or dlc > 64:
raise ValueError(f"DLC 超出 CANFD 上限 64: {dlc}")
fid = fr.get("id", 0)
if isinstance(fid, str):
fid = int(fid, 0)
data = _hex_to_bytes64(fr.get("data"))
out += struct.pack("<BBHI", ch & 0xFF, dlc & 0xFF, 0, int(fid) & 0xFFFFFFFF)
out += data
return bytes(out)
def unpack_fsok(resp8: bytes) -> dict:
if not resp8 or len(resp8) < 8:
return {"ok": False, "queued": 0, "backlog": 0, "log": ["FSOK 响应过短"]}
magic, ok, queued, backlog = struct.unpack_from("<IBBH", resp8, 0)
if magic != FSOK_MAGIC:
return {"ok": False, "queued": 0, "backlog": 0, "log": ["FSOK magic 不匹配"]}
return {
"ok": bool(ok),
"queued": int(queued),
"backlog": int(backlog),
"log": [] if ok else ["native 快发拒绝(未连接/解析失败)"],
}
def _note_fast_port(obj) -> None:
global g_fast_send_port
if isinstance(obj, dict):
p = obj.get("fast_send_port")
if isinstance(p, (int, float)) and int(p) > 0:
g_fast_send_port = int(p)
def _discover_fast_send_port(timeout=2.0, force=False) -> int:
global g_fast_send_port
if g_fast_send_port and not force:
return g_fast_send_port
try:
_status, data = _ecat_proxy(ECAT_PROXY_PREFIX + "native/health", b"{}", timeout)
obj = json.loads(data.decode("utf-8", "replace") or "{}")
_note_fast_port(obj)
if force and not isinstance(obj, dict):
g_fast_send_port = 0
except Exception:
if force:
g_fast_send_port = 0
return g_fast_send_port
def _close_fast_sock() -> None:
global g_fast_sock
with g_fast_sock_lock:
s = g_fast_sock
g_fast_sock = None
if s is not None:
try:
s.close()
except Exception:
pass
def _fast_sock_connect(port: int, timeout: float):
sock = socket.create_connection(("127.0.0.1", int(port)), timeout=None)
sock.settimeout(timeout)
try:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except Exception:
pass
return sock
def ecat_send_raw_frames(frames, timeout_ms=2000) -> dict:
"""ESF1 快发:**仅**持久 TCP → native fast_send_port。
2026-09-23 用户要求堵死慢发:删除 HTTP send_raw / JSON motor/send 回落。
失败即 ok:false,由前端暴露错误;不自动走慢路径。
持久连接上的 sendall/recv **必须整段持锁**,否则并发请求会把 FSOK
错配到别的帧。连接失败时清掉缓存端口并重探一次(仍只走 TCP)。
"""
global g_fast_sock, g_fast_send_port
try:
body = pack_esf(frames)
except Exception as e:
return {"ok": False, "queued": 0, "backlog": 0, "path": "none",
"log": [f"ESF 打包失败: {e}"]}
timeout = max(0.2, float(timeout_ms or 2000) / 1000.0)
def _tcp_once(port: int) -> dict:
nonlocal body, timeout
global g_fast_sock
r = None
with g_fast_sock_lock:
if g_fast_sock is None:
g_fast_sock = _fast_sock_connect(port, timeout)
sock = g_fast_sock
sock.sendall(body)
resp = b""
while len(resp) < 8:
chunk = sock.recv(8 - len(resp))
if not chunk:
raise OSError("快发口连接已关闭")
resp += chunk
r = unpack_fsok(resp)
r["path"] = "fast_tcp"
return r
last_err = None
for attempt in (0, 1):
port = _discover_fast_send_port(timeout, force=(attempt == 1))
if not port:
last_err = "native 无 fast_send_port(服务未起或版本过旧)"
break
try:
return _tcp_once(port)
except Exception as e:
last_err = f"快发口异常: {e}"
_close_fast_sock()
g_fast_send_port = 0
if attempt == 0:
continue
return {
"ok": False,
"queued": 0,
"backlog": 0,
"path": "fast_tcp",
"log": [last_err or "二进制快发失败(已禁用 HTTP/JSON 回落)"],
}
# ── 幂等判定 / 连接拒绝判定(P0 保留)────────────────────────────────
def _ecat_target_idempotent(target: str) -> bool:
"""/api/<tail> 是否允许 autostart 后自动重发。默认 False(偏安全)。"""
tail = target[len("/api/"):] if target.startswith("/api/") else target.lstrip("/")
tail = tail.split("?", 1)[0].strip("/")
if tail in _ECAT_IDEMPOTENT_TAILS:
return True
# 显式只读前缀
if tail.startswith(("native/", "diag/")) and "stop" not in tail and "kill" not in tail:
return True
return False
def _ecat_connect_refused(exc) -> bool:
"""异常是否属于「根本没连上」(请求未发出,重发安全)。"""
if isinstance(exc, ConnectionRefusedError):
return True
if isinstance(exc, OSError):
if getattr(exc, "winerror", None) in (10061, 10065):
return True
if getattr(exc, "errno", None) in (111, 10061):
return True
msg = str(exc).lower()
return ("refused" in msg) or ("actively refused" in msg)
# ── 裸 socket HTTP 客户端 + 代理 ──────────────────────────────────────
def _ecat_post(port, path, body, timeout, connect_timeout=None, ctype=None):
"""向 EtherCAT 后端(native)转发一次 POST, 返回 (status, body_bytes)。
**不要改回 urllib/http.client 并给它传 timeout。** 那样每次请求会多花约 10ms:
给 socket 设 timeout 会让 Python 把它切成非阻塞模式, 而 Windows 上非阻塞
connect 要靠 select 轮询完成, 于是撞上约 10ms 的系统调度粒度。实测
(本机回环, /api/motor/state):
urllib.urlopen(req, timeout=5) p50 11.0 ms
http.client + timeout p50 10.7 ms
裸 socket 连接阶段不设 timeout p50 0.58 ms
后端自己只用 0.58ms, 那 10ms 全是这一层白交的过路费 ——
电机页 EtherCAT 链路的刷新上限直接被它按在 ~76Hz。
所以这里手写最小 HTTP/1.0 客户端: 连接阶段**不设** timeout(本机回环连接
要么立刻成功要么立刻 ECONNREFUSED, 不会挂住), 连上之后才设读写超时。
connect_timeout: 只给冷路径(逐个试漂移端口)用的连接超时。**热路径(缓存
端口)必须保持 None(阻塞连接)**——本机对死端口无论阻塞还是 settimeout 都
不是秒拒(阻塞 ~2s, settimeout(0.3) 吃满 0.3s 才回 WOULDBLOCK), 但活端口
的阻塞 connect 只要 ~0.6ms, 两害相权取其轻。
"""
payload = (
f"POST {path} HTTP/1.0\r\n"
f"Host: 127.0.0.1:{port}\r\n"
f"Content-Type: {ctype or 'application/json'}\r\n"
f"Content-Length: {len(body or b'')}\r\n"
"Connection: close\r\n"
"\r\n"
).encode("ascii") + (body or b"")
sock = socket.create_connection(("127.0.0.1", port), timeout=connect_timeout)
try:
sock.settimeout(timeout) # 连上之后再设: 只作用于收发, 不影响已完成的 connect
sock.sendall(payload)
chunks = []
while True:
chunk = sock.recv(65536)
if not chunk:
break
chunks.append(chunk)
finally:
sock.close()
raw = b"".join(chunks)
head, _, data = raw.partition(b"\r\n\r\n")
first = head.split(b"\r\n", 1)[0].split(b" ", 2)
if len(first) < 2 or not first[1].isdigit():
raise OSError(f"EtherCAT 后端回了无法解析的响应头: {head[:120]!r}")
# 404 等业务错误要原样透传: 后端的 404 体里带
# "服务在改动之后没重启" 的排查提示, 吞掉它会让人白排查半天。
return int(first[1]), data
def ecat_proxy(path, body, timeout=6.0, ctype=None):
"""把 /ecat/api/xxx 转发到 EtherCAT 后端的 /api/xxx。
返回 (status, body_bytes)。路径归一化与桌面桥共用 normalize_ecat_api_path(R9)。
端口在 ECAT_PORTS 里探测并缓存到 g_ecat_port:
EtherCAT 后端会在 8090..8095 之间漂移(端口被占就挪),
写死 8090 会在它挪了之后表现为"页面点了没反应"。
连不上时返回 503 + 与后端同构的 {"ok":false,"log":[...]} 信封,
这样前端只有一条错误处理路径 —— 它不需要知道中间还隔着一层代理。
2026-09-12: 先用 _port_taken 快探,全灭则立刻 autostart;
不再对 6 个死端口各等 timeout 秒(那样要 30s+ 才会拉起服务)。
ctype: 可选 Content-Type(ESF1 二进制 body 用 application/octet-stream)。
P0/R1: 非幂等请求 autostart 后**不**自动重发,返回 503 由用户/前端显式重试。
"""
global g_ecat_port
# R9: 与 GuiApi 桥共用归一化。"motor/state" | "/api/motor/state"
# | "/ecat/api/motor/state" → target="/api/motor/state"
_proxy_path, target = normalize_ecat_api_path(path)
idem = _ecat_target_idempotent(target)
# 端口尝试顺序: 先缓存口, 失败再试其余漂移口(带短连接超时), 全灭才拉起。
# **健康路径上绝不能对死端口做预探测** —— 本机 Windows 对回环死端口的
# connect 不是秒拒(阻塞 ~2s, settimeout(0.3) 吃满 0.3s 回 WOULDBLOCK),
# 2026-09-12 实测每请求预探 5 个死端口 = 恒定 1.5s, 电机页"本页请求耗时
# 1517ms"、刷新被按到 0.66Hz 皆此。缓存口命中的代价只有 ~0.6ms。
ports = ([g_ecat_port] if g_ecat_port else []) + \
[p for p in ECAT_PORTS if p != g_ecat_port]
last = None
for port in ports:
try:
# 缓存口连不上大概率是后端挪了/死了, 阻塞 connect ~2s 可接受;
# 其余候选口用 0.3s 连接超时, 避免逐口 2s(全灭时 12s+)。
status, data = _ecat_post(
port, target, body, timeout,
connect_timeout=None if port == g_ecat_port else 0.3,
ctype=ctype)
g_ecat_port = port
return status, data
except Exception as e: # 连不上/超时 → 视情况试下一个端口
last = e
if g_ecat_port == port:
g_ecat_port = None # 缓存端口失效, 下轮重新全量探测
# P0/R1: 非幂等请求一旦可能已到达旧实例(超时/非连接拒绝),
# 禁止再打其它端口——否则 flash/motor/send 等会被重复执行。
if not idem and not _ecat_connect_refused(e):
break
# 无存活端口或全失败 → 单实例拉起;**非幂等路径禁止自动重发**
start_logs = []
try:
port, start_logs = _autostart()
except Exception as e:
last = e
port = None
start_logs = [f"autostart 异常: {type(e).__name__}: {e}"]
if port:
g_ecat_port = port
if not idem:
# 源请求可能已在旧实例上执行,或在超时途中被接收。
# 自动重发会二次写 flash / 重复 motor/start。返回 503,
# 由前端/用户显式重试(后端已就绪,重试通常一次即成功)。
msg = (
f"EtherCAT 服务已自动拉起,但 {target} 为非幂等路径,"
"已按约定不自动重发,请重试一次"
)
body = json.dumps(
{"ok": False, "log": start_logs + [msg]},
ensure_ascii=False).encode("utf-8")
return 503, body
try:
status, data = _ecat_post(port, target, body, timeout, ctype=ctype)
return status, data
except Exception as e:
last = e
msg = (f"EtherCAT 服务不可用: {type(last).__name__ if last else 'unknown'}")
body = json.dumps({"ok": False, "log": start_logs + [msg]},
ensure_ascii=False).encode("utf-8")
return 503, body
# 原名兼容:server / 脚本 / 测试 patch 点
_ecat_proxy = ecat_proxy
def _body_to_bytes(body) -> bytes:
"""桥/代理共用的 body 序列化:dict → JSON,str/bytes 原样。"""
if body is None:
return b"{}"
if isinstance(body, (bytes, bytearray)):
return bytes(body)
if isinstance(body, str):
return body.encode("utf-8")
try:
return json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
except Exception as e:
raise ValueError(f"body 序列化失败: {type(e).__name__}: {e}") from e
def api_call(path, body=None, timeout_ms=4000) -> dict:
"""统一 ECAT API 调用入口(R10)。桌面桥 GuiApi.ecat_api 直接调这里。
浏览器路径不经过本函数(HTTP → Handler → ecat_proxy),但两者最终
共用 normalize_ecat_api_path + ecat_proxy/_ecat_post,无第二套协议栈。
返回与 native 同构的 dict(ok/log/...)。
"""
proxy_path, _target = normalize_ecat_api_path(path)
try:
raw = _body_to_bytes(body)
except ValueError as e:
return {"ok": False, "log": [f"原生桥 {e}"]}
timeout = max(0.2, float(timeout_ms or 4000) / 1000.0)
try:
status, data = ecat_proxy(proxy_path, raw, timeout)
except Exception as e:
return {"ok": False, "log": [f"原生桥异常: {type(e).__name__}: {e}"]}
try:
obj = json.loads(data.decode("utf-8", "replace") or "{}")
except Exception:
obj = {"ok": False, "log": ["原生桥响应非 JSON"]}
if not isinstance(obj, dict):
obj = {"ok": False, "log": ["原生桥响应不是对象"]}
obj.setdefault("ok", status == 200)
if status >= 400 and "log" not in obj:
obj["log"] = [f"HTTP {status}"]
_note_fast_port(obj)
return obj
+314
View File
@@ -0,0 +1,314 @@
#!/usr/bin/env python3
"""ecat_proc.py — ecat_native 进程治理(跨进程锁 / autostart / kill / RealTime)。
从 esp32_server.py 外提(R8)。职责是「谁有权把 ecat_native 拉起来/杀掉」,
不碰 HTTP/静态/lease——那些留在 esp32_server.py。
路径注入:BASE_DIR **只在 esp32_server 定义一次**,启动时调用
`init(BASE_DIR)` 设置本模块的 ECAT_DIR / ECAT_NATIVE_EXE,并向
ecat_client 注册 autostart 钩子。
保留 P0 行为(不得回退):
- 跨进程命名互斥锁 _EcatProcXLock(GUI 进程 + 服务子进程串行 kill/start)
- **先复用健康监听实例**,绝不 kill 对方刚拉起的 native
- 默认网卡 I210(与 js/common/device-link.js / pages/motor.html 一致,R2)
- 拉起时带 --parent-pid,native 在本服务退出后自我了断
- stdout 落盘(DEVNULL 会吞掉崩溃现场)
"""
from __future__ import annotations
import ctypes
import os
import subprocess
import threading
import time
import ecat_client
# 网卡默认子串:与 js/common/device-link.js / pages/motor.html 保持一致(R2)
ECAT_DEFAULT_ADAPTER = "I210"
ECAT_DIR = None
ECAT_NATIVE_EXE = None
g_ecat_child = None # 由本服务拉起的 ecat 服务子进程(复用已有实例时为 None)
g_ecat_log_fp = None # native stdout 落盘句柄(见 autostart 内注释)
g_ecat_lock = threading.Lock() # 串行化自动启动, 避免并发请求拉起多个进程
# R3: 跨进程互斥(GUI 进程 + 服务子进程都可能 autostart/kill native)
_ECAT_XPROC_MUTEX_NAME = "Local\\DCU_Esp32Tools_EcatNativeProc"
def _resolve_ecat_dir(base_dir):
"""定位 tools/ecat(ecat_native.exe 的仓库,与本仓库平级)。
从 git worktree(.../tools/ESP32/.worktrees/<slug>)启动时,BASE_DIR 的
上一级是 .worktrees 而不是 tools,直接 join("..","ecat") 会拼成
.../ESP32/.worktrees/ecat,找不到 exe。先走回主检出再取兄弟目录。
"""
d = os.path.abspath(base_dir)
parent = os.path.dirname(d)
if os.path.basename(parent) == ".worktrees":
d = os.path.dirname(parent) # → .../tools/ESP32 主检出
return os.path.normpath(os.path.join(d, "..", "ecat"))
def init(base_dir: str) -> dict:
"""由 esp32_server 在导入时调用:注入 BASE_DIR,注册 autostart 钩子。"""
global ECAT_DIR, ECAT_NATIVE_EXE
ECAT_DIR = _resolve_ecat_dir(base_dir)
ECAT_NATIVE_EXE = os.path.join(ECAT_DIR, "build_native", "ecat_native.exe")
ecat_client.set_autostart_hook(autostart)
return {"ecat_dir": ECAT_DIR, "native_exe": ECAT_NATIVE_EXE}
class _EcatProcXLock:
"""Windows 命名互斥锁:串行化所有会 kill/拉起 ecat_native 的路径。"""
def __init__(self, timeout_ms=60000):
self.timeout_ms = timeout_ms
self._handle = None
self._owned = False
def __enter__(self):
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
self._handle = k32.CreateMutexW(None, False, _ECAT_XPROC_MUTEX_NAME)
if not self._handle:
return self
r = k32.WaitForSingleObject(self._handle, self.timeout_ms)
# WAIT_OBJECT_0=0, WAIT_ABANDONED=0x80
self._owned = r in (0, 0x80)
return self
def __exit__(self, *exc):
if not self._handle:
return False
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
try:
if self._owned:
k32.ReleaseMutex(self._handle)
finally:
k32.CloseHandle(self._handle)
self._handle = None
return False
def _set_ecat_realtime(pid: int) -> bool:
"""把 ecat_native 提到 RealTime 进程优先级。
native 自己 SetPriorityClass(RT) 会被系统静默钳成 High(令牌无
SeIncreaseBasePriorityPrivilege)。由本服务(启动方)从外部设置才能粘住。
pdo 线程已是 TIME_CRITICAL:进程类 High→线程 15,RealTime→线程 31,
后者才扛得住桌面 DPC/调度抢占。实测 1ms 档 late ~12/s → 个位数。
失败返回 False,不影响启动。
"""
try:
import ctypes.wintypes as wt # noqa: F401
PROCESS_SET_INFORMATION = 0x0200
REALTIME_PRIORITY_CLASS = 0x00000100
HIGH_PRIORITY_CLASS = 0x00000080
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
h = k32.OpenProcess(PROCESS_SET_INFORMATION, False, int(pid))
if not h:
return False
try:
if k32.SetPriorityClass(h, REALTIME_PRIORITY_CLASS):
return True
return bool(k32.SetPriorityClass(h, HIGH_PRIORITY_CLASS))
finally:
k32.CloseHandle(h)
except Exception:
return False
def _ecat_reuse_listening_native():
"""端口上已有健康 native → 直接复用(可能是另一进程刚拉起的),不 kill。"""
g_port = ecat_client.g_ecat_port
ports = ([g_port] if g_port else []) + \
[p for p in ecat_client.ECAT_PORTS if p != g_port]
for p in ports:
if not ecat_client._port_taken(p):
continue
try:
status, data = ecat_client._ecat_post(
p, "/api/native/health", b"{}", 1.0,
connect_timeout=0.4)
if status == 200:
return p
except Exception:
continue
return None
def _kill_all_ecat_native():
"""强杀所有 ecat_native.exe,返回杀掉的个数。
两个 native 并存会抢 Npcap 适配器:后起的抓不到从站、HTTP 超时、
PDO 空转。启动新实例前必须保证旧的死透。只按进程名杀,不碰 python。
"""
try:
out = subprocess.check_output(
["tasklist", "/FI", "IMAGENAME eq ecat_native.exe",
"/FO", "CSV", "/NH"],
text=True, stderr=subprocess.DEVNULL, timeout=8,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) or ""
except Exception:
return 0
pids = []
for line in (out or "").splitlines():
parts = [p.strip().strip('"') for p in line.split(",")]
if len(parts) >= 2 and parts[0].lower() == "ecat_native.exe":
try:
pids.append(int(parts[1]))
except ValueError:
pass
if not pids:
return 0
for pid in pids:
try:
subprocess.run(
["taskkill", "/F", "/PID", str(pid)],
capture_output=True, timeout=5,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
except Exception:
pass
# 等 Npcap 句柄随进程退出由内核回收
deadline = time.time() + 3.0
while time.time() < deadline:
try:
again = subprocess.check_output(
["tasklist", "/FI", "IMAGENAME eq ecat_native.exe",
"/FO", "CSV", "/NH"],
text=True, stderr=subprocess.DEVNULL, timeout=4,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
) or ""
if "ecat_native.exe" not in (again or "").lower():
break
except Exception:
break
time.sleep(0.2)
else:
time.sleep(0.3)
return len(pids)
def autostart():
"""按需拉起 tools/ecat 的服务进程。返回 (port, log_lines)。
单实例策略(2026-09-12,R3 加固 2026-09-23):
- 跨进程命名互斥锁串行化 kill/start(GUI 进程与服务子进程共用)。
- **先探测健康监听实例并复用**,绝不 kill 对方刚拉起的 native。
- 本进程子进程仍存活 → 只等它听口,**绝不 kill_all**(否则启动中的实例
会被自己下一次请求杀掉,形成 8s kill_stale 死循环)。
- 子进程已死 / 从未拉起且无健康监听 → 先 taskkill 全部残留 ecat_native,再起唯一实例。
port 为 None 表示拉不起来。
"""
global g_ecat_child, g_ecat_log_fp
if not ECAT_NATIVE_EXE:
return None, ["ecat_proc.init 未调用(BASE_DIR 未注入)"]
with _EcatProcXLock():
with g_ecat_lock:
logs = []
# ── R3: 另一进程(GUI/服务)已在跑的健康实例 → 复用 ──
reuse = _ecat_reuse_listening_native()
if reuse:
ecat_client.g_ecat_port = reuse
return reuse, [f"复用已在监听的 EtherCAT native (端口 {reuse})"]
# ── 我们自己拉起的还活着:只复用/等它,不杀 ──
if g_ecat_child is not None and g_ecat_child.poll() is None:
port = ecat_client._ecat_find_port()
if port:
ecat_client.g_ecat_port = port
return port, []
# 仍在启动(kill_stale/绑口最多约 10s):等完再判
deadline = time.time() + 10
while time.time() < deadline:
if g_ecat_child.poll() is not None:
break
port = ecat_client._ecat_find_port()
if port:
ecat_client.g_ecat_port = port
return port, []
time.sleep(0.3)
if g_ecat_child.poll() is None:
return None, logs + ["EtherCAT 服务仍在启动中(未监听端口),请稍候再试"]
g_ecat_child = None
if g_ecat_child is not None and g_ecat_child.poll() is not None:
g_ecat_child = None
if not os.path.isfile(ECAT_NATIVE_EXE):
return None, [
f"找不到 EtherCAT 原生守护进程: {ECAT_NATIVE_EXE}",
"请先运行 tools\\ecat\\build_native.bat 编译(需 VS2022 + Npcap)。",
]
# 无健康监听才清残留:禁止双实例抢网卡
killed = _kill_all_ecat_native()
if killed:
logs.append(f"已清掉 {killed} 个残留 ecat_native(单实例,防抢网卡)")
try:
# 日志落文件而不是 DEVNULL:native 的 stdout 是崩溃/楔死/自动重拉的
# 唯一现场(WEDGE、kill_stale、进 OP 全打这里),DEVNULL 吞掉后只能靠
# Windows 事件日志反推(2026-09-12 教训)。追加写,重启服务不清零。
if g_ecat_log_fp is None:
g_ecat_log_fp = open(
os.path.join(ECAT_DIR, "build_native", "native_stdout.log"),
"ab", buffering=0)
g_ecat_child = subprocess.Popen(
[ECAT_NATIVE_EXE, "--substr", ECAT_DEFAULT_ADAPTER,
"--mask", "0x50",
"--web", "src/web", "--parent-pid", str(os.getpid())],
cwd=ECAT_DIR,
stdout=g_ecat_log_fp, stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
logs.append(f"已自动启动 EtherCAT 原生守护进程(pid {g_ecat_child.pid})")
if _set_ecat_realtime(g_ecat_child.pid):
logs.append(f"EtherCAT 进程优先级已提到 RealTime(pid {g_ecat_child.pid})")
else:
logs.append("EtherCAT 进程 RealTime 提权失败(保持默认/High, 可手动设)")
except OSError as e:
return None, logs + [f"启动 EtherCAT 服务失败: {type(e).__name__}: {e}"]
deadline = time.time() + 12
while time.time() < deadline:
if g_ecat_child.poll() is not None:
return None, logs + [
f"EtherCAT 守护进程启动后立刻退出(exit={g_ecat_child.returncode}) —— "
"多为 Npcap 未安装或 8090..8095 全被占; "
"可双击 tools/ecat/ecat_web.bat 看具体报错"]
port = ecat_client._ecat_find_port()
if port:
ecat_client.g_ecat_port = port
logs.append(f"EtherCAT 服务已就绪(端口 {port})")
return port, logs
time.sleep(0.3)
return None, logs + ["EtherCAT 服务启动超时(12s 内未监听端口)"]
# 旧名兼容(server / 脚本)
_ecat_autostart = autostart
def watch_parent(ppid):
"""兜底看门狗(Job Object 万一不可用时): 父 GUI 进程消失就自我了断,
避免窗口关了服务还在后台占端口。"""
def _loop():
k32 = ctypes.windll.kernel32
h = k32.OpenProcess(0x00100000, False, ppid) # SYNCHRONIZE
if not h:
return
k32.WaitForSingleObject(h, 0xFFFFFFFF) # 父进程退出即返回
os._exit(0)
threading.Thread(target=_loop, daemon=True).start()
_watch_parent = watch_parent
+126 -793
View File
File diff suppressed because it is too large Load Diff
+6 -1
View File
@@ -17,13 +17,14 @@ import urllib.request
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import esp32_server as srv # noqa: E402
import ecat_client # noqa: E402 # P2: _ecat_post/_ecat_find_port 实现在此模块
WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
calls = [] # 假 _ecat_post 记录到的请求路径
calls_lock = threading.Lock()
def fake_post(port, path, body, timeout):
def fake_post(port, path, body, timeout, **kwargs):
with calls_lock:
calls.append(path)
# 带空格的 "ok": true —— 故意与真 native 的紧凑格式不同,
@@ -89,8 +90,12 @@ def wait_flag(want, timeout=8.0):
def main():
# 加速测试: GRACE 从 3s 缩到 0.3s(重试轮次间隔同用该值, 一并缩短无妨)
srv.APP_LEASE_GRACE_S = 0.3
# P2 拆分后:lease 停循环路径用 esp32_server 命名空间;
# HTTP 代理路径在 ecat_client 内部解析 _ecat_post —— 两处都要 patch。
srv._ecat_post = fake_post
srv._ecat_find_port = lambda: 9999
ecat_client._ecat_post = fake_post
ecat_client._ecat_find_port = lambda: 9999
# 找空闲端口起真服务
with socket.socket() as probe: