open62541开发：添加sqlite3 历史数据库

            

    历史数据库在OPCUA  应用中十分重要，例如OPCUA 网关和OPCUA 汇聚服务器中都需要历史数据库功能。但是open62541 协议栈中仅包含了基于内存的历史数据库，在实际应用中是不够的。本博文讨论open62541 中添加sqlite3 为基础的历史数据库若干问题。

借鉴

        Github上有一些open62541 添加sqlite 的项目的项目，在CSDN 上

学习open62541 --- [58] 使用数据库存储历史数据

的博文中，介绍了如何将sqlite 改成Linux 下的实现。它的项目在：

GitHub - nicolasr75/open62541_sqlite:  

      他的项目对我帮助很大，但是，它的sqlite 不是加载到open62541 的Plugin 中，而是单独地添加在应用程序的项目中的。 他将c程序和h 文件合在了一起，只有使用sqlite 历史数据时才会被include 并且编译。另一方面，该项目只支持单变量历史数据存储。

      我的项目是将open62541 安装到系统目录中的，所以希望将sqlite 结合到open61541 程序中。另外扩展为多变量历史数据存储。下面记录我的实现过程。

在open62541 包中加入sqlite3 

安装sqlite3 库

sudo apt -y install libsqlite3-dev

        将SQLiteBackend.h 放置到open62541 的Plugins/include /plugin/history_data 中。

重新编译open62541 就可以了。也可以单独地将SQLiteBackend.h放置在应用程序中。

修改成多节点存储

        在参考文章中只是单变量存储的方法。不支持多个变量的存储。我将它进行了修改：

方法1：使用NodeId区分不同变量

将MearsuringPointID 改为NodeId.节点的字符串（ns=1；i=5055）。注意要带单括号（‘）。

 strncat(query, "') AND (MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);

方法2 将不同变量的数据构建不同的表（table）

这种方法的好处是访问数据库效率高。

在这里，我们仅实现第一种方式。

源代码（SQLiteBackend.h)

