Completed
Push — master ( d538b5...98e417 )
by Olivier
01:58
created

MonitoredItemService.trigger_datachange()   A

Complexity

Conditions 1

Size

Total Lines 4

Duplication

Lines 0
Ratio 0 %

Code Coverage

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