跳到主要内容

PDO 输入输出

通过 slave.pdo 访问过程映像。​

PDO 就是过程映像里的一段数据，内核每周期自动收发。

映射成 ctypes 结构体后，读写字段就是直接读写过程映像本身 —— 0 拷贝：不用拷贝、不用刷新，读到的永远是当前周期，写下的下个周期发出。

不想定义结构体时，用元素视图 input_array / output_array 按字节偏移单点读写。

PDO 周期与邮箱的关系

邮箱通信（SDO/CoE/FoE 等）自动附带在 PDO 周期中，不占用额外帧。邮箱响应延迟约 2-3 个 PDO 周期。

结构体契约
  • 必须 _pack_ = 1，字段顺序与 PDO 映射逐字节一致；存在可判定的填充字节（漏写 _pack_）时映射抛 RuntimeError，不静默错位。
  • 不得大于对应方向的映像字节数（slave.pdo.inputs_mapping_info["size"] / outputs_mapping_info["size"]）。
    • 超尺寸、映像不可用、offset 越界都抛 RuntimeError（消息带从站号、方向与字节数）。
    • 小于合法，按偏移覆盖。
  • 字段只能是 ctypes 定宽类型；字符串用 ctypes.c_char * n，变长数据用整帧 inputs 解。
  • 映射一次长期有效；close() / 重新进 OP 后旧视图作废，重进 OP 时重取。

使用案例​

日常使用​

① 映射到结构体（0 拷贝，持续用）：进 OP 后映射一次，拿到钉在过程映像上的结构体实例，之后当变量读写。

import ctypes
from darra_ethercat import EtherCATMaster, EcState

class InPdo(ctypes.Structure):
    _pack_ = 1                                       # 必须；字段顺序与 PDO 逐字节一致
    _fields_ = [
        ("status_word", ctypes.c_uint16),
        ("actual_position", ctypes.c_int32),
    ]

class OutPdo(ctypes.Structure):
    _pack_ = 1
    _fields_ = [
        ("control_word", ctypes.c_uint16),
        ("target_position", ctypes.c_int32),
    ]

master = (EtherCATMaster()
          .set_eni(r"C:\EtherCAT\MyProject.deni")    # DENI 一行加载
          .build())
master.set_state(EcState.OP)                         # 进 OP

slave = master[1]                                    # 从站（1-based 索引）

inputs  = slave.pdo.inputs_mapping(InPdo)            # 只读视图（进 OP 后映射一次）
outputs = slave.pdo.outputs_mapping(OutPdo)

outputs.control_word    = 0x000F                     # 写：下个周期发出
outputs.target_position = inputs.actual_position     # 读：永远是当前周期
print(f"状态字=0x{inputs.status_word:04X} 位置={inputs.actual_position}")

② 元素视图（不定义结构体）：按字节偏移取一个元素，元素的每个成员就是一个定宽标量读写口。

# 单点：索引越界抛 IndexError
pos = slave.pdo.input_array[2].as_int32              # 偏移 2 起 4 字节
sw  = slave.pdo.input_array[0].as_uint16             # 偏移 0 起 2 字节（StatusWord）

slave.pdo.output_array[0].as_uint16 = 0x000F         # 写：下个周期发出
slave.pdo.output_array[2].as_int32  = 10000

# 整段：一份完整输入帧快照，同一拍一致，同一份 bytes 里解多个字段
import struct
frame = slave.pdo.inputs                             # bytes；取不到帧返回 None
if frame is not None:
    sw2, pos2 = struct.unpack_from("<Hi", frame, 0)
    print(f"状态字=0x{sw2:04X} 位置={pos2}")

周期变化​

映射取一次就够，同一份映射放进你自己的周期循环（或循环里的调度点）直接用：

import time

while running:
    outputs.control_word    = 0x000F if inputs.status_word & 0x0004 else 0x0006
    outputs.target_position = inputs.actual_position
    time.sleep(0.001)                                # 你自己的节奏，例如 1 ms

