merge conflict

opcua/client/ua_client.py

@@ -477,3 +477,13 @@ class UaClient(object):
         self.logger.debug(response)
         response.ResponseHeader.ServiceResult.check()
         return response.Results
+
+    def modify_monitored_items(self, params):
+        self.logger.info("modify_monitored_items")
+        request = ua.ModifyMonitoredItemsRequest()
+        request.Parameters = params
+        data = self._uasocket.send_request(request)
+        response = ua.ModifyMonitoredItemsResponse.from_binary(data)
+        self.logger.debug(response)
+        response.ResponseHeader.ServiceResult.check()
+        return response.Results

opcua/common/subscription.py

@@ -239,6 +239,7 @@ class Subscription(object):
                 data.client_handle = mi.RequestedParameters.ClientHandle
                 data.node = Node(self.server, mi.ItemToMonitor.NodeId)
                 data.attribute = mi.ItemToMonitor.AttributeId
+                #TODO: Either use the filter from request or from response. Here it uses from request, in modify it uses from response
                 data.mfilter = mi.RequestedParameters.Filter
                 self._monitoreditems_map[mi.RequestedParameters.ClientHandle] = data
         results = self.server.create_monitored_items(params)
@@ -272,4 +273,60 @@ class Subscription(object):
                     del(self._monitoreditems_map[k])
                     return
 
+    def modify_monitored_item(self, handle, new_samp_time, new_queuesize=0, mod_filter_val=-1):
+        """
+        Modify a monitored item.
+        :param handle: Handle returned when originally subscribing
+        :param new_samp_time: New wanted sample time
+        :param new_queuesize: New wanted queuesize, default is 0
+        :param mod_filter_val: New deadband filter value
+        :return: Return a Modify Monitored Item Result
+        """
+        for monitored_item_index in self._monitoreditems_map:
+            if self._monitoreditems_map[monitored_item_index].server_handle == handle:
+                item_to_change = self._monitoreditems_map[monitored_item_index]
+                break
+        if mod_filter_val == None:
+            mod_filter = None
+        elif mod_filter_val < 0:
+            mod_filter = item_to_change.mfilter
+        else:
+            mod_filter = ua.DataChangeFilter()
+            mod_filter.Trigger = ua.DataChangeTrigger(1)  # send notification when status or value change
+            mod_filter.DeadbandType = 1
+            mod_filter.DeadbandValue = mod_filter_val  # absolute float value or from 0 to 100 for percentage deadband
+        modif_item = ua.MonitoredItemModifyRequest()
+        modif_item.MonitoredItemId = handle
+        modif_item.RequestedParameters = self._modify_monitored_item_request(new_queuesize, new_samp_time,
+                                                                             mod_filter)
+        params = ua.ModifyMonitoredItemsParameters()
+        params.SubscriptionId = self.subscription_id
+        params.ItemsToModify.append(modif_item)
+        results = self.server.modify_monitored_items(params)
+        item_to_change.mfilter = results[0].FilterResult
+        return results
+
+    def _modify_monitored_item_request(self, new_queuesize, new_samp_time, mod_filter):
+        req_params = ua.MonitoringParameters()
+        with self._lock:
+            req_params.ClientHandle = self._client_handle
+        req_params.QueueSize = new_queuesize
+        req_params.Filter = mod_filter
+        req_params.SamplingInterval = new_samp_time
+        return req_params
 
+    def deadband_monitor(self, var, deadband_val, deadbandtype=1, queuesize=0, attr=ua.AttributeIds.Value):
+        """
+        Method to create a subscription with a Deadband Value.
+        Default deadband value type is absolute.
+        Return a handle which can be used to unsubscribe
+        :param var: Variable to which you want to subscribe
+        :param deadband_val: Absolute float value
+        :param deadbandtype: Default value is 1 (absolute), change to 2 for percentage deadband
+        :param queuesize: Wanted queue size, default is 1
+        """
+        deadband_filter = ua.DataChangeFilter()
+        deadband_filter.Trigger = ua.DataChangeTrigger(1)  # send notification when status or value change
+        deadband_filter.DeadbandType = deadbandtype
+        deadband_filter.DeadbandValue = deadband_val  # absolute float value or from 0 to 100 for percentage deadband
+        return self._subscribe(var, attr, deadband_filter, queuesize)

opcua/server/internal_subscription.py

@@ -16,11 +16,29 @@ class MonitoredItemData(object):
         self.callback_handle = None
         self.monitored_item_id = None
         self.mode = None
-        self.filter = None  
+        self.filter = None
+        self.mvalue = MonitoredItemValues()
         self.where_clause_evaluator = None
         self.queue_size = 0
 
 
+class MonitoredItemValues(object):
+
+    def __init__(self):
+        self.current_value = None
+        self.old_value = None
+
+    def set_current_value(self, cur_val):
+        self.old_value = self.current_value
+        self.current_value = cur_val
+
+    def get_current_value(self):
+        return self.current_value
+
+    def get_old_value(self):
+        return self.old_value
+
+
 class MonitoredItemService(object):
 
     """
@@ -161,19 +179,32 @@ class MonitoredItemService(object):
 
     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.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)
+            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.queue_size)
+                mdata.mvalue.set_current_value(value.Value.Value)
+                if mdata.filter is not None:
+                    deadband_flag_pass = self.deadband_callback(mdata.mvalue, mdata.filter)
+                else:
+                    deadband_flag_pass = True
+                if deadband_flag_pass:
+                    event.ClientHandle = mdata.client_handle
+                    event.Value = value
+                    self.isub.enqueue_datachange_event(mid, event, mdata.queue_size)
+
+    def deadband_callback(self, values, flt):
+        if (values.get_old_value() is not None) or \
+                ((abs(values.get_current_value() - values.get_old_value())) > flt.DeadbandValue):
+            return True
+        else:
+            return False
 
     def trigger_event(self, event):
         with self._lock:
@@ -255,7 +286,6 @@ class InternalSubscription(object):
         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)