跳到主要内容

FSoE 功能安全

FSoE (Functional Safety over EtherCAT) 是 ETH 主站内的 FSoE 状态机源码与协议接口，在标准 PDO 通道上叠加安全协议层。周期路径曾 fail-closed。这不是已认证安全产品，也不是 PLC 能力；周期最短 125µs（柔性）。

通过 slave.fsoe 访问。是否支持 FSoE 用 is_capable 查询。​

FSoE 设备检测

FSoE 设备检测由 SDK 自动完成，结果经 slave.fsoe.is_capable 查询。检测过程：

  1. CoE 前置条件 — 从站必须支持 CoE 邮箱协议
  2. 0xF980:01 检测 — 尝试 SDO 读取设备级 FSoE 安全地址，成功则确认支持
  3. 0x9001:02 检测 — 若上步失败，尝试读取 MDP 连接参数（适用于多模块安全设备）

仅检查 CoE 支持是不够的，因为很多非安全设备也支持 CoE。FSoE 设备必须实现特定的安全对象索引才能被正确识别。

SDK 提供三个层级的 FSoE 支持：

  • slave.fsoe — 基础接口，通过从站访问，提供单连接 FSoE 操作
  • SafeMdp — MDP 多连接管理器，支持单从站上的多个 FSoE 连接
  • SafetyManager — 多从站多连接协调器，提供 bind_safe_io 一步初始化

FSoE 工作流程​

FSoE 在 EtherCAT PDO 通道上建立独立的安全连接。主站负责协议管理，从站负责安全逻辑。

状态机流程：

  • Reset — 初始状态，等待主站发起会话
  • Session — 会话建立，交换连接 ID
  • Connection — 连接建立，验证安全地址
  • Parameter — 参数下载阶段（SRA CRC 校验）
  • Data — 正常安全数据交换
  • Failsafe — 失效安全，从站输出安全值
自动状态推进

SafetyManager.bind_safe_io() 一行完成初始化 + 启动数据交换，中间状态（Session → Connection → Parameter）由 SDK 自动推进。

快速开始​

推荐方式

使用 SafetyManager.bind_safe_io() 一步完成初始化，无需手动管理状态机。

SafetyManager / SafeMdp 为高级 API

本页示例中出现的 SafetyManager(master._dll, master.master_index) 与 SafeMdp(master._dll, master.master_index, slave_index=...) 写法属于内部高级 API, 仅在需要多 FSoE 连接协调或 MDP 多模块场景时使用。常规单从站安全数据读写走 slave.fsoe（基础接口）即可。 高级 API 的句柄参数后续可能调整，请勿在生产代码中长期固化该构造形式。

from darra_ethercat import EtherCATMaster, EcState, SafetyManager

master = EtherCATMaster()
master.set_eni("config.deni")
master.build()
master.set_state(EcState.OP)

# 一步绑定安全 IO
mgr = SafetyManager(master._dll, master.master_index)
mgr.bind_safe_io(
    slave_index=1,
    safe_input_offset=0,
    safe_input_size=2,
    safe_output_offset=0,
    safe_output_size=1,
    safety_address=0x0100,
    watchdog_ms=100
)

# PDO 回调中读写安全数据（bind_safe_io 内部 connection_id = slave_index + 1）
def on_pdo(mi):
    data = mgr.read_input_frame(1, 2)
    if data:
        mgr.write_output_frame(1, 2, bytes([0x01]))

master.on_pdo_cycle(on_pdo)

FSoE 与普通 PDO 的区别​

FSoE 安全数据不能像普通 PDO 那样直接映射。原因在于协议层的差异：

使用体验对比

尽管底层机制不同，bind_safe_io + 数据交换回调的使用体验已经非常接近普通 PDO。

普通 PDO​

IOmap 中的数据就是用户数据，映射结构体直接读写即可

FSoE PDO​

过程数据帧内存放的是 FSoE 协议帧，用户的安全数据被包裹在帧内部，且需要经过校验

SDK 的 FSoE 协议引擎负责：

  • CRC 校验 — 验证输入帧完整性、计算输出帧 CRC
  • 状态机管理 — Session -> Connection -> Parameter -> Data 自动推进
  • 看门狗 — 监控通信超时，超时自动进入 Failsafe
  • 序列号 — 检测帧丢失和重复

绕过协议引擎直接改 PDO 帧会破坏安全协议完整性，因此 FSoE 必须通过独立缓冲区访问。

基础接口属性​

基础 FSoE (slave.fsoe) 暴露下列只读属性：

类别属性类型读写说明
连接connection_idint只读最近一次 init_connection 实际使用的连接 ID（传 0 时由原生层自动分配并回写）
状态stateint（FSoEState 取值）只读当前 FSoE 状态
能力is_capablebool只读从站是否支持 FSoE
连接connection_countint只读主站的 FSoE 连接数
连接级状态查询

连接级字段不是属性，通过 get_status() 返回的 dict 读取。dict 键：state、last_error、error_count、frames_sent、frames_received、crc_errors、watchdog_errors、watchdog_expired、in_failsafe、toggle_bit、initialized（注意：下文 FSoEConnectionStatus 数据类的同名字段叫 is_initialized，dict 键是 initialized）。看门狗 / 最后错误另有 check_watchdog() / get_last_error() 方法。

基础接口 (slave.fsoe)​

通过 slave.fsoe 访问。​

连接管理​

slave = master[1]
fsoe = slave.fsoe

# 初始化 FSoE 连接
fsoe.init_connection(
    connection_id=0,
    safety_address=0x0100,
    watchdog_time_ms=100,
    safe_input_size=2,
    safe_output_size=1
)

# 获取连接状态
status = fsoe.get_status()
if status:
    print(f"FSoE 状态: {status['state']}")
    print(f"失效安全: {status['in_failsafe']}")

# 请求状态转换
fsoe.request_state(FSoEState.DATA)  # DATA 状态

# 重置连接
fsoe.reset()

# 关闭连接
fsoe.close_connection()

安全数据读写​

# 写入安全输出
fsoe.write_safe_output(bytes([0x01]))

# 读取安全输入
data = fsoe.read_safe_input()
if data:
    print(f"安全输入: {data.hex()}")

参数下载​

# 下载安全参数，返回 SRA CRC
sra_crc = fsoe.download_parameters(param_data)
if sra_crc is not None:
    print(f"SRA CRC: 0x{sra_crc:08X}")

# 设置失效安全输出
fsoe.set_failsafe_output(bytes([0x00]))

看门狗与错误​

# 检查看门狗状态 (True=正常)
if not fsoe.check_watchdog():
    print("FSoE 看门狗超时!")

# 获取最后的错误代码
error = fsoe.get_last_error()
if error != FSoEError.NONE:
    desc = FSoEHelper.get_error_description(FSoEError(error))
    print(f"FSoE 错误: {desc}")

# 清除错误
fsoe.clear_error()

状态控制​

request_state()​

def request_state(self, target_state: int) -> bool

请求 FSoE 状态转换。自动验证状态转换有效性。

参数：

  • target_state (int) — 目标状态（FSoEState 枚举值）

返回值：

  • bool — 成功返回 True

reset()​

def reset(self) -> bool

重置 FSoE 连接到初始状态，重新开始状态机。

close_connection()​

def close_connection(self) -> bool

关闭 FSoE 连接，释放资源。

clear_error()​

def clear_error(self) -> None

清除 FSoE 错误。清除后可尝试重新建立连接。

参数和配置​

download_parameters()​

def download_parameters(self, param_data: bytes) -> Optional[int]

下载安全参数。

备注

仅在 Parameter 状态下可调用。参数内容由安全配置工具生成。

参数：

  • param_data (bytes) — 参数数据

返回值：

  • int | None — 成功返回 SRA CRC 校验值，失败返回 None

set_failsafe_output()​

def set_failsafe_output(self, data: bytes) -> bool

预设失效安全时的输出值。当 FSoE 进入 Failsafe 状态时，从站自动切换到此值。

示例：

# 失效安全时所有输出关闭
fsoe.set_failsafe_output(bytes([0x00]))

# 使用 ctypes 结构体
safe_out = SafeDriveOutput()
safe_out.safety_control_word = 0x0001  # 请求 STO（安全扭矩关闭）
fsoe.set_failsafe_output(bytes(safe_out))

