martin/aqhomecontrol
Public Access
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

aqhome-data: adding datapoints basically works now.

Browse Source
This commit is contained in:
Martin Preuss
2023-08-17 00:24:38 +02:00
parent f9ae85b9ad
commit 3bfb39966f
19 changed files with 727 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/aqhome-data/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
test/

2
apps/aqhome-data/0BUILD
View File

@@ -2,7 +2,7 @@
<gwbuild>
<target type="Program" name="aqhome-storage" install="$(bindir)" >
<target type="Program" name="aqhome-data" install="$(bindir)" >
<includes type="c" >
$(gwenhywfar_cflags)

4
apps/aqhome-data/aqhome_data_p.h
View File

@@ -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 {

1
apps/aqhome-data/fini.c
View File

@@ -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);

48
apps/aqhome-data/init.c
View File

@@ -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 */

125
apps/aqhome-data/loop.c
View File

@@ -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 <gwenhywfar/gwenhywfar.h>
@@ -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; i<numValues; i++) {
uint64_t timestamp;
union {double f; uint64_t i;} u;
int rv;
timestamp=*(dataPoints++);
u.i=*(dataPoints++);
rv=AQH_Storage_AddDatapoint(aqh->storage, 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);
}