Add filter support for numeric absolute filters for monitored items

include/ua_types.h

@@ -814,6 +814,8 @@ struct UA_DataType {
     UA_DataTypeMember *members;
 };
 
+UA_Boolean isDataTypeNumeric(const UA_DataType *type);
+
 /* The following is used to exclude type names in the definition of UA_DataType
  * structures if the feature is disabled. */
 #ifdef UA_ENABLE_TYPENAMES

src/server/ua_services_subscription.c

@@ -153,10 +153,36 @@ Service_SetPublishingMode(UA_Server *server, UA_Session *session,
                                            &response->resultsSize, &UA_TYPES[UA_TYPES_STATUSCODE]);
 }
 
-static void
+static UA_StatusCode
 setMonitoredItemSettings(UA_Server *server, UA_MonitoredItem *mon,
                          UA_MonitoringMode monitoringMode,
-                         const UA_MonitoringParameters *params) {
+                         const UA_MonitoringParameters *params,
+                         // This parameter is optional and used only if mon->lastValue is not set yet.
+                         // Then numeric type will be detected from this value. Set null as defaut.
+                         const UA_DataType* dataType) {
+
+    /* Filter */
+    if(params->filter.encoding != UA_EXTENSIONOBJECT_DECODED) {
+        UA_DataChangeFilter_init(&(mon->filter));
+        mon->filter.trigger = UA_DATACHANGETRIGGER_STATUSVALUE;
+    } else if(params->filter.content.decoded.type != &UA_TYPES[UA_TYPES_DATACHANGEFILTER]) {
+        return UA_STATUSCODE_BADMONITOREDITEMFILTERINVALID;
+    } else {
+        UA_DataChangeFilter *filter = (UA_DataChangeFilter *)params->filter.content.decoded.data;
+        // TODO implement EURange to support UA_DEADBANDTYPE_PERCENT
+        if (filter->deadbandType == UA_DEADBANDTYPE_PERCENT) {
+            return UA_STATUSCODE_BADMONITOREDITEMFILTERUNSUPPORTED;
+        }
+        if (UA_Variant_isEmpty(&mon->lastValue)) {
+            if (!dataType || !isDataTypeNumeric(dataType))
+                return UA_STATUSCODE_BADFILTERNOTALLOWED;
+        } else
+        if (!isDataTypeNumeric(mon->lastValue.type)) {
+            return UA_STATUSCODE_BADFILTERNOTALLOWED;
+        }
+        UA_DataChangeFilter_copy(filter, &(mon->filter));
+    }
+
     MonitoredItem_unregisterSampleCallback(server, mon);
     mon->monitoringMode = monitoringMode;
 
@@ -184,15 +210,6 @@ setMonitoredItemSettings(UA_Server *server, UA_MonitoredItem *mon,
     if(samplingInterval != samplingInterval) /* Check for nan */
         mon->samplingInterval = server->config.samplingIntervalLimits.min;
 
-    /* Filter */
-    if(params->filter.encoding != UA_EXTENSIONOBJECT_DECODED ||
-       params->filter.content.decoded.type != &UA_TYPES[UA_TYPES_DATACHANGEFILTER]) {
-        /* Default: Trigger only on the value and the statuscode */
-        mon->trigger = UA_DATACHANGETRIGGER_STATUSVALUE;
-    } else {
-        UA_DataChangeFilter *filter = (UA_DataChangeFilter *)params->filter.content.decoded.data;
-        mon->trigger = filter->trigger;
-    }
 
     /* QueueSize */
     UA_BOUNDEDVALUE_SETWBOUNDS(server->config.queueSizeLimits,
@@ -204,6 +221,7 @@ setMonitoredItemSettings(UA_Server *server, UA_MonitoredItem *mon,
     /* Register sample callback if reporting is enabled */
     if(monitoringMode == UA_MONITORINGMODE_REPORTING)
         MonitoredItem_registerSampleCallback(server, mon);
+    return UA_STATUSCODE_GOOD;
 }
 
 static const UA_String binaryEncoding = {sizeof("Default Binary") - 1, (UA_Byte *)"Default Binary"};
@@ -241,13 +259,13 @@ Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct cre
         UA_DataValue_deleteMembers(&v);
         return;
     }
-    UA_DataValue_deleteMembers(&v);
 
     /* Check if the encoding is supported */
     if(request->itemToMonitor.dataEncoding.name.length > 0 &&
        (!UA_String_equal(&binaryEncoding, &request->itemToMonitor.dataEncoding.name) ||
         request->itemToMonitor.dataEncoding.namespaceIndex != 0)) {
         result->statusCode = UA_STATUSCODE_BADDATAENCODINGUNSUPPORTED;
+        UA_DataValue_deleteMembers(&v);
         return;
     }
 
@@ -255,6 +273,7 @@ Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct cre
     if(request->itemToMonitor.attributeId != UA_ATTRIBUTEID_VALUE &&
        request->itemToMonitor.dataEncoding.name.length > 0) {
         result->statusCode = UA_STATUSCODE_BADDATAENCODINGINVALID;
+        UA_DataValue_deleteMembers(&v);
         return;
     }
 
@@ -262,6 +281,7 @@ Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct cre
     UA_MonitoredItem *newMon = UA_MonitoredItem_new(UA_MONITOREDITEMTYPE_CHANGENOTIFY);
     if(!newMon) {
         result->statusCode = UA_STATUSCODE_BADOUTOFMEMORY;
+        UA_DataValue_deleteMembers(&v);
         return;
     }
     UA_StatusCode retval = UA_NodeId_copy(&request->itemToMonitor.nodeId,
@@ -269,6 +289,7 @@ Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct cre
     if(retval != UA_STATUSCODE_GOOD) {
         result->statusCode = retval;
         MonitoredItem_delete(server, newMon);
+        UA_DataValue_deleteMembers(&v);
         return;
     }
     newMon->subscription = cmc->sub;
@@ -276,8 +297,15 @@ Operation_CreateMonitoredItem(UA_Server *server, UA_Session *session, struct cre
     UA_String_copy(&request->itemToMonitor.indexRange, &newMon->indexRange);
     newMon->monitoredItemId = ++cmc->sub->lastMonitoredItemId;
     newMon->timestampsToReturn = cmc->timestampsToReturn;
-    setMonitoredItemSettings(server, newMon, request->monitoringMode,
-                             &request->requestedParameters);
+    retval = setMonitoredItemSettings(server, newMon, request->monitoringMode,
+                             &request->requestedParameters, v.value.type);
+    UA_DataValue_deleteMembers(&v);
+    if(retval != UA_STATUSCODE_GOOD) {
+        result->statusCode = retval;
+        MonitoredItem_delete(server, newMon);
+        --cmc->sub->lastMonitoredItemId;
+        return;
+    }
 
     UA_Subscription_addMonitoredItem(cmc->sub, newMon);
 
@@ -337,8 +365,13 @@ Operation_ModifyMonitoredItem(UA_Server *server, UA_Session *session, UA_Subscri
         result->statusCode = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
         return;
     }
+    UA_StatusCode retval;
+    retval = setMonitoredItemSettings(server, mon, mon->monitoringMode, &request->requestedParameters, NULL);
+    if(retval != UA_STATUSCODE_GOOD) {
+        result->statusCode = retval;
+        return;
+    }
 
-    setMonitoredItemSettings(server, mon, mon->monitoringMode, &request->requestedParameters);
     result->revisedSamplingInterval = mon->samplingInterval;
     result->revisedQueueSize = mon->maxQueueSize;
 
@@ -430,6 +463,7 @@ Operation_SetMonitoringMode(UA_Server *server, UA_Session *session,
 
         /* Initialize lastSampledValue */
         UA_ByteString_deleteMembers(&mon->lastSampledValue);
+        UA_Variant_deleteMembers(&mon->lastValue);
     }
 }
 

src/server/ua_subscription.h

@@ -81,7 +81,8 @@ struct UA_MonitoredItem {
     UA_UInt32 maxQueueSize;
     UA_Boolean discardOldest;
     // TODO: dataEncoding is hardcoded to UA binary
-    UA_DataChangeTrigger trigger;
+    UA_DataChangeFilter filter;
+    UA_Variant lastValue;
 
     /* Sample Callback */
     UA_UInt64 sampleCallbackId;

src/server/ua_subscription_datachange.c

@@ -64,6 +64,7 @@ MonitoredItem_delete(UA_Server *server, UA_MonitoredItem *monitoredItem) {
     LIST_REMOVE(monitoredItem, listEntry);
     UA_String_deleteMembers(&monitoredItem->indexRange);
     UA_ByteString_deleteMembers(&monitoredItem->lastSampledValue);
+    UA_Variant_deleteMembers(&monitoredItem->lastValue);
     UA_NodeId_deleteMembers(&monitoredItem->monitoredNodeId);
     UA_Server_delayedFree(server, monitoredItem);
 }
@@ -146,10 +147,90 @@ void MonitoredItem_ensureQueueSpace(UA_MonitoredItem *mon) {
     /* TODO: Infobits for Events? */
 }
 
+#define ABS_SUBTRACT_TYPE_INDEPENDENT(a,b) ((a)>(b)?(a)-(b):(b)-(a))
+
+static UA_INLINE UA_Boolean
+outOfDeadBand(const void *data1, const void *data2, const size_t index, const UA_DataType *type, const UA_Double deadbandValue) {
+    if (type == &UA_TYPES[UA_TYPES_SBYTE]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_SByte*)data1)[index], ((const UA_SByte*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_BYTE]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Byte*)data1)[index], ((const UA_Byte*)data2)[index]) <= deadbandValue)
+                return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_INT16]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Int16*)data1)[index], ((const UA_Int16*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_UINT16]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_UInt16*)data1)[index], ((const UA_UInt16*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_INT32]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Int32*)data1)[index], ((const UA_Int32*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_UINT32]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_UInt32*)data1)[index], ((const UA_UInt32*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_INT64]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Int64*)data1)[index], ((const UA_Int64*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_UINT64]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_UInt64*)data1)[index], ((const UA_UInt64*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_FLOAT]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Float*)data1)[index], ((const UA_Float*)data2)[index]) <= deadbandValue)
+            return false;
+    } else
+    if (type == &UA_TYPES[UA_TYPES_DOUBLE]) {
+        if (ABS_SUBTRACT_TYPE_INDEPENDENT(((const UA_Double*)data1)[index], ((const UA_Double*)data2)[index]) <= deadbandValue)
+            return false;
+    }
+    return true;
+}
+
+static UA_INLINE UA_Boolean
+updateNeededForFilteredValue(const UA_Variant *value, const UA_Variant *oldValue, const UA_Double deadbandValue) {
+    if (value->arrayLength != oldValue->arrayLength) {
+        return true;
+    }
+    if (value->type != oldValue->type) {
+        return true;
+    }
+    if (UA_Variant_isScalar(value)) {
+        return outOfDeadBand(value->data, oldValue->data, 0, value->type, deadbandValue);
+    } else {
+        for (size_t i = 0; i < value->arrayLength; ++i) {
+            if (outOfDeadBand(value->data, oldValue->data, i, value->type, deadbandValue))
+                return true;
+        }
+    }
+    return false;
+}
+
 /* Errors are returned as no change detected */
 static UA_Boolean
 detectValueChangeWithFilter(UA_MonitoredItem *mon, UA_DataValue *value,
                             UA_ByteString *encoding) {
+    if (isDataTypeNumeric(value->value.type)
+            && (mon->filter.trigger == UA_DATACHANGETRIGGER_STATUSVALUE
+                || mon->filter.trigger == UA_DATACHANGETRIGGER_STATUSVALUETIMESTAMP)) {
+        if (mon->filter.deadbandType == UA_DEADBANDTYPE_ABSOLUTE) {
+            if (!updateNeededForFilteredValue(&value->value, &mon->lastValue, mon->filter.deadbandValue))
+                return false;
+        } /*else if (mon->filter.deadbandType == UA_DEADBANDTYPE_PERCENT) {
+            // TODO where do this EURange come from ?
+            UA_Double deadbandValue = fabs(mon->filter.deadbandValue * (EURange.high-EURange.low));
+            if (!updateNeededForFilteredValue(value->value, mon->lastValue, deadbandValue))
+                return false;
+        }*/
+    }
+
     /* Encode the data for comparison */
     size_t binsize = UA_calcSizeBinary(value, &UA_TYPES[UA_TYPES_DATAVALUE]);
     if(binsize == 0)
@@ -179,7 +260,7 @@ static UA_Boolean
 detectValueChange(UA_MonitoredItem *mon, UA_DataValue *value, UA_ByteString *encoding) {
     /* Apply Filter */
     UA_Boolean hasValue = value->hasValue;
-    if(mon->trigger == UA_DATACHANGETRIGGER_STATUS)
+    if(mon->filter.trigger == UA_DATACHANGETRIGGER_STATUS)
         value->hasValue = false;
 
     UA_Boolean hasServerTimestamp = value->hasServerTimestamp;
@@ -189,7 +270,7 @@ detectValueChange(UA_MonitoredItem *mon, UA_DataValue *value, UA_ByteString *enc
 
     UA_Boolean hasSourceTimestamp = value->hasSourceTimestamp;
     UA_Boolean hasSourcePicoseconds = value->hasSourcePicoseconds;
-    if(mon->trigger < UA_DATACHANGETRIGGER_STATUSVALUETIMESTAMP) {
+    if(mon->filter.trigger < UA_DATACHANGETRIGGER_STATUSVALUETIMESTAMP) {
         value->hasSourceTimestamp = false;
         value->hasSourcePicoseconds = false;
     }
@@ -272,6 +353,8 @@ sampleCallbackWithValue(UA_Server *server, UA_Subscription *sub,
     newNotification->mon = monitoredItem;
 
     /* Replace the encoding for comparison */
+    UA_Variant_deleteMembers(&monitoredItem->lastValue);
+    UA_Variant_copy(&value->value, &monitoredItem->lastValue);
     UA_ByteString_deleteMembers(&monitoredItem->lastSampledValue);
     monitoredItem->lastSampledValue = *valueEncoding;
 

src/ua_types.c

@@ -1079,3 +1079,12 @@ UA_Array_delete(void *p, size_t size, const UA_DataType *type) {
     }
     UA_free((void*)((uintptr_t)p & ~(uintptr_t)UA_EMPTY_ARRAY_SENTINEL));
 }
+
+UA_Boolean
+isDataTypeNumeric(const UA_DataType *type) {
+    // All data types ids between UA_TYPES_SBYTE and UA_TYPES_DOUBLE are numeric
+    for (int i = UA_TYPES_SBYTE; i <= UA_TYPES_DOUBLE; ++i)
+        if (&UA_TYPES[i] == type)
+            return true;
+    return false;
+}