aqhomecontrol/apps/aqhome-data/s_updatedata.c

/****************************************************************************
 * This file is part of the project AqHome.
 * AqHome (c) by 2025 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 <config.h>
#endif


#include "./s_updatedata.h"

#include "./server_p.h"
#include <aqhome/data/value.h>
#include "aqhome/ipc2/endpoint.h"
#include "aqhome/msg/ipc/m_ipc.h"
#include "aqhome/msg/ipc/data/m_ipcd.h"
#include "aqhome/msg/ipc/data/m_ipcd_multidata.h"
#include "aqhome/msg/ipc/m_ipc_result.h"
#include "aqhome/msg/ipc/m_ipc_tag16.h"

#include <gwenhywfar/debug.h>



/* ------------------------------------------------------------------------------------------------
 * defines
 * ------------------------------------------------------------------------------------------------
 */

#define DISABLE_DEBUGLOG



/* ------------------------------------------------------------------------------------------------
 * forward declarations
 * ------------------------------------------------------------------------------------------------
 */

static int _storeDataPoints(AQHOME_SERVER *xo, const AQH_VALUE *v, const uint64_t *dataPoints, unsigned int numValues);
static void _sendDataChangedMsgToAllClients(AQHOME_SERVER *xo, AQH_OBJECT *epSrc,
                                            const AQH_VALUE *v, const uint64_t *dataPoints, int numValues);



/* ------------------------------------------------------------------------------------------------
 * implementations
 * ------------------------------------------------------------------------------------------------
 */

void AqHomeDataServer_HandleUpdateData(AQH_OBJECT *o, AQH_OBJECT *ep, const AQH_MESSAGE *msg, const GWEN_TAG16_LIST *tagList)
{
  if (tagList) {
    AQHOME_SERVER *xo;
  
    xo=AqHomeDataServer_GetServerData(o);
    if (xo) {
      AQH_MESSAGE *outMsg;
      int resultCode=AQH_MSGDATA_RESULT_SUCCESS;
      AQH_VALUE *recvdValue;
  
      recvdValue=AQH_IpcdMessageMultiData_ReadValue(tagList);
      if (recvdValue) {
        const char *valueName;
        const uint64_t *dataPoints=NULL;
        uint64_t numberOfPoints=0;
  
        valueName=recvdValue?AQH_Value_GetName(recvdValue):NULL;
        AQH_IpcdMessageMultiData_ReadDatapoints(tagList, &dataPoints, &numberOfPoints);
        if (numberOfPoints>0) {
          AQH_VALUE *value;
  
          value=AqHomeDataServer_GetOrCreateValueForDriverWithTemplate(o, ep, recvdValue);
          if (value) {
            if (AQH_Endpoint_GetPermissions(ep) & AQH_ENDPOINT_PERMS_ADDDATA) {
              resultCode=_storeDataPoints(xo, value, dataPoints, numberOfPoints);
              if (resultCode==AQH_MSGDATA_RESULT_SUCCESS)
                _sendDataChangedMsgToAllClients(xo, ep, value, dataPoints, numberOfPoints);
            }
            else {
              DBG_INFO(NULL, "No permissions to add data to value \"%s\"", valueName);
              resultCode=AQH_MSGDATA_RESULT_ERROR_PERMS;
            }
          }
          else {
            DBG_INFO(NULL, "No permissions to add/create value \"%s\"", valueName);
            resultCode=AQH_MSGDATA_RESULT_ERROR_PERMS;
          }
        }
        else {
          DBG_INFO(NULL, "No datapoints");
          resultCode=AQH_MSGDATA_RESULT_ERROR_INVALID;
        }
        AQH_Value_free(recvdValue);
      }
      else {
        DBG_INFO(NULL, "No value");
        resultCode=AQH_MSGDATA_RESULT_ERROR_INVALID;
      }
  
      outMsg=AQH_IpcMessageResult_new(AQH_IPC_PROTOCOL_DATA_ID,
                                      AQH_IPC_PROTOCOL_DATA_VERSION,
                                      AQH_MSGTYPE_IPC_DATA_RESULT,
                                      AQH_Endpoint_GetNextMessageId(ep),
                                      AQH_IpcMessage_GetMsgId(msg),
                                      resultCode, NULL);
      AQH_Endpoint_AddMsgOut(ep, outMsg);
    }
  }
}



int _storeDataPoints(AQHOME_SERVER *xo, const AQH_VALUE *v, const uint64_t *dataPoints, unsigned int numValues)
{
  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(xo->storage, AQH_Value_GetId(v), timestamp, u.f);
    if (rv<0) {
      DBG_INFO(NULL, "here (%d)", rv);
      return AQH_MSGDATA_RESULT_ERROR_GENERIC;
    }
    else {
      DBG_INFO(NULL, "Datapoint added for value \"%s\"", AQH_Value_GetNameForSystem(v));
    }
  } /* for */

  return AQH_MSGDATA_RESULT_SUCCESS;
}



void _sendDataChangedMsgToAllClients(AQHOME_SERVER *xo, AQH_OBJECT *epSrc, const AQH_VALUE *v, const uint64_t *dataPoints, int numValues)
{
  AQH_OBJECT *ep;

  ep=AQH_Object_List_First(xo->tcpClientList);
  while(ep) {
    if (ep!=epSrc) {
      if (AQH_Endpoint_GetFlags(ep) & AQH_ENDPOINT_FLAGS_WANTUPDATES) {
        AQH_MESSAGE *msg;

        DBG_DEBUG(AQH_LOGDOMAIN, "Sending update msg to endpoint");
        msg=AQH_IpcdMessageMultiData_new(AQH_MSGTYPE_IPC_DATA_DATACHANGED,
                                         AQH_Endpoint_GetNextMessageId(ep), 0, 0,
                                         v, dataPoints, numValues);
        AQH_Endpoint_AddMsgOut(ep, msg);
      }
      else {
        DBG_DEBUG(AQH_LOGDOMAIN, "Endpoint doesn't want updates");
      }
    }
    else {
      DBG_DEBUG(AQH_LOGDOMAIN, "Not sending update msg to source of updates");
    }
    ep=AQH_Object_List_Next(ep);
  }
}