邮箱网关 (Mailbox Gateway)

邮箱网关实现 ETG.8200 标准，允许外部诊断工具通过 UDP/IP 访问 EtherCAT 主站和从站的对象字典。

通过 master.mailbox_gateway 访问。

ETG.8200标准

邮箱网关服务符合 ETG.8200 标准规范，使用标准 UDP 端口 0x88A4 (34980)，帧结构完全符合 Table 1 要求。

服务控制

port

@property
def port(self) -> int

@port.setter
def port(self, value: int) -> None

UDP 端口号，默认 0x88A4 (34980)。启动后会更新为实际绑定的端口。

多实例自动偏移

运行多个主站实例时，端口会根据 master_number 自动偏移，避免端口冲突：

• 实例 1: 34980（默认）
• 实例 2: 34981
• 实例 N: 34980 + (N - 1)

如需手动指定端口，在 start() 前设置 port 属性即可覆盖自动值。

警告

端口号必须在服务启动前设置。服务运行时无法更改端口。如果期望端口被占用，会自动回退到可用端口，port 会更新为实际值。

is_running

@property
def is_running(self) -> bool

服务是否正在运行。

start()

def start(self) -> None

启动邮箱网关服务，开始监听 UDP 请求。如果期望端口被占用，会自动尝试备用端口，最终由系统分配。

示例:

gateway = master.mailbox_gateway
gateway.start()
print(f"邮箱网关已启动，端口: {gateway.port}")

stop()

def stop(self) -> None

停止邮箱网关服务，释放 UDP 端口。停止后可再次调用 start() 重新启动。

示例:

master.mailbox_gateway.stop()

支持的协议

协议	类型	说明
CoE	0x03	SDO 读写，支持主站（ETG.1510）和从站对象字典
SoE	0x04	IDN 参数读写（ETG.1020 标准）
FoE	0x05	文件传输（当前仅支持单包）
VoE	0x0F	厂商特定协议透传

地址路由

• Address = 0x0000 → 访问主站对象字典（仅 CoE）
• Address = 从站站地址 → 透传到对应从站（CoE / SoE / FoE / VoE）

异步邮箱事务

2.5.x 新增 — MailboxTransaction 提供单次邮箱事务的真异步执行能力, 直连底层 G-5 异步 Worker (mbx_submit_async / mbx_wait / mbx_cancel / mbx_get_result)。适用于需要把整帧邮箱报文 (含 6 字节 MbxHeader) 投递并在不阻塞主循环前提下等待结果的场景。

模块位置

MailboxTransaction 位于 darra_ethercat.abstractions 子包, 不在顶层 darra_ethercat 命名空间。常规 CoE/SoE/FoE 读写请优先使用 slave.coe / slave.soe 等协议接口, 本 API 面向自定义邮箱帧透传。

MailboxStatus 枚举

from darra_ethercat.abstractions import MailboxStatus

class MailboxStatus(IntEnum):
    PENDING         = 0    # 事务待执行
    SUCCESS         = 1    # 事务成功
    TIMEOUT         = -1   # 超时
    CANCELLED       = -2   # 被取消
    MAILBOX_ERROR   = -3   # 邮箱协议错误
    PROTO_MISMATCH  = -4   # 协议类型不匹配
    COUNTER_FAIL    = -5   # 邮箱计数器校验失败
    WKC_FAIL        = -6   # WKC 校验失败
    NO_MEMORY       = -7   # 内存不足
    INVALID_ARG     = -8   # 参数非法
    NOT_IMPLEMENTED = -9   # DLL 未导出异步 API

MailboxTransaction

from darra_ethercat.abstractions import MailboxTransaction, MailboxStatus

txn = MailboxTransaction(dll, master_index=0, slave_index=1,
                         protocol_type=0x03, timeout_ms=5000)

参数	类型	默认	说明
dll	DarraCoreDLL	—	DLL 接口实例 (master._dll)
master_index	int	—	主站索引 (0-based)
slave_index	int	—	从站索引 (1-based)
protocol_type	int	—	ECT_MBXT_* 协议类型 (CoE=3, FoE=4, SoE=5, AoE=1, EoE=2, VoE=0x0F, FSoE=8)
timeout_ms	int	5000	事务超时 (毫秒)

