Subscription.__init__() - Code Metrics - Inspection of "WhereClause support" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#179)

by Olivier

created 2016-05-06 19:46 UTC

Subscription.init() A

↳ Parent: Subscription

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	1

Metric	Value
cc	1
dl	0
loc	13
rs	9.4285
ccs	13
cts	13
cp	1
crap	1

"""
high level interface to subscriptions
"""
import time
import logging
from threading import Lock

from opcua import ua
from opcua import Node


class SubHandler(object):
    """
    Subscription Handler. To receive events from server for a subscription
    This class is just a sample class. Whatever class having these methods can be used
    """

    def data_change(self, handle, node, val, attr):
        """
        Deprecated, use datachange_notification
        """
        pass

    def datachange_notification(self, node, val, data):
        """
        called for every datachange notfication from server
        """
        pass

    def event_notification(self, event):
        """
        called for every event notfication from server
        """
        pass

    def status_change_notification(self, status):
        """
        called for every status change notfication from server
        """
        pass


class EventResult(object):
    """
    To be sent to clients for every events from server
    """

    def __init__(self):
        self.server_handle = None

    def __str__(self):
        return "EventResult({})".format([str(k) + ":" + str(v) for k, v in self.__dict__.items()])
    __repr__ = __str__


class SubscriptionItemData(object):
    """
    To store usefull data from a monitored item
    """
    def __init__(self):
        self.node = None
        self.client_handle = None
        self.server_handle = None
        self.attribute = None
        self.mfilter = None


class DataChangeNotif(object):
    """
    To be send to clients for every datachange notification from server
    """
    def __init__(self, subscription_data, monitored_item):
        self.monitored_item = monitored_item
        self.subscription_data = subscription_data

    def __str__(self):
        return "DataChangeNotfication({}, {})".format(self.subscription_data, self.monitored_item)
    __repr__ = __str__