内核每周期搬运过程数据：循环里读到的永远是当前周期、写下的下个周期发出，不需要任何额外调用，也不要每拍重新映射（close() / 重新进 OP 后旧视图才作废，重进 OP 时重取一次即可）。

结构体映射（零拷贝）​

inputs_mapping() / outputs_mapping()​

def inputs_mapping(self, struct_type, offset=0)      # TxPDO（从站→主站）：只读
def outputs_mapping(self, struct_type, offset=0)     # RxPDO（主站→从站）：可读可写

把 ctypes 结构体钉在 GetIO live 过程映像指针上，映射一次后读字段即读当前周期值（0 拷贝，不是快照）。

参数：

  • struct_type （ctypes.Structure 子类） — 必须 _pack_ = 1，字段顺序与该方向 PDO 映射逐字节一致
  • offset (int) — 相对该方向过程映像起始的字节偏移（可选），默认 0

返回值：

  • 输入方向 — 钉在 live 输入过程映像上的结构体实例；只读，给字段赋值抛 RuntimeError
  • 输出方向 — 钉在 live 输出过程映像上的结构体实例；改字段即改 PDO，下个周期发出

说明：

  • 进 OP 后调用。
  • 下列情况一律抛 RuntimeError：
    • GetIO 失败（主站未启动 / 未进 OP / 该方向无映射）
    • 映像为空
    • offset 为负
    • offset + ctypes.sizeof(struct_type) 超过映像字节数
    • 结构体存在可判定的填充字节

示例：

inputs  = slave.pdo.inputs_mapping(InPdo)           # 只映射一次（进 OP 后）
outputs = slave.pdo.outputs_mapping(OutPdo)

outputs.control_word = 0x000F                        # 写：下个周期发出
pos = inputs.actual_position                         # 读：当前周期值

inputs_slice_mapping() / outputs_slice_mapping()​

从指定字节偏移处钉页，供 MDP 等按模块切段的设备使用。

def inputs_slice_mapping(self, struct_type, offset)   # 输入切片：只读
def outputs_slice_mapping(self, struct_type, offset)  # 输出切片：可写

参数：

  • struct_type （ctypes.Structure 子类） — 必须 _pack_ = 1，字段顺序与 PDO 映射逐字节一致
  • offset (int) — 相对该方向过程映像起始的字节偏移，必填（不接受默认 0，避免把切片调用误写成整段调用）

返回值：

  • 与 inputs_mapping / outputs_mapping 同类型 — 钉在 live 过程映像 offset 处的结构体实例（非快照）；输入方向只读

说明：

  • 与 inputs_mapping(struct_type, offset) 同一实现、同一越界口径。
  • offset 为负、或 offset + ctypes.sizeof(struct_type) 超过映像字节数，抛 RuntimeError。

示例：

# MDP 设备：模块 1 从偏移 0 开始，模块 2 从偏移 8 开始（模块各自定义 ctypes 结构体）
mod1 = slave.pdo.inputs_slice_mapping(Module1In, 0)
mod2 = slave.pdo.inputs_slice_mapping(Module2In, 8)

out2 = slave.pdo.outputs_slice_mapping(Module2Out, 8)
out2.setpoint = 1500

bind_pdo_struct()​

def bind_pdo_struct(self, struct_type, is_input=True, offset=0)

按方向参数转调 inputs_mapping / outputs_mapping，供同时需要两个方向的调用点用一个入口。

参数：

  • struct_type （ctypes.Structure 子类） — 必须 _pack_ = 1
  • is_input (bool) — True 取输入方向（只读），False 取输出方向（可写），默认 True
  • offset (int) — 相对该方向过程映像起始的字节偏移，默认 0

返回值：

  • 结构体实例 — 钉在 live 过程映像上（非快照）；输入方向只读

示例：

input_ref = slave.pdo.bind_pdo_struct(InPdo, is_input=True)
print(input_ref.actual_position)                     # 每次读即最新值

inputs_mapping_info / outputs_mapping_info​