#ifndef BACKEND_H
#define BACKEND_H#include <time.h>
#include <stdio.h>
#include <open62541/plugin/historydata/history_data_backend.h>
#include "sqlite3.h"static const size_t END_OF_DATA = SIZE_MAX;
static const size_t QUERY_BUFFER_SIZE = 500;char *UA_String2string(UA_String uaString)
{char *convert = (char *)UA_malloc(sizeof(char) * uaString.length + 1);memcpy(convert, uaString.data, uaString.length);convert[uaString.length] = '\0';return convert;
}
UA_Int64 convertTimestampStringToUnixSeconds(const char* timestampString)
{UA_DateTimeStruct dts;memset(&dts, 0, sizeof(dts));//   printf("convertTimestampStringToUnixSeconds:%s\n",timestampString);sscanf(timestampString, "%hu-%hu-%hu %hu:%hu:%hu", &dts.year, &dts.month, &dts.day, &dts.hour, &dts.min, &dts.sec);// printf("%d %d\n",dts.year, dts.month);UA_DateTime dt = UA_DateTime_fromStruct(dts);UA_Int64 t = UA_DateTime_toUnixTime(dt);//   printf("convertTimestampStringToUnixSeconds=%lu\n",t);return t;
}const char* convertUnixSecondsToTimestampString(UA_Int64 unixSeconds)
{static char buffer[20];//  printf("convertUnixSecondsToTimestampString=%lu\n",unixSeconds);UA_DateTime dt = UA_DateTime_fromUnixTime(unixSeconds);UA_DateTimeStruct dts = UA_DateTime_toStruct(dt);struct tm tm;memset(&tm, 0, sizeof(tm));tm.tm_year = dts.year- 1900;tm.tm_mon  = dts.month - 1;tm.tm_mday = dts.day;tm.tm_hour = dts.hour;tm.tm_min  = dts.min;tm.tm_sec  = dts.sec;// printf("=%d %d\n",tm.tm_year,tm.tm_mon);memset(buffer, 0, 20);strftime(buffer, 20, "%Y-%m-%d %H:%M:%S", &tm);//  printf("convertUnixSecondsToTimestampString:%s\n",buffer);return buffer;
}//Context that is needed for the SQLite callback for copying data.
struct context_copyDataValues {size_t maxValues;size_t counter;UA_DataValue *values;
};typedef  struct context_copyDataValues  context_copyDataValues;struct context_sqlite {sqlite3* sqlite;const char* measuringPointID;
};static struct context_sqlite*
generateContext_sqlite(const char* filename) 
{sqlite3* handle;char *errorMessage;int res = sqlite3_open(filename, &handle);if (res != SQLITE_OK)return NULL;struct context_sqlite* ret = (struct context_sqlite*)UA_calloc(1, sizeof(struct context_sqlite));if (ret == NULL){return NULL;}const char *sql = "DROP TABLE IF EXISTS PeriodicValues;" "CREATE TABLE PeriodicValues(MeasuringPointID STRING, Value DOUBLE, Timestamp DATETIME DEFAULT CURRENT_TIMESTAMP);";res = sqlite3_exec(handle, sql, NULL, NULL, &errorMessage);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);sqlite3_close(handle);return NULL;}ret->sqlite = handle;//For this demo we have only one source measuring point which we hardcode in the context.//A more advanced demo should determine the available measuring points from the source//itself or maybe an external configuration file.ret->measuringPointID = "1";return ret;
}static UA_StatusCode
serverSetHistoryData_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,UA_Boolean historizing,const UA_DataValue *value)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;size_t result;char* errorMessage;char query[QUERY_BUFFER_SIZE];//  UA_String Id=nodeId->identifier.string;char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncpy(query, "INSERT INTO PeriodicValues VALUES('", QUERY_BUFFER_SIZE);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);if (value->hasValue && value->status == UA_STATUSCODE_GOOD &&value->value.type == &UA_TYPES[UA_TYPES_DOUBLE]){char remaining[60];snprintf(remaining, 60, "',%f, CURRENT_TIMESTAMP);", *(double*)(value->value.data));//datetime(CURRENT_TIMESTAMP,'localtime')strncat(query, remaining, QUERY_BUFFER_SIZE);}else{printf("%s | Error | historical value is invalid\n", __func__);return UA_STATUSCODE_BADINTERNALERROR;}printf("serverSetHistoryData_sqliteHDB:%s\n",query);int res = sqlite3_exec(context->sqlite, query, NULL, NULL, &errorMessage);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return UA_STATUSCODE_BADINTERNALERROR;}return UA_STATUSCODE_GOOD;
}static size_t
getEnd_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId)
{return END_OF_DATA;
}//This is a callback for all queries that return a single timestamp as the number of Unix seconds
static int timestamp_callback(void* result, int count, char **data, char **columns)
{*(UA_Int64*)result = convertTimestampStringToUnixSeconds(data[0]);printf("timestamp_callback:%s\n",data[0]);//   printf("timestamp_callback%lu\n",*(UA_Int64*)result);return 0;
}static int resultSize_callback(void* result, int count, char **data, char **columns)
{*(size_t*)result = strtol(data[0], NULL, 10);return 0;
}static size_t
lastIndex_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;size_t result;char* errorMessage;char query[QUERY_BUFFER_SIZE];strncpy(query, "SELECT Timestamp FROM PeriodicValues WHERE MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);strncat(query, "'  ORDER BY Timestamp DESC LIMIT 1", QUERY_BUFFER_SIZE);//DESCprintf("lastIndex:%s\n",query);int res = sqlite3_exec(context->sqlite, query, timestamp_callback, &result, &errorMessage);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return END_OF_DATA;}return result;
}static size_t
firstIndex_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;size_t result;char* errorMessage;char query[QUERY_BUFFER_SIZE];strncpy(query, "SELECT Timestamp FROM PeriodicValues WHERE MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);strncat(query, "' ORDER BY Timestamp LIMIT 1", QUERY_BUFFER_SIZE);int res = sqlite3_exec(context->sqlite, query, timestamp_callback, &result, &errorMessage);printf("firsrIndex:%s\n",query);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return END_OF_DATA;}return result;
}static UA_Boolean
search_sqlite(struct context_sqlite* context,const UA_NodeId *nodeId,UA_Int64 unixSeconds, MatchStrategy strategy,size_t *index) 
{	*index = END_OF_DATA; // TODOchar* errorMessage;char query[QUERY_BUFFER_SIZE];strncpy(query, "SELECT Timestamp FROM PeriodicValues WHERE MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);strncat(query, "' AND ", QUERY_BUFFER_SIZE);switch (strategy){case MATCH_EQUAL_OR_AFTER:strncat(query, "Timestamp>='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(unixSeconds), QUERY_BUFFER_SIZE);strncat(query, "' ORDER BY Timestamp LIMIT 1", QUERY_BUFFER_SIZE);break;case MATCH_AFTER:strncat(query, "Timestamp>'", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(unixSeconds), QUERY_BUFFER_SIZE);strncat(query, "' ORDER BY Timestamp LIMIT 1", QUERY_BUFFER_SIZE);break;case MATCH_EQUAL_OR_BEFORE:strncat(query, "Timestamp<='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(unixSeconds), QUERY_BUFFER_SIZE);strncat(query, "' ORDER BY Timestamp DESC LIMIT 1", QUERY_BUFFER_SIZE);break;case MATCH_BEFORE:strncat(query, "Timestamp<'", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(unixSeconds), QUERY_BUFFER_SIZE);strncat(query, "' ORDER BY Timestamp DESC LIMIT 1", QUERY_BUFFER_SIZE);break;default:return false;}printf("search_sqlite:%s\n",query);int res = sqlite3_exec(context->sqlite, query, timestamp_callback, index, &errorMessage);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return false;}else{return true;}}static size_t
getDateTimeMatch_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,const UA_DateTime timestamp,const MatchStrategy strategy)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;UA_Int64 ts =UA_DateTime_toUnixTime(timestamp);printf("getDateTimeMatch_sqliteHDB:%s\n",convertUnixSecondsToTimestampString(ts));printf("strategy:%u\n",strategy);size_t result = END_OF_DATA;UA_Boolean res = search_sqlite(context,nodeId, ts, strategy, &result);return result;
}static size_t
resultSize_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,size_t startIndex,size_t endIndex)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;char* errorMessage;size_t result = 0;printf("resultSize_sqliteHDB:startIndex:%lu\n",startIndex);printf("resultSize_sqliteHDB:endIndex:%lu\n",endIndex);char query[QUERY_BUFFER_SIZE];strncpy(query, "SELECT COUNT(*) FROM PeriodicValues WHERE ", QUERY_BUFFER_SIZE);strncat(query, "(Timestamp>='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(startIndex), QUERY_BUFFER_SIZE);strncat(query, "') AND (Timestamp<='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(endIndex), QUERY_BUFFER_SIZE);strncat(query, "') AND (MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);strncat(query, "')", QUERY_BUFFER_SIZE);printf("resultSize_sqliteHDB:%s\n",query);int res = sqlite3_exec(context->sqlite, query, resultSize_callback, &result, &errorMessage);if (res != SQLITE_OK){printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return 0; // no data}return result;
}static int copyDataValues_callback(void* result, int count, char **data, char **columns)
{UA_DataValue dv;UA_DataValue_init(&dv);dv.status = UA_STATUSCODE_GOOD;dv.hasStatus = true;dv.sourceTimestamp = UA_DateTime_fromUnixTime(convertTimestampStringToUnixSeconds(data[0]));dv.hasSourceTimestamp = true;dv.serverTimestamp = dv.sourceTimestamp;dv.hasServerTimestamp = true;double value = strtod(data[1], NULL);UA_Variant_setScalarCopy(&dv.value, &value, &UA_TYPES[UA_TYPES_DOUBLE]);dv.hasValue = true;context_copyDataValues* ctx = (context_copyDataValues*)result;UA_DataValue_copy(&dv, &ctx->values[ctx->counter]);ctx->counter++;if (ctx->counter == ctx->maxValues){return 1;}else{return 0;}
}static UA_StatusCode
copyDataValues_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,size_t startIndex,size_t endIndex,UA_Boolean reverse,size_t maxValues,UA_NumericRange range,UA_Boolean releaseContinuationPoints,const UA_ByteString *continuationPoint,UA_ByteString *outContinuationPoint,size_t *providedValues,UA_DataValue *values)
{//NOTE: this demo does not support continuation points!!!struct context_sqlite* context = (struct context_sqlite*)hdbContext;char* errorMessage;//  const char* measuringPointID = "1";char query[QUERY_BUFFER_SIZE];strncpy(query, "SELECT Timestamp, Value FROM PeriodicValues WHERE ", QUERY_BUFFER_SIZE);strncat(query, "(Timestamp>='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(startIndex), QUERY_BUFFER_SIZE);strncat(query, "') AND (Timestamp<='", QUERY_BUFFER_SIZE);strncat(query, convertUnixSecondsToTimestampString(endIndex), QUERY_BUFFER_SIZE);strncat(query, "') AND (MeasuringPointID='", QUERY_BUFFER_SIZE);char measuringPointID[60];snprintf(measuringPointID, 60, "ns=%d;i=%d",nodeId->namespaceIndex,nodeId->identifier.numeric);strncat(query, measuringPointID, QUERY_BUFFER_SIZE);strncat(query, "')", QUERY_BUFFER_SIZE);context_copyDataValues ctx;ctx.maxValues = maxValues;ctx.counter = 0;ctx.values = values;int res = sqlite3_exec(context->sqlite, query, copyDataValues_callback, &ctx, &errorMessage);if (res != SQLITE_OK){if (res == SQLITE_ABORT) // if reach maxValues, then request abort, so this is not error{sqlite3_free(errorMessage);return UA_STATUSCODE_GOOD;}else{printf("%s | Error | %s\n", __func__, errorMessage);sqlite3_free(errorMessage);return UA_STATUSCODE_BADINTERNALERROR;}}else{return UA_STATUSCODE_GOOD;}
}static const UA_DataValue*
getDataValue_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,size_t index)
{struct context_sqlite* context = (struct context_sqlite*)hdbContext;return NULL;
}static UA_Boolean
boundSupported_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId)
{return false; // We don't support returning bounds in this demo
}static UA_Boolean
timestampsToReturnSupported_sqliteHDB(UA_Server *server,void *hdbContext,const UA_NodeId *sessionId,void *sessionContext,const UA_NodeId *nodeId,const UA_TimestampsToReturn timestampsToReturn)
{return true;
}UA_HistoryDataBackend
UA_HistoryDataBackend_sqlite(const char* filename)
{UA_HistoryDataBackend result;memset(&result, 0, sizeof(UA_HistoryDataBackend));result.serverSetHistoryData = &serverSetHistoryData_sqliteHDB;result.resultSize = &resultSize_sqliteHDB;result.getEnd = &getEnd_sqliteHDB;result.lastIndex = &lastIndex_sqliteHDB;result.firstIndex = &firstIndex_sqliteHDB;result.getDateTimeMatch = &getDateTimeMatch_sqliteHDB;result.copyDataValues = &copyDataValues_sqliteHDB;result.getDataValue = &getDataValue_sqliteHDB;result.boundSupported = &boundSupported_sqliteHDB;result.timestampsToReturnSupported = &timestampsToReturnSupported_sqliteHDB;result.deleteMembers = NULL; // We don't support deleting in this demoresult.getHistoryData = NULL; // We don't support the high level API in this demoresult.context = generateContext_sqlite(filename);return result;
}#endif