class Subscription(object):
    """
    Subscription object returned by Server or Client objects.
    The object represent a subscription to an opc-ua server.
    This is a high level class, especially subscribe_data_change
    and subscribe_events methods. If more control is necessary look at
    code and/or use create_monitored_items method.
    """

    def __init__(self, server, params, handler):
        self.logger = logging.getLogger(__name__)
        self.server = server
        self._client_handle = 200
        self._handler = handler
        self.parameters = params  # move to data class
        self._monitoreditems_map = {}
        self._lock = Lock()
        self.subscription_id = None
        response = self.server.create_subscription(params, self.publish_callback)
        self.subscription_id = response.SubscriptionId  # move to data class
        self.server.publish()
        self.server.publish()

    def delete(self):
        """
        Delete subscription on server. This is automatically done by Client and Server classes on exit
        """
        results = self.server.delete_subscriptions([self.subscription_id])
        results[0].check()

    def publish_callback(self, publishresult):
        self.logger.info("Publish callback called with result: %s", publishresult)
        while self.subscription_id is None:
            time.sleep(0.01)

        for notif in publishresult.NotificationMessage.NotificationData:
            if isinstance(notif, ua.DataChangeNotification):
                self._call_datachange(notif)
            elif isinstance(notif, ua.EventNotificationList):
                self._call_event(notif)
            elif isinstance(notif, ua.StatusChangeNotification):
                self._call_status(notif)
            else:
                self.logger.warning("Notification type not supported yet for notification %s", notif)

        ack = ua.SubscriptionAcknowledgement()
        ack.SubscriptionId = self.subscription_id
        ack.SequenceNumber = publishresult.NotificationMessage.SequenceNumber
        self.server.publish([ack])

    def _call_datachange(self, datachange):
        for item in datachange.MonitoredItems:
            with self._lock:
                if item.ClientHandle not in self._monitoreditems_map:
                    self.logger.warning("Received a notification for unknown handle: %s", item.ClientHandle)
                    continue
                data = self._monitoreditems_map[item.ClientHandle]
            if hasattr(self._handler, "datachange_notification"):
                event_data = DataChangeNotif(data, item)
                try:
                    self._handler.datachange_notification(data.node, item.Value.Value.Value, event_data)
                except Exception:
                    self.logger.exception("Exception calling data change handler")
            elif hasattr(self._handler, "data_change"):  # deprecated API
                self.logger.warning("data_change method is deprecated, use datachange_notification")
                try:
                    self._handler.data_change(data.server_handle, data.node, item.Value.Value.Value, data.attribute)
                except Exception:
                    self.logger.exception("Exception calling deprecated data change handler")
            else:
                self.logger.error("DataChange subscription created but handler has no datachange_notification method")

    def _call_event(self, eventlist):
        for event in eventlist.Events:
            with self._lock:
                data = self._monitoreditems_map[event.ClientHandle]
            result = EventResult()
            result.server_handle = data.server_handle
            for idx, sattr in enumerate(data.mfilter.SelectClauses):
                if len(sattr.BrowsePath) == 0:
                    setattr(result, sattr.AttributeId.name, event.EventFields[idx].Value)
                else:
                    setattr(result, sattr.BrowsePath[0].Name, event.EventFields[idx].Value)
            if hasattr(self._handler, "event_notification"):
                try:
                    self._handler.event_notification(result)
                except Exception:
                    self.logger.exception("Exception calling event handler")
            elif hasattr(self._handler, "event"):  # depcrecated API
                try:
                    self._handler.event(data.server_handle, result)
                except Exception:
                    self.logger.exception("Exception calling deprecated event handler")
            else:
                self.logger.error("Event subscription created but handler has no event_notification method")

    def _call_status(self, status):
        try:
            self._handler.status_change_notification(status.Status)
        except Exception:
            self.logger.exception("Exception calling status change handler")

    def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value):
        """
        Subscribe for data change events for a node or list of nodes.
        default attribute is Value.
        Return a handle which can be used to unsubscribe
        If more control is necessary use create_monitored_items method
        """
        return self._subscribe(nodes, attr, queuesize=0)

    def _get_node(self, nodeid):
        if isinstance(nodeid, ua.NodeId):
            node = Node(self.server, nodeid)
        elif isinstance(nodeid, Node):
            node = nodeid
        else:
            node = Node(self.server, ua.NodeId(nodeid))
        return node

    def _get_filter_from_event_type(self, eventtype):
        eventtype = self._get_node(eventtype)
        evfilter = ua.EventFilter()
        evfilter.SelectClauses = self._select_clauses_from_evtype(eventtype)
        evfilter.WhereClause = self._where_clause_from_evtype(eventtype)
        return evfilter

    def _select_clauses_from_evtype(self, evtype):
        clauses = []
        for property in get_event_properties_from_type_node(evtype):
            op = ua.SimpleAttributeOperand()
            op.TypeDefinitionId = evtype.nodeid
            op.AttributeId = ua.AttributeIds.Value
            op.BrowsePath = [property.get_browse_name()]
            clauses.append(op)
        return clauses

    def _where_clause_from_evtype(self, evtype):
        cf = ua.ContentFilter()
        el = ua.ContentFilterElement()
        # operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
        op = ua.SimpleAttributeOperand()
        op.AttributeId = ua.AttributeIds.NodeId
        el.FilterOperands.append(op)
        for subtypeid in [st.nodeid for st in self._get_subtypes(evtype)]:
            op = ua.LiteralOperand()
            op.Value = ua.Variant(subtypeid)
            el.FilterOperands.append(op)
        el.FilterOperator = ua.FilterOperator.InList

        cf.Elements.append(el)
        return cf

    def _get_subtypes(self, parent, nodes=None):
        if nodes is None:
            nodes = [parent]
        for child in parent.get_children(refs=ua.ObjectIds.HasSubtype):
            nodes.append(child)
            self._get_subtypes(child, nodes)
        return nodes

    def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtype=ua.ObjectIds.BaseEventType, evfilter=None):
        """
        Subscribe to events from a node. Default node is Server node.
        In most servers the server node is the only one you can subscribe to.
        if evfilter is provided, evtype is ignored
        Return a handle which can be used to unsubscribe
        """
        sourcenode = self._get_node(sourcenode)
        if evfilter is None:
            evfilter = self._get_filter_from_event_type(evtype)
        return self._subscribe(sourcenode, ua.AttributeIds.EventNotifier, evfilter)

    def _subscribe(self, nodes, attr, mfilter=None, queuesize=0):
        is_list = True
        if not type(nodes) in (list, tuple):
            is_list = False
            nodes = [nodes]
        mirs = []
        for node in nodes:
            mir = self._make_monitored_item_request(node, attr, mfilter, queuesize)
            mirs.append(mir)

        mids = self.create_monitored_items(mirs)
        if is_list:
            return mids
        if type(mids[0]) == ua.StatusCode:
            mids[0].check()
        return mids[0]

    def _make_monitored_item_request(self, node, attr, mfilter, queuesize):
        rv = ua.ReadValueId()
        rv.NodeId = node.nodeid
        rv.AttributeId = attr
        # rv.IndexRange //We leave it null, then the entire array is returned
        mparams = ua.MonitoringParameters()
        with self._lock:
            self._client_handle += 1
            mparams.ClientHandle = self._client_handle
        mparams.SamplingInterval = self.parameters.RequestedPublishingInterval
        mparams.QueueSize = queuesize
        mparams.DiscardOldest = True
        if mfilter:
            mparams.Filter = mfilter
        mir = ua.MonitoredItemCreateRequest()
        mir.ItemToMonitor = rv
        mir.MonitoringMode = ua.MonitoringMode.Reporting
        mir.RequestedParameters = mparams
        return mir

    def create_monitored_items(self, monitored_items):
        """
        low level method to have full control over subscription parameters
        Client handle must be unique since it will be used as key for internal registration of data
        """
        params = ua.CreateMonitoredItemsParameters()
        params.SubscriptionId = self.subscription_id
        params.ItemsToCreate = monitored_items
        params.TimestampsToReturn = ua.TimestampsToReturn.Neither

        # insert monitoried item into map to avoid notification arrive before result return
        # server_handle is left as None in purpose as we don't get it yet.
        with self._lock:
            for mi in monitored_items:
                data = SubscriptionItemData()
                data.client_handle = mi.RequestedParameters.ClientHandle
                data.node = Node(self.server, mi.ItemToMonitor.NodeId)
                data.attribute = mi.ItemToMonitor.AttributeId
                data.mfilter = mi.RequestedParameters.Filter
                self._monitoreditems_map[mi.RequestedParameters.ClientHandle] = data
        results = self.server.create_monitored_items(params)
        mids = []
        # process result, add server_handle, or remove it if failed
        with self._lock:
            for idx, result in enumerate(results):

                mi = params.ItemsToCreate[idx]
                if not result.StatusCode.is_good():
                    del self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
                    mids.append(result.StatusCode)
                    continue
                data = self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
                data.server_handle = result.MonitoredItemId
                mids.append(result.MonitoredItemId)
        return mids

    def unsubscribe(self, handle):
        """
        unsubscribe to datachange or events using the handle returned while subscribing
        if you delete subscription, you do not need to unsubscribe
        """
        params = ua.DeleteMonitoredItemsParameters()
        params.SubscriptionId = self.subscription_id
        params.MonitoredItemIds = [handle]
        results = self.server.delete_monitored_items(params)
        results[0].check()
        with self._lock:
            for k, v in self._monitoreditems_map.items():
                if v.server_handle == handle:
                    del(self._monitoreditems_map[k])
                    return