SafetyManager（多从站协调器）​

SafetyManager 管理多个从站的 FSoE 连接，提供一步绑定和批量操作。

from darra_ethercat import SafetyManager
mgr = SafetyManager(master._dll, master.master_index)

bind_safe_io()​

def bind_safe_io(self, slave_index, safe_input_offset, safe_input_size,
                 safe_output_offset, safe_output_size,
                 safety_address=0, watchdog_ms=100) -> bool

一步绑定安全 IO。自动完成连接初始化，状态机（Session → Connection → Parameter → Data）随后由协议引擎随周期自动推进。

connection_id 取值

bind_safe_io 内部以 connection_id = slave_index + 1 注册连接（bind_mdp_safe_input / bind_mdp_safe_output 用 module_number + 1，bind_mdp_drive_axis 用 axis_index + 1）。后续 read_input_frame / write_output_frame 必须传同一个 connection_id。需要显式指定 ID 时用 bind_safe_input / bind_safe_output。

bind_mdp_safe_input() / bind_mdp_safe_output()​

def bind_mdp_safe_input(self, slave_index, safety_address, input_size,
                        module_number=0, watchdog_ms=100) -> bool
def bind_mdp_safe_output(self, slave_index, safety_address, output_size,
                         module_number=0, watchdog_ms=100) -> bool

MDP 模块绑定安全输入/输出。

bind_mdp_drive_axis()​

def bind_mdp_drive_axis(self, slave_index, axis_index,
                        safety_address=0, watchdog_ms=100) -> bool

绑定 MDP 伺服驱动器的安全轴连接。

批量操作​

# 检查所有连接是否在 DATA 状态
if mgr.all_in_data_state():
    print("所有安全连接正常")

# 检查是否有连接进入失效安全（三态返回）
failsafe = mgr.any_in_failsafe()
if failsafe is None:
    print("有连接状态读失败, 按未知处理")
elif failsafe:
    print("有连接进入失效安全!")

# 所有连接进入失效安全
mgr.enter_all_failsafe()

# 重置所有连接
mgr.reset_all()

# 关闭所有连接
mgr.close_all()

# 获取状态摘要
print(mgr.get_status_summary())

# 连接数量
print(f"连接数: {mgr.connection_count}")

查找连接​

# 按安全地址查找
mdp = mgr.find_by_address(0x0100)

# 按连接 ID 查找
mdp = mgr.find_by_connection_id(1)

安全帧读写​

# 写入安全输出数据（connection_id 用绑定时的连接 ID，bind_safe_io 下为 slave_index + 1）
mgr.write_output_frame(slave_index=1, connection_id=2, frame_data=bytes([0x01]))

# 读取安全输入数据
data = mgr.read_input_frame(slave_index=1, connection_id=2)

DataExchange 回调​

从站级别的安全数据交换回调，在 SafeMdp 实例上注册（SafetyManager 场景用 mgr.get_mdp(slave_index, connection_id) 取到对应实例）。回调参数为 FSoEDataExchangeEvent 快照（字段见下文「事件参数」）。

# 注册 / 注销（每个 PDO 周期处理中触发一次）
mdp.add_data_exchanged_callback(lambda ev:
    print(f"状态={ev.current_state}, 输入={ev.safe_inputs.hex()}"))
mdp.remove_data_exchanged_callback(cb)

回调由 FSoE 周期处理在安全输入刷新后、安全输出提交前触发。

时序：

PDO 周期回调
+-- FSoE 前处理
|   +-- 对每个 FSoE 从站：
|       +-- 刷新安全输入缓冲区     <- 输入已刷新
|       +-- 检查状态变化和错误事件
|       +-- DataExchange 回调      <- 在此读写安全数据
+-- on_pdo_cycle 回调              <- 普通 PDO 读写（主站级）
+-- FSoE 后处理
    +-- 提交安全输出缓冲区         <- 自动提交输出
PDO 周期时序

绑定后，PDO 周期路径上可读写安全数据（并非标配每周期自动交换安全数据）：

  1. 周期开始 — 刷新所有 FSoE 从站的安全输入缓冲区、检查状态变化并触发事件
  2. DataExchange 回调 — 从站级别回调，读取输入（已是最新数据），修改输出
  3. PDO 回调 — 主站级别回调（普通 PDO 读写）
  4. 周期结束 — 提交所有 FSoE 从站的安全输出缓冲区

