跳到主要内容

OpcUaNode

OpcUaNode 是 NodeCollection 索引器返回的包装对象, 表示远程一个节点. Lazy Load — 只有访问属性才触发 RPC.

前置阅读

公共属性​

类别属性类型访问说明
标识 (无 RPC)node_idstrR节点 ID 字符串
数据 (触发 RPC)valueOpcUaDataValueR当前值 — 触发 read(Value)
display_nameOptional[str]R显示名 — 触发 read(DisplayName)
browse_nameOptional[str]R浏览名 — 触发 read(BrowseName)
descriptionOptional[str]R描述 — 触发 read(Description)
node_classNodeClassR节点类别 — 触发 read(NodeClass)
data_type_node_idOptional[str]R数据类型 NodeId — 触发 read(DataType)
is_readableboolRAccessLevel bit0
is_writableboolRAccessLevel bit1
子节点 (触发 RPC)childrenList[Reference]R子节点列表 — 触发 browse

索引器​

索引返回说明
node[browse_name]Optional[OpcUaNode]按 BrowseName 取直接子节点, 未找到返回 None

公共方法​

方法说明
read_attribute(attr)读指定 Attribute, 返回 OpcUaDataValue
write(value)写 Value 属性, 返回 StatusCode
write_attribute(attr, value)写任意属性
resolve(path)按 BrowsePath 深层解析 (相对当前节点)
call(parent_object_node_id, *inputs)调用本节点为方法

例子​

from darra_opcua import OpcUaVariant, AttributeId

# 1. 拿节点 (无 RPC)
temp = ua.nodes["ns=2;s=Temperature"]

# 2. 读基本信息 (各 1 次 RPC)
print(f"NodeId       = {temp.node_id}")
print(f"display_name = {temp.display_name}")
print(f"node_class   = {temp.node_class.name}")
print(f"data_type    = {temp.data_type_node_id}")

# 3. 读值 (with 自动 dispose)
with temp.value as dv:
    print(f"Value = {dv.value} °C @ {dv.source_timestamp}")

# 4. 写
temp.write(OpcUaVariant().set_double(25.5))

# 5. 浏览子节点
for child in temp.children:
    print(f"  {child.browse_name} -> {child.node_id}")

# 6. 链式访问 (Server.ServerStatus.CurrentTime)
with ua.nodes.server["ServerStatus"]["CurrentTime"].value as dv:
    print(f"Server time = {dv.value}")

注意事项​

  • 所有触发 RPC 的属性都是"读一次返回一次", 不缓存. 想缓存自己存变量
  • 链式访问 a["b"]["c"] 每段都是一次 browse RPC, 总计 N 次
  • OpcUaDataValue 必须 dispose, 否则泄漏 native 内存. 用 with.
  • 跨线程使用 OpcUaNode 安全 (它只持有 NodeId 字符串), 但 OpcUaDataValue 不要跨线程

与 resolve 的对比​

# 方式 A: 链式索引器 - 4 次 RPC
with ua.nodes.server["ServerStatus"]["CurrentTime"].value as dv:  # 1+2+3+4
    print(dv.value)

# 方式 B: 路径解析 - 2 次 RPC
node = ua.resolve("/Objects/Server/ServerStatus/CurrentTime")     # 1
with node.value as dv:                                            # 1
    print(dv.value)

# 方式 C: 直接 NodeId - 1 次 RPC
with ua.read("i=2258") as dv:
    print(dv.value)

如果路径已知 → 方式 C; 探索式开发 → 方式 A.

下一步​