EoE (Ethernet over EtherCAT)
EoE 协议通过 EtherCAT 总线实现以太网通信,支持配置 IP/MAC/DNS 地址和以太网帧收发,符合 ETG.1000.6 和 ETG.1020 标准。
通过 slave.eoe 访问。从站不支持 EoE 时为 None。
属性
读写属性实时操作: getter 从设备读取当前值, setter 校验格式后立即写入设备。
| 属性 | 类型 | 访问 | 说明 |
|---|---|---|---|
| can_send_frame | bool | 只读 | 是否支持发送以太网帧 |
| can_receive_frame | bool | 只读 | 是否支持接收以太网帧 |
| can_set_ip | bool | 只读 | 是否支持设置 IP 参数 |
| can_get_ip | bool | 只读 | 是否支持获取 IP 参数 |
| ip | int | 读写 | IP 地址 (uint32 网络字节序) |
| subnet | int | 读写 | 子网掩码 (uint32) |
| gateway | int | 读写 | 默认网关 (uint32) |
| mac | bytes | 读写 | MAC 地址 (6 字节) |
| dns | int | 读写 | DNS 服务器地址 (uint32) |
| dns_name | str | 读写 | DNS 名称 |
EoE 同时提供成对 setter/getter 方法用于批量配置: set_ip(ip, subnet, gateway) / get_ip(), set_mac(mac) / get_mac(), set_dns(dns_ip, dns_name) / get_dns(), 以及 set_full_param() / get_full_param() 一次性读写完整网络配置。
示例:
import struct, socket
if slave.eoe is not None:
# 读取
result = slave.eoe.get_ip()
if result is not None:
print(f"IP: {socket.inet_ntoa(struct.pack('!I', result[0]))}")
# 写入
ip = struct.unpack('!I', socket.inet_aton('192.168.1.100'))[0]
subnet = struct.unpack('!I', socket.inet_aton('255.255.255.0'))[0]
gateway = struct.unpack('!I', socket.inet_aton('192.168.1.1'))[0]
slave.eoe.set_ip(ip, subnet, gateway)
slave.eoe.set_mac(bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]))
地址过滤器 (ETG.1020)
get_address_filters()
def get_address_filters(self, max_filters: int = 16) -> Optional[List[bytes]]
获取 MAC 地址过滤器列表, 每个过滤器为 6 字节 MAC 地址。
set_address_filters()
def set_address_filters(self, filters: List[bytes]) -> bool
设置 MAC 地址过滤器列表。每个元素为 6 字节 MAC 地址, 空列表清空过滤器。
示例:
mac = bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF])
slave.eoe.set_address_filters([mac])
filters = slave.eoe.get_address_filters()
for f in filters or []:
print(f"过滤器: {':'.join(f'{b:02X}' for b in f)}")
slave.eoe.set_address_filters([]) # 清空
以太网帧收发
send_frame()
def send_frame(self, data: bytes) -> bool
发送以太网帧。
receive_frame()
def receive_frame(self) -> Optional[bytes]
接收以太网帧, 返回帧数据, 失败返回 None。
Ping 测试
ping()
def ping(self, target_ip: str, timeout_ms: int = 5000,
ttl: int = 64) -> EoEPingResult
通过 EoE 通道发送 ICMP Echo Request 并等待响应, 返回 EoEPingResult 结果对象 (无论成功或失败均返回, 用 success 字段判断)。
参数:
target_ip(str) — 目标 IP 地址 (点分十进制, 如"192.168.1.1")timeout_ms(int) — 超时时间 (毫秒), 默认 5000ttl(int) — IP TTL, 默认 64
相关结构:
@dataclass
class EoEPingResult:
success: bool = False # Ping 是否成功
round_trip_time_ms: float = 0.0 # 往返时间 (毫秒)
target_address: str = "" # 目标地址
ttl: int = 0 # 生存时间
error_message: str = "" # 错误消息 (失败时)
示例:
result = slave.eoe.ping("192.168.1.1", timeout_ms=2000)
if result.success:
print(f"RTT={result.round_trip_time_ms:.2f}ms TTL={result.ttl}")
else:
print(f"Ping 失败: {result.error_message}")
完整示例
网络配置与 Ping
import struct, socket
if slave.eoe is not None:
print(f"当前 IP: {slave.eoe.get_ip()}")
print(f"当前 MAC: {slave.eoe.get_mac()}")
new_ip = struct.unpack('!I', socket.inet_aton('10.0.0.100'))[0]
new_subnet = struct.unpack('!I', socket.inet_aton('255.255.255.0'))[0]
new_gw = struct.unpack('!I', socket.inet_aton('10.0.0.1'))[0]
slave.eoe.set_ip(new_ip, new_subnet, new_gw)
slave.eoe.set_dns(struct.unpack('!I', socket.inet_aton('10.0.0.1'))[0])
ping = slave.eoe.ping("10.0.0.1")
if ping.success:
print(f"Ping 成功: {ping.round_trip_time_ms:.1f}ms")
ARP 缓存管理
set_arp_entry()
def set_arp_entry(self, ip: str, mac: bytes) -> None
设置 ARP 缓存条目, 将指定 IP 地址与 MAC 地址绑定。
参数:
ip(str) — IP 地址 (如"192.168.1.1")mac(bytes) — MAC 地址 (6 字节)
clear_arp_cache()
def clear_arp_cache(self) -> None
清除从站的 ARP 缓存。
示例:
if slave.eoe is not None:
# 设置静态 ARP 条目
slave.eoe.set_arp_entry("192.168.1.1",
bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]))
# 清除 ARP 缓存
slave.eoe.clear_arp_cache()
异步帧接收 Hook
EoEAdvanced.set_receive_hook() 注册一个 native 级 EoE 帧接收回调, 由 PDO 主循环在收到 EoE Fragment Data 帧时直接触发, 无需轮询 receive_frame_data()。对齐 C# EoEInstance.FrameReceived 事件。
set_receive_hook(callback)
def set_receive_hook(self, callback) -> bool
注册 EoE 异步接收 Hook, 每个 master 共享一个 native 回调; 重复调用会覆盖前一次注册。
参数:
callback(Callable) — 回调签名callback(slave_index: int, frame: bytes) -> None
返回值:
bool— DLL 是否成功绑定 hook (DLL 未导出EOESetReceiveHook时返回False)
clear_receive_hook()
def clear_receive_hook(self) -> bool
清除 EoE 异步接收 Hook。
示例:
from ethercat.slave.eoe import EoEAdvanced
eoe_adv = EoEAdvanced(slave._dll, slave._mi, slave._si)
def on_frame(slave_index: int, frame: bytes):
print(f"[EoE] slave={slave_index} len={len(frame)}")
if eoe_adv.set_receive_hook(on_frame):
print("Hook 注册成功")
# ... PDO 周期内每收到 EoE 帧自动触发回调 ...
eoe_adv.clear_receive_hook()
回调跑在 DLL 主循环线程, 必须保证耗时极短 (几十 µs 以内), 否则会拖慢实时 PDO 周期。建议在回调内只做入队, 真正处理转入业务线程。
ARP 表条目 (EoEPinger)
EoEPinger 是上层 ICMP Ping 工具, 自带独立的 ARP 缓存:
from ethercat.slave.eoe import EoEPinger
pinger = EoEPinger(slave.eoe, local_ip='192.168.100.1')
pinger.update_arp_cache("192.168.100.5",
bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]))
rtt = pinger.ping("192.168.100.5", timeout_ms=1000)
print(f"RTT: {rtt:.2f} ms" if rtt is not None else "超时")
EoEPinger.set_arp_entry(ip, mac) 与 EoE.set_arp_entry() 同名但作用域不同 — Pinger 的缓存仅供 Ping 使用, EoE 实例的缓存供其自身帧拼装使用。