跳到主要内容

添加 MonitoredItem

前置阅读

sub.add_node(...)​

def add_node(self,
             node_id: str,
             callback: Optional[Callable[[DataChangeEventArgs], None]] = None,
             attribute: AttributeId = AttributeId.Value,
             sampling_interval_ms: float = -1.0) -> int: ...
参数默认说明
node_id—监控的 NodeId
callbackNone该 MI 专用回调 (容器级 data_changed 仍会被触发)
attributeValue (13)监控的 Attribute
sampling_interval_ms-1 (跟订阅)服务端采样间隔. 0 = 尽快

返回: int MonitoredItem handle (本地句柄, 用于后续 modify / remove)

异常: OpcUaException (BadNodeIdUnknown 等)

h1 = sub.add_node("ns=2;s=Temperature")
h2 = sub.add_node("ns=2;s=Pressure", sampling_interval_ms=100)

# 带专用回调
def on_temp(e):
    print(f"Temperature: {e.value_string}")

h3 = sub.add_node("ns=2;s=Temperature2", callback=on_temp)

add_many​

一次 RPC 添加多个 MI, 比循环 add_node 快得多:

def add_many(self,
             node_ids: List[str],
             callback: Optional[Callable[[DataChangeEventArgs], None]] = None,
             attribute: AttributeId = AttributeId.Value
             ) -> List[Tuple[int, StatusCode]]: ...

返回每项 (mi_handle, status) — 失败项 mi=0.

nodes = [f"ns=2;s=T{i}" for i in range(1, 101)]
results = sub.add_many(nodes)
for nid, (mi, st) in zip(nodes, results):
    if st == StatusCode.Good:
        print(f"  {nid} -> handle={mi}")
    else:
        print(f"  {nid} FAILED: {st.name}")

性能: 100 个 MI 用 add_many 约 1 次 RPC (~10 ms), 用循环 add_node 约 100 次 RPC (~500-1000 ms).

DataChangeEventArgs 字段​

订阅的回调参数 DataChangeEventArgs (dataclass, 跨线程安全):

字段类型说明
monitored_item_handleint触发的 MI 句柄
node_idstr该 MI 的 NodeId
value_stringOptional[str]值的字符串表示 (预抽取, 跨线程安全)
data_type_nameOptional[str]内置数据类型枚举名 (如 Double)
statusStatusCodeDataValue 的 Status
source_timestampOptional[datetime]数据源时间戳 (UTC)
server_timestampOptional[datetime]服务端时间戳 (UTC)
安全设计

value_string / data_type_name 已在 C 层 Publish 线程同步抽取为 Python str/枚举, 跨线程使用安全. 不再暴露原始 native DataValue 指针, 避免 use-after-free 闪退.

移除​

sub.remove(handle)
sub.remove_by_node_id("ns=2;s=Temperature")  # 按 NodeId 移除全部对应

不同 Attribute 的监控​

除了默认监控 Value, 也能监控 Status / Quality:

sub.add_node("ns=2;s=T1", attribute=AttributeId.Value)
sub.add_node("ns=2;s=T1", attribute=AttributeId.StatusCode)  # 只关心状态变化

DataChangeFilter (死区)​

当前 SDK 默认不设 filter (任何变化都推). 如需 Absolute / Percent 死区, 走底层 API (后续版本会暴露).

下一步​