应用程序测试

server 修改为添加两个变量。NodeId 使用Numeric 方式。

源代码（server.h)

#include <signal.h>
#include <open62541/server.h>
#include <open62541/server_config_default.h>
#include <open62541/plugin/log_stdout.h>
#include <open62541/plugin/historydata/history_data_gathering_default.h>
#include <open62541/plugin/historydata/history_database_default.h>
#include <open62541/plugin/historydatabase.h>
#include "SQLiteBackend.h"static UA_Boolean running = true;static void stopHandler(int sign) 
{(void)sign;UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");running = false;
}UA_NodeId addHistoryVariable(UA_Server *server,char *VairableName){/* Define the attribute of the uint32 variable node */UA_VariableAttributes attr = UA_VariableAttributes_default;UA_Double myDouble = 17.2;UA_Variant_setScalar(&attr.value, &myDouble, &UA_TYPES[UA_TYPES_DOUBLE]);attr.description = UA_LOCALIZEDTEXT("en-US", VairableName);attr.displayName = UA_LOCALIZEDTEXT("en-US", VairableName);attr.dataType = UA_TYPES[UA_TYPES_DOUBLE].typeId;/** We set the access level to also support history read* This is what will be reported to clients*/attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE | UA_ACCESSLEVELMASK_HISTORYREAD;/** We also set this node to historizing, so the server internals also know from it.*/attr.historizing = true;/* Add the variable node to the information model */UA_NodeId doubleNodeId = UA_NODEID_STRING(1, VairableName);UA_QualifiedName doubleName = UA_QUALIFIEDNAME(1, VairableName);UA_NodeId parentNodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER);UA_NodeId parentReferenceNodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES);UA_NodeId outNodeId;UA_NodeId_init(&outNodeId);UA_StatusCode retval = UA_Server_addVariableNode(server,UA_NODEID_NULL,parentNodeId,parentReferenceNodeId,doubleName,UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),attr,NULL,&outNodeId);UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "UA_Server_addVariableNode %s", UA_StatusCode_name(retval));UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,"addHistoryVariable:%d",outNodeId.identifier.numeric);
return outNodeId;
}
int main(void) 
{signal(SIGINT, stopHandler);signal(SIGTERM, stopHandler);UA_Server *server = UA_Server_new();UA_ServerConfig *config = UA_Server_getConfig(server);UA_ServerConfig_setDefault(config);UA_HistoryDataGathering gathering = UA_HistoryDataGathering_Default(1);config->historyDatabase = UA_HistoryDatabase_default(gathering);UA_NodeId VariableANodeId=addHistoryVariable(server,"myDoubleValueA");
UA_NodeId VariableBNodeId=addHistoryVariable(server,"myDoubleValueB");UA_HistorizingNodeIdSettings setting;setting.historizingBackend = UA_HistoryDataBackend_sqlite("database.sqlite");setting.maxHistoryDataResponseSize = 100;setting.historizingUpdateStrategy = UA_HISTORIZINGUPDATESTRATEGY_VALUESET;UA_StatusCode   retval = gathering.registerNodeId(server, gathering.context, &VariableANodeId, setting);retval = gathering.registerNodeId(server, gathering.context, &VariableBNodeId, setting);UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "registerNodeId %s", UA_StatusCode_name(retval));retval = UA_Server_run(server, &running);UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "UA_Server_run %s", UA_StatusCode_name(retval));UA_Server_delete(server);return (int)retval;}

使用DB Browser for SQLite 工具查看数据库数据

下载一个DB Browser for SQLite 工具可以查看数据库中的数据：

使用uaExpert

         使用uaExpert 修改变量的数据 ，并且查看历史数据。具体方法可以参照引用的博文。特别注意的是；在数据库中存储的时间标签是UTC标准时间，而uaExpert 使用的是北京时间，它们相差了8小时。在查看历史数据时，要注意起始时间和结束时间,设置错了，不会读出。可以通过DB Browser 查看记录数据的时间标签，并将小时加8。

结束语

        开源软件并不完全是拿来就够了。真正实际应用，需要大量的开发工作。要充分估计开源软件的学习和扩展功能的工作量。