MDP 连接 — 按 connection_id 寻址​

SafeMdp 管理单从站上的多个 FSoE 连接，不返回独立的连接对象：所有连接级操作都把 connection_id 作为首参传入 SafeMdp 方法。

# SafeMdp: 每个方法首参为 connection_id
mdp.request_state(connection_id=1, target_state=FSoEState.DATA)  # 请求状态转换
mdp.reset(connection_id=1)                               # 重置连接
mdp.clear_error(connection_id=1)                         # 清除错误
mdp.close_connection(connection_id=1)                    # 关闭连接
mdp.download_parameters(connection_id=1, param_data=...)  # 下载安全参数
mdp.set_failsafe_output(connection_id=1, data=...)        # 设置失效安全输出
mdp.write_safe_output(connection_id=1, data=...)          # 写安全输出
data = mdp.read_safe_input(connection_id=1)               # 读安全输入
state = mdp.get_state(connection_id=1)                    # 读连接状态（FSoEState 取值）

连接级状态 / 错误 / 看门狗经 mdp.get_status(connection_id) 返回的 dict 读取。

FSoEConnectionMode​

class FSoEConnectionMode(IntEnum):
    SINGLE   = 0  # 单连接模式 - 所有模块共用一个 FSoE 连接和安全地址
    MULTIPLE = 1  # 多连接模式 - 每个模块有独立的 FSoE 连接和安全地址
选择指南
  • 单连接（bind_safe_io）— 简单设备、单轴驱动器
  • 多连接（bind_mdp_safe_input / bind_mdp_safe_output / bind_mdp_drive_axis）— 模块需要独立监控/恢复时（如多轴驱动器各轴独立安全连接）

MDP 多连接模式 (SafeMdp)​

SafeMdp 管理单个从站上的多个独立 FSoE 连接，适用于安全 I/O 模块等支持 MDP 的从站。

from darra_ethercat import SafeMdp

mdp = SafeMdp(master._dll, master.master_index, slave_index=1)

连接探测与管理​

# 探测可用连接
conn_count = mdp.detect_connections()
print(f"支持 {conn_count} 个 FSoE 连接")

# 获取设备安全地址
addr = mdp.get_device_address()
print(f"设备安全地址: 0x{addr:04X}")

# 获取从站连接数量
total = mdp.get_slave_connection_count()

连接初始化​

# 初始化连接（支持 MDP 扩展参数）
# connection_id=0 时由原生层自动分配，实际 ID 用 mdp.actual_connection_id(0) 查回
mdp.init_connection(
    connection_id=1,
    safety_address=0x0100,
    watchdog_time_ms=100,
    safe_input_size=2,
    safe_output_size=1,
    pdo_input_offset=0,      # PDO 输入偏移
    pdo_output_offset=0,     # PDO 输出偏移
    module_number=0,         # MDP 模块编号
    module_profile=0,        # MDP 模块配置文件
    axis_number=0,           # 驱动器轴号
    connection_type=0        # 连接类型
)

# 各连接独立操作
mdp.write_safe_output(connection_id=1, data=bytes([0x01]))
data = mdp.read_safe_input(connection_id=1)
mdp.request_state(connection_id=1, target_state=FSoEState.DATA)
mdp.close_connection(connection_id=1)

模块诊断​

模块通信参数 / 诊断读取（形参 connection_id 对应 MDP 模块编号，从 0 起）：

# 读取模块通信参数
comm_param = mdp.get_module_comm_param(connection_id=0)

# 读取模块诊断
diag = mdp.get_module_diagnosis(connection_id=0)
if diag:
    state, diagnosis = diag
    print(f"连接状态: {state}, 诊断码: 0x{diagnosis:04X}")

安全数据读写​

read_safe_input() / write_safe_output() 读写原始字节缓冲区。用户按设备厂商提供的安全 PDO 布局解析或拼装字节，常用方式包括：

  • int.from_bytes() / to_bytes() 处理标量字段
  • struct.pack / struct.unpack 处理多字段记录
  • ctypes.Structure 定义类型化布局