属性:

类别	属性	类型	读写	说明
请求	request_frame	Optional[bytes]	读写	完整邮箱帧 (含 6 字节 MbxHeader)
	response_capacity	int	读写	响应缓冲容量 (字节, 默认 1500)
	timeout_ms	int	读写	事务超时 (毫秒)
响应	response_frame	Optional[bytes]	只读	成功后填充的响应帧
	response_length	int	只读	响应帧实际长度 (字节, 含 MbxHeader)
	response_counter	int	只读	响应帧邮箱计数器 (低 3 位)
状态	status	MailboxStatus	只读	事务当前状态
	error_code	int	只读	协议层错误码 (0 表示无)
	slave_index	int	只读	从站索引
	protocol_type	int	只读	邮箱协议类型
时间	submit_time	Optional[float]	只读	投递时间戳 (time.time())
	complete_time	Optional[float]	只读	完成时间戳

方法:

方法	返回	说明
execute()	MailboxStatus	同步执行事务 (内部走异步 Worker: submit → wait → get_result), 返回最终状态
cancel(ticket)	MailboxStatus	取消正在进行的异步事务 (置 C 层 cancel_flag)

示例:

from darra_ethercat.abstractions import MailboxTransaction, MailboxStatus

# 构造一个 CoE 邮箱事务 (protocol_type=0x03)
txn = MailboxTransaction(master._dll, master_index=master.master_index,
                         slave_index=1, protocol_type=0x03)

# 整帧邮箱报文 (含 6 字节 MbxHeader, 自行按 ETG.1000.6 组装)
txn.request_frame = bytes(...)

status = txn.execute()
if status == MailboxStatus.SUCCESS:
    print(f"响应 {txn.response_length} 字节: {txn.response_frame.hex()}")
else:
    print(f"事务失败: {status.name} (error_code=0x{txn.error_code:04X})")

底层未导出异步 API 时降级

若底层 native 库未导出 G-5 异步 Worker, execute() 返回 MailboxStatus.NOT_IMPLEMENTED, 不抛异常。应用层据此判断是否可用。

帧结构

EtherCAT Header:

• Length: Mailbox Header + Data 的总长度
• Data Type: 固定为 0x05 (Mailbox communication)

Mailbox Header:

• Length: 邮箱数据部分长度
• Address: 0x0000=主站, 其他=从站站地址
• Type: 0x03=CoE, 0x04=SoE, 0x05=FoE, 0x0F=VoE
• Cnt: 邮箱计数器 (1-7 循环)

完整示例

场景 A — 集成现有配置/调试工具

让第三方配置工具（或工程师工具）通过 UDP 访问主站/从站对象字典，无需修改被控设备固件。

场景 B — 自定义诊断或自动化脚本

通过脚本/服务定期采集从站参数、批量读取对象或执行远程配置。

场景 C — 远程监控与告警采集

把网关用于远程被动监控（周期性读取或订阅），并把关键数据汇报到集中监控系统。

启动网关

from darra_ethercat import EtherCATMaster, EcState

with EtherCATMaster() as master:
    master.set_eni("config.deni")
    master.set_network(r"\\Device\\NPF_{GUID}")
    master.set_state(EcState.OP)
    master.start()

    # 启动邮箱网关（可选：自定义端口）
    gateway = master.mailbox_gateway
    # gateway.port = 12345  # 默认 34980，如需修改请在 start 前设置
    gateway.start()
    print(f"邮箱网关已启动，端口: {gateway.port}")

    # PDO 自动在后台线程运行，网关在 OP 状态下提供邮箱服务
    input("按回车停止...")

    gateway.stop()

Python 外部客户端

import socket
import struct

