/**************************************************************************** * 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 "./c_getdatapoints.h" #include "./aqhome_data_p.h" #include "aqhome/ipc/data/ipc_data.h" #include "aqhome/ipc/data/msg_data_multidata.h" #include "aqhome/ipc/data/msg_data_getdata.h" #include "aqhome/ipc/endpoint_ipc.h" #include "aqhome/ipc/msg_ipc_result.h" #include "aqhome/ipc/msg_ipc_tag16.h" #include /* ------------------------------------------------------------------------------------------------ * defines * ------------------------------------------------------------------------------------------------ */ #define AQHOMEDATA_HANDLEGETDATAPOINTS_MAXTABLEENTRIES 2048 #define AQHOMEDATA_HANDLEGETDATAPOINTS_MAXDATAPOINTS 1024 /* ------------------------------------------------------------------------------------------------ * forward declarations * ------------------------------------------------------------------------------------------------ */ static int _getAndSendDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t tsBegin, uint64_t tsEnd, uint64_t num, uint32_t refMsgId); static int _getAndSendDataPointsNoNum(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t tsBegin, uint64_t tsEnd, uint32_t refMsgId); static int _getAndSendDataPointsWithNum(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t num, uint32_t refMsgId); static void _sendDataPointsResponse(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, const uint64_t *tablePtr, uint32_t refMsgId); static void _getAndSendLastDatapoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint32_t refMsgId); /* ------------------------------------------------------------------------------------------------ * implementations * ------------------------------------------------------------------------------------------------ */ void AqHomeData_HandleGetDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, GWEN_MSG *recvdMsg) { GWEN_MSG *outMsg; int resultCode=AQH_MSG_IPC_SUCCESS; if (AQH_IpcEndpoint_GetPermissions(ep) & AQH_IPCENDPOINT_PERMS_READDATA) { AQH_VALUE *value; char *valueName; uint64_t tsBegin; uint64_t tsEnd; uint64_t numRequested; AQH_GetDataDataIpcMsg_Parse(recvdMsg, 0); valueName=AQH_Tag16IpcMsg_GetTagDataAsNewString(recvdMsg, AQH_MSGDATA_GETDATA_TAGS_NAME, NULL); tsBegin=AQH_Tag16IpcMsg_GetTagDataAsUint64(recvdMsg, AQH_MSGDATA_GETDATA_TAGS_BEGIN, 0); tsEnd=AQH_Tag16IpcMsg_GetTagDataAsUint64(recvdMsg, AQH_MSGDATA_GETDATA_TAGS_END, 0); numRequested=AQH_Tag16IpcMsg_GetTagDataAsUint64(recvdMsg, AQH_MSGDATA_GETDATA_TAGS_NUM, 0); value=AQH_Storage_GetValueByNameForSystem(aqh->storage, valueName); if (value) { resultCode=_getAndSendDataPoints(aqh, ep, value, tsBegin, tsEnd, numRequested, GWEN_IpcMsg_GetMsgId(recvdMsg)); if (resultCode==AQH_MSG_IPC_SUCCESS) return; } else { DBG_INFO(NULL, "Value \"%s\" does not exist", valueName); resultCode=AQH_MSG_IPC_ERROR_NOTFOUND; } free(valueName); } else { DBG_ERROR(AQH_LOGDOMAIN, "No permissions to read data"); resultCode=AQH_MSG_IPC_ERROR_PERMS; } outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, GWEN_MsgEndpoint_GetNextMessageId(ep), GWEN_IpcMsg_GetMsgId(recvdMsg), resultCode); GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); } int _getAndSendDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t tsBegin, uint64_t tsEnd, uint64_t num, uint32_t refMsgId) { if (num==0) return _getAndSendDataPointsNoNum(aqh, ep, value, tsBegin, tsEnd, refMsgId); else if (num==1) { _getAndSendLastDatapoint(aqh, ep, value, refMsgId); return AQH_MSG_IPC_SUCCESS; } else return _getAndSendDataPointsWithNum(aqh, ep, value, num, refMsgId); } int _getAndSendDataPointsNoNum(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t tsBegin, uint64_t tsEnd, uint32_t refMsgId) { uint64_t valueId; uint64_t *tablePtr; valueId=AQH_Value_GetId(value); tablePtr=AQH_Storage_GetDataPoints(aqh->storage, valueId, tsBegin, tsEnd, AQHOMEDATA_HANDLEGETDATAPOINTS_MAXTABLEENTRIES); if (tablePtr) { _sendDataPointsResponse(aqh, ep, value, tablePtr, refMsgId); free(tablePtr); return AQH_MSG_IPC_SUCCESS; } else { DBG_INFO(NULL, "No matching datapoints for value \"%s\"", AQH_Value_GetNameForSystem(value)); return AQH_MSG_IPC_ERROR_NODATA; } } int _getAndSendDataPointsWithNum(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint64_t num, uint32_t refMsgId) { uint64_t valueId; uint64_t *tablePtr; if (num>AQHOMEDATA_HANDLEGETDATAPOINTS_MAXDATAPOINTS) num=AQHOMEDATA_HANDLEGETDATAPOINTS_MAXDATAPOINTS; valueId=AQH_Value_GetId(value); tablePtr=AQH_Storage_GetLastNDataPoints(aqh->storage, valueId, num); if (tablePtr) { _sendDataPointsResponse(aqh, ep, value, tablePtr, refMsgId); free(tablePtr); return AQH_MSG_IPC_SUCCESS; } else { DBG_INFO(NULL, "No matching datapoints for value \"%s\"", AQH_Value_GetNameForSystem(value)); return AQH_MSG_IPC_ERROR_NODATA; } } void _sendDataPointsResponse(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, const uint64_t *tablePtr, uint32_t refMsgId) { int numTableEntries; int numDataPoints; GWEN_MSG *outMsg; numTableEntries=(int)(tablePtr[0]); numDataPoints=numTableEntries/2; outMsg=AQH_MultiDataDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_GETDATA_RSP, GWEN_MsgEndpoint_GetNextMessageId(ep), refMsgId, value, &(tablePtr[1]), numDataPoints); GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); } void _getAndSendLastDatapoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE *value, uint32_t refMsgId) { GWEN_MSG *outMsg; int resultCode=AQH_MSG_IPC_SUCCESS; int rv; uint64_t timestamp=0; double data=0.0; rv=AQH_Storage_GetLastDataPoint(aqh->storage, AQH_Value_GetId(value), ×tamp, &data); if (rv<0) { switch(rv) { case GWEN_ERROR_INVALID: resultCode=AQH_MSG_IPC_ERROR_INVALID; break; case GWEN_ERROR_NO_DATA: resultCode=AQH_MSG_IPC_ERROR_NODATA; break; default: resultCode=AQH_MSG_IPC_ERROR_GENERIC; break; } } else { outMsg=AQH_MultiDataDataIpcMsg_newForOne(AQH_MSGTYPE_IPC_DATA_GETDATA_RSP, GWEN_MsgEndpoint_GetNextMessageId(ep), refMsgId, value, timestamp, data); GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); return; } outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, GWEN_MsgEndpoint_GetNextMessageId(ep), refMsgId, resultCode); GWEN_MsgEndpoint_AddSendMessage(ep, outMsg); }