示例：

data = fsoe.read_safe_input()
if data:
    input_bits = data[0]
    ch0_safe = bool(input_bits & 0x01)

fsoe.write_safe_output(bytes([0x01]))  # 通道 0 输出

枚举与数据类​

FSoEState​

class FSoEState(IntEnum):
    RESET      = 0x100   # 初始/重置状态
    SESSION    = 0x101   # 会话建立
    CONNECTION = 0x102   # 连接建立
    PARAMETER  = 0x103   # 参数下载
    DATA       = 0x104   # 数据交换（正常工作）
    FAILSAFE   = 0x105   # 失效安全

同一取值集合在 darra_ethercat.FsoEState（utils/dll.py）中同名存在，get_status() 返回的 state 取后者类型。

FSoEError​

class FSoEError(IntEnum):
    NONE                = 0x0000  # 无错误
    WRONG_COMMAND       = 0x0001  # 错误的命令
    UNKNOWN_COMMAND     = 0x0002  # 未知命令
    WRONG_CONNECTION_ID = 0x0003  # 连接 ID 不匹配
    CRC_ERROR           = 0x0004  # CRC 校验失败
    WATCHDOG            = 0x0005  # 看门狗超时
    WRONG_ADDRESS       = 0x0006  # 错误的 FSoE 地址
    WRONG_DATA          = 0x0007  # 无效数据
    COMM_PARAM_LENGTH   = 0x0008  # 通信参数长度错误
    COMM_PARAM          = 0x0009  # 通信参数错误
    APP_PARAM_LENGTH    = 0x000A  # 应用参数长度错误
    APP_PARAM           = 0x000B  # 应用参数错误
    UNEXPECTED_SESSION  = 0x000C  # 意外的会话命令
    FAILSAFE_DATA       = 0x000D  # 收到失效安全数据
    NOT_INITIALIZED     = 0x0100  # FSoE 未初始化
    MAX_CONNECTIONS     = 0x0101  # 达到最大连接数
    INVALID_STATE_TRANSITION = 0x0102  # 无效状态转换
底层同值枚举

darra_ethercat.FsoEError（utils/dll.py，get_status() 的 last_error 取该类型）与本枚举取值相同，其中 0x0003 / 0x0004 / 0x0102 三个成员名分别为 WRONG_CONN_ID / CRC / INVALID_STATE。FSoEHelper.get_error_description() 用的是本节的 FSoEError。

FSoEFailsafeReason​

class FSoEFailsafeReason(IntEnum):
    WATCHDOG_TIMEOUT    = 0  # 看门狗超时
    CRC_ERROR           = 1  # CRC 校验错误
    COMMUNICATION_ERROR = 2  # 通信错误
    APPLICATION_REQUEST = 3  # 应用请求
    SLAVE_REQUEST       = 4  # 从站请求
    MASTER_REQUEST      = 5  # 主站请求
    RECOVERY_TO_DATA    = 6  # 恢复到数据交换

SafeModuleProfile​

class SafeModuleProfile(IntEnum):
    DIGITAL_INPUT   = 190  # 数字输入模块
    DIGITAL_IN_OUT  = 195  # 数字输入输出模块
    DIGITAL_OUTPUT  = 290  # 数字输出模块
    DRIVE_CONNECTION = 790 # 伺服驱动器连接
    MASTER          = 6900 # 主站模块

FSoEConnectionConfig​

@dataclass
class FSoEConnectionConfig:
    connection_id: int = 0           # 连接 ID
    safety_address: int = 0          # 安全地址
    watchdog_time_ms: int = 100      # 看门狗超时 (ms)
    safe_input_size: int = 0         # 安全输入大小 (字节)
    safe_output_size: int = 0        # 安全输出大小 (字节)
    pdo_input_offset: int = 0        # PDO 输入偏移
    pdo_output_offset: int = 0       # PDO 输出偏移

FSoEConnectionStatus​