class MailboxGatewayClient:
    """ETG.8200 邮箱网关 UDP 客户端"""

    def __init__(self, host: str = "127.0.0.1", port: int = 0x88A4):
        self.addr = (host, port)
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(3.0)

    def build_frame(self, address: int, mb_type: int, mb_data: bytes) -> bytes:
        """组装 ETG.8200 邮箱网关 UDP 帧 (示例参考, 可按需重命名)"""
        mb_len = len(mb_data)
        ecat_len = 6 + mb_len
        ecat_header = (0x05 << 12) | (ecat_len & 0x07FF)

        frame = struct.pack('<H', ecat_header)          # EtherCAT Header
        frame += struct.pack('<H', mb_len)              # Mailbox Length
        frame += struct.pack('<H', address)             # Address
        frame += bytes([0x00])                           # Channel + Priority
        frame += bytes([mb_type << 4])                   # Type
        frame += mb_data
        return frame

    def send_request(self, address: int, mb_type: int, mb_data: bytes) -> bytes:
        frame = self.build_frame(address, mb_type, mb_data)
        self.sock.sendto(frame, self.addr)
        resp, _ = self.sock.recvfrom(4096)
        return resp

    def coe_sdo_read(self, address: int, index: int, subindex: int) -> bytes:
        """CoE SDO Upload（读取对象字典）"""
        co_data = bytearray(8)
        co_data[1] = 0x20                                # CoE Type: SDO Request
        co_data[2] = 0x40                                # SDO Upload (Read)
        struct.pack_into('<H', co_data, 3, index)
        co_data[5] = subindex
        return self.send_request(address, 0x03, bytes(co_data))

    def coe_sdo_write(self, address: int, index: int, subindex: int, value: bytes) -> bytes:
        """CoE SDO Download（写入对象字典）"""
        co_data = bytearray(6 + len(value))
        co_data[1] = 0x20                                # CoE Type: SDO Request
        co_data[2] = 0x20                                # SDO Download (Write)
        struct.pack_into('<H', co_data, 3, index)
        co_data[5] = subindex
        co_data[6:6+len(value)] = value
        return self.send_request(address, 0x03, bytes(co_data))

    def soe_read(self, address: int, idn: int, drive_no: int = 0, element: int = 0x05) -> bytes:
        """SoE Read（读取 IDN 参数）
        element_flags: 0x00=Name, 0x01=Attribute, 0x02=Unit, 0x03=Min, 0x04=Max, 0x05=Value, 0x06=Default, 0x07=DataType
        """
        soe_data = bytearray(4)
        soe_data[0] = 0x01                               # OpCode: Read
        soe_data[1] = (drive_no << 3) | (element & 0x07)
        struct.pack_into('<H', soe_data, 2, idn)
        return self.send_request(address, 0x04, bytes(soe_data))

    def soe_write(self, address: int, idn: int, value: bytes, drive_no: int = 0, element: int = 0x05) -> bytes:
        """SoE Write（写入 IDN 参数）"""
        soe_data = bytearray(4 + len(value))
        soe_data[0] = 0x02                               # OpCode: Write
        soe_data[1] = (drive_no << 3) | (element & 0x07)
        struct.pack_into('<H', soe_data, 2, idn)
        soe_data[4:4+len(value)] = value
        return self.send_request(address, 0x04, bytes(soe_data))

    def foe_read(self, address: int, filename: str) -> bytes:
        """FoE Read（从从站下载文件，仅支持单包）"""
        name_bytes = filename.encode('ascii') + b'\x00'
        foe_data = bytearray(5 + len(name_bytes))
        foe_data[0] = 0x01                               # OpCode: Read Request
        foe_data[5:] = name_bytes
        return self.send_request(address, 0x05, bytes(foe_data))

    def voe_send(self, address: int, vendor_id: int, vendor_type: int, data: bytes) -> bytes:
        """VoE 发送/接收（厂商特定协议）"""
        voe_data = bytearray(6 + len(data))
        struct.pack_into('<I', voe_data, 0, vendor_id)   # VendorID
        struct.pack_into('<H', voe_data, 4, vendor_type) # VendorType
        voe_data[6:6+len(data)] = data
        return self.send_request(address, 0x0F, bytes(voe_data))

# --- 使用示例 ---
client = MailboxGatewayClient("192.168.1.100")

# CoE: 读取主站对象字典 (address=0x0000)
resp = client.coe_sdo_read(0x0000, 0x1018, 0x01)
print(f"主站 VendorID: {resp.hex()}")

# CoE: 读取/写入从站对象字典
resp = client.coe_sdo_read(0x03E9, 0x6040, 0x00)
client.coe_sdo_write(0x03E9, 0x6040, 0x00, struct.pack('<H', 0x0006))

# SoE: 读取从站 IDN 参数值
resp = client.soe_read(0x03E9, idn=32, drive_no=0, element=0x05)

