Subscription.unsubscribe() - 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-17 07:16 UTC

Subscription.unsubscribe() A

↳ Parent: Subscription

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	4

Metric	Value
cc	4
dl	0
loc	15
ccs	1
cts	1
cp	1
crap	4
rs	9.2

"""
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 notification from server
        """
        pass

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

    def status_change_notification(self, status):
        """
        called for every status change notification 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__

    def get_event_props_as_fields_dict(self):
        """
        convert all properties of the EventResult class to a dict of variants
        """
        field_vars = {}
        for key, value in vars(self).items():
            if not key.startswith("__") and key is not "server_handle":
                field_vars[key] = ua.Variant(value)
        return field_vars


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

FreeOpcUa / python-opcua

Pull Request — master (#179)

Subscription.unsubscribe() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like