read / write

前置阅读

• 数据类型 OpcUaVariant 和 OpcUaDataValue 是所有读写的基础.
• 高频访问相同节点请用 register_nodes 加速.
• 实时变化通知用 Subscription 而不是循环 read.

read

def read(self,
         node_id: str,
         attribute: AttributeId = AttributeId.Value) -> OpcUaDataValue: ...

读单节点的指定 Attribute.

参数	类型	默认	说明
node_id	str	—	NodeId 字符串 (i=2258 / ns=2;s=...)
attribute	AttributeId	Value (13)	要读的属性

返回: OpcUaDataValue (context manager, 强烈推荐 with)

异常: OpcUaException (BadNodeIdUnknown / BadNotReadable / BadCommunicationError / ...)

from darra_opcua import AttributeId

# 读 Value
with ua.read("ns=2;s=Temperature") as dv:
    print(dv.value)                    # Pythonic 解码 (按 DataType 自动)
    print(dv.variant.as_double)        # 强类型抽取

# 读 DisplayName
with ua.read("ns=2;s=Temperature", AttributeId.DisplayName) as dv:
    print(dv.value)

---

read_many

def read_many(self,
              node_ids: List[str],
              attribute: AttributeId = AttributeId.Value
              ) -> List[Optional[OpcUaDataValue]]: ...

一次 RPC 批量读多个节点的同一 Attribute.

nodes = ["ns=2;s=T1", "ns=2;s=T2", "ns=2;s=T3"]
results = ua.read_many(nodes)
for nid, dv in zip(nodes, results):
    if dv is None:
        print(f"{nid}: <failed>")
        continue
    with dv:
        print(f"{nid} = {dv.value}")

返回长度 = len(node_ids), 失败槽位为 None. 每个非 None 元素都是 OpcUaDataValue (持有 native 内存, 必须 dispose).

性能: 批量比循环单读快 ~N 倍 (省 N-1 次 RPC 往返). 100 节点典型 ~10 ms vs 循环 ~500 ms.

---

write

def write(self,
          node_id: str,
          value: OpcUaVariant,
          attribute: AttributeId = AttributeId.Value) -> StatusCode: ...

写单节点.

from darra_opcua import OpcUaVariant, StatusCode

st = ua.write("ns=2;s=Setpoint", OpcUaVariant().set_double(42.5))
if st != StatusCode.Good:
    print(f"Write failed: {st.name}")

返回 StatusCode (Good / BadNotWritable / BadTypeMismatch / BadUserAccessDenied / ...).

业务级失败不抛异常

write 不抛异常 (除非 transport 级错误), 业务级失败通过返回值判断. 必须检查返回值, 不要假定 Good.

write 内部不消费传入的 OpcUaVariant; 用户构造的 Variant 仍由用户负责 dispose. 推荐:

v = OpcUaVariant().set_int32(42)
try:
    ua.write("ns=2;s=Setpoint", v)
finally:
    v.close()

# 或 with 简写
with OpcUaVariant().set_int32(42) as v:
    ua.write("ns=2;s=Setpoint", v)

---

OpcUaVariant 构造速查

from darra_opcua import OpcUaVariant
import datetime as dt

# 流式 Setter (链式调用)
OpcUaVariant().set_boolean(True)
OpcUaVariant().set_sbyte(-1)
OpcUaVariant().set_byte(255)
OpcUaVariant().set_int16(42)
OpcUaVariant().set_uint16(42)
OpcUaVariant().set_int32(42)
OpcUaVariant().set_uint32(42)
OpcUaVariant().set_int64(42)
OpcUaVariant().set_float(3.14)
OpcUaVariant().set_double(3.14)
OpcUaVariant().set_string("hello")
OpcUaVariant().set_datetime(dt.datetime.now(dt.timezone.utc))

详见 OpcUaVariant 数据类型.

---

OpcUaDataValue 字段速查

字段	类型	说明
value	Any	Pythonic 解码 (按 DataType 自动转 bool/int/float/str/datetime)
variant	OpcUaVariant	原始 Variant (借用, 无需 dispose)
status	StatusCode	Good / Bad* / Uncertain*
is_good	bool	== (status == StatusCode.Good)
source_timestamp	Optional[datetime]	数据源时间戳 (UTC)
server_timestamp	Optional[datetime]	服务端打的时间戳 (UTC)

with ua.read("ns=2;s=T1") as dv:
    if not dv.is_good:
        print(f"Bad value: {dv.status.name}")
        return
    print(f"{dv.value} @ {dv.source_timestamp.isoformat()}")

---

AttributeId 常用值

Id	名称	说明
1	NodeId	节点 ID
2	NodeClass	节点类别
3	BrowseName	浏览名
4	DisplayName	显示名
5	Description	描述
13	Value	变量值 (默认)
14	DataType	数据类型 NodeId
17	AccessLevel	访问权限位
20	Historizing	是否在记录历史
21	Executable	Method 是否可执行

---

最佳实践

• 批量优先: 监控 100+ Tag 用 read_many, 不要循环 read
• 必须 dispose: OpcUaDataValue 持有 native 内存, 漏 dispose 会泄漏. 用 with.
• write 检查返回值: 不要假定一定 Good
• 实时变化用订阅: 不要 1 秒一次轮询 read, 改用 create_subscription DataChange 事件 (服务端推送)

下一步

• Subscription 订阅
• register_nodes 高频访问加速