@dataclass
class FSoEConnectionStatus:
    state: FSoEState = FSoEState.RESET   # 当前状态
    last_error: FSoEError = FSoEError.NONE  # 最后错误
    error_count: int = 0                 # 错误计数
    frames_sent: int = 0                 # 发送帧数
    frames_received: int = 0             # 接收帧数
    crc_errors: int = 0                  # CRC 错误数
    watchdog_errors: int = 0             # 看门狗错误数
    watchdog_expired: bool = False       # 看门狗是否过期
    in_failsafe: bool = False            # 是否在失效安全
    toggle_bit: int = 0                  # 当前切换位值
    is_initialized: bool = False         # 是否已初始化

事件参数​

@dataclass
class FSoEStateChangedEventArgs:
    slave_index: int             # 从站索引
    connection_index: int        # 连接索引
    old_state: FSoEState         # 变化前状态
    new_state: FSoEState         # 变化后状态

@dataclass
class FSoEErrorEventArgs:
    slave_index: int             # 从站索引
    connection_index: int        # 连接索引
    error: FSoEError             # 错误代码
    current_state: FSoEState     # 当前状态
    @property
    def description(self) -> str  # 错误描述文本

@dataclass
class FSoEFailsafeEventArgs:
    slave_index: int             # 从站索引
    connection_index: int        # 连接索引
    reason: FSoEFailsafeReason   # 失效安全原因
    entering_failsafe: bool      # True=进入, False=恢复

@dataclass
class FSoEDataUpdatedEventArgs:
    slave_index: int             # 从站索引
    connection_index: int        # 连接索引
    is_input: bool               # 是否为输入数据
    data: bytes                  # 安全数据

@dataclass
class FSoEDataExchangeEvent:
    slave_index: int                  # 从站索引
    current_state: FSoEState          # 本周期当前 FSoE 状态
    safe_inputs: bytes                # 本周期安全输入快照（拷贝）
    safe_outputs: bytes               # 本周期安全输出快照（拷贝）
    in_failsafe: bool                 # 是否处于失效安全模式
    cycle_count: int                  # 累计 PDO 周期计数
    timestamp: Optional[datetime]     # 事件触发时间戳（UTC）

从站事件​

从站级 FSoE 事件通过 slave.events 注册：

slave = master[1]

# FSoE 状态变化
slave.events.add_fsoe_state_changed(lambda old, new:
    print(f"FSoE 状态: {old} -> {new}"))

# FSoE 错误
slave.events.add_fsoe_error(lambda code:
    print(f"FSoE 错误: 0x{code:04X}"))

# FSoE 进入/恢复失效安全
slave.events.add_fsoe_failsafe_triggered(lambda:
    print("进入失效安全模式!"))

# FSoE 安全数据更新
slave.events.add_fsoe_safe_data_updated(lambda data:
    print(f"安全数据: {data.hex()}"))

# FSoE 数据交换
slave.events.add_fsoe_data_exchange(lambda inp, out:
    print(f"输入: {inp.hex()}, 输出: {out.hex()}"))

工具类​

FSoEHelper​

from darra_ethercat.slave.fsoe import FSoEHelper

# 检查状态转换是否合法
valid = FSoEHelper.is_valid_state_transition(FSoEState.RESET, FSoEState.SESSION)

# 获取错误描述
desc = FSoEHelper.get_error_description(FSoEError.WATCHDOG)

# 解析命令字节（Command 为 Table 5 整字节值）
cmd = FSoEHelper.get_command_type(0x36)  # FSoECommand.PROCESS_DATA

CRC 校验​

from darra_ethercat import fsoe_crc16, fsoe_crc16_fast

data = bytes([0x01, 0x02, 0x03])
crc = fsoe_crc16(data)           # 标准计算
crc_fast = fsoe_crc16_fast(data) # 查找表加速

is_connection_id_available()​

from darra_ethercat.slave.fsoe import is_connection_id_available

def is_connection_id_available(conn_id: int) -> bool

模块级函数，校验指定 ConnectionID 是否可用（未被其他活动 FSoE 连接占用）。

参数：

  • conn_id (int) — 待校验的连接 ID, 合法范围 [1, 65535]（0 保留）

返回值：

  • bool — True 可用（available）; False 已占用 / 非法
ETG.5120 §5.2.3

ConnectionID = 0 为保留值（不可分配），合法范围 [1, 65535]。 同一从站网络内每个 FSoE 连接必须使用唯一 ConnID, 重复会导致 WRONG_CONNECTION_ID (FSoEError 0x0003)。

