跳到主要内容

事件订阅

OPC UA 事件 (Alarms & Conditions) 通过 subscribe_events 订阅. 这是 OPC UA 协议事件, 与 SDK 内部 ua.events.* 不是一回事.

区别
  • ua.subscribe_events(...) — OPC UA 协议事件 (报警 / 条件, 服务端推送, 本节)
  • ua.events.* — SDK 内部事件 (会话 / 订阅生命周期, 与协议无关)

签名​

def subscribe_events(self,
                     node_id: Optional[str] = None,
                     existing_subscription: Optional[Subscription] = None
                     ) -> EventSubscription: ...
参数默认说明
node_idNone (Server, i=2253)事件源节点
existing_subscriptionNone复用已存在的 Subscription, None 则内部新建

EventSubscription​

成员说明
event_arrivedList[Callable[[EventArrivedEventArgs], None]] 事件到达回调列表
dispose() / close()移除该 Event MI; 如果是内部新建容器还会销毁容器

支持 with context manager.

EventArrivedEventArgs​

@dataclass
class EventArrivedEventArgs:
    severity: int = 0
    message: Optional[str] = None
    source_name: Optional[str] = None
    time: Optional[datetime] = None
    event_type: Optional[str] = None
字段类型说明
severityint严重度 1-1000
messageOptional[str]事件文本
source_nameOptional[str]事件源名
timeOptional[datetime]事件发生时间 (UTC)
event_typeOptional[str]事件类型 NodeId

用法​

默认订阅 Server 节点​

import time

with ua.subscribe_events() as ev:
    def on_event(e):
        print(f"[{e.severity}] {e.source_name}: {e.message}")
        print(f"  EventType={e.event_type} Time={e.time}")
    ev.event_arrived.append(on_event)

    time.sleep(300)  # 5 分钟

订阅指定节点的事件​

with ua.subscribe_events("ns=2;s=Line1") as ev:   # Line1 必须是 EventNotifier
    ev.event_arrived.append(lambda e: print(e.message))
    time.sleep(60)

复用已有 Subscription 容器​

with ua.create_subscription(500) as sub:
    sub.add_node("ns=2;s=Counter")    # 数据订阅

    # 把 Event MI 挂到同一个容器
    with ua.subscribe_events("ns=2;s=Line1", existing_subscription=sub) as ev:
        ev.event_arrived.append(lambda e: print(e.message))
        time.sleep(60)
        # ev.dispose 只移除 Event MI, 不销毁 sub

EventNotifier​

只有 EventNotifier Attribute (12) 非 0 的节点才能产生事件. 默认 Server (i=2253) 是 EventNotifier. 自定义节点是否有事件源能力, 看服务端实现.

EventFilter (高级)​

OPC UA EventFilter 可以过滤 EventType / Severity / SourceNode, 当前 SDK 默认订阅 BaseEventType (全部事件), 复杂过滤待后续版本暴露.

报警操作 (Acknowledge / Confirm)​

收到 AlarmConditionType 事件后, 操作员通常要 Acknowledge (确认). 通过 call 调用报警节点的 Acknowledge Method:

from darra_opcua import OpcUaVariant

ua.call(
    object_node_id=condition_id,
    method_node_id=f"{condition_id}.Acknowledge",
    inputs=[
        OpcUaVariant().set_string(event_id),                # bytes(EventId) 视服务端实现
        OpcUaVariant().set_string("Operator confirmed"),    # LocalizedText 简化为 string
    ])

下一步​