diff --git a/.gitignore b/.gitignore index 9201dfd..4264ace 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,5 @@ aqhomed.vg aqhome-mqttlog.vg aqhome-storage.vg aqhome-storage.pid +aqhome-data.vg +aqhome-data.pid diff --git a/apps/aqhome-data/.gitignore b/apps/aqhome-data/.gitignore new file mode 100644 index 0000000..65e3ba2 --- /dev/null +++ b/apps/aqhome-data/.gitignore @@ -0,0 +1 @@ +test/ diff --git a/apps/aqhome-data/0BUILD b/apps/aqhome-data/0BUILD index 4c92afa..26f2031 100644 --- a/apps/aqhome-data/0BUILD +++ b/apps/aqhome-data/0BUILD @@ -2,7 +2,7 @@ - + $(gwenhywfar_cflags) diff --git a/apps/aqhome-data/aqhome_data_p.h b/apps/aqhome-data/aqhome_data_p.h index d2b24bf..c10e08f 100644 --- a/apps/aqhome-data/aqhome_data_p.h +++ b/apps/aqhome-data/aqhome_data_p.h @@ -17,10 +17,10 @@ #define AQHOME_DATA_DEFAULT_PIDFILE "/var/run/aqhome-data.pid" #define AQHOME_DATA_DEFAULT_DATADIR "/var/lib/aqhome-data/data" -#define AQHOME_DATA_DEFAULT_STATEFILE "/var/lib/aqhome-data/statefile" - #define AQHOME_DATA_DEFAULT_IPC_PORT 45456 +#define AQHOME_DATA_STATEFILENAME "statefile" + struct AQHOME_DATA { diff --git a/apps/aqhome-data/fini.c b/apps/aqhome-data/fini.c index 764a867..6bd434c 100644 --- a/apps/aqhome-data/fini.c +++ b/apps/aqhome-data/fini.c @@ -53,6 +53,7 @@ void AqHomeData_Fini(AQHOME_DATA *aqh) GWEN_MsgEndpoint_Disconnect(aqh->ipcdEndpoint); } GWEN_MsgEndpoint_free(aqh->ipcdEndpoint); + aqh->ipcdEndpoint=NULL; if (aqh->pidFile) remove(aqh->pidFile); diff --git a/apps/aqhome-data/init.c b/apps/aqhome-data/init.c index f22c27c..c2b717b 100644 --- a/apps/aqhome-data/init.c +++ b/apps/aqhome-data/init.c @@ -114,31 +114,26 @@ int AqHomeData_Init(AQHOME_DATA *aqh, int argc, char **argv) int _setupStorage(AQHOME_DATA *aqh, GWEN_DB_NODE *dbArgs) { const char *dataFolder; - const char *stateFile; + GWEN_BUFFER *nameBuf; + AQH_STORAGE *sto; + int rv; dataFolder=GWEN_DB_GetCharValue(dbArgs, "dataFolder", 0, AQHOME_DATA_DEFAULT_DATADIR); - stateFile=GWEN_DB_GetCharValue(dbArgs, "stateFile", 0, NULL); - if (stateFile && *stateFile) { - AQH_STORAGE *sto; - int rv; + nameBuf=GWEN_Buffer_new(0, 256, 0, 1); + GWEN_Buffer_AppendArgs(nameBuf, "%s%s%s", dataFolder, GWEN_DIR_SEPARATOR_S, AQHOME_DATA_STATEFILENAME); - sto=AQH_Storage_new(); - AQH_Storage_SetStateFile(sto, stateFile); + sto=AQH_Storage_new(); + AQH_Storage_SetStateFile(sto, GWEN_Buffer_GetStart(nameBuf)); + AQH_Storage_SetDataFileFolder(sto, dataFolder); + GWEN_Buffer_free(nameBuf); - AQH_Storage_SetDataFileFolder(sto, (dataFolder && *dataFolder)?dataFolder:NULL); - - rv=AQH_Storage_Init(sto); - if (rv<0) { - DBG_INFO(NULL, "here (%d)", rv); - AQH_Storage_free(sto); - return rv; - } - aqh->storage=sto; - } - else { - DBG_ERROR(NULL, "No state file given"); - return GWEN_ERROR_GENERIC; + rv=AQH_Storage_Init(sto); + if (rv<0) { + DBG_INFO(NULL, "here (%d)", rv); + AQH_Storage_free(sto); + return rv; } + aqh->storage=sto; return 0; } @@ -155,10 +150,10 @@ void _setupIpc(AQHOME_DATA *aqh, GWEN_DB_NODE *dbArgs) if (tcpAddress && *tcpAddress && tcpPort) { GWEN_MSG_ENDPOINT *ep; + DBG_INFO(NULL, "Starting TCP service on \"%s\":%d", tcpAddress, tcpPort); ep=GWEN_TcpdEndpoint_new(tcpAddress, tcpPort, NULL, 0); GWEN_TcpdEndpoint_SetAcceptFn(ep, _acceptIpcFn, aqh); - GWEN_MsgEndpoint_Tree2_AddChild(aqh->ipcdEndpoint, ep); aqh->ipcdEndpoint=ep; } } @@ -254,17 +249,6 @@ int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs) I18S("Folder where data files are stored"), I18S("Folder where data files are stored") }, - { - GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ - GWEN_ArgsType_Char, /* type */ - "statefile", /* name */ - 0, /* minnum */ - 1, /* maxnum */ - "S", /* short option */ - "statefile", /* long option */ - I18S("File where rooms, devices and values etc. are stored"), - I18S("File where rooms, devices and values etc. are stored") - }, { GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ GWEN_ArgsType_Char, /* type */ diff --git a/apps/aqhome-data/loop.c b/apps/aqhome-data/loop.c index 18de2c3..3500010 100644 --- a/apps/aqhome-data/loop.c +++ b/apps/aqhome-data/loop.c @@ -15,6 +15,7 @@ #include "./aqhome_data_p.h" #include "aqhome/ipc/data/ipc_data.h" #include "aqhome/ipc/data/msg_data_values.h" +#include "aqhome/ipc/data/msg_data_datapoints.h" #include "aqhome/ipc/msg_ipc_result.h" #include @@ -61,7 +62,7 @@ static void _handleGetLastDataPoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, con void AqHomeData_Loop(AQHOME_DATA *aqh, int timeoutInMsecs) { if (aqh) { - GWEN_MsgEndpoint_ChildrenIoLoop(aqh->ipcdEndpoint, timeoutInMsecs); + GWEN_MsgEndpoint_IoLoop(aqh->ipcdEndpoint, timeoutInMsecs); _readAndHandleIpcMessages(aqh); } } @@ -218,12 +219,13 @@ void _handleAddValue(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *re s=AQH_ValuesDataIpcMsg_GetValueUnits(recvdMsg, 0); if (s && *s) AQH_Value_SetValueUnits(v, s); - DBG_INFO(NULL, "Adding value \"%s\"", s); + DBG_INFO(NULL, "Adding value \"%s\" (%s)", AQH_Value_GetName(v), AQH_Value_GetValueUnits(v)); AQH_Storage_AddValue(aqh->storage, v); + AQH_Storage_AddRuntimeFlags(aqh->storage, AQH_STORAGE_RTFLAGS_MODIFIED); resultCode=AQH_MSG_IPC_SUCCESS; } else { - DBG_INFO(NULL, "Value \"%s\" already exists", s); + DBG_INFO(NULL, "Value \"%s\" already exists", s); resultCode=AQH_MSG_IPC_ERROR_EXISTS; } } @@ -248,14 +250,127 @@ void _handleAddValue(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *re -void _handleEditValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg) +void _handleEditValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *recvdMsg) { + GWEN_MSG *outMsg; + int resultCode=0; + + if (AQH_ValuesDataIpcMsg_IsValid(recvdMsg)) { + uint32_t numValues; + + numValues=AQH_ValuesDataIpcMsg_GetNumValues(recvdMsg); + if (numValues==1) { + const char *s; + + s=AQH_ValuesDataIpcMsg_GetValueName(recvdMsg, 0); + if (s && *s) { + AQH_VALUE *v; + + v=AQH_Storage_GetValueByName(aqh->storage, s); + if (v==NULL) { + DBG_INFO(NULL, "Value \"%s\" doesn't not exist", s); + resultCode=AQH_MSG_IPC_ERROR_EXISTS; + } + else { + DBG_INFO(NULL, "Updating value \"%s\" (%s)", AQH_Value_GetName(v), AQH_Value_GetValueUnits(v)); + s=AQH_ValuesDataIpcMsg_GetValueUnits(recvdMsg, 0); + if (s && *s) + AQH_Value_SetValueUnits(v, s); + AQH_Storage_AddRuntimeFlags(aqh->storage, AQH_STORAGE_RTFLAGS_MODIFIED); + resultCode=AQH_MSG_IPC_SUCCESS; + } + } + else { + DBG_INFO(NULL, "Value without name "); + resultCode=AQH_MSG_IPC_ERROR_INVALID; + } + } + else { + DBG_INFO(NULL, "Invalid number of values in message (%d)", numValues); + resultCode=AQH_MSG_IPC_ERROR_INVALID; + } + } + else { + DBG_INFO(NULL, "Invalid message received"); + resultCode=AQH_MSG_IPC_ERROR_BADDATA; + } + + outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, resultCode); + GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); } -void _handleAddDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg) +void _handleAddDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *recvdMsg) { + GWEN_MSG *outMsg; + int resultCode=0; + + if (AQH_DataPointsDataIpcMsg_IsValid(recvdMsg)) { + uint32_t numValues; + + numValues=AQH_DataPointsDataIpcMsg_GetNumValues(recvdMsg); + if (numValues) { + const char *s; + + s=AQH_DataPointsDataIpcMsg_GetValueName(recvdMsg); + if (s && *s) { + AQH_VALUE *v; + + v=AQH_Storage_GetValueByName(aqh->storage, s); + if (v==NULL) { + // TODO: maybe create the value on the fly + DBG_INFO(NULL, "Value \"%s\" doesn't not exist", s); + resultCode=AQH_MSG_IPC_ERROR_EXISTS; + } + else { + const uint64_t *dataPoints; + + dataPoints=AQH_DataPointsDataIpcMsg_GetDataPoints(recvdMsg); + if (dataPoints) { + uint32_t i; + + for(i=0; istorage, AQH_Value_GetId(v), timestamp, u.f); + if (rv<0) { + DBG_INFO(NULL, "here (%d)", rv); + resultCode=AQH_MSG_IPC_ERROR_GENERIC; + } + else { + DBG_INFO(NULL, "Datapoint added for value \"%s\"", s); + resultCode=0; + } + } /* for */ + } /* if datapoints */ + else { + DBG_INFO(NULL, "No datapoints"); + resultCode=AQH_MSG_IPC_ERROR_BADDATA; + } + } + } + else { + DBG_INFO(NULL, "Value without name "); + resultCode=AQH_MSG_IPC_ERROR_INVALID; + } + } + else { + DBG_INFO(NULL, "No datapoints"); + resultCode=AQH_MSG_IPC_ERROR_BADDATA; + } + } + else { + DBG_INFO(NULL, "Invalid message received"); + resultCode=AQH_MSG_IPC_ERROR_BADDATA; + } + + outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, resultCode); + GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); } diff --git a/apps/aqhome-tool/data/0BUILD b/apps/aqhome-tool/data/0BUILD index 3aeb0d0..2185d69 100644 --- a/apps/aqhome-tool/data/0BUILD +++ b/apps/aqhome-tool/data/0BUILD @@ -34,12 +34,16 @@ getvalues.h + addvalue.h + adddata.h $(local/typefiles) getvalues.c + addvalue.c + adddata.c diff --git a/apps/aqhome-tool/data/adddata.c b/apps/aqhome-tool/data/adddata.c new file mode 100644 index 0000000..9c9fd48 --- /dev/null +++ b/apps/aqhome-tool/data/adddata.c @@ -0,0 +1,275 @@ +/**************************************************************************** + * This file is part of the project AqHome. + * AqHome (c) by 2023 Martin Preuss, all rights reserved. + * + * The license for this file can be found in the file COPYING which you + * should have received along with this file. + ****************************************************************************/ + +#ifdef HAVE_CONFIG_H +# include +#endif + +#include "./adddata.h" +#include "../utils.h" + +#include "aqhome/msg/msg_node.h" +#include "aqhome/ipc/msg_ipc_result.h" +#include "aqhome/ipc/data/msg_data_datapoints.h" +#include "aqhome/ipc/data/ipc_data.h" + +#include +#include +#include +#include + +#include +#include + + +#define I18S(msg) msg +#define I18N(msg) GWEN_I18N_Translate(PACKAGE, msg) + + +static int _doAddData(GWEN_DB_NODE *dbArgs); +static void _sendCommand(GWEN_MSG_ENDPOINT *epTcp, const char *valueName, const char *valueUnits, + uint64_t timestampToSend, double dataToSend); + + + + +int AQH_Tool_AddDataPoint(GWEN_DB_NODE *dbGlobalArgs, int argc, char **argv) +{ + GWEN_DB_NODE *dbLocalArgs; + int rv; + const GWEN_ARGS args[]= { + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "tcpAddress", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "t", /* short option */ + "tcpaddress", /* long option */ + I18S("Specify TCP address to connect to (defaults to 127.0.0.1)"), + I18S("Specify TCP address to connect to (defaults to 127.0.0.1)") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Int, /* type */ + "tcpPort", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "P", /* short option */ + "tcpport", /* long option */ + I18S("Specify the TCP port to listen on"), + I18S("Specify the TCP port to listen on") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Int, /* type */ + "timeout", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "T", /* short option */ + NULL, /* long option */ + I18S("Specify timeout in seconds for response"), + I18S("Specify timeout in seconds for response") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "valueName", /* name */ + 1, /* minnum */ + 1, /* maxnum */ + "N", /* short option */ + "valuename", /* long option */ + I18S("Name/path of the value to add (e.g. server/temp/system)"), + I18S("Name/path of the value to add (e.g. server/temp/system)") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "valueUnits", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "U", /* short option */ + "valueunits", /* long option */ + I18S("Units of the value to add (e.g. \"Grad Celsius\")"), + I18S("Units of the value to add (e.g. \"Grad Celsius\")") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "timestamp", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "t", /* short option */ + "timestamp", /* long option */ + I18S("Timestamp of the data (now if omitted)"), + I18S("Timestamp of the data (now if omitted)") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "value", /* name */ + 1, /* minnum */ + 1, /* maxnum */ + "v", /* short option */ + "value", /* long option */ + I18S("Value to write"), + I18S("Value to write") + }, + { + GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST, /* flags */ + GWEN_ArgsType_Int, /* type */ + "help", /* name */ + 0, /* minnum */ + 0, /* maxnum */ + "h", /* short option */ + "help", /* long option */ + "Show this help screen", /* short description */ + "Show this help screen" /* long description */ + } + }; + + dbLocalArgs=GWEN_DB_GetGroup(dbGlobalArgs, GWEN_DB_FLAGS_DEFAULT, "local"); + rv=GWEN_Args_Check(argc, argv, 1, + GWEN_ARGS_MODE_ALLOW_FREEPARAM, + args, + dbLocalArgs); + if (rv==GWEN_ARGS_RESULT_ERROR) { + fprintf(stderr, "ERROR: Could not parse arguments\n"); + return 1; + } + else if (rv==GWEN_ARGS_RESULT_HELP) { + GWEN_BUFFER *ubuf; + + ubuf=GWEN_Buffer_new(0, 1024, 0, 1); + if (GWEN_Args_Usage(args, ubuf, GWEN_ArgsOutType_Txt)) { + fprintf(stderr, "ERROR: Could not create help string\n"); + return 1; + } + fprintf(stderr, "%s\n", GWEN_Buffer_GetStart(ubuf)); + GWEN_Buffer_free(ubuf); + return 0; + } + + return _doAddData(dbLocalArgs); +} + + + +int _doAddData(GWEN_DB_NODE *dbArgs) +{ + GWEN_MSG_ENDPOINT *epTcp; + int timeoutInSeconds; + const char *valueName; + const char *valueUnits; + const char *valueTimestamp; + const char *valueData; + uint64_t timestampToSend; + double dataToSend; + GWEN_MSG *msg; + int rv; + + timeoutInSeconds=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 5); + valueName=GWEN_DB_GetCharValue(dbArgs, "valueName", 0, NULL); + valueUnits=GWEN_DB_GetCharValue(dbArgs, "valueUnits", 0, NULL); + valueTimestamp=GWEN_DB_GetCharValue(dbArgs, "timestamp", 0, NULL); + valueData=GWEN_DB_GetCharValue(dbArgs, "value", 0, NULL); + if (!(valueName && *valueName)) { + DBG_ERROR(NULL, "ERROR: Missing value name"); + return 1; + } + + if (!(valueData && *valueData)) { + DBG_ERROR(NULL, "ERROR: Missing data"); + return 1; + } + + if (valueTimestamp && *valueTimestamp) { + unsigned long int x; + + if (1!=sscanf("%lu", valueTimestamp, &x)) { + DBG_ERROR(NULL, "ERROR: Invalid timestamp"); + return 1; + } + timestampToSend=(uint64_t) x; + } + else + timestampToSend=(uint64_t) time(NULL); + + rv=GWEN_Text_StringToDouble(valueData, &dataToSend); + if (rv<0) { + DBG_ERROR(NULL, "ERROR: Invalid data"); + return 1; + } + + fprintf(stdout, "Sending AddData request\n"); + + + epTcp=Utils_SetupIpcEndpoint(dbArgs); + if (epTcp==NULL) { + DBG_ERROR(NULL, "ERROR creating TCP connection"); + return 2; + } + + _sendCommand(epTcp, valueName, valueUnits, timestampToSend, dataToSend); + + for (;;) { + uint16_t code; + + msg=Utils_WaitForSpecificIpcMessage(epTcp, AQH_MSGTYPE_IPC_DATA_ADDDATAPOINTS_RSP, timeoutInSeconds); + if (msg==NULL) { + DBG_ERROR(NULL, "No response received"); + return 2; + } + code=GWEN_IpcMsg_GetCode(msg); + if (code==AQH_MSGTYPE_IPC_DATA_ADDDATAPOINTS_RSP || + code==AQH_MSGTYPE_IPC_DATA_RESULT) { + uint32_t resultCode; + + resultCode=AQH_ResultIpcMsg_GetResultCode(msg); + if (resultCode!=AQH_MSG_IPC_SUCCESS) { + fprintf(stderr, "ERROR: %d\n", resultCode); + GWEN_MsgEndpoint_free(epTcp); + return 3; + } + else { + fprintf(stdout, "Data added.\n"); + break; + } + } + else { + DBG_INFO(NULL, "Unexpected message \"%d\"", code); + GWEN_MsgEndpoint_free(epTcp); + return 3; + } + } /* for */ + + GWEN_MsgEndpoint_free(epTcp); + return 0; +} + + + +void _sendCommand(GWEN_MSG_ENDPOINT *epTcp, const char *valueName, const char *valueUnits, uint64_t timestampToSend, double dataToSend) +{ + GWEN_MSG *msgOut; + union {double f; uint64_t i;} u; + uint64_t arrayToSend[2]; + + u.f=dataToSend; + arrayToSend[0]=timestampToSend; + arrayToSend[1]=u.i; + + msgOut=AQH_DataPointsDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_ADDDATAPOINTS_REQ, 0, 0, valueName, valueUnits, arrayToSend, 1); + GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut); +} + + + + + + diff --git a/apps/aqhome-tool/data/adddata.h b/apps/aqhome-tool/data/adddata.h new file mode 100644 index 0000000..1606d65 --- /dev/null +++ b/apps/aqhome-tool/data/adddata.h @@ -0,0 +1,21 @@ +/**************************************************************************** + * This file is part of the project AqHome. + * AqHome (c) by 2023 Martin Preuss, all rights reserved. + * + * The license for this file can be found in the file COPYING which you + * should have received along with this file. + ****************************************************************************/ + +#ifndef AQHOME_TOOL_ADDDATA_H +#define AQHOME_TOOL_ADDDATA_H + + +#include + + + +int AQH_Tool_AddDataPoint(GWEN_DB_NODE *dbGlobalArgs, int argc, char **argv); + + +#endif + diff --git a/apps/aqhome-tool/data/addvalue.c b/apps/aqhome-tool/data/addvalue.c new file mode 100644 index 0000000..7821f51 --- /dev/null +++ b/apps/aqhome-tool/data/addvalue.c @@ -0,0 +1,220 @@ +/**************************************************************************** + * This file is part of the project AqHome. + * AqHome (c) by 2023 Martin Preuss, all rights reserved. + * + * The license for this file can be found in the file COPYING which you + * should have received along with this file. + ****************************************************************************/ + +#ifdef HAVE_CONFIG_H +# include +#endif + +#include "./addvalue.h" +#include "../utils.h" + +#include "aqhome/msg/msg_node.h" +#include "aqhome/ipc/msg_ipc_result.h" +#include "aqhome/ipc/data/msg_data_values.h" +#include "aqhome/ipc/data/ipc_data.h" + +#include +#include +#include +#include + +#include +#include + + +#define I18S(msg) msg +#define I18N(msg) GWEN_I18N_Translate(PACKAGE, msg) + + +static int _doAddValue(GWEN_DB_NODE *dbArgs); +static void _sendCommand(GWEN_MSG_ENDPOINT *epTcp, const char *valueName, const char *valueUnits); + + + + +int AQH_Tool_AddValue(GWEN_DB_NODE *dbGlobalArgs, int argc, char **argv) +{ + GWEN_DB_NODE *dbLocalArgs; + int rv; + const GWEN_ARGS args[]= { + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "tcpAddress", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "t", /* short option */ + "tcpaddress", /* long option */ + I18S("Specify TCP address to connect to (defaults to 127.0.0.1)"), + I18S("Specify TCP address to connect to (defaults to 127.0.0.1)") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Int, /* type */ + "tcpPort", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "P", /* short option */ + "tcpport", /* long option */ + I18S("Specify the TCP port to listen on"), + I18S("Specify the TCP port to listen on") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Int, /* type */ + "timeout", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "T", /* short option */ + NULL, /* long option */ + I18S("Specify timeout in seconds for response"), + I18S("Specify timeout in seconds for response") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "valueName", /* name */ + 1, /* minnum */ + 1, /* maxnum */ + "N", /* short option */ + "valuename", /* long option */ + I18S("Name/path of the value to add (e.g. server/temp/system)"), + I18S("Name/path of the value to add (e.g. server/temp/system)") + }, + { + GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ + GWEN_ArgsType_Char, /* type */ + "valueUnits", /* name */ + 0, /* minnum */ + 1, /* maxnum */ + "U", /* short option */ + "valueunits", /* long option */ + I18S("Units of the value to add (e.g. \"Grad Celsius\")"), + I18S("Units of the value to add (e.g. \"Grad Celsius\")") + }, + { + GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST, /* flags */ + GWEN_ArgsType_Int, /* type */ + "help", /* name */ + 0, /* minnum */ + 0, /* maxnum */ + "h", /* short option */ + "help", /* long option */ + "Show this help screen", /* short description */ + "Show this help screen" /* long description */ + } + }; + + dbLocalArgs=GWEN_DB_GetGroup(dbGlobalArgs, GWEN_DB_FLAGS_DEFAULT, "local"); + rv=GWEN_Args_Check(argc, argv, 1, + GWEN_ARGS_MODE_ALLOW_FREEPARAM, + args, + dbLocalArgs); + if (rv==GWEN_ARGS_RESULT_ERROR) { + fprintf(stderr, "ERROR: Could not parse arguments\n"); + return 1; + } + else if (rv==GWEN_ARGS_RESULT_HELP) { + GWEN_BUFFER *ubuf; + + ubuf=GWEN_Buffer_new(0, 1024, 0, 1); + if (GWEN_Args_Usage(args, ubuf, GWEN_ArgsOutType_Txt)) { + fprintf(stderr, "ERROR: Could not create help string\n"); + return 1; + } + fprintf(stderr, "%s\n", GWEN_Buffer_GetStart(ubuf)); + GWEN_Buffer_free(ubuf); + return 0; + } + + return _doAddValue(dbLocalArgs); +} + + + +int _doAddValue(GWEN_DB_NODE *dbArgs) +{ + GWEN_MSG_ENDPOINT *epTcp; + int timeoutInSeconds; + const char *valueName; + const char *valueUnits; + GWEN_MSG *msg; + + timeoutInSeconds=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 5); + valueName=GWEN_DB_GetCharValue(dbArgs, "valueName", 0, NULL); + valueUnits=GWEN_DB_GetCharValue(dbArgs, "valueUnits", 0, NULL); + if (!(valueName && *valueName)) { + DBG_ERROR(NULL, "ERROR: Missing value name"); + return 1; + } + fprintf(stdout, "Sending AddValue request\n"); + + epTcp=Utils_SetupIpcEndpoint(dbArgs); + if (epTcp==NULL) { + DBG_ERROR(NULL, "ERROR creating TCP connection"); + return 2; + } + + _sendCommand(epTcp, valueName, valueUnits); + + for (;;) { + uint16_t code; + + msg=Utils_WaitForSpecificIpcMessage(epTcp, AQH_MSGTYPE_IPC_DATA_ADDVALUES_RSP, timeoutInSeconds); + if (msg==NULL) { + DBG_ERROR(NULL, "No response received"); + return 2; + } + code=GWEN_IpcMsg_GetCode(msg); + if (code==AQH_MSGTYPE_IPC_DATA_ADDVALUES_RSP || + code==AQH_MSGTYPE_IPC_DATA_RESULT) { + uint32_t resultCode; + + resultCode=AQH_ResultIpcMsg_GetResultCode(msg); + if (resultCode!=AQH_MSG_IPC_SUCCESS) { + fprintf(stderr, "ERROR: %d\n", resultCode); + GWEN_MsgEndpoint_free(epTcp); + return 3; + } + else { + fprintf(stdout, "Value added.\n"); + break; + } + } + else { + DBG_INFO(NULL, "Unexpected message \"%d\"", code); + GWEN_MsgEndpoint_free(epTcp); + return 3; + } + } /* for */ + + GWEN_MsgEndpoint_free(epTcp); + return 0; +} + + + +void _sendCommand(GWEN_MSG_ENDPOINT *epTcp, const char *valueName, const char *valueUnits) +{ + GWEN_MSG *msgOut; + AQH_VALUE *v; + + v=AQH_Value_new(); + AQH_Value_SetName(v, valueName); + AQH_Value_SetValueUnits(v, valueUnits); + + msgOut=AQH_ValuesDataIpcMsg_newForOneValue(AQH_MSGTYPE_IPC_DATA_ADDVALUES_REQ, AQH_MSGDATA_VALUES_FLAGS_LASTMSG, v); + GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut); + AQH_Value_free(v); +} + + + + + + diff --git a/apps/aqhome-tool/data/addvalue.h b/apps/aqhome-tool/data/addvalue.h new file mode 100644 index 0000000..ce64691 --- /dev/null +++ b/apps/aqhome-tool/data/addvalue.h @@ -0,0 +1,21 @@ +/**************************************************************************** + * This file is part of the project AqHome. + * AqHome (c) by 2023 Martin Preuss, all rights reserved. + * + * The license for this file can be found in the file COPYING which you + * should have received along with this file. + ****************************************************************************/ + +#ifndef AQHOME_TOOL_ADDVALUE_H +#define AQHOME_TOOL_ADDVALUE_H + + +#include + + + +int AQH_Tool_AddValue(GWEN_DB_NODE *dbGlobalArgs, int argc, char **argv); + + +#endif + diff --git a/apps/aqhome-tool/data/getvalues.c b/apps/aqhome-tool/data/getvalues.c index a333a31..8a76bf7 100644 --- a/apps/aqhome-tool/data/getvalues.c +++ b/apps/aqhome-tool/data/getvalues.c @@ -129,7 +129,7 @@ int _doGetValues(GWEN_DB_NODE *dbArgs) timeoutInSeconds=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 5); - fprintf(stdout, "Sending PING request\n"); + fprintf(stdout, "Sending GetValues request\n"); _sendCommand(epTcp); diff --git a/apps/aqhome-tool/main.c b/apps/aqhome-tool/main.c index 07b283e..664b190 100644 --- a/apps/aqhome-tool/main.c +++ b/apps/aqhome-tool/main.c @@ -14,6 +14,8 @@ #include "./nodes/flash.h" #include "./nodes/getdevices.h" #include "./data/getvalues.h" +#include "./data/addvalue.h" +#include "./data/adddata.h" #include #include @@ -71,6 +73,8 @@ int main(int argc, char **argv) GWEN_FE_DAH("flash", AQH_Tool_Flash, I18N("Flash a given node on the network")), GWEN_FE_DAH("getdevices", AQH_Tool_GetDevices, I18N("Request list of known devices on the network")), GWEN_FE_DAH("getvalues", AQH_Tool_GetValues, I18N("Request list of known values on the data server")), + GWEN_FE_DAH("addvalue", AQH_Tool_AddValue, I18N("Add a value to the data server")), + GWEN_FE_DAH("adddata", AQH_Tool_AddDataPoint, I18N("Send a datapoint to the data server")), GWEN_FE_END(), }; const GWEN_FUNCS *func; @@ -87,6 +91,7 @@ int main(int argc, char **argv) GWEN_Logger_Open(0, "aqhome-tool", 0, GWEN_LoggerType_Console, GWEN_LoggerFacility_User); GWEN_Logger_SetLevel(0, GWEN_LoggerLevel_Notice); + GWEN_Logger_SetLevel(0, GWEN_LoggerLevel_Info); rv=AQH_Init(); if (rv<0) { diff --git a/aqhome-data.sh b/aqhome-data.sh new file mode 100755 index 0000000..75d8ec3 --- /dev/null +++ b/aqhome-data.sh @@ -0,0 +1,12 @@ +#!/bin/bash + +export AQHOME_LOGLEVEL=info +export LD_LIBRARY_PATH="0-build/aqhome/:$LD_LIBRARY_PATH" + +# 0-build/apps/aqhome-storage/aqhome-storage $* + +0-build/apps/aqhome-data/aqhome-data \ + --datafolder=apps/aqhome-data/test/data \ + -t 127.0.0.1 -P 1899 \ + -p ./aqhome-data.pid \ + $* diff --git a/aqhome/data/storage.c b/aqhome/data/storage.c index b1f11fd..8eac62a 100644 --- a/aqhome/data/storage.c +++ b/aqhome/data/storage.c @@ -359,6 +359,23 @@ void AQH_Storage_SetDataFileFolder(AQH_STORAGE *sto, const char *s) +int AQH_Storage_AddDatapoint(AQH_STORAGE *sto, uint64_t valueId, uint64_t timestamp, double dataPoint) +{ + AQH_DATAFILE *df; + + df=_getDataFileByValueId(sto, valueId); + if (df) { + DBG_INFO(AQH_LOGDOMAIN, "Appending record to datafile"); + AQH_DataFile_AppendRecord(df, timestamp, dataPoint); + return 0; + } + else { + DBG_ERROR(AQH_LOGDOMAIN, "Error getting data file for value %lu", (unsigned long int) valueId); + return GWEN_ERROR_GENERIC; + } +} + + void AQH_Storage_HandleMqttPublish(AQH_STORAGE *sto, const char *sTopic, const char *sValue) { diff --git a/aqhome/data/storage.h b/aqhome/data/storage.h index 8aee5f4..c1a1675 100644 --- a/aqhome/data/storage.h +++ b/aqhome/data/storage.h @@ -86,6 +86,8 @@ AQHOME_API int AQH_Storage_Fini(AQH_STORAGE *sto); AQHOME_API int AQH_Storage_WriteState(AQH_STORAGE *sto); +AQHOME_API int AQH_Storage_AddDatapoint(AQH_STORAGE *sto, uint64_t valueId, uint64_t timestamp, double dataPoint); + AQHOME_API void AQH_Storage_HandleMqttPublish(AQH_STORAGE *sto, const char *topic, const char *value); diff --git a/aqhome/ipc/data/msg_data_values.c b/aqhome/ipc/data/msg_data_values.c index 7172ae2..fc5fa2a 100644 --- a/aqhome/ipc/data/msg_data_values.c +++ b/aqhome/ipc/data/msg_data_values.c @@ -18,6 +18,7 @@ #include #include +#include #include diff --git a/aqhome/ipc/msg_ipc_result.h b/aqhome/ipc/msg_ipc_result.h index 71ceaaa..91a417d 100644 --- a/aqhome/ipc/msg_ipc_result.h +++ b/aqhome/ipc/msg_ipc_result.h @@ -23,10 +23,11 @@ #define AQH_MSG_IPC_SUCCESS 0 -#define AQH_MSG_IPC_ERROR_INVALID 1 -#define AQH_MSG_IPC_ERROR_EXISTS 2 -#define AQH_MSG_IPC_ERROR_NODATA 3 -#define AQH_MSG_IPC_ERROR_BADDATA 4 +#define AQH_MSG_IPC_ERROR_GENERIC 1 +#define AQH_MSG_IPC_ERROR_INVALID 2 +#define AQH_MSG_IPC_ERROR_EXISTS 3 +#define AQH_MSG_IPC_ERROR_NODATA 4 +#define AQH_MSG_IPC_ERROR_BADDATA 5 AQHOME_API GWEN_MSG *AQH_ResultIpcMsg_new(uint16_t code, uint32_t resultCode);