示例 — 初始化前占用检查：

from darra_ethercat.slave.fsoe import is_connection_id_available
from darra_ethercat import SafetyManager

conn_id = 0x1234
if not is_connection_id_available(conn_id):
    print(f"ConnID 0x{conn_id:04X} 不可用（已占用或非法），请选择其他 ID")
else:
    mgr = SafetyManager(master._dll, master.master_index)
    # 显式指定 connection_id 的绑定入口
    mgr.bind_safe_input(slave_index=1, connection_id=conn_id,
                        safety_address=0x0100, input_size=2, watchdog_ms=100)
    mgr.bind_safe_output(slave_index=1, connection_id=conn_id,
                         safety_address=0x0100, output_size=1, watchdog_ms=100)

示例 — 动态分配可用 ID:

def allocate_conn_id(start: int = 1) -> int:
    """从 start 开始扫描首个可用 ConnID"""
    for cid in range(start, 0x10000):
        if is_connection_id_available(cid):
            return cid
    raise RuntimeError("无可用 FSoE ConnectionID")

my_id = allocate_conn_id(0x1000)

完整示例​

单连接安全 IO​

from darra_ethercat import EtherCATMaster, EcState
from darra_ethercat.slave.fsoe import FSoEState

with EtherCATMaster() as master:
    master.set_eni("config.deni")
    master.build()
    master.set_state(EcState.OP)

    slave = master[1]
    fsoe = slave.fsoe

    if fsoe.is_capable:
        # 初始化连接
        fsoe.init_connection(
            connection_id=0,
            safety_address=0x0100,
            watchdog_time_ms=100,
            safe_input_size=2,
            safe_output_size=1
        )

        # 设置失效安全输出
        fsoe.set_failsafe_output(bytes([0x00]))

        # 请求进入数据交换状态
        fsoe.request_state(FSoEState.DATA)

        # 注册事件
        slave.events.add_fsoe_error(lambda code:
            print(f"FSoE 错误: 0x{code:04X}"))

        # PDO 回调中读写安全数据
        def on_pdo(mi):
            data = fsoe.read_safe_input()
            if data:
                ch0 = (data[0] & 0x01) != 0
                fsoe.write_safe_output(bytes([0x01 if ch0 else 0x00]))

        master.on_pdo_cycle(on_pdo)

        import time
        time.sleep(10)
        fsoe.close_connection()

SafetyManager 一步绑定​

from darra_ethercat import EtherCATMaster, EcState, SafetyManager

master = EtherCATMaster()
master.set_eni("config.deni")
master.build()
master.set_state(EcState.OP)

mgr = SafetyManager(master._dll, master.master_index)

# 一步绑定（状态机由协议引擎自动推进 Session→Connection→Parameter→Data）
mgr.bind_safe_io(
    slave_index=1,
    safe_input_offset=0, safe_input_size=2,
    safe_output_offset=0, safe_output_size=1,
    safety_address=0x0100, watchdog_ms=100
)

# MDP 伺服驱动器安全轴绑定
mgr.bind_mdp_drive_axis(slave_index=2, axis_index=0,
                        safety_address=0x0200, watchdog_ms=100)

# 检查所有连接状态
print(mgr.get_status_summary())
if mgr.all_in_data_state():
    print("所有安全连接已进入数据交换状态")

# PDO 回调（bind_safe_io 的 connection_id = slave_index + 1 = 2）
def on_pdo(mi):
    data = mgr.read_input_frame(1, 2)
    if data:
        mgr.write_output_frame(1, 2, bytes([0x01]))

master.on_pdo_cycle(on_pdo)

import time
time.sleep(30)
mgr.close_all()
master.close()

安全驱动器（带类型化读写）​

import ctypes
from darra_ethercat import EtherCATMaster, EcState, SafetyManager

class SafeDriveInput(ctypes.Structure):
    _pack_ = 1
    _fields_ = [
        ("safety_status_word", ctypes.c_uint16),
        ("safe_actual_position", ctypes.c_int32),
        ("safe_actual_velocity", ctypes.c_int32),
    ]