def get_event_properties_from_type_node(node):
    properties = []
    curr_node = node

    while True:
        properties.extend(curr_node.get_properties())

        if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
            break

        parents = curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
        if len(parents) != 1:  # Something went wrong
            return None
        curr_node = parents[0]

    return properties


1			"""
2			high level interface to subscriptions
3			"""
4	1		import time
5	1		import logging
6	1		from threading import Lock
7
8	1		from opcua import ua
9	1		from opcua import Node
10
11
12	1		class SubHandler(object):
13			"""
14			Subscription Handler. To receive events from server for a subscription
15			This class is just a sample class. Whatever class having these methods can be used
16			"""
17
18	1		def data_change(self, handle, node, val, attr):
19			"""
20			Deprecated, use datachange_notification
21			"""
22			pass
23
24	1		def datachange_notification(self, node, val, data):
25			"""
26			called for every datachange notfication from server
27			"""
28			pass
29
30	1		def event_notification(self, event):
31			"""
32			called for every event notfication from server
33			"""
34			pass
35
36	1		def status_change_notification(self, status):
37			"""
38			called for every status change notfication from server
39			"""
40			pass
41
42
43	1		class EventResult(object):
44			"""
45			To be sent to clients for every events from server
46			"""
47
48	1		def __init__(self):
49	1		self.server_handle = None
50
51	1		def __str__(self):
52			return "EventResult({})".format([str(k) + ":" + str(v) for k, v in self.__dict__.items()])
53	1		__repr__ = __str__
54
55
56	1		class SubscriptionItemData(object):
57			"""
58			To store usefull data from a monitored item
59			"""
60	1		def __init__(self):
61	1		self.node = None
62	1		self.client_handle = None
63	1		self.server_handle = None
64	1		self.attribute = None
65	1		self.mfilter = None
66
67
68	1		class DataChangeNotif(object):
69			"""
70			To be send to clients for every datachange notification from server
71			"""
72	1		def __init__(self, subscription_data, monitored_item):
73	1		self.monitored_item = monitored_item
74	1		self.subscription_data = subscription_data
75
76	1		def __str__(self):
77			return "DataChangeNotfication({}, {})".format(self.subscription_data, self.monitored_item)
78	1		__repr__ = __str__
79
80
81	1		class Subscription(object):
82			"""
83			Subscription object returned by Server or Client objects.
84			The object represent a subscription to an opc-ua server.
85			This is a high level class, especially subscribe_data_change
86			and subscribe_events methods. If more control is necessary look at
87			code and/or use create_monitored_items method.
88			"""
89
90	1		def __init__(self, server, params, handler):
91	1		self.logger = logging.getLogger(__name__)
92	1		self.server = server
93	1		self._client_handle = 200
94	1		self._handler = handler
95	1		self.parameters = params # move to data class
96	1		self._monitoreditems_map = {}
97	1		self._lock = Lock()
98	1		self.subscription_id = None
99	1		response = self.server.create_subscription(params, self.publish_callback)
100	1		self.subscription_id = response.SubscriptionId # move to data class
101	1		self.server.publish()
102	1		self.server.publish()
103
104	1		def delete(self):
105			"""
106			Delete subscription on server. This is automatically done by Client and Server classes on exit
107			"""
108	1		results = self.server.delete_subscriptions([self.subscription_id])
109	1		results[0].check()
110
111	1		def publish_callback(self, publishresult):
112	1		self.logger.info("Publish callback called with result: %s", publishresult)
113	1		while self.subscription_id is None:
114			time.sleep(0.01)
115
116	1		for notif in publishresult.NotificationMessage.NotificationData:
117	1		if isinstance(notif, ua.DataChangeNotification):
118	1		self._call_datachange(notif)
119	1		elif isinstance(notif, ua.EventNotificationList):
120	1		self._call_event(notif)
121			elif isinstance(notif, ua.StatusChangeNotification):
122			self._call_status(notif)
123			else:
124			self.logger.warning("Notification type not supported yet for notification %s", notif)
125
126	1		ack = ua.SubscriptionAcknowledgement()
127	1		ack.SubscriptionId = self.subscription_id
128	1		ack.SequenceNumber = publishresult.NotificationMessage.SequenceNumber
129	1		self.server.publish([ack])
130
131	1		def _call_datachange(self, datachange):
132	1		for item in datachange.MonitoredItems:
133	1		with self._lock:
134	1		if item.ClientHandle not in self._monitoreditems_map:
135	1		self.logger.warning("Received a notification for unknown handle: %s", item.ClientHandle)
136	1		continue
137	1		data = self._monitoreditems_map[item.ClientHandle]
138	1		if hasattr(self._handler, "datachange_notification"):
139	1		event_data = DataChangeNotif(data, item)
140	1		try:
141	1		self._handler.datachange_notification(data.node, item.Value.Value.Value, event_data)
142			except Exception:
143			self.logger.exception("Exception calling data change handler")
144	1		elif hasattr(self._handler, "data_change"): # deprecated API
145	1		self.logger.warning("data_change method is deprecated, use datachange_notification")
146	1		try:
147	1		self._handler.data_change(data.server_handle, data.node, item.Value.Value.Value, data.attribute)
148			except Exception:
149			self.logger.exception("Exception calling deprecated data change handler")
150			else:
151			self.logger.error("DataChange subscription created but handler has no datachange_notification method")
152
153	1		def _call_event(self, eventlist):
154	1		for event in eventlist.Events:
155	1		with self._lock:
156	1		data = self._monitoreditems_map[event.ClientHandle]
157	1		result = EventResult()
158	1		result.server_handle = data.server_handle
159	1		for idx, sattr in enumerate(data.mfilter.SelectClauses):
160	1		if len(sattr.BrowsePath) == 0:
161			setattr(result, sattr.AttributeId.name, event.EventFields[idx].Value)
162			else:
163	1		setattr(result, sattr.BrowsePath[0].Name, event.EventFields[idx].Value)
164	1		if hasattr(self._handler, "event_notification"):
165	1		try:
166	1		self._handler.event_notification(result)
167			except Exception:
168			self.logger.exception("Exception calling event handler")
169			elif hasattr(self._handler, "event"): # depcrecated API
170			try:
171			self._handler.event(data.server_handle, result)
172			except Exception:
173			self.logger.exception("Exception calling deprecated event handler")
174			else:
175			self.logger.error("Event subscription created but handler has no event_notification method")
176
177	1		def _call_status(self, status):
178			try:
179			self._handler.status_change_notification(status.Status)
180			except Exception:
181			self.logger.exception("Exception calling status change handler")
182
183	1		def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value):
184			"""
185			Subscribe for data change events for a node or list of nodes.
186			default attribute is Value.
187			Return a handle which can be used to unsubscribe
188			If more control is necessary use create_monitored_items method
189			"""
190	1		return self._subscribe(nodes, attr, queuesize=0)
191
192	1		def _get_node(self, nodeid):
193	1		if isinstance(nodeid, ua.NodeId):
194			node = Node(self.server, nodeid)
195	1		elif isinstance(nodeid, Node):
196	1		node = nodeid
197			else:
198	1		node = Node(self.server, ua.NodeId(nodeid))
199	1		return node
200
201	1		def _get_filter_from_event_type(self, eventtype):
202	1		eventtype = self._get_node(eventtype)
203	1		evfilter = ua.EventFilter()
204	1		evfilter.SelectClauses = self._select_clauses_from_evtype(eventtype)
205	1		evfilter.WhereClause = self._where_clause_from_evtype(eventtype)
206	1		return evfilter
207	1
208	1		def _select_clauses_from_evtype(self, evtype):
209	1		clauses = []
210	1		for property in get_event_properties_from_type_node(evtype):
211			op = ua.SimpleAttributeOperand()
212	1		op.TypeDefinitionId = evtype.nodeid
213			op.AttributeId = ua.AttributeIds.Value
214			op.BrowsePath = [property.get_browse_name()]
215			clauses.append(op)
216			return clauses
217
218	1		def _where_clause_from_evtype(self, evtype):
219	1		cf = ua.ContentFilter()
220	1		el = ua.ContentFilterElement()
221			# operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
222	1		op = ua.SimpleAttributeOperand()
223	1		op.AttributeId = ua.AttributeIds.NodeId
224	1		el.FilterOperands.append(op)
225	1		for subtypeid in [st.nodeid for st in self._get_subtypes(evtype)]:
226	1		op = ua.LiteralOperand()
227	1		op.Value = ua.Variant(subtypeid)
228	1		el.FilterOperands.append(op)
229	1		el.FilterOperator = ua.FilterOperator.InList
230	1
231			cf.Elements.append(el)
232	1		return cf
233	1
234	1		def _get_subtypes(self, parent, nodes=None):
235	1		if nodes is None:
236	1		nodes = [parent]
237	1		for child in parent.get_children(refs=ua.ObjectIds.HasSubtype):
238			nodes.append(child)
239	1		self._get_subtypes(child, nodes)
240	1		return nodes
241	1
242	1		def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtype=ua.ObjectIds.BaseEventType, evfilter=None):
243			"""
244	1		Subscribe to events from a node. Default node is Server node.
245	1		In most servers the server node is the only one you can subscribe to.
246	1		if evfilter is provided, evtype is ignored
247	1		Return a handle which can be used to unsubscribe
248	1		"""
249	1		sourcenode = self._get_node(sourcenode)
250	1		if evfilter is None:
251	1		evfilter = self._get_filter_from_event_type(evtype)
252	1		return self._subscribe(sourcenode, ua.AttributeIds.EventNotifier, evfilter)
253	1
254	1		def _subscribe(self, nodes, attr, mfilter=None, queuesize=0):
255	1		is_list = True
256	1		if not type(nodes) in (list, tuple):
257	1		is_list = False
258			nodes = [nodes]
259	1		mirs = []
260			for node in nodes:
261			mir = self._make_monitored_item_request(node, attr, mfilter, queuesize)
262			mirs.append(mir)
263
264	1		mids = self.create_monitored_items(mirs)
265	1		if is_list:
266	1		return mids
267	1		if type(mids[0]) == ua.StatusCode:
268			mids[0].check()
269			return mids[0]
270
271	1		def _make_monitored_item_request(self, node, attr, mfilter, queuesize):
272	1		rv = ua.ReadValueId()
273	1		rv.NodeId = node.nodeid
274	1		rv.AttributeId = attr
275	1		# rv.IndexRange //We leave it null, then the entire array is returned
276	1		mparams = ua.MonitoringParameters()
277	1		with self._lock:
278	1		self._client_handle += 1
279	1		mparams.ClientHandle = self._client_handle
280	1		mparams.SamplingInterval = self.parameters.RequestedPublishingInterval
281			mparams.QueueSize = queuesize
282	1		mparams.DiscardOldest = True
283	1		if mfilter:
284	1		mparams.Filter = mfilter
285	1		mir = ua.MonitoredItemCreateRequest()
286	1		mir.ItemToMonitor = rv
287	1		mir.MonitoringMode = ua.MonitoringMode.Reporting
288	1		mir.RequestedParameters = mparams
289	1		return mir
290	1
291	1		def create_monitored_items(self, monitored_items):
292	1		"""
293			low level method to have full control over subscription parameters
294	1		Client handle must be unique since it will be used as key for internal registration of data
295			"""
296			params = ua.CreateMonitoredItemsParameters()
297			params.SubscriptionId = self.subscription_id
298			params.ItemsToCreate = monitored_items
299	1		params.TimestampsToReturn = ua.TimestampsToReturn.Neither
300	1
301	1		# insert monitoried item into map to avoid notification arrive before result return
302	1		# server_handle is left as None in purpose as we don't get it yet.
303	1		with self._lock:
304	1		for mi in monitored_items:
305	1		data = SubscriptionItemData()
306	1		data.client_handle = mi.RequestedParameters.ClientHandle
307	1		data.node = Node(self.server, mi.ItemToMonitor.NodeId)
308	1		data.attribute = mi.ItemToMonitor.AttributeId
309			data.mfilter = mi.RequestedParameters.Filter
310			self._monitoreditems_map[mi.RequestedParameters.ClientHandle] = data
311	1		results = self.server.create_monitored_items(params)
312	1		mids = []
313	1		# process result, add server_handle, or remove it if failed
314			with self._lock:
315	1	View Code Duplication	for idx, result in enumerate(results):
			0 ignored issues – show Duplication introduced 2016-05-03 15:34 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
316	1		mi = params.ItemsToCreate[idx]
317			if not result.StatusCode.is_good():
318	1		del self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
319	1		mids.append(result.StatusCode)
320			continue
321	1		data = self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
322	1		data.server_handle = result.MonitoredItemId
323			mids.append(result.MonitoredItemId)
324	1		return mids
325
326	1		def unsubscribe(self, handle):
327			"""
328			unsubscribe to datachange or events using the handle returned while subscribing
329			if you delete subscription, you do not need to unsubscribe
330			"""
331			params = ua.DeleteMonitoredItemsParameters()
332			params.SubscriptionId = self.subscription_id
333			params.MonitoredItemIds = [handle]
334			results = self.server.delete_monitored_items(params)
335			results[0].check()
336			with self._lock:
337			for k, v in self._monitoreditems_map.items():
338			if v.server_handle == handle:
339			del(self._monitoreditems_map[k])
340			return
341
342
343			def get_event_properties_from_type_node(node):
344			properties = []
345			curr_node = node
346
347			while True:
348			properties.extend(curr_node.get_properties())
349
350			if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
351			break
352
353			parents = curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
354			if len(parents) != 1: # Something went wrong
355			return None
356			curr_node = parents[0]
357
358			return properties
359

FreeOpcUa / python-opcua

Pull Request — master (#179)

Subscription.__init__() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like

Subscription.init() A