core.subscription_manager.SubscriptionManager

class documentation

class SubscriptionManager:

Constructor: SubscriptionManager()

View In Hierarchy

Centrally manages all OPC UA subscription needs.

It bundles subscriptions using only one subscription to an OPC UA server to not stay within max. subscription limits of certain OPC UA servers.

Method __init__ Create a new instance.
Async Method clear_subscriptions Undocumented
Async Method datachange_notification Handle OPC UA data change notifications callback.
Async Method event_notification Handle OPC UA event notifications callback.
Async Method init Undocumented
Async Method status_change_notification Handle OPC UA status notifications callback.
Async Method subscribe_data_change Add given handler to receive data change notifications for given nodes.
Async Method subscribe_event Add given handler to receive event notifications for given node.
Instance Variable logger Undocumented
Instance Variable period Publishing interval in milliseconds for OPC UA subscription.
Property no_of_monitored_items Return the number of monitored nodes.
Async Method _create_subscription Undocumented
Instance Variable _data_subscription_map Maps a OPC UA node to a list of handlers for data change notifications.
Instance Variable _event_subscription_map Maps a OPC UA node to a list of handlers for event notifications.
Instance Variable _ua_client Undocumented
Instance Variable _ua_subscription Undocumented
def __init__(self):

Create a new instance.

async def clear_subscriptions(self):

Undocumented

async def datachange_notification(self, node: Node, val: Any, data: DataChangeNotif):

Handle OPC UA data change notifications callback.

async def event_notification(self, ua_event: UaEvent):

Handle OPC UA event notifications callback.

async def init(self, ua_client: Client, period: float = 100):

Undocumented

async def status_change_notification(self, status: ua.StatusChangeNotification):

Handle OPC UA status notifications callback.

async def subscribe_data_change(self, handler: UaDataChangeSubscriber, nodes: Node | NodeId | Iterable[Node | NodeId]):

Add given handler to receive data change notifications for given nodes.

async def subscribe_event(self, handler: UaEventSubscriber, source_node: Node | NodeId):

Add given handler to receive event notifications for given node.

logger =

Undocumented

period: float =

Publishing interval in milliseconds for OPC UA subscription.

@property
no_of_monitored_items: int =

Return the number of monitored nodes.

async def _create_subscription(self, ua_client: Client):

Undocumented

_data_subscription_map: dict[NodeId, list[UaDataChangeSubscriber]] =

Maps a OPC UA node to a list of handlers for data change notifications.

_event_subscription_map: dict[NodeId, list[UaEventSubscriber]] =

Maps a OPC UA node to a list of handlers for event notifications.

_ua_client: Client =

Undocumented

_ua_subscription: Subscription =

Undocumented