class SafeDriveOutput(ctypes.Structure):
    _pack_ = 1
    _fields_ = [
        ("safety_control_word", ctypes.c_uint16),
    ]

master = EtherCATMaster()
master.set_eni("config.deni")
master.build()
master.set_state(EcState.OP)

mgr = SafetyManager(master._dll, master.master_index)

# 一步绑定安全驱动器（10 字节输入 + 2 字节输出）
mgr.bind_safe_io(
    slave_index=1,
    safe_input_offset=0,
    safe_input_size=ctypes.sizeof(SafeDriveInput),   # 10
    safe_output_offset=0,
    safe_output_size=ctypes.sizeof(SafeDriveOutput),  # 2
    safety_address=0x0100,
    watchdog_ms=100
)

# PDO 回调中进行安全驱动控制（bind_safe_io 的 connection_id = slave_index + 1 = 2）
def on_pdo(mi):
    data = mgr.read_input_frame(1, 2)
    if data and len(data) >= ctypes.sizeof(SafeDriveInput):
        drive_in = SafeDriveInput.from_buffer_copy(data)

        drive_out = SafeDriveOutput()
        if drive_in.safety_status_word & 0x0100:  # Bit8: 安全故障
            drive_out.safety_control_word = 0x0080  # 确认故障
            print(f"安全故障: 位置={drive_in.safe_actual_position}")
        else:
            drive_out.safety_control_word = 0x0000  # 清除 STO

        mgr.write_output_frame(1, 2, bytes(drive_out))

# 失效安全事件
slave = master[1]
slave.events.add_fsoe_failsafe_triggered(lambda:
    print("驱动器安全停止!"))

master.on_pdo_cycle(on_pdo)

import time
time.sleep(30)
mgr.close_all()
master.close()

MDP 多连接​

from darra_ethercat import SafeMdp

mdp = SafeMdp(master._dll, master.master_index, slave_index=1)

# 探测并初始化所有连接
conn_count = mdp.detect_connections()
print(f"支持 {conn_count} 个 FSoE 连接")

# 获取设备安全地址
addr = mdp.get_device_address()
print(f"设备安全地址: 0x{addr:04X}")

for i in range(conn_count):
    mdp.init_connection(
        connection_id=i + 1,
        safety_address=0x0100 + i,
        watchdog_time_ms=100,
        safe_input_size=2,
        safe_output_size=1
    )

# 查询各连接状态
for i in range(conn_count):
    status = mdp.get_status(i + 1)
    if status:
        print(f"连接 {i + 1}: 状态={status['state']}, 失效安全={status['in_failsafe']}")

# 模块诊断（形参为模块编号）
for i in range(conn_count):
    diag = mdp.get_module_diagnosis(i)
    if diag:
        state, diagnosis = diag
        print(f"模块 {i}: 诊断码=0x{diagnosis:04X}")

# 关闭
for i in range(conn_count):
    mdp.close_connection(i + 1)

MDP 多轴安全驱动器​

from darra_ethercat import EtherCATMaster, EcState, SafetyManager

master = EtherCATMaster()
master.set_eni("config.deni")
master.build()
master.set_state(EcState.OP)

mgr = SafetyManager(master._dll, master.master_index)

# 绑定 2 个轴（每轴独立 FSoE 连接）
mgr.bind_mdp_drive_axis(slave_index=1, axis_index=0,
                        safety_address=0x0100, watchdog_ms=100)
mgr.bind_mdp_drive_axis(slave_index=1, axis_index=1,
                        safety_address=0x0200, watchdog_ms=100)

# 检查所有连接状态
print(mgr.get_status_summary())
if mgr.all_in_data_state():
    print("所有安全连接已进入数据交换状态")

# PDO 回调（bind_mdp_drive_axis 的 connection_id = axis_index + 1）
def on_pdo(mi):
    # 各轴独立读写安全数据
    for axis in range(2):
        data = mgr.read_input_frame(1, axis + 1)
        if data:
            # 处理安全数据...
            pass

master.on_pdo_cycle(on_pdo)

# 事件监控: 每轴独立
slave = master[1]
slave.events.add_fsoe_failsafe_triggered(lambda:
    print("轴失效安全!"))

import time
time.sleep(30)
mgr.close_all()
master.close()