SubHandler.event() - Code Metrics - Inspection of "Events implementation proposal" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#161)

by Denis

created 2016-04-26 12:52 UTC

SubHandler.event() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Metric	Value
cc	1
dl	0
loc	5
rs	9.4285
ccs	1
cts	2
cp	0.5
crap	1.125

"""
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()
        for desc in eventtype.get_children_descriptions(refs=ua.ObjectIds.HasProperty, nodeclassmask=ua.NodeClass.Variable):
            op = ua.SimpleAttributeOperand()
            op.TypeDefinitionId = eventtype.nodeid
            op.AttributeId = ua.AttributeIds.Value
            op.BrowsePath = [desc.BrowseName]
            evfilter.SelectClauses.append(op)
        return evfilter

    def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtype=ua.ObjectIds.BaseEventType):
        """
        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.
        Return a handle which can be used to unsubscribe
        """
        sourcenode = self._get_node(sourcenode)
        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



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	1
43		class EventResult(object):
44		"""
45		To be sent to clients for every events from server
46		"""
47
48		def __init__(self):
49	1	self.server_handle = None
50
51		def __str__(self):
52		return "EventResult({})".format([str(k) + ":" + str(v) for k, v in self.__dict__.items()])
53		__repr__ = __str__
54	1
55	1
56		class SubscriptionItemData(object):
57	1	"""
58		To store usefull data from a monitored item
59	1	"""
60		def __init__(self):
61		self.node = None
62	1	self.client_handle = None
63		self.server_handle = None
64		self.attribute = None
65		self.mfilter = None
66	1
67	1
68	1	class DataChangeNotif(object):
69	1	"""
70	1	To be send to clients for every datachange notification from server
71	1	"""
72		def __init__(self, subscription_data, monitored_item):
73		self.monitored_item = monitored_item
74	1	self.subscription_data = subscription_data
75
76		def __str__(self):
77		return "DataChangeNotfication({}, {})".format(self.subscription_data, self.monitored_item)
78	1	__repr__ = __str__
79	1
80	1
81		class Subscription(object):
82	1	"""
83		Subscription object returned by Server or Client objects.
84	1	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	1	code and/or use create_monitored_items method.
88		"""
89
90		def __init__(self, server, params, handler):
91		self.logger = logging.getLogger(__name__)
92		self.server = server
93		self._client_handle = 200
94		self._handler = handler
95		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	1
104	1	def delete(self):
105	1	"""
106	1	Delete subscription on server. This is automatically done by Client and Server classes on exit
107	1	"""
108	1	results = self.server.delete_subscriptions([self.subscription_id])
109		results[0].check()
110	1
111		def publish_callback(self, publishresult):
112		self.logger.info("Publish callback called with result: %s", publishresult)
113		while self.subscription_id is None:
114	1	time.sleep(0.01)
115	1
116		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	1	self._call_status(notif)
123	1	else:
124	1	self.logger.warning("Notification type not supported yet for notification %s", notif)
125	1
126	1	ack = ua.SubscriptionAcknowledgement()
127		ack.SubscriptionId = self.subscription_id
128		ack.SequenceNumber = publishresult.NotificationMessage.SequenceNumber
129		self.server.publish([ack])
130
131		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		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	1	except Exception:
143	1	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	1	else:
151	1	self.logger.error("DataChange subscription created but handler has no datachange_notification method")
152	1
153	1	def _call_event(self, eventlist):
154		for event in eventlist.Events:
155		with self._lock:
156		data = self._monitoreditems_map[event.ClientHandle]
157		result = EventResult()
158		result.server_handle = data.server_handle
159	1	for idx, sattr in enumerate(data.mfilter.SelectClauses):
160	1	if len(sattr.BrowsePath) == 0:
161	1	setattr(result, sattr.AttributeId.name, event.EventFields[idx].Value)
162	1	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	1	elif hasattr(self._handler, "event"): # depcrecated API
170	1	try:
171	1	self._handler.event(data.server_handle, result)
172	1	except Exception:
173		self.logger.exception("Exception calling deprecated event handler")
174		else:
175	1	self.logger.error("Event subscription created but handler has no event_notification method")
176	1
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	1	"""
190		return self._subscribe(nodes, attr, queuesize=0)
191
192		def _get_node(self, nodeid):
193		if isinstance(nodeid, ua.NodeId):
194		node = Node(self.server, nodeid)
195		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		evfilter = ua.EventFilter()
204	1	for desc in eventtype.get_children_descriptions(refs=ua.ObjectIds.HasProperty, nodeclassmask=ua.NodeClass.Variable):
205	1	op = ua.SimpleAttributeOperand()
206		op.TypeDefinitionId = eventtype.nodeid
207	1	op.AttributeId = ua.AttributeIds.Value
208	1	op.BrowsePath = [desc.BrowseName]
209	1	evfilter.SelectClauses.append(op)
210	1	return evfilter
211	1
212	1	def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtype=ua.ObjectIds.BaseEventType):
213	1	"""
214	1	Subscribe to events from a node. Default node is Server node.
215	1	In most servers the server node is the only one you can subscribe to.
216	1	Return a handle which can be used to unsubscribe
217		"""
218	1	sourcenode = self._get_node(sourcenode)
219		evfilter = self._get_filter_from_event_type(evtype)
220		return self._subscribe(sourcenode, ua.AttributeIds.EventNotifier, evfilter)
221
222		def _subscribe(self, nodes, attr, mfilter=None, queuesize=0):
223		is_list = True
224	1	if not type(nodes) in (list, tuple):
225	1	is_list = False
226	1	nodes = [nodes]
227		mirs = []
228	1	for node in nodes:
229	1	mir = self._make_monitored_item_request(node, attr, mfilter, queuesize)
230	1	mirs.append(mir)
231	1
232	1	mids = self.create_monitored_items(mirs)
233	1	if is_list:
234	1	return mids
235	1	if type(mids[0]) == ua.StatusCode:
236	1	mids[0].check()
237		return mids[0]
238	1
239	1	def _make_monitored_item_request(self, node, attr, mfilter, queuesize):
240	1	rv = ua.ReadValueId()
241	1	rv.NodeId = node.nodeid
242	1	rv.AttributeId = attr
243	1	# rv.IndexRange //We leave it null, then the entire array is returned
244		mparams = ua.MonitoringParameters()
245	1	with self._lock:
246	1	self._client_handle += 1
247	1	mparams.ClientHandle = self._client_handle
248	1	mparams.SamplingInterval = self.parameters.RequestedPublishingInterval
249		mparams.QueueSize = queuesize
250	1	mparams.DiscardOldest = True
251	1	if mfilter:
252	1	mparams.Filter = mfilter
253	1	mir = ua.MonitoredItemCreateRequest()
254	1	mir.ItemToMonitor = rv
255	1	mir.MonitoringMode = ua.MonitoringMode.Reporting
256	1	mir.RequestedParameters = mparams
257	1	return mir
258	1
259	1	def create_monitored_items(self, monitored_items):
260	1	"""
261	1	low level method to have full control over subscription parameters
262	1	Client handle must be unique since it will be used as key for internal registration of data
263	1	"""
264		params = ua.CreateMonitoredItemsParameters()
265	1	params.SubscriptionId = self.subscription_id
266		params.ItemsToCreate = monitored_items
267		params.TimestampsToReturn = ua.TimestampsToReturn.Neither
268
269		# insert monitoried item into map to avoid notification arrive before result return
270	1	# server_handle is left as None in purpose as we don't get it yet.
271	1	with self._lock:
272	1	for mi in monitored_items:
273	1	data = SubscriptionItemData()
274		data.client_handle = mi.RequestedParameters.ClientHandle
275		data.node = Node(self.server, mi.ItemToMonitor.NodeId)
276		data.attribute = mi.ItemToMonitor.AttributeId
277	1	data.mfilter = mi.RequestedParameters.Filter
278	1	self._monitoreditems_map[mi.RequestedParameters.ClientHandle] = data
279	1	results = self.server.create_monitored_items(params)
280	1	mids = []
281	1	# process result, add server_handle, or remove it if failed
282	1	with self._lock:
283	1	for idx, result in enumerate(results):
284	1	mi = params.ItemsToCreate[idx]
285	1	if not result.StatusCode.is_good():
286	1	del self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
287		mids.append(result.StatusCode)
288	1	continue
289	1	data = self._monitoreditems_map[mi.RequestedParameters.ClientHandle]
290	1	data.server_handle = result.MonitoredItemId
291	1	mids.append(result.MonitoredItemId)
292	1	return mids
293	1
294	1	def unsubscribe(self, handle):
295	1	"""
296	1	unsubscribe to datachange or events using the handle returned while subscribing
297	1	if you delete subscription, you do not need to unsubscribe
298	1	"""
299		params = ua.DeleteMonitoredItemsParameters()
300	1	params.SubscriptionId = self.subscription_id
301		params.MonitoredItemIds = [handle]
302		results = self.server.delete_monitored_items(params)
303		results[0].check()
304		with self._lock:
305	1	for k, v in self._monitoreditems_map.items():
306	1	if v.server_handle == handle:
307	1	del(self._monitoreditems_map[k])
308	1	return
309
310

FreeOpcUa / python-opcua

Pull Request — master (#161)

SubHandler.event() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like