@property
def inputs_mapping_info(self) -> dict                # slave.pdo
@property
def outputs_mapping_info(self) -> dict

本方向过程映像的 live 指针与字节数，用来在映射前核对结构体大小。

返回值：

  • dict — {"pointer": int, "size": int, "slave_index": int}；size 是本方向过程映像字节数，未进 OP / 未映射时为 0

示例：

import ctypes

if ctypes.sizeof(InPdo) > slave.pdo.inputs_mapping_info["size"]:
    raise RuntimeError("结构体大于输入 PDO")

直接访问（不映射也能用）​

下面这组调用与结构体映射彼此独立：不映射可以直接用，映射了也常并用（结构体放不下的字段用它解）。

input_array / output_array​

@property
def input_array(self) -> PDOArrayInstance            # TxPDO（从站→主站）：只读
@property
def output_array(self) -> PDOArrayInstance           # RxPDO（主站→从站）：可读可写

按字节偏移索引过程数据元素：input_array[i] 读输入，output_array[i] 读写输出，i 相对本方向过程映像起点。

返回值：

  • PDOArrayInstance — 按字节索引的过程映像视图；length 是本方向过程映像字节数，未进 OP / 未映射时为 0（此时任何索引都抛 IndexError）

说明：

  • input_array 是 TxPDO，主站每周期覆盖，只读：写它的元素成员或 input_array[i] = v 一律抛 RuntimeError。
  • output_array 是 RxPDO，每次成员赋值走一次版本化事务（Copy → 改帧 → 提交，冲突有界重试），成功的写下个周期发出。
  • 提交失败抛 RuntimeError。
  • 元素视图按偏移缓存复用，读取每次取一份当前帧快照后按偏移解析 —— 与 slave.pdo.inputs 同一份快照语义，不是 inputs_mapping() 的 0 拷贝钉页引用。
  • 索引越界抛 IndexError；偏移或宽度落在本方向过程映像之外抛 ValueError（与索引越界刻意分开，便于区分选错字节与选错类型）。
  • 索引不是整数、整型成员（content 与 as_int8 … as_uint64）的赋值不是整数（bool 按 0 / 1）抛 TypeError。
  • as_float / as_double 另收 int / float / bool。
  • 快照取不到（未进 OP / 该方向无映射）抛 RuntimeError。

新鲜度：

  • 一次成员访问 = 一份新快照。
  • item.as_int16 与 item.as_bool 各读一次，两次之间可能已经过了一个总线周期，同一元素的两个字段可能来自不同拍。
  • 要多字段同拍一致，先取一份整帧 frame = slave.pdo.inputs，再从这一份 bytes 里按偏移解析。
  • 要 0 拷贝的整段视图用 inputs_mapping(struct_type)。

字节序： 多字节一律小端（EtherCAT 线序），本层不做任何换序。

0 拷贝：

  • 元素访问器不是 0 拷贝路径 —— 每次读取都先复制一份完整过程映像帧（length 字节）再按偏移解析。
  • 0 拷贝只用以下入口（都钉在 live 过程映像上，不是快照）：
    • inputs_mapping
    • outputs_mapping
    • inputs_slice_mapping
    • outputs_slice_mapping
    • bind_pdo_struct

相关结构：

class PDOArrayInstance:        # slave.pdo.input_array / output_array
    length                     # int，只读：本方向过程映像字节数；未进 OP / 未映射 = 0
    def __getitem__(index)     # → PDODataItem；index ∉ [0, length) 抛 IndexError
    def __setitem__(index, value)   # output_array[i] = v ≡ output_array[i].content = v
    def __len__()              # ≡ length

