Subscription.publish_callback() - Code Metrics - Inspection of "Event implementation moved to server" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#177)

by Denis

created 2016-05-03 15:31 UTC

Subscription.publish_callback() B

↳ Parent: Subscription

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	6.5625

Metric	Value
cc	6
dl	0
loc	19
ccs	12
cts	16
cp	0.75
crap	6.5625
rs	8

"""
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 property in get_event_properties_from_type_node(eventtype):
            op = ua.SimpleAttributeOperand()
            op.TypeDefinitionId = eventtype.nodeid
            op.AttributeId = ua.AttributeIds.Value
            op.BrowsePath = [property.get_browse_name()]
            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


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		for property in get_event_properties_from_type_node(eventtype):
205	1		op = ua.SimpleAttributeOperand()
206	1		op.TypeDefinitionId = eventtype.nodeid
207	1		op.AttributeId = ua.AttributeIds.Value
208	1		op.BrowsePath = [property.get_browse_name()]
209	1		evfilter.SelectClauses.append(op)
210	1		return evfilter
211
212	1		def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtype=ua.ObjectIds.BaseEventType):
213			"""
214			Subscribe to events from a node. Default node is Server node.
215			In most servers the server node is the only one you can subscribe to.
216			Return a handle which can be used to unsubscribe
217			"""
218	1		sourcenode = self._get_node(sourcenode)
219	1		evfilter = self._get_filter_from_event_type(evtype)
220	1		return self._subscribe(sourcenode, ua.AttributeIds.EventNotifier, evfilter)
221
222	1		def _subscribe(self, nodes, attr, mfilter=None, queuesize=0):
223	1		is_list = True
224	1		if not type(nodes) in (list, tuple):
225	1		is_list = False
226	1		nodes = [nodes]
227	1		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
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	1		return mids[0]
238
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			# rv.IndexRange //We leave it null, then the entire array is returned
244	1		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	1		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
259	1		def create_monitored_items(self, monitored_items):
260			"""
261			low level method to have full control over subscription parameters
262			Client handle must be unique since it will be used as key for internal registration of data
263			"""
264	1		params = ua.CreateMonitoredItemsParameters()
265	1		params.SubscriptionId = self.subscription_id
266	1		params.ItemsToCreate = monitored_items
267	1		params.TimestampsToReturn = ua.TimestampsToReturn.Neither
268
269			# insert monitoried item into map to avoid notification arrive before result return
270			# 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	1		data.client_handle = mi.RequestedParameters.ClientHandle
275	1		data.node = Node(self.server, mi.ItemToMonitor.NodeId)
276	1		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			# 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	1		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
294	1		def unsubscribe(self, handle):
295			"""
296			unsubscribe to datachange or events using the handle returned while subscribing
297			if you delete subscription, you do not need to unsubscribe
298			"""
299	1		params = ua.DeleteMonitoredItemsParameters()
300	1		params.SubscriptionId = self.subscription_id
301	1		params.MonitoredItemIds = [handle]
302	1		results = self.server.delete_monitored_items(params)
303	1		results[0].check()
304	1		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
311	1		def get_event_properties_from_type_node(node):
312	1		properties = []
313	1		curr_node = node
314
315	1	View Code Duplication	while True:
			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		properties.extend(curr_node.get_properties())
317
318	1		if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
319	1		break
320
321	1		parents = curr_node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
322	1		if len(parents) != 1: # Something went wrong
323			return None
324	1		curr_node = parents[0]
325
326			return properties
327

FreeOpcUa / python-opcua

Pull Request — master (#177)

Subscription.publish_callback() B

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like