Respect monitored item trigger (DataChangeFilter)

opcua/common/subscription.py

@@ -177,6 +177,17 @@ def subscribe_data_change(self, nodes, attr=ua.AttributeIds.Value):
         """
         return self._subscribe(nodes, attr, queuesize=0)
 
+    def subscribe_data_timestamp_change(self, nodes, attr=ua.AttributeIds.Value):
+        """
+        Subscribe for data and timestamp change events for a node or list of nodes.
+        default attribute is Value.
+        Return a handle which can be used to unsubscribe
+        If more control is necessary use create_monitored_items method
+        """
+        timestamp_filter = ua.DataChangeFilter()
+        timestamp_filter.Trigger = ua.DataChangeTrigger(2) # send notification when status, value or timestamp change
+        return self._subscribe(nodes, attr, mfilter=timestamp_filter, queuesize=0)
+
     def subscribe_events(self, sourcenode=ua.ObjectIds.Server, evtypes=ua.ObjectIds.BaseEventType, evfilter=None, queuesize=0):
         """
         Subscribe to events from a node. Default node is Server node.

opcua/server/address_space.py

@@ -641,7 +641,7 @@ def set_attribute_value(self, nodeid, attr, value):
             old = attval.value
             attval.value = value
             cbs = []
-            if old.Value != value.Value:  # only send call callback when a value change has happend
+            if old.SourceTimestamp != value.SourceTimestamp or old.ServerTimestamp != value.ServerTimestamp:
                 cbs = list(attval.datachange_callbacks.items())
 
         for k, v in cbs:

opcua/server/internal_subscription.py

@@ -197,9 +197,11 @@ def datachange_callback(self, handle, value, error=None):
                 mdata.mvalue.set_current_value(value.Value.Value)
                 if mdata.filter:
                     deadband_flag_pass = self.deadband_callback(mdata.mvalue, mdata.filter)
+                    trigger_flag_pass = self.datachange_trigger_callback(mdata.mvalue, mdata.filter.Trigger)
                 else:
                     deadband_flag_pass = True
-                if deadband_flag_pass:
+                    trigger_flag_pass = self.datachange_trigger_callback(mdata.mvalue, ua.DataChangeTrigger.StatusValue)
+                if deadband_flag_pass and trigger_flag_pass:
                     event.ClientHandle = mdata.client_handle
                     event.Value = value
                     self.isub.enqueue_datachange_event(mid, event, mdata.queue_size)
@@ -216,6 +218,13 @@ def deadband_callback(self, values, flt):
         else:
             return False
 
+    def datachange_trigger_callback(self, values, trigger):
+        if trigger == ua.DataChangeTrigger.StatusValue and \
+                values.get_current_value() == values.get_old_value():
+            return False
+        else:
+            return True
+
     def trigger_event(self, event):
         with self._lock:
             if event.SourceNode not in self._monitored_events:

opcua/tools.py

@@ -357,11 +357,11 @@ def uasubscribe():
                         "--eventtype",
                         dest="eventtype",
                         default="datachange",
-                        choices=['datachange', 'event'],
+                        choices=['datachange', 'timeordatachange', 'event'],
                         help="Event type to subscribe to")
 
     args = parse_args(parser, requirenodeid=False)
-    if args.eventtype == "datachange":
+    if args.eventtype == "datachange" or args.eventtype == "timeordatachange":
         _require_nodeid(parser, args)
     else:
         # FIXME: this is broken, someone may have written i=84 on purpose
@@ -377,6 +377,8 @@ def uasubscribe():
         sub = client.create_subscription(500, handler)
         if args.eventtype == "datachange":
             sub.subscribe_data_change(node)
+        elif args.eventtype == "timeordatachange":
+            sub.subscribe_data_timestamp_change(node)
         else:
             sub.subscribe_events(node)
         print("Type Ctr-C to exit")

tests/tests_subscriptions.py

@@ -157,6 +157,48 @@ def test_subscription_count_empty(self):
         self.assertEqual(myhandler.datachange_count, 4)
         sub.delete()
 
+    def test_subscription_count_timestamp_trigger(self):
+        myhandler = MySubHandlerCounter()
+        sub = self.opc.create_subscription(1, myhandler)
+        o = self.opc.get_objects_node()
+        var = o.add_variable(3, 'SubVarCounter', 0.1)
+        sub.subscribe_data_timestamp_change(var)
+        nb = 12
+        for i in range(nb):
+            val = var.get_value()
+            var.set_value(val +1)
+        time.sleep(0.2)  # let last event arrive
+        self.assertEqual(myhandler.datachange_count, nb + 1)
+        sub.delete()
+
+    def test_subscription_count_timestamp_trigger_no_change(self):
+        myhandler = MySubHandlerCounter()
+        sub = self.opc.create_subscription(1, myhandler)
+        o = self.opc.get_objects_node()
+        var = o.add_variable(3, 'SubVarCounter', 0.1)
+        sub.subscribe_data_timestamp_change(var)
+        nb = 12
+        for i in range(nb):
+            val = var.get_value()
+            var.set_value(val)
+        time.sleep(0.2)  # let last event arrive
+        self.assertEqual(myhandler.datachange_count, nb + 1)
+        sub.delete()
+
+    def test_subscription_count_timestamp_trigger_same_timestamp(self):
+        myhandler = MySubHandlerCounter()
+        sub = self.opc.create_subscription(1, myhandler)
+        o = self.opc.get_objects_node()
+        var = o.add_variable(3, 'SubVarCounter', 0.1)
+        sub.subscribe_data_timestamp_change(var)
+        nb = 12
+        for i in range(nb):
+            val = var.get_data_value() # use exactly the same value including timestamps
+            var.set_value(val)
+        time.sleep(0.2)  # let last event arrive
+        self.assertEqual(myhandler.datachange_count, 1)
+        sub.delete()
+
     def test_subscription_overload_simple(self):
         nb = 10
         myhandler = MySubHandler()
@@ -502,7 +544,7 @@ def test_several_different_events(self):
         propertystring2 = "This is my test 2"
         evgen2.event.PropertyNum = propertynum2
         evgen2.event.PropertyString = propertystring2
-        
+
         for i in range(3):
             evgen1.trigger()
             evgen2.trigger()
@@ -560,7 +602,7 @@ def test_several_different_events_2(self):
         propertystring3 = "This is my test 3"
         evgen3.event.PropertyNum3 = propertynum3
         evgen3.event.PropertyString = propertystring2
-        
+
         for i in range(3):
             evgen1.trigger()
             evgen2.trigger()