# SoE: 写入从站 IDN 参数值
client.soe_write(0x03E9, idn=32, value=struct.pack('<I', 1000))

# FoE: 从从站下载文件
resp = client.foe_read(0x03E9, "firmware.bin")

# VoE: 厂商特定协议
resp = client.voe_send(0x03E9, vendor_id=0x00001164, vendor_type=0x0001, data=b'\x01\x02')

注意事项

网络安全

默认网关是强透传的，允许网络访问主站和从站的全部对象字典。建议：

• 务必在受控网络中使用，或通过 IP 白名单 / VPN 隧道 限制访问
• 配合操作系统防火墙规则，仅放行可信来源 IP
• 生产环境中谨慎开启，避免暴露到公网

性能影响

邮箱网关运行在独立线程，对主循环性能影响极小。但大量同步 SDO/邮箱操作仍可能影响总体延迟——请合理限制并发与速率。

协议详解

CoE (CAN over EtherCAT)

支持的操作:

• SDO Upload (0x40) - 读取对象字典
• SDO Download (0x20) - 写入对象字典
• SDO Abort (0x80) - 错误响应

错误码:

• 0x06020000 - 对象不存在
• 0x06010000 - 不支持的访问
• 0x06010002 - 写保护
• 0x05040001 - 命令未实现
• 0x08000000 - 一般错误

SoE (Servo over EtherCAT)

支持的操作:

• IDN Read (0x01) - 读取 IDN 参数
• IDN Write (0x02) - 写入 IDN 参数

支持的元素:

• Value (0x05) - 参数值
• Name (0x00) - 参数名称
• Attribute (0x01) - 参数属性
• Unit (0x02) - 单位
• Min/Max (0x03/0x04) - 最小/最大值
• Default (0x06) - 默认值
• Data Type (0x07) - 数据类型

错误码:

• 0x8001 - 服务不可用
• 0x1001 - 无效命令
• 0x1009 - IDN不存在
• 0x3002 - 无效数据大小
• 0x7002 - 写入失败
• 0x8000 - 一般错误

FoE (File over EtherCAT)

支持的操作:

• Read Request (0x01) - 文件下载请求
• Write Request (0x02) - 文件上传准备
• Data (0x03) - 文件数据传输
• Ack (0x04) - 确认
• Error (0x05) - 错误响应

当前限制:

• 仅支持单包文件传输（小文件）
• 完整分段传输需要会话状态管理（未来版本）

错误码:

• 0x00008001 - 无效操作码 / 未实现
• 0x00008002 - 文件未找到
• 0x00008003 - 非法文件名
• 0x00008000 - 一般错误

VoE (Vendor over EtherCAT)

帧格式:

• VendorID (4 bytes) - 厂商标识
• VendorType (2 bytes) - 厂商类型
• Data (n bytes) - 厂商特定数据

功能:

• 完整透传到从站 VoE 接口
• 自动处理请求/响应
• 支持任意长度数据

错误码:

• 0x0001 - 服务不可用
• 0x0002 - 无效帧
• 0x0003 - 无响应
• 0x0000 - 一般错误

错误处理

协议	错误类型	错误码	说明
CoE	对象不存在	0x06020000	请求的对象或从站不存在
CoE	不支持的访问	0x06010000	协议类型不支持或从站无对应功能
CoE	写保护	0x06010002	尝试写入只读对象
CoE	命令未实现	0x05040001	SDO 命令规范不支持
CoE	一般错误	0x08000000	其他处理错误
SoE	IDN不存在	0x1009	请求的IDN参数不存在
SoE	写入失败	0x7002	IDN参数写入失败
FoE	文件未找到	0x00008002	请求的文件不存在
FoE	非法文件名	0x00008003	文件名格式错误
VoE	无响应	0x0003	从站无VoE响应

限制

已实现功能

• CoE 完整透传
• SoE IDN 参数访问
• FoE 单包文件传输
• VoE 厂商协议透传
• 主站对象字典访问
• 从站邮箱透传
• 错误处理和错误码映射

待实现功能

• FoE 分段传输（大文件支持）
• Address 映射表 (+0x8000 虚拟映射)
• 多播/广播请求
• 会话状态管理