Adding the methods necessary for Modifying Subscriptions, make deadband...

Adding the methods necessary for Modifying Subscriptions, make deadband filters and also added support to deadband filter on server side.

opcua/common/subscription.py

@@ -272,18 +272,27 @@ class Subscription(object):
                     del(self._monitoreditems_map[k])
                     return
 
-    def modify_monitored_item(self, handle, new_samp_time, new_queuesize=0, mod_filter=None):
+    def modify_monitored_item(self, handle, new_samp_time, new_queuesize=0, mod_filter_val=None):
         """
         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: New wanted filter, default is None
+        :param mod_filter_val: New deadband filter value
         :return: Return a Modify Monitored Item Result
         """
+        if mod_filter_val < 0:
+            mod_filter = ua.DataChangeFilter()
+            mod_filter.DeadbandValue = -1
+        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)
+        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)
@@ -301,7 +310,7 @@ class Subscription(object):
 
     def deadband_monitor(self, var, deadband_val, deadbandtype=1, queuesize=0, attr=ua.AttributeIds.Value):
         """
-        Function to create a subscription with a Deadband 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
@@ -310,8 +319,18 @@ class Subscription(object):
         :param queuesize: Wanted queue size, default is 1
         """
         deadband_filter = ua.DataChangeFilter()
-        deadband_filter.Trigger = DataChangeTrigger(1)  # send notification when status or value change
+        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)
 
+    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
+
+

opcua/server/address_space.py

@@ -553,7 +553,7 @@ class AddressSpace(object):
 
         for k, v in cbs:
             try:
-                v(k, value)
+                v(k, value, old)
             except Exception as ex:
                 self.logger.exception("Error calling datachange callback %s, %s, %s", k, v, ex)
 

opcua/server/internal_subscription.py

@@ -60,7 +60,7 @@ class MonitoredItemService(object):
     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)
+        self.datachange_callback(handle, variant, variant)
 
     def _modify_monitored_item(self, params):
         with self._lock:
@@ -71,6 +71,7 @@ class MonitoredItemService(object):
                     result.RevisedSamplingInterval = self.isub.data.RevisedPublishingInterval
                     result.RevisedQueueSize = params.RequestedParameters.QueueSize
                     result.FilterResult = params.RequestedParameters.Filter
+                    mdata.mfilter = result.FilterResult
                     mdata.parameters = result
                     return result
             result = ua.MonitoredItemModifyResult()
@@ -162,21 +163,32 @@ class MonitoredItemService(object):
         self._monitored_items.pop(mid)
         return ua.StatusCode()
 
-    def datachange_callback(self, handle, value, error=None):
+    def datachange_callback(self, handle, value, value_old, 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.parameters.RevisedQueueSize)
+                if mdata.mfilter != None:
+                    deadband_flag_pass = self.deadband_callback(value, value_old, mdata.mfilter)
+                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.parameters.RevisedQueueSize)
+
+    def deadband_callback(self, value, value_old, filter):
+        if ((abs(value.Value.Value - value_old.Value.Value)) > filter.DeadbandValue):
+            return True
+        else:
+            return False
 
     def trigger_event(self, event):
         with self._lock: