MonitoredItemService._make_monitored_item_common() - Code Metrics - Inspection of "split create_monitored_item method as an attemp tp..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 5ef587...a61443 )

by Olivier

created 2016-05-16 19:28 UTC

MonitoredItemService._make_monitored_item_common() A

↳ Parent: MonitoredItemService

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	1

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
dl	0
loc	16
ccs	13
cts	13
cp	1
crap	1
rs	9.4285
c	1
b	0
f	0

"""
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
        self.mfilter = 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:
            with self._lock:
                if item.ItemToMonitor.AttributeId == ua.AttributeIds.EventNotifier:
                    result = self._create_events_monitored_item(item)
                else:
                    result = self._create_data_change_monitored_item(item)
            results.append(result)
        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.MonitoredItemModifyResult()
                if mdata.monitored_item_id == params.MonitoredItemId:
                    self.isub.data.RevisedPublishingInterval = params.RequestedParameters.SamplingInterval
                    result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
                    result.RevisedQueueSize = params.RequestedParameters.QueueSize
                    result.FilterResult = params.RequestedParameters.Filter
                    mdata.parameters = result
                    return result
            result = ua.MonitoredItemModifyResult()
            result.StatusCode(ua.StatusCodes.BadMonitoredItemIdInvalid)
            return result

    def _commit_monitored_item(self, result, mdata):
        if result.StatusCode.is_good():
            self._monitored_items[result.MonitoredItemId] = mdata
            self._monitored_item_counter += 1

    def _make_monitored_item_common(self, params):
        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

        return result, mdata

    def _create_events_monitored_item(self, params):
        self.logger.info("request to subscribe to events for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)

        result, mdata = self._make_monitored_item_common(params)
        ev_notify_byte = self.aspace.get_attribute_value(params.ItemToMonitor.NodeId, ua.AttributeIds.EventNotifier).Value.Value
        if ev_notify_byte is None or ev_notify_byte & 1 == 0:
            result.StatusCode = ua.StatusCode(ua.StatusCodes.BadServiceUnsupported)
            return result
        result.FilterResult = ua.EventFilterResult()
        for _ in params.RequestedParameters.Filter.SelectClauses:
            result.FilterResult.SelectClauseResults.append(ua.StatusCode())
        # FIXME: where clause result
        self._commit_monitored_item(result, mdata)
        self._monitored_events[params.ItemToMonitor.NodeId] = result.MonitoredItemId
        return result

    def _create_data_change_monitored_item(self, params):
        self.logger.info("request to subscribe to datachange for node %s and attribute %s", params.ItemToMonitor.NodeId, params.ItemToMonitor.AttributeId)

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

FreeOpcUa / python-opcua

Push — master ( 5ef587...a61443 )

MonitoredItemService._make_monitored_item_common() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like