class PDODataItem:             # 单个 PDO 项的视图：slave.pdo.input_array[i] / output_array[i]
    content                    # int，读写：原始字节（无符号 8 位）
    as_int8                    # int，读写：有符号 8 位
    as_uint8                   # 与 content 同义同字节
    as_int16                   # 多字节一律小端（EtherCAT 线序）
    as_uint16
    as_int32
    as_uint32
    as_int64
    as_uint64
    as_float                   # IEEE754 单精度
    as_double                  # IEEE754 双精度
    as_bool                    # bool，读写：bit0 语义，只有最低位为 1 才是 True
    def get_bit_in_byte(bit_index)     # → bool；bit_index 0-7，0 = LSB
    def set_bit_in_byte(bit_index, value)   # 仅输出方向；只改目标位，同字节其余位不动
    def get_bit(bit_index)     # 旧名别名（DeprecationWarning）→ get_bit_in_byte
    def set_bit(bit_index, value)   # 旧名别名（DeprecationWarning）→ set_bit_in_byte
    def as_value(data_type)    # 按 EcDataType 读：只支持 11 个定宽类型
    def write_value(data_type, value)   # 按 EcDataType 写
    def as_base_data(data_type)         # → BaseData：位串 / 变长类型的出口
    def write_from_base_data(data)      # → bool；失败返回 False（不抛）
    def __int__()              # int(item) ≡ .content

as_bool 只认最低位：字节 0x02 与 0x80 都读 False。

写 as_bool 是整字节写：True 写 0x01、False 写 0x00（不是改单个位），该字节的 bit1–bit7 会被一并清掉；要按位读写请用 get_bit_in_byte() / set_bit_in_byte()。

要按整字节 0 / 非 0 判真假，读 content 自行比较 —— 没有第二个「非 0 即真」的口子。

as_value / write_value 只支持与定宽类型一一对应的 11 个数据类型：

  • BOOLEAN
  • INTEGER8 / INTEGER16 / INTEGER32 / INTEGER64
  • UNSIGNED8 / UNSIGNED16 / UNSIGNED32 / UNSIGNED64
  • REAL32 / REAL64

位串（BIT1…BIT8）、24 / 40 / 48 / 56 位整数、字符串、DOMAIN 一律抛 ValueError。

示例：

item = slave.pdo.input_array[2]                  # 偏移 2 处的元素视图
pos  = item.as_int32                             # 读：当前帧快照
sw   = slave.pdo.input_array[0].as_uint16        # 偏移 0 起 2 字节（StatusWord）

slave.pdo.output_array[2].as_int32 = 10000       # 写：版本化局部提交，下个周期发出
slave.pdo.output_array[0].set_bit_in_byte(0, True)       # 只置 bit0，同字节其余位不动
slave.pdo.output_array[0] = 0x11                 # ≡ output_array[0].content = 0x11
flag = slave.pdo.input_array[3].as_bool          # bit0 语义
byte0 = int(slave.pdo.input_array[0])            # ≡ .content

for i in range(slave.pdo.output_array.length):   # 遍历整个输出映像
    slave.pdo.output_array[i].set_bit_in_byte(0, False)

get_bit_in_byte() / set_bit_in_byte()​

读 / 写元素所在那个字节的第 bit_index 位。旧名 get_bit() / set_bit() 仍可用（转调新名，触发 DeprecationWarning），新代码用新名。

def get_bit_in_byte(self, bit_index) -> bool
def set_bit_in_byte(self, bit_index, value) -> None

参数：

  • bit_index (int) — 位号 0-7，0 是最低位（LSB）；不在 0-7 抛 ValueError
  • value (bool) — 目标位的新值（仅 set_bit_in_byte）

返回值：

  • bool — 该位当前是否为 1（get_bit_in_byte）
  • None — set_bit_in_byte 无返回值

说明：

  • 位操作按字节寻址，与元素宽度无关：无论元素按 1 / 2 / 4 / 8 字节哪种类型解读，bit_index 始终指向元素偏移处的那一个字节，不会扩展到相邻字节。
  • get_bit_in_byte 输入 / 输出方向都可读。
  • set_bit_in_byte 只用于输出方向（输入方向抛 RuntimeError），走一次版本化事务，只改目标位，同一字节其余位保持不变。
  • offset 超出当前帧抛 ValueError，事务提交失败抛 RuntimeError。

