opcua.server.MonitoredItemService - Code Metrics - Inspection of "Events debugging" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#133)

by Denis

created 2016-03-08 09:52 UTC

opcua.server.MonitoredItemService B

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	169
Duplicated Lines	0 %

Test Coverage

Coverage

80%

Metric	Value
dl	0
loc	169
ccs	112
cts	140
cp	0.8
rs	8.2608
wmc	40

13 Methods

Rating	Name	Size	Complexity
B	_delete_monitored_items()	14	6
A	delete_all_monitored_items()	2	2
C	_create_monitored_item()	43	7
A	trigger_datachange()	4	1
A	_modify_monitored_item()	14	4
A	__init__()	9	1
A	create_monitored_items()	5	2
A	modify_monitored_items()	5	2
A	delete_monitored_items()	7	3
A	trigger_event()	16	4
A	_get_event_fields()	17	4
A	datachange_callback()	13	3
A	trigger_statuschange()	2	1

How to fix Complexity

"""
server side implementation of a subscription object
"""

from threading import RLock
import logging
import copy

from opcua import ua


class MonitoredItemData(object):

    def __init__(self):
        self.client_handle = None
        self.callback_handle = None
        self.monitored_item_id = None
        self.parameters = None
        self.mode = None


class MonitoredItemService(object):

    """
    implement monitoreditem service for 1 subscription
    """

    def __init__(self, isub, aspace):
        self.logger = logging.getLogger(__name__ + str(isub.data.SubscriptionId))
        self.isub = isub
        self.aspace = aspace
        self._lock = RLock()
        self._monitored_items = {}
        self._monitored_events = {}
        self._monitored_datachange = {}
        self._monitored_item_counter = 111

    def delete_all_monitored_items(self):
        self.delete_monitored_items([mdata.monitored_item_id for mdata in self._monitored_items.values()])

    def create_monitored_items(self, params):
        results = []
        for item in params.ItemsToCreate:
            results.append(self._create_monitored_item(item))
        return results

    def modify_monitored_items(self, params):
        results = []
        for item in params.ItemsToModify:
            results.append(self._modify_monitored_item(item))
        return results

    def trigger_datachange(self, handle, nodeid, attr):
        self.logger.debug("triggering datachange for handle %s, nodeid %s, and attribute %s", handle, nodeid, attr)
        variant = self.aspace.get_attribute_value(nodeid, attr)
        self.datachange_callback(handle, variant)

    def _modify_monitored_item(self, params):
        with self._lock:
            for mdata in self._monitored_items.values():
                result = ua.MonitoredItemCreateResult()
                if mdata.monitored_item_id == params.MonitoredItemId:
                    result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
                    result.RevisedQueueSize = params.RequestedParameters.QueueSize
                    result.FilterResult = params.RequestedParameters.Filter
                    mdata.parameters = result
                    return result
            # FIXME modify event subscriptions
            result = ua.MonitoredItemCreateResult()
            result.StatusCode(ua.StatusCodes.BadMonitoredItemIdInvalid)
            return result

    def _create_monitored_item(self, params):
        with self._lock:
            result = ua.MonitoredItemCreateResult()
            result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
            result.RevisedQueueSize = params.RequestedParameters.QueueSize
            self._monitored_item_counter += 1
            result.MonitoredItemId = self._monitored_item_counter
            self.logger.debug("Creating MonitoredItem with id %s", result.MonitoredItemId)

            mdata = MonitoredItemData()
            mdata.parameters = result
            mdata.mode = params.MonitoringMode
            mdata.client_handle = params.RequestedParameters.ClientHandle
            mdata.mfilter = params.RequestedParameters.Filter
            mdata.monitored_item_id = result.MonitoredItemId

            self._monitored_items[result.MonitoredItemId] = mdata

            if params.ItemToMonitor.AttributeId == ua.AttributeIds.EventNotifier:
                self.logger.info("request to subscribe to events for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)
                if self.aspace.get_attribute_value(params.ItemToMonitor.NodeId, ua.AttributeIds.EventNotifier).Value.Value != 1:
                    result.StatusCode = ua.StatusCode(ua.StatusCodes.BadServiceUnsupported)
                result.FilterResult = ua.EventFilterResult()
                for _ in params.RequestedParameters.Filter.SelectClauses:
                    result.FilterResult.SelectClauseResults.append(ua.StatusCode())
                # FIXME: where clause result
                self._monitored_events[params.ItemToMonitor.NodeId] = result.MonitoredItemId
            else:
                self.logger.info("request to subscribe to datachange for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)
                result.FilterResult = params.RequestedParameters.Filter
                result.StatusCode, handle = self.aspace.add_datachange_callback(params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId, self.datachange_callback)
                self.logger.debug("adding callback return status %s and handle %s", result.StatusCode, handle)
                mdata.callback_handle = handle
                self._monitored_datachange[handle] = result.MonitoredItemId
                if result.StatusCode.is_good():
                    # force data change event generation
                    self.trigger_datachange(handle, params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)

            if not result.StatusCode.is_good():
                del(self._monitored_items[result.MonitoredItemId])
                self._monitored_item_counter -= 1

            return result

    def delete_monitored_items(self, ids):
        self.logger.debug("delete monitored items %s", ids)
        with self._lock:
            results = []
            for mid in ids:
                results.append(self._delete_monitored_items(mid))
            return results

    def _delete_monitored_items(self, mid):
        if not mid in self._monitored_items:
            return ua.StatusCode(ua.StatusCodes.BadMonitoredItemIdInvalid)
        for k, v in self._monitored_events.items():
            if v == mid:
                self._monitored_events.pop(k)
                break
        for k, v in self._monitored_datachange.items():
            if v == mid:
                self.aspace.delete_datachange_callback(k)
                self._monitored_datachange.pop(k)
                break
        self._monitored_items.pop(mid)
        return ua.StatusCode()

    def datachange_callback(self, handle, value, error=None):
        if error:
            self.logger.info("subscription %s: datachange callback called with handle '%s' and erorr '%s'", self, handle, error)
            self.trigger_statuschange(error)
        else:
            self.logger.info("subscription %s: datachange callback called with handle '%s' and value '%s'", self, handle, value.Value)
            event = ua.MonitoredItemNotification()
            with self._lock:
                mid = self._monitored_datachange[handle]
                mdata = self._monitored_items[mid]
                event.ClientHandle = mdata.client_handle
                event.Value = value
                self.isub.enqueue_datachange_event(mid, event, mdata.parameters.RevisedQueueSize)

    def trigger_event(self, event):
        with self._lock:
            if not event.SourceNode in self._monitored_events:
                self.logger.debug("%s has no subscription for events %s from node: %s", self, event, event.SourceNode)
                return False
            self.logger.debug("%s has subscription for events %s from node: %s", self, event, event.SourceNode)
            mid = self._monitored_events[event.SourceNode]
            if not mid in self._monitored_items:
                self.logger.debug("Could not find monitored items for id %s for event %s in subscription %s", mid, event, self)
                return False
            mdata = self._monitored_items[mid]
            fieldlist = ua.EventFieldList()
            fieldlist.ClientHandle = mdata.client_handle
            fieldlist.EventFields = self._get_event_fields(mdata.mfilter, event)
            self.isub.enqueue_event(mid, fieldlist, mdata.parameters.RevisedQueueSize)
            return True

    def _get_event_fields(self, evfilter, event):
        fields = []
        for sattr in evfilter.SelectClauses:
            try:
                if not sattr.BrowsePath:
                    #val = getattr(event, ua.AttributeIdsInv[sattr.Attribute])
                    val = getattr(event, sattr.Attribute.name)
                    val = copy.deepcopy(val)
                    fields.append(ua.Variant(val))
                else:
                    name = sattr.BrowsePath[0].Name
                    val = getattr(event, name)
                    val = copy.deepcopy(val)
                    fields.append(ua.Variant(val))
            except AttributeError:
                fields.append(ua.Variant())
        return fields

    def trigger_statuschange(self, code):
        self.isub.enqueue_statuschange(code)


class InternalSubscription(object):

    def __init__(self, subservice, data, addressspace, callback):
        self.logger = logging.getLogger(__name__ + str(data.SubscriptionId))
        self.aspace = addressspace
        self.subservice = subservice
        self.data = data
        self.callback = callback
        self.monitored_item_srv = MonitoredItemService(self, addressspace)
        self.task = None
        self._lock = RLock()
        self._triggered_datachanges = {}
        self._triggered_events = {}
        self._triggered_statuschanges = []
        self._notification_seq = 1
        self._not_acknowledged_results = {}
        self._startup = True
        self._keep_alive_count = 0
        self._publish_cycles_count = 0
        self._stopev = False

    def __str__(self):
        return "Subscription(id:{})".format(self.data.SubscriptionId)

    def start(self):
        self.logger.debug("starting subscription %s", self.data.SubscriptionId)
        self._subscription_loop()

    def stop(self):
        self.logger.debug("stopping subscription %s", self.data.SubscriptionId)
        self._stopev = True
        self.monitored_item_srv.delete_all_monitored_items()

    def _subscription_loop(self):
        #self.logger.debug("%s loop", self)
        if not self._stopev:
            self.subservice.loop.call_later(self.data.RevisedPublishingInterval / 1000.0, self._sub_loop)

    def _sub_loop(self):
        if self._stopev:
            return
        self.publish_results()
        self._subscription_loop()

    def has_published_results(self):
        with self._lock:
            if self._startup or self._triggered_datachanges or self._triggered_events:
                return True
            if self._keep_alive_count > self.data.RevisedMaxKeepAliveCount:
                self.logger.debug("keep alive count %s is > than max keep alive count %s, sending publish event", self._keep_alive_count, self.data.RevisedMaxKeepAliveCount)
                return True
            self._keep_alive_count += 1
            return False

    def publish_results(self):
        if self._publish_cycles_count > self.data.RevisedLifetimeCount:
            self.logger.warning("Subscription %s has expired, publish cycle count(%s) > lifetime count (%s)", self, self._publish_cycles_count, self.data.RevisedLifetimeCount)
            # FIXME this will never be send since we do not have publish request anyway
            self.monitored_item_srv.trigger_statuschange(ua.StatusCode(ua.StatusCodes.BadTimeout))
            self._stopev = True
        with self._lock:
            if self.has_published_results():  # FIXME: should we pop a publish request here? or we do not care?
                self._publish_cycles_count += 1
                result = self._pop_publish_result()
                self.callback(result)

    def _pop_publish_result(self):
        result = ua.PublishResult()
        result.SubscriptionId = self.data.SubscriptionId
        self._pop_triggered_datachanges(result)
        self._pop_triggered_events(result)
        self._pop_triggered_statuschanges(result)
        self._keep_alive_count = 0
        self._startup = False
        result.NotificationMessage.SequenceNumber = self._notification_seq
        if len(result.NotificationMessage.NotificationData) != 0:
            self._notification_seq += 1
            self._not_acknowledged_results[result.NotificationMessage.SequenceNumber] = result
        result.MoreNotifications = False
        result.AvailableSequenceNumbers = list(self._not_acknowledged_results.keys())
        return result

    def _pop_triggered_datachanges(self, result):
        if self._triggered_datachanges:
            notif = ua.DataChangeNotification()
            notif.MonitoredItems = [item for sublist in self._triggered_datachanges.values() for item in sublist]
            self._triggered_datachanges = {}
            self.logger.debug("sending datachanges notification with %s events", len(notif.MonitoredItems))
            result.NotificationMessage.NotificationData.append(notif)

    def _pop_triggered_events(self, result):
        if self._triggered_events:
            notif = ua.EventNotificationList()
            notif.Events = [item for sublist in self._triggered_events.values() for item in sublist]
            self._triggered_events = {}
            result.NotificationMessage.NotificationData.append(notif)
            self.logger.debug("sending event notification with %s events", len(notif.Events))

    def _pop_triggered_statuschanges(self, result):
        if self._triggered_statuschanges:
            notif = ua.StatusChangeNotification()
            notif.Status = self._triggered_statuschanges.pop(0)
            result.NotificationMessage.NotificationData.append(notif)
            self.logger.debug("sending event notification %s", notif.Status)

    def publish(self, nb):
        with self._lock:
            self._publish_cycles_count = 0
            if nb in self._not_acknowledged_results:
                self._not_acknowledged_results.pop(nb)

    def republish(self, nb):
        self.logger.info("re-publish request for ack %s in subscription %s", nb, self)
        with self._lock:
            if nb in self._not_acknowledged_results:
                self.logger.info("re-publishing ack %s in subscription %s", nb, self)
                return self._not_acknowledged_results[nb].NotificationMessage
            else:
                self.logger.info("Error request to re-published non existing ack %s in subscription %s", nb, self)
                return ua.NotificationMessage()

    def enqueue_datachange_event(self, mid, eventdata, maxsize):
        self._enqueue_event(mid, eventdata, maxsize, self._triggered_datachanges)

    def enqueue_event(self, mid, eventdata, maxsize):
        self._enqueue_event(mid, eventdata, maxsize, self._triggered_events)

    def enqueue_statuschange(self, code):
        self._triggered_statuschanges.append(code)

    def _enqueue_event(self, mid, eventdata, size, queue):
        if mid not in queue:
            queue[mid] = [eventdata]
            return
        if size != 0:
            if len(queue[mid]) >= size:
                queue[mid].pop(0)
        queue[mid].append(eventdata)


1		"""
2		server side implementation of a subscription object
3		"""
4
5	1	from threading import RLock
6	1	import logging
7	1	import copy
8
9	1	from opcua import ua
10
11
12	1	class MonitoredItemData(object):
13
14	1	def __init__(self):
15	1	self.client_handle = None
16	1	self.callback_handle = None
17	1	self.monitored_item_id = None
18	1	self.parameters = None
19	1	self.mode = None
20
21
22	1	class MonitoredItemService(object):
23
24		"""
25		implement monitoreditem service for 1 subscription
26		"""
27
28	1	def __init__(self, isub, aspace):
29	1	self.logger = logging.getLogger(__name__ + str(isub.data.SubscriptionId))
30	1	self.isub = isub
31	1	self.aspace = aspace
32	1	self._lock = RLock()
33	1	self._monitored_items = {}
34	1	self._monitored_events = {}
35	1	self._monitored_datachange = {}
36	1	self._monitored_item_counter = 111
37
38	1	def delete_all_monitored_items(self):
39	1	self.delete_monitored_items([mdata.monitored_item_id for mdata in self._monitored_items.values()])
40
41	1	def create_monitored_items(self, params):
42	1	results = []
43	1	for item in params.ItemsToCreate:
44	1	results.append(self._create_monitored_item(item))
45	1	return results
46
47	1	def modify_monitored_items(self, params):
48		results = []
49		for item in params.ItemsToModify:
50		results.append(self._modify_monitored_item(item))
51		return results
52
53	1	def trigger_datachange(self, handle, nodeid, attr):
54	1	self.logger.debug("triggering datachange for handle %s, nodeid %s, and attribute %s", handle, nodeid, attr)
55	1	variant = self.aspace.get_attribute_value(nodeid, attr)
56	1	self.datachange_callback(handle, variant)
57
58	1	def _modify_monitored_item(self, params):
59		with self._lock:
60		for mdata in self._monitored_items.values():
61		result = ua.MonitoredItemCreateResult()
62		if mdata.monitored_item_id == params.MonitoredItemId:
63		result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
64		result.RevisedQueueSize = params.RequestedParameters.QueueSize
65		result.FilterResult = params.RequestedParameters.Filter
66		mdata.parameters = result
67		return result
68		# FIXME modify event subscriptions
69		result = ua.MonitoredItemCreateResult()
70		result.StatusCode(ua.StatusCodes.BadMonitoredItemIdInvalid)
71		return result
72
73	1	def _create_monitored_item(self, params):
74	1	with self._lock:
75	1	result = ua.MonitoredItemCreateResult()
76	1	result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
77	1	result.RevisedQueueSize = params.RequestedParameters.QueueSize
78	1	self._monitored_item_counter += 1
79	1	result.MonitoredItemId = self._monitored_item_counter
80	1	self.logger.debug("Creating MonitoredItem with id %s", result.MonitoredItemId)
81
82	1	mdata = MonitoredItemData()
83	1	mdata.parameters = result
84	1	mdata.mode = params.MonitoringMode
85	1	mdata.client_handle = params.RequestedParameters.ClientHandle
86	1	mdata.mfilter = params.RequestedParameters.Filter
87	1	mdata.monitored_item_id = result.MonitoredItemId
88
89	1	self._monitored_items[result.MonitoredItemId] = mdata
90
91	1	if params.ItemToMonitor.AttributeId == ua.AttributeIds.EventNotifier:
92	1	self.logger.info("request to subscribe to events for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)
93	1	if self.aspace.get_attribute_value(params.ItemToMonitor.NodeId, ua.AttributeIds.EventNotifier).Value.Value != 1:
94	1	result.StatusCode = ua.StatusCode(ua.StatusCodes.BadServiceUnsupported)
95	1	result.FilterResult = ua.EventFilterResult()
96	1	for _ in params.RequestedParameters.Filter.SelectClauses:
97	1	result.FilterResult.SelectClauseResults.append(ua.StatusCode())
98		# FIXME: where clause result
99	1	self._monitored_events[params.ItemToMonitor.NodeId] = result.MonitoredItemId
100		else:
101	1	self.logger.info("request to subscribe to datachange for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)
102	1	result.FilterResult = params.RequestedParameters.Filter
103	1	result.StatusCode, handle = self.aspace.add_datachange_callback(params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId, self.datachange_callback)
104	1	self.logger.debug("adding callback return status %s and handle %s", result.StatusCode, handle)
105	1	mdata.callback_handle = handle
106	1	self._monitored_datachange[handle] = result.MonitoredItemId
107	1	if result.StatusCode.is_good():
108		# force data change event generation
109	1	self.trigger_datachange(handle, params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)
110
111	1	if not result.StatusCode.is_good():
112	1	del(self._monitored_items[result.MonitoredItemId])
113	1	self._monitored_item_counter -= 1
114
115	1	return result
116
117	1	def delete_monitored_items(self, ids):
118	1	self.logger.debug("delete monitored items %s", ids)
119	1	with self._lock:
120	1	results = []
121	1	for mid in ids:
122	1	results.append(self._delete_monitored_items(mid))
123	1	return results
124
125	1	def _delete_monitored_items(self, mid):
126	1	if not mid in self._monitored_items:
127	1	return ua.StatusCode(ua.StatusCodes.BadMonitoredItemIdInvalid)
128	1	for k, v in self._monitored_events.items():
129	1	if v == mid:
130	1	self._monitored_events.pop(k)
131	1	break
132	1	for k, v in self._monitored_datachange.items():
133	1	if v == mid:
134	1	self.aspace.delete_datachange_callback(k)
135	1	self._monitored_datachange.pop(k)
136	1	break
137	1	self._monitored_items.pop(mid)
138	1	return ua.StatusCode()
139
140	1	def datachange_callback(self, handle, value, error=None):
141	1	if error:
142		self.logger.info("subscription %s: datachange callback called with handle '%s' and erorr '%s'", self, handle, error)
143		self.trigger_statuschange(error)
144		else:
145	1	self.logger.info("subscription %s: datachange callback called with handle '%s' and value '%s'", self, handle, value.Value)
146	1	event = ua.MonitoredItemNotification()
147	1	with self._lock:
148	1	mid = self._monitored_datachange[handle]
149	1	mdata = self._monitored_items[mid]
150	1	event.ClientHandle = mdata.client_handle
151	1	event.Value = value
152	1	self.isub.enqueue_datachange_event(mid, event, mdata.parameters.RevisedQueueSize)
153
154	1	def trigger_event(self, event):
155	1	with self._lock:
156	1	if not event.SourceNode in self._monitored_events:
157		self.logger.debug("%s has no subscription for events %s from node: %s", self, event, event.SourceNode)
158		return False
159	1	self.logger.debug("%s has subscription for events %s from node: %s", self, event, event.SourceNode)
160	1	mid = self._monitored_events[event.SourceNode]
161	1	if not mid in self._monitored_items:
162		self.logger.debug("Could not find monitored items for id %s for event %s in subscription %s", mid, event, self)
163		return False
164	1	mdata = self._monitored_items[mid]
165	1	fieldlist = ua.EventFieldList()
166	1	fieldlist.ClientHandle = mdata.client_handle
167	1	fieldlist.EventFields = self._get_event_fields(mdata.mfilter, event)
168	1	self.isub.enqueue_event(mid, fieldlist, mdata.parameters.RevisedQueueSize)
169	1	return True
170
171	1	def _get_event_fields(self, evfilter, event):
172	1	fields = []
173	1	for sattr in evfilter.SelectClauses:
174	1	try:
175	1	if not sattr.BrowsePath:
176		#val = getattr(event, ua.AttributeIdsInv[sattr.Attribute])
177		val = getattr(event, sattr.Attribute.name)
178		val = copy.deepcopy(val)
179		fields.append(ua.Variant(val))
180		else:
181	1	name = sattr.BrowsePath[0].Name
182	1	val = getattr(event, name)
183	1	val = copy.deepcopy(val)
184	1	fields.append(ua.Variant(val))
185		except AttributeError:
186		fields.append(ua.Variant())
187	1	return fields
188
189	1	def trigger_statuschange(self, code):
190		self.isub.enqueue_statuschange(code)
191
192
193	1	class InternalSubscription(object):
194
195	1	def __init__(self, subservice, data, addressspace, callback):
196	1	self.logger = logging.getLogger(__name__ + str(data.SubscriptionId))
197	1	self.aspace = addressspace
198	1	self.subservice = subservice
199	1	self.data = data
200	1	self.callback = callback
201	1	self.monitored_item_srv = MonitoredItemService(self, addressspace)
202	1	self.task = None
203	1	self._lock = RLock()
204	1	self._triggered_datachanges = {}
205	1	self._triggered_events = {}
206	1	self._triggered_statuschanges = []
207	1	self._notification_seq = 1
208	1	self._not_acknowledged_results = {}
209	1	self._startup = True
210	1	self._keep_alive_count = 0
211	1	self._publish_cycles_count = 0
212	1	self._stopev = False
213
214	1	def __str__(self):
215		return "Subscription(id:{})".format(self.data.SubscriptionId)
216
217	1	def start(self):
218	1	self.logger.debug("starting subscription %s", self.data.SubscriptionId)
219	1	self._subscription_loop()
220
221	1	def stop(self):
222	1	self.logger.debug("stopping subscription %s", self.data.SubscriptionId)
223	1	self._stopev = True
224	1	self.monitored_item_srv.delete_all_monitored_items()
225
226	1	def _subscription_loop(self):
227		#self.logger.debug("%s loop", self)
228	1	if not self._stopev:
229	1	self.subservice.loop.call_later(self.data.RevisedPublishingInterval / 1000.0, self._sub_loop)
230
231	1	def _sub_loop(self):
232	1	if self._stopev:
233	1	return
234	1	self.publish_results()
235	1	self._subscription_loop()
236
237	1	def has_published_results(self):
238	1	with self._lock:
239	1	if self._startup or self._triggered_datachanges or self._triggered_events:
240	1	return True
241	1	if self._keep_alive_count > self.data.RevisedMaxKeepAliveCount:
242		self.logger.debug("keep alive count %s is > than max keep alive count %s, sending publish event", self._keep_alive_count, self.data.RevisedMaxKeepAliveCount)
243		return True
244	1	self._keep_alive_count += 1
245	1	return False
246
247	1	def publish_results(self):
248	1	if self._publish_cycles_count > self.data.RevisedLifetimeCount:
249		self.logger.warning("Subscription %s has expired, publish cycle count(%s) > lifetime count (%s)", self, self._publish_cycles_count, self.data.RevisedLifetimeCount)
250		# FIXME this will never be send since we do not have publish request anyway
251		self.monitored_item_srv.trigger_statuschange(ua.StatusCode(ua.StatusCodes.BadTimeout))
252		self._stopev = True
253	1	with self._lock:
254	1	if self.has_published_results(): # FIXME: should we pop a publish request here? or we do not care?
255	1	self._publish_cycles_count += 1
256	1	result = self._pop_publish_result()
257	1	self.callback(result)
258
259	1	def _pop_publish_result(self):
260	1	result = ua.PublishResult()
261	1	result.SubscriptionId = self.data.SubscriptionId
262	1	self._pop_triggered_datachanges(result)
263	1	self._pop_triggered_events(result)
264	1	self._pop_triggered_statuschanges(result)
265	1	self._keep_alive_count = 0
266	1	self._startup = False
267	1	result.NotificationMessage.SequenceNumber = self._notification_seq
268	1	if len(result.NotificationMessage.NotificationData) != 0:
269	1	self._notification_seq += 1
270	1	self._not_acknowledged_results[result.NotificationMessage.SequenceNumber] = result
271	1	result.MoreNotifications = False
272	1	result.AvailableSequenceNumbers = list(self._not_acknowledged_results.keys())
273	1	return result
274
275	1	def _pop_triggered_datachanges(self, result):
276	1	if self._triggered_datachanges:
277	1	notif = ua.DataChangeNotification()
278	1	notif.MonitoredItems = [item for sublist in self._triggered_datachanges.values() for item in sublist]
279	1	self._triggered_datachanges = {}
280	1	self.logger.debug("sending datachanges notification with %s events", len(notif.MonitoredItems))
281	1	result.NotificationMessage.NotificationData.append(notif)
282
283	1	def _pop_triggered_events(self, result):
284	1	if self._triggered_events:
285	1	notif = ua.EventNotificationList()
286	1	notif.Events = [item for sublist in self._triggered_events.values() for item in sublist]
287	1	self._triggered_events = {}
288	1	result.NotificationMessage.NotificationData.append(notif)
289	1	self.logger.debug("sending event notification with %s events", len(notif.Events))
290
291	1	def _pop_triggered_statuschanges(self, result):
292	1	if self._triggered_statuschanges:
293		notif = ua.StatusChangeNotification()
294		notif.Status = self._triggered_statuschanges.pop(0)
295		result.NotificationMessage.NotificationData.append(notif)
296		self.logger.debug("sending event notification %s", notif.Status)
297
298	1	def publish(self, nb):
299	1	with self._lock:
300	1	self._publish_cycles_count = 0
301	1	if nb in self._not_acknowledged_results:
302	1	self._not_acknowledged_results.pop(nb)
303
304	1	def republish(self, nb):
305		self.logger.info("re-publish request for ack %s in subscription %s", nb, self)
306		with self._lock:
307		if nb in self._not_acknowledged_results:
308		self.logger.info("re-publishing ack %s in subscription %s", nb, self)
309		return self._not_acknowledged_results[nb].NotificationMessage
310		else:
311		self.logger.info("Error request to re-published non existing ack %s in subscription %s", nb, self)
312		return ua.NotificationMessage()
313
314	1	def enqueue_datachange_event(self, mid, eventdata, maxsize):
315	1	self._enqueue_event(mid, eventdata, maxsize, self._triggered_datachanges)
316
317	1	def enqueue_event(self, mid, eventdata, maxsize):
318	1	self._enqueue_event(mid, eventdata, maxsize, self._triggered_events)
319
320	1	def enqueue_statuschange(self, code):
321		self._triggered_statuschanges.append(code)
322
323	1	def _enqueue_event(self, mid, eventdata, size, queue):
324	1	if mid not in queue:
325	1	queue[mid] = [eventdata]
326	1	return
327	1	if size != 0:
328	1	if len(queue[mid]) >= size:
329	1	queue[mid].pop(0)
330		queue[mid].append(eventdata)
331

FreeOpcUa / python-opcua

Pull Request — master (#133)

opcua.server.MonitoredItemService B

Complexity

Size/Duplication

Test Coverage

13 Methods

How to fix Complexity

Complex Class

Duplication Side-by-Side

Filter issues like