初始化
强烈建议使用 Darra 配置工具导出的 DENI 文件进行初始化。 DENI 文件包含完整的网络配置(网口、从站、PDO 映射、DC 设置等),可确保配置的一致性和正确性。
构造函数
EtherCATMaster()
class EtherCATMaster:
def __init__(self, dll_path: Optional[str] = None,
master_index: Optional[int] = None) -> None
创建 EtherCAT 主站实例。
参数:
dll_path(str, 可选) — 原生库路径,None 时自动搜索master_index(int, 可选) — 指定主站索引,None 时自动分配
示例:
from darra_ethercat import EtherCATMaster, EcState
with EtherCATMaster() as master:
# 加载配置文件
master.set_eni(r"C:\EtherCAT\MyProject.deni")
# 设置网络并扫描从站
slave_count = master.set_network(r"\Device\NPF_{GUID}")
print(f"发现 {slave_count} 个从站")
# 状态转换到 OP
master.set_state(EcState.OP)
# 启动 PDO 循环
master.start()
上下文管理器
with EtherCATMaster() as master:
master.set_eni("config.deni")
count = master.set_network(r"\Device\NPF_{GUID}")
if count > 0:
master.set_state(EcState.OP)
master.start()
# master 退出 with 块时自动调用 close()
配置方法
set_eni()
def set_eni(self, eni_path: str) -> "EtherCATMaster"
加载 ENI / DENI / JSON 配置文件。读文件 UTF-8 后交给 DLL LoadConfigJson。原生层检测到 DENI XML 走 ParseDeniXmlConfig(选网卡 + 活扫 + 全配置),JSON 走 JSON 引擎。返回自身实例,支持链式调用。
用户加载总线配置只走 set_eni() / load_config_json()。save_xml / apply_xml / MasterXMLConfiguration 不是产品 API。
load_config_json()
def load_config_json(self, json_str: str) -> int
从 JSON 或 DENI XML 字符串加载配置(DLL LoadConfigJson)。
参数:
json_str(str) — JSON 或 DENI XML 文本
返回值:
int— DLL 返回码
示例:
with open(r"C:\EtherCAT\MyProject.deni", encoding="utf-8") as f:
rc = master.load_config_json(f.read())
set_network()
def set_network(self, adapter: str, redundant_adapter: str = "") -> int
设置网络适配器并扫描从站。
参数:
adapter(str) — 主网卡名称(网卡设备名)redundant_adapter(str) — 冗余网卡名称,空字符串表示不使用冗余
返回值:
int— 发现的从站数量0— 失败(原生0xFFFF失败哨兵归一化)或成功但总线上没有从站。两种含义用master.last_error区分:21= 驱动未就绪(瞬态,可重试),1168= NDIS 协议未绑定网卡,5023= 数据闸关闭(授权未通过,与网卡无关)
示例:
# Windows
slave_count = master.set_network(r"\Device\NPF_{GUID}")
# Linux
# slave_count = master.set_network("eth0")
# 冗余模式
slave_count = master.set_network(
r"\Device\NPF_{PRIMARY-GUID}",
r"\Device\NPF_{SECONDARY-GUID}"
)
set_esi_file()
def set_esi_file(self, file_path: str) -> "EtherCATMaster"
加载单个 ESI 文件或目录。立即调用 DLL EcEsi_LoadFile / EcEsi_LoadDirectory(目录则整目录加载)。ESI 解析在 DLL(EcEsi_*)。扫站后 build() 会 AutoMatchAll。返回自身实例,支持链式调用。
如果使用 DENI 文件,通常不需要单独加载 ESI。
native EcEsi_* 是 char*(ANSI)入口。Windows 必须用 mbcs(系统 ANSI 代码页),不能 UTF-8 — 路径含中文时 CreateFile / FindFirstFile 会找不到。Linux 用 UTF-8。
set_esi_files()
def set_esi_files(self, path: str) -> "EtherCATMaster"
加载 ESI 文件或目录。立即调用 DLL:目录 → EcEsi_LoadDirectory,文件 → EcEsi_LoadFile。Windows 路径同样按 mbcs 传入 native。返回自身实例,支持链式调用。
enable_auto_startup()
def enable_auto_startup(self) -> "EtherCATMaster"
标记需要自动配置从站。返回自身实例,支持链式调用。
自动配置(build() 时执行 AutoConfigureSM)按各从站启动参数中的 PDO Assignment / Mapping 元数据计算并写入 SM2/SM3 物理长度,只处理有启动参数的从站;没有启动参数的从站跳过。
使用 DENI 文件时,set_network() 阶段的扫描结果会自动与 DENI 中的从站配置匹配(设备与版本匹配按 ESI 设备声明的 CheckRevisionNo 策略执行)。网络拓扑与 DENI 不一致时 set_state() 返回 False,未到目标的从站与卡住的状态步写入日志。
set_eni() 和 set_network() 都可以设置网口。当两者同时使用时,后调用的方法覆盖前面的冲突配置:
# set_eni 在后 -> 使用 DENI 文件中的网口配置
master.set_network(r"\Device\NPF_{GUID}")
master.set_eni(r"C:\config.deni") # <- DENI 中的网口覆盖 set_network
# set_network 在后 -> set_network 的网口覆盖 DENI
master.set_eni(r"C:\config.deni")
master.set_network(r"\Device\NPF_{GUID}") # <- 覆盖 DENI 中的网口
DENI 文件中的从站配置、PDO 映射、DC 设置等始终保留,仅网口配置受覆盖规则影响。
set_cycle_time()
def set_cycle_time(self, time_ns: int) -> None
设置 PDO 循环周期(纳秒)。
示例:
master.set_cycle_time(1_000_000) # 1ms
build()
def build(self) -> "EtherCATMaster"
收尾链式初始化:若已通过 set_esi_file() / set_esi_files() 加载 ESI,扫站后执行 EcEsi_AutoMatchAll 绑定从站;enable_auto_startup() 标记过则执行自动配置;随后记录驱动兼容性日志。返回自身实例。
一步初始化
EtherCATMaster.from_json()
@classmethod
def from_json(cls, json_config: str, dll_path: Optional[str] = None) -> "EtherCATMaster"
通过 JSON 配置字符串一步完成主站初始化(内部执行 Initialize → SetNetwork → LoadConfigJson → AutoConfigureSM → 状态转换到 target_state → 启动 PDO)。
参数:
json_config(str) — JSON 格式的配置字符串dll_path(str, 可选) — 原生库路径
返回值:
EtherCATMaster— 已初始化的主站实例
JSON 配置格式:
{
"adapter": "\\Device\\NPF_{GUID}",
"adapter_redundant": "",
"target_state": 8,
"slaves": [
{
"index": 1,
"startup_params": [
{
"index": 7186, "sub_index": 0,
"data_hex": "00", "transition": 0, "timing": 1
}
]
}
]
}
字段说明:
adapter(必填)主网卡名;adapter_redundant冗余网卡名,空串表示不用冗余target_state数值状态码,缺省 8(EcState.OP);取值见EcState枚举slaves[].index从站索引(1 起);startup_params[]为该从站的启动参数- 启动参数:
index/sub_index用数值(0x1C12 写 7186);data_hex是十六进制字节串;transition取TRANS_IP=0/TRANS_PS=1/TRANS_SO=2/TRANS_OS=3/TRANS_SP=4/TRANS_PI=5(见darra_ethercat.utils.dll);timing取TIMING_BEFORE=0/TIMING_AFTER=1;可选priority(数值小优先)与ca(Complete Access) - PDO 周期不在 JSON 中配置,用
set_cycle_time()/config.loop_cycle设置
示例:
import json
config = json.dumps({
"adapter": r"\Device\NPF_{GUID}",
"target_state": 8 # EcState.OP
})
with EtherCATMaster.from_json(config) as master:
# 一步初始化完成:主站已进入 OP 并启动 PDO
print(f"从站数: {master.slave_count}")
EtherCATMaster.from_json_file()
@classmethod
def from_json_file(cls, json_file_path: str, dll_path: Optional[str] = None) -> "EtherCATMaster"
通过 JSON 文件一步初始化(与 from_json() 同语义,读文件后走同一初始化链)。
示例:
with EtherCATMaster.from_json_file("config.json") as master:
# 一步初始化完成:主站已进入 OP 并启动 PDO
print(f"从站数: {master.slave_count}")
资源释放
close()
def close(self) -> bool
关闭并释放主站资源(安全停机 → 注销回调 → EcClose 释放;旧 DLL 回退 Stop + Dispose)。返回 True 表示已释放(重复调用幂等);返回 False 表示未完成释放(native 资源仍被占用),可再次调用重试,原因见 last_close_error。
__enter__ / __exit__
支持 with 语句自动释放资源。close() 未完成释放且无在飞异常时,__exit__ 抛 RuntimeError 如实上报。
EtherCATMaster.emergency_cleanup()
@staticmethod
def emergency_cleanup(dll_path: Optional[str] = None) -> None
紧急清理所有主站资源。在进程异常退出时调用,释放网络句柄。
示例:
import atexit
atexit.register(EtherCATMaster.emergency_cleanup)
多实例管理
EtherCATMaster.get_active_instance_count()
@staticmethod
def get_active_instance_count(dll_path: Optional[str] = None) -> int
当前活跃的主站实例数量。
EtherCATMaster.get_max_instance_count()
@staticmethod
def get_max_instance_count(dll_path: Optional[str] = None) -> int
允许的最大主站实例数量。基于可用隔离核心数和驱动上限决定。
网络扫描
枚举只含物理有线口:虚拟网卡、WiFi、TAP 不进列表,ASIX 等 USB 有线网卡保留。每张网卡用 bound_protocol 标记是否已绑定 DarraRT_Eth(1=已绑定,0=未绑定),未绑定口不要当作主口候选。
get_network_info()
from darra_ethercat import get_network_info
from darra_ethercat.utils import DarraCoreDLL
def get_network_info(dll: DarraCoreDLL,
is_redundant: bool = False,
need_slaves_num: bool = False) -> List[NetworkInfo]
获取网卡列表(物理有线口,绑定状态见 bound_protocol)。第一个参数为 DarraCoreDLL 实例。
参数:
dll(DarraCoreDLL) — DLL 接口实例is_redundant(bool) — 是否按冗余模式枚举,默认Falseneed_slaves_num(bool) — 是否同时探测各网卡从站数量 (较慢),默认False
NetworkInfo 数据类:
@dataclass
class NetworkInfo:
name: str # 网卡适配器名称
desc: str # 适配器描述
mac: str # MAC 地址 (格式 aa:bb:cc:dd:ee:ff)
slave_num: Optional[int] # 从站总数 (need_slaves_num=True 时有值)
redundant_slave_num: Optional[int] # 受冗余保护的从站数量 (need_slaves_num=True 时有值)
bound_protocol: Optional[int] # 是否已绑定 DarraRT_Eth (1=已绑定, 0=未绑定, need_slaves_num=True 时有值)
示例:
from darra_ethercat import get_network_info
from darra_ethercat.utils import DarraCoreDLL
dll = DarraCoreDLL()
adapters = get_network_info(dll, need_slaves_num=True)
for adapter in adapters:
print(f"{adapter.desc}: {adapter.slave_num or 0} 个从站")
scan_slaves()
from darra_ethercat import scan_slaves
from darra_ethercat.utils import DarraCoreDLL
def scan_slaves(dll: DarraCoreDLL, adapter_name: str,
secondary_adapter: Optional[str] = None) -> List[ScannedSlaveInfo]
完整扫描:填名字/Vendor,供选用从站。DENI 快速计数请用 quick_slave_count()。count > 0 时列表不得为空。
示例:
from darra_ethercat import scan_slaves
from darra_ethercat.utils import DarraCoreDLL
dll = DarraCoreDLL()
slaves = scan_slaves(dll, r"\Device\NPF_{GUID}")
for s in slaves:
print(f"[{s.index}] {s.name} VID=0x{s.vendor_id:08X}")
abort_scan() / reset_scan_abort()
from darra_ethercat import abort_scan, reset_scan_abort
def abort_scan(dll: DarraCoreDLL) -> None
def reset_scan_abort(dll: DarraCoreDLL) -> None
abort_scan() 中止当前正在进行的从站扫描 (用于关闭窗口 / 取消操作时快速中断阻塞的扫描); reset_scan_abort() 在下次扫描前重置中止标志。
示例:
import threading
from darra_ethercat import scan_slaves, abort_scan
from darra_ethercat.utils import DarraCoreDLL
dll = DarraCoreDLL()
# 在后台线程中扫描
scan_thread = threading.Thread(target=lambda: scan_slaves(dll, r"\Device\NPF_{GUID}"))
scan_thread.start()
# 用户点击取消
abort_scan(dll)
scan_thread.join()
EtherCATMaster.quick_slave_count()
@staticmethod
def quick_slave_count(adapter: str, dll_path: Optional[str] = None) -> int
快速从站计数(不读取 EEPROM,仅使用广播读取)。
EtherCATMaster.get_serial_number()
@staticmethod
def get_serial_number(dll_path: Optional[str] = None) -> str
获取当前设备的序列号(用于授权验证)。
完整示例
手动初始化流程
from darra_ethercat import EtherCATMaster, EcState
with EtherCATMaster() as master:
# 设置网络适配器
slave_count = master.set_network(r"\Device\NPF_{GUID}")
if slave_count == 0:
print("未发现从站")
exit()
# 设置 PDO 周期
master.set_cycle_time(1_000_000)
# 状态转换
master.set_state(EcState.OP)
# 启动 PDO 循环
master.start()
# 访问从站
slave = master[1]
status = slave.coe.sdo_read_value(0x6041, 0, dtype='u16')
print(f"StatusWord = 0x{status:04X}" if status else "读取失败")
import time
time.sleep(5)
master.stop()
使用 DENI 文件(推荐)
from darra_ethercat import EtherCATMaster, EcState
with EtherCATMaster() as master:
master.set_eni(r"C:\EtherCAT\MyProject.deni")
slave_count = master.set_network(r"\Device\NPF_{GUID}")
master.set_state(EcState.OP)
master.start()
动态扫描网口
from darra_ethercat import EtherCATMaster, EcState, get_network_info
from darra_ethercat.utils import DarraCoreDLL
adapters = get_network_info(DarraCoreDLL(), need_slaves_num=True)
target = next((a for a in adapters if (a.slave_num or 0) > 0), None)
if target is None:
exit()
with EtherCATMaster() as master:
master.set_esi_files(r"C:\ESI")
master.enable_auto_startup()
slave_count = master.set_network(target.name)
if slave_count > 0:
master.set_state(EcState.OP)
master.start()
冗余模式
from darra_ethercat import EtherCATMaster, EcState
with EtherCATMaster() as master:
master.set_redundancy_mode(2)
master.set_eni(r"C:\config.deni")
slave_count = master.set_network(
r"\Device\NPF_{PRIMARY}",
r"\Device\NPF_{SECONDARY}"
)
master.set_state(EcState.OP)
master.start()
多实例(多网卡独立总线)
每个主站实例管理一条独立的 EtherCAT 总线,使用不同的网卡。
- 每个实例必须使用不同的网卡 — 同一网卡不能被多个实例同时使用
- CPU 核心自动分配 — 每个实例自动占用不同的 CPU 核心
from darra_ethercat import EtherCATMaster, EcState, get_network_info
from darra_ethercat.utils import DarraCoreDLL
adapters = [a for a in get_network_info(DarraCoreDLL(), need_slaves_num=True)
if (a.slave_num or 0) > 0]
master1 = EtherCATMaster()
master1.set_network(adapters[0].name)
master1.set_state(EcState.OP)
master1.start()
master2 = EtherCATMaster()
master2.set_network(adapters[1].name)
master2.set_state(EcState.OP)
master2.start()
print(f"活跃主站数: {EtherCATMaster.get_active_instance_count()}")