示例：

slave.pdo.output_array[0].set_bit_in_byte(0, True)       # 偏移 0 那个字节的 bit0
slave.pdo.output_array[0].set_bit_in_byte(7, False)      # 同一个字节的 bit7
ready = slave.pdo.input_array[4].get_bit_in_byte(2)      # 偏移 4 那个字节的 bit2 —— 与元素宽度无关

as_value() / write_value()​

def as_value(self, data_type)                        # → int | float | bool
def write_value(self, data_type, value)              # 仅输出方向

按 ESI / ENI 里那条 PDO 映射条目的数据类型读写，不用自己记宽度。

参数：

  • data_type (EcDataType) — 该位置的映射类型，见 EcDataType
  • value — 要写入的值（仅 write_value），按目标类型换算，不要求与 data_type 同类

返回值：

  • int / float / bool — 读到的值；BOOLEAN 返回 bool（bit0 语义）

说明：

  • 白名单恰好 11 个定宽类型，其余一律抛 ValueError —— 不猜宽度、不静默按 1 字节读（BIT8 有意不落到 as_uint8：位串与无符号 8 位语义不同）。
  • write_value 的值按目标类型换算：浮点转整数就近取偶，NaN / 无穷 / 超出目标类型取值范围抛 OverflowError。
  • value=None 或来源类型不受支持抛 TypeError。

示例：

from darra_ethercat import EcDataType

pos  = slave.pdo.input_array[2].as_value(EcDataType.INTEGER32)
flag = slave.pdo.input_array[3].as_value(EcDataType.BOOLEAN)      # bit0 语义

slave.pdo.output_array[2].write_value(EcDataType.INTEGER32, 10000)

as_base_data() / write_from_base_data()​

宽类型出口：as_value / write_value 只覆盖 11 个定宽类型，位串与变长字段用这一对进出。

def as_base_data(self, data_type) -> BaseData
def write_from_base_data(self, data) -> bool

参数：

  • data_type (EcDataType) — 该位置的映射类型；字节宽度按数据类型表取（向上取整到字节），变长 / 未知类型按 1 字节处理
  • data (BaseData) — 要写回的整段原始字节

返回值：

  • BaseData — 按 data_type 宽度从当前位置拷出的原始字节包装
  • bool — 是否写成功；失败返回 False（不抛），输入方向固定返回 False

说明：

  • as_base_data 在偏移 / 宽度超出当前帧时抛 ValueError。
  • write_from_base_data 走输出 PDO 的版本化提交，下个周期发出。
  • 越界、传入对象不是 BaseData、数据为空都以 False 返回，不抛异常。

示例：

from darra_ethercat import BaseData, EcDataType

wide = slave.pdo.input_array[0].as_base_data(EcDataType.BIT3)     # 位串：定宽标量覆盖不到
ok   = slave.pdo.output_array[0].write_from_base_data(wide)       # 失败返回 False（不抛）

blob = BaseData(b"\x01\x02", dtype=EcDataType.UNSIGNED16)         # 也可以自己构造整段写回
slave.pdo.output_array[0].write_from_base_data(blob)

inputs / outputs / write_outputs()​

按字节访问过程映像整帧，用于结构体放不下的字段（字符串、变长数据）。

@property
def inputs(self) -> Optional[bytes]                  # 输入整帧：每次访问返回一份独占快照
@property
def outputs(self) -> Optional[bytes]                 # 输出整帧快照
def write_outputs(self, data) -> bool                # 整帧替换

返回值：

  • bytes — 快照，不是 0 拷贝：同一拍一致，拷出来不跟着总线变；取不到帧返回 None，该方向长度为 0 时返回 b""
  • bool — 是否写入成功（write_outputs）

说明：

  • 这三个成员声明在 slave.pdo 上（slave.pdo.inputs），不在 slave 上。
  • write_outputs 是单写者的权威整帧替换：长度与该方向过程映像字节数不一致时拒绝返回 False，且不截断。
  • 它是整帧写，不要拿它做局部 / 批量的读改写（那类更新用元素视图成员，走版本化事务）。

示例：

frame = slave.pdo.inputs
if frame is not None:
    name = frame[6:22].split(b"\x00", 1)[0].decode("ascii", "replace")
    print(f"status=0x{frame[0] | (frame[1] << 8):04X} name={name}")

out = bytearray(slave.pdo.outputs or b"")
if out:
    out[0:2] = (0x000F).to_bytes(2, "little")
    slave.pdo.write_outputs(bytes(out))              # 整帧替换

batch_read() / batch_write()​

一次取多个区间 / 一次提交多个区间。

def batch_read(self, requests) -> list               # 一份快照里解多个区间
def batch_write(self, requests) -> int               # 一个事务写多个区间

参数：

  • requests (list) — batch_read 为 (offset, size) 元组列表；batch_write 为 (offset, data) 元组列表
  • offset (int) — 字节偏移，相对该方向过程映像起点

返回值：

  • list — batch_read 返回 bytes 列表，失败项为 None
  • int — batch_write 返回成功写入的项数（整批一个事务：全部成功返回总项数，任一项失败整批返回 0）

说明：

  • batch_read 一次取一份完整输入帧快照，多个区间从同一拍解出。
  • batch_write 是一个事务、一次提交：任一项越界即整批失败，不部分提交、不静默截断。
  • 单字段读写走元素视图（见 input_array / output_array）：读 slave.pdo.input_array[0].as_uint16，写 slave.pdo.output_array[0].as_int32 = 10000；越界抛异常，不返回默认值。
  • 字符串等结构体放不下的字段用整帧 inputs / outputs 自己解（见上文 inputs / outputs / write_outputs()）。

示例：

values = slave.pdo.batch_read([(0, 2), (2, 4)])      # 一次快照，多个字段
ok = slave.pdo.batch_write([(0, (0x000F).to_bytes(2, "little")),
                            (2, (10000).to_bytes(4, "little"))])

完整示例​

import ctypes
import time
from darra_ethercat import EcDataType, EtherCATMaster, EcState

class InPdo(ctypes.Structure):
    _pack_ = 1
    _fields_ = [("status_word", ctypes.c_uint16),
                ("actual_position", ctypes.c_int32)]

class OutPdo(ctypes.Structure):
    _pack_ = 1
    _fields_ = [("control_word", ctypes.c_uint16),
                ("target_position", ctypes.c_int32)]

master = (EtherCATMaster()
          .set_eni(r"C:\EtherCAT\MyProject.deni")    # DENI 一行加载
          .build())
master.set_state(EcState.OP)                         # 进 OP

slave = master[1]                                    # 从站（1-based 索引）

# ① 0 拷贝：映射一次，结构体钉在 live 过程映像上
inputs  = slave.pdo.inputs_mapping(InPdo)            # 只读视图
outputs = slave.pdo.outputs_mapping(OutPdo)

# ② 元素视图：不定义结构体，按字节偏移单点读写
print(slave.pdo.input_array[0].as_uint16)            # 偏移 0 起 2 字节，小端
slave.pdo.output_array[2].write_value(EcDataType.INTEGER32, 10000)
slave.pdo.output_array[0].set_bit_in_byte(0, True)           # 只改偏移 0 那个字节的 bit0

# ③ 整帧快照：同一拍一致，用来解结构体放不下的字段
frame = slave.pdo.inputs
if frame is not None:
    print(f"快照 {len(frame)} 字节, status=0x{frame[0] | (frame[1] << 8):04X}")

# 周期循环：读到的永远是当前周期，写下的下个周期发出
for _ in range(1000):
    outputs.control_word    = 0x000F if inputs.status_word & 0x0004 else 0x0006
    outputs.target_position = inputs.actual_position
    time.sleep(0.001)                                # 你自己的节奏，例如 1 ms

print(f"状态字=0x{inputs.status_word:04X} 位置={inputs.actual_position}")
master.close()