跳到主要内容

MethodBuilder

概述

OPC UA Call 服务要求把入参打包成 Variant[], 顺序与服务端 InputArguments 一一对应。MethodBuilder 提供链式 API, 一行串起 load_input_args / arg / invoke, 自动按 Python 类型选 Variant 编码。

对应规范段: Part 4 §5.11.2 (Call) / Part 5 §6.4 (Argument)。

API

方法类别读写说明
MethodBuilder(ua, object_id, method_id)构造绑定到方法
.load_input_args()链式读取 InputArguments 描述
.arg(name_or_index, value)链式按名 / 按索引设值
.invoke()终端返回 (StatusCode, list[Variant])
.input_args属性LoadInputArgs 后参数清单

支持 Python 类型: bool / int / float / str / datetime / bytes / list

代码示例

from opcua import DarraOpcUa, MethodBuilder, StatusCode

with DarraOpcUa("opc.tcp://localhost:4840") as ua:
ua.connect()

# 1) 流式调用 - 按名设参
st, outs = (MethodBuilder(ua, "ns=2;s=Robot", "ns=2;s=Robot.MoveTo")
.load_input_args()
.arg("x", 100.0)
.arg("y", 200.0)
.arg("speed", 50)
.invoke())

if st != StatusCode.GOOD:
print(f"MoveTo 失败: {st}")

# 2) 按索引
st2, outs2 = (MethodBuilder(ua, "ns=2;s=Calc", "ns=2;s=Calc.Add")
.arg(0, 3.14).arg(1, 2.71).invoke())
print(f"Add → {outs2[0].as_double}")

# 3) 复用 builder
b = MethodBuilder(ua, "ns=2;s=Pump", "ns=2;s=Pump.SetSpeed").load_input_args()
for rpm in (100, 500, 1000, 1500):
s, _ = b.arg("rpm", rpm).invoke()
if s != StatusCode.GOOD:
break
time.sleep(0.5)

与 call 的对比

# 旧写法: 手撸 list
inputs = [Variant(100.0), Variant(200.0), Variant(50)] # 整型? 浮点?
outs = ua.call("ns=2;s=Robot", "ns=2;s=Robot.MoveTo", inputs)

# 新写法: MethodBuilder
st, outs = (MethodBuilder(ua, "ns=2;s=Robot", "ns=2;s=Robot.MoveTo")
.load_input_args()
.arg("x", 100.0).arg("y", 200.0).arg("speed", 50)
.invoke())

实现限制

  • load_input_args() 依赖 ExtensionObject 解码, 部分 Server 上 input_args 列表可能为空 (但按索引仍可工作)
  • 不支持复杂结构入参, 只支持基本类型 + 数组
  • invokecall, 失败抛 OpcUaException. 想 Safe 包 try/except

跨语言对照

C#PythonJavaC++RustC
MethodBuilderMethodBuilderMethodBuilderMethodBuilderMethodBuilderDarraUa_MethodBuilder_*
.Arg(name, value).arg(name, value).arg(name, value).Arg(name, value).arg(name, value)DarraUa_MethodBuilder_ArgByName

下一步