跳到主要内容

call

前置 / 配套
  • 数据类型 OpcUaVariant 是入参 / 出参的容器.
  • 找方法 NodeId 用 browse 配 filter=NodeClass.Method.

签名​

def call(self,
         object_node_id: str,
         method_node_id: str,
         inputs: Optional[List[OpcUaVariant]] = None) -> List[OpcUaVariant]: ...
参数说明
object_node_id方法所在 Object 节点 NodeId
method_node_id方法本身 NodeId
inputs入参 Variant 列表 (None / [] 表示无参)

返回 List[OpcUaVariant] 输出参数列表.

内存所有权 (重要)​

返回列表里每个 OpcUaVariant 都持有 native handle (_owns=True), 必须 dispose, 否则泄漏 native Variant 内存.

推荐写法:

outputs = ua.call(obj_id, mtd_id, args)
for v in outputs:
    with v:
        print(v.as_int32)

或显式:

outputs = ua.call(obj_id, mtd_id, args)
try:
    for v in outputs:
        # 用 v.as_xxx 等
        ...
finally:
    for v in outputs:
        v.close()

例子​

from darra_opcua import OpcUaVariant

# 调 Calculator.Add(3, 4)
inputs = [OpcUaVariant().set_int32(3), OpcUaVariant().set_int32(4)]
outputs = ua.call(
    object_node_id="ns=2;s=Calculator",
    method_node_id="ns=2;s=Calculator.Add",
    inputs=inputs)

for v in outputs:
    with v:
        print(f"sum = {v.as_int32}")

# 入参用完也 dispose (call 内部不消费)
for v in inputs:
    v.close()

简化版:

def call_simple(ua, obj, mtd, *vs):
    outputs = ua.call(obj, mtd, list(vs))
    try:
        return [v.value for v in outputs]   # value = Pythonic 解码
    finally:
        for v in outputs: v.close()
        for v in vs: v.close()

result = call_simple(ua, "ns=2;s=Calc", "ns=2;s=Calc.Add",
                     OpcUaVariant().set_int32(3), OpcUaVariant().set_int32(4))
print(result)   # [7]

异常​

异常含义
OpcUaException(BadNodeIdInvalid)NodeId 解析失败
OpcUaException(BadCommunicationError)Transport 错
OpcUaException(BadMethodInvalid)该 NodeId 不是 Method
OpcUaException(BadInvalidArgument)入参类型 / 个数不符
OpcUaException(BadUserAccessDenied)无权限调
OpcUaException(BadXxx)方法返回非 Good 业务码

业务级失败 (方法返回非 Good) 也抛异常, ex.status_code 即业务码.

找方法的 NodeId​

from darra_opcua import NodeClass

# 1. browse Object 子节点过滤 Method
methods = ua.browse("ns=2;s=Calculator", filter=NodeClass.Method)
for m in methods:
    print(f"  {m.browse_name} -> {m.node_id}")

# 2. 看入参 / 出参签名 (browse 该方法节点的 Property)
props = ua.browse(methods[0].node_id)   # 含 InputArguments / OutputArguments

输入参数构造​

按方法签名构造 OpcUaVariant:

# Add(Int32, Int32) -> Int32
OpcUaVariant().set_int32(3), OpcUaVariant().set_int32(4)

# SetTemperature(string deviceId, double value)
OpcUaVariant().set_string("Boiler1"), OpcUaVariant().set_double(85.5)

# 复杂结构 (ExtensionObject) — 高级用法, 后续版本完善

异步 (后续版本)​

当前 call 是同步阻塞 (内部走 RPC + 等响应). 异步版本待后续版本提供.

如果想异步, 用 concurrent.futures / asyncio.to_thread:

import asyncio

result = await asyncio.to_thread(ua.call, obj_id, mtd_id, args)

下一步​