aqhomecontrol/aqhome/msg/endpoint_node.c

/****************************************************************************
 * 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 <config.h>
#endif


#include "aqhome/msg/endpoint_node_p.h"
#include "aqhome/msg/msg_node.h"

#include <gwenhywfar/inherit.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/text.h>

#include <fcntl.h>
#include <termios.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <unistd.h>


#define AQH_MSG_ENDPOINT_NODE_NAME            "node"

#define AQH_MSG_ENDPOINT_NODE_BAUDRATE        B19200
#define AQH_MSG_ENDPOINT_NODE_BYTE_MICROSECS  520

#define AQH_MSG_ENDPOINT_NODE_BUFFERSIZE      128



GWEN_INHERIT(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE)




static int _getReadFd(GWEN_MSG_ENDPOINT *ep);
static int _getWriteFd(GWEN_MSG_ENDPOINT *ep);
static int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);
static int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr);

static void GWENHYWFAR_CB _freeData(void *bp, void *p);

static int _startMsg(GWEN_MSG_ENDPOINT *ep);
static int _endMsg(GWEN_MSG_ENDPOINT *ep);
static int _isLineBusy(GWEN_MSG_ENDPOINT *ep);

static int _openDevice(GWEN_MSG_ENDPOINT *ep);
static int _attnLow(GWEN_MSG_ENDPOINT *ep);
static int _attnHigh(GWEN_MSG_ENDPOINT *ep);
static int _isAttnLow(GWEN_MSG_ENDPOINT *ep);





GWEN_MSG_ENDPOINT *AQH_MsgEndpointNode_new(const char *devicePath, int groupId)
{
  GWEN_MSG_ENDPOINT *ep;
  AQH_MSG_ENDPOINT_NODE *xep;
  int fd;

  ep=GWEN_MsgEndpoint_new(AQH_MSG_ENDPOINT_NODE_NAME, groupId);
  GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT_NODE, xep);
  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep, xep, _freeData);

  GWEN_MsgEndpoint_SetHandleReadableFn(ep, _handleReadable);
  GWEN_MsgEndpoint_SetHandleWritableFn(ep, _handleWritable);
  GWEN_MsgEndpoint_SetGetReadFdFn(ep, _getReadFd);
  GWEN_MsgEndpoint_SetGetWriteFdFn(ep, _getWriteFd);

  xep->deviceName=strdup(devicePath);

  fd=_openDevice(ep);
  if (fd<0) {
    DBG_INFO(NULL, "here (%d)", fd);
    GWEN_MsgEndpoint_free(ep);
    return NULL;
  }
  GWEN_MsgEndpoint_SetFd(ep, fd);
  _attnHigh(ep);

  return ep;
}



void _freeData(void *bp, void *p)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=(AQH_MSG_ENDPOINT_NODE*) p;
  GWEN_FREE_OBJECT(xep);
}



uint32_t AQH_NodeEndpoint_GetAcceptedMsgGroups(const GWEN_MSG_ENDPOINT *ep)
{
  const AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep)
    return xep->acceptedMsgGroups;
  return 0;
}



void AQH_NodeEndpoint_SetAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep)
    xep->acceptedMsgGroups=f;
}



void AQH_NodeEndpoint_AddAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep)
    xep->acceptedMsgGroups|=f;
}



void AQH_NodeEndpoint_DelAcceptedMsgGroups(GWEN_MSG_ENDPOINT *ep, uint32_t f)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep)
    xep->acceptedMsgGroups&=~f;
}



int _getReadFd(GWEN_MSG_ENDPOINT *ep)
{
  return GWEN_MsgEndpoint_GetFd(ep);
}



int _getWriteFd(GWEN_MSG_ENDPOINT *ep)
{
  return GWEN_MsgEndpoint_HaveMessageToSend(ep)?GWEN_MsgEndpoint_GetFd(ep):GWEN_ERROR_NO_DATA;
}



int _handleReadable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
{
  int rv;
  uint8_t buffer[AQH_MSG_ENDPOINT_NODE_BUFFERSIZE];
  int len;
  int i;

  DBG_DEBUG(GWEN_LOGDOMAIN, "Reading from endpoint %s", GWEN_MsgEndpoint_GetName(ep));
  do {
    rv=read(GWEN_MsgEndpoint_GetFd(ep), buffer, sizeof(buffer));
  } while( (rv<0) && errno==EINTR);
  if (rv<0) {
    if (errno==EAGAIN || errno==EWOULDBLOCK)
      return GWEN_ERROR_TRY_AGAIN;
    DBG_ERROR(GWEN_LOGDOMAIN, "Error on read(): %s (%d)", strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  else if (rv==0) {
    DBG_ERROR(GWEN_LOGDOMAIN, "EOF met on read()");
    return GWEN_ERROR_IO;
  }
  len=rv;

  for (i=0; i<len; i++) {
    GWEN_MSG *msg;

    msg=GWEN_MsgEndpoint_GetCurrentlyReceivedMsg(ep);
    if (msg==NULL) {
      msg=GWEN_Msg_new(AQH_MSG_ENDPOINT_NODE_BUFFERSIZE);
      GWEN_Msg_SetGroupId(msg, GWEN_MsgEndpoint_GetGroupId(ep));
      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, msg);
    }
    rv=GWEN_Msg_AddByte(msg, buffer[i]);
    if (rv<0) {
      DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
      return rv;
    }
    rv=AQH_NodeMsg_IsMsgComplete(msg);
    if (rv<0) {
      /* invalid message */
      DBG_ERROR(GWEN_LOGDOMAIN, "Invalid message, discarding");
      GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
      rv=GWEN_MsgEndpoint_DiscardInput(ep);
      if (rv<0) {
        DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
        return rv;
      }
    }
    else if (rv>0) {
      /* complete msg received, add to list */
      if (!AQH_NodeMsg_IsChecksumValid(msg)) {
        DBG_ERROR(AQH_LOGDOMAIN, "Invalid checksum, discarding message");
        GWEN_Text_DumpString(GWEN_Msg_GetBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 6);
        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
        rv=GWEN_MsgEndpoint_DiscardInput(ep);
        if (rv<0) {
          DBG_ERROR(GWEN_LOGDOMAIN, "here (%d)", rv);
          return rv;
        }
      }
      else {
        GWEN_Msg_Attach(msg);
        GWEN_MsgEndpoint_SetCurrentlyReceivedMsg(ep, NULL);
        GWEN_MsgEndpoint_AddReceivedMessage(ep, msg);
      }
    }
  } /* for */

  return 0;
}



int _handleWritable(GWEN_MSG_ENDPOINT *ep, GWEN_UNUSED GWEN_MSG_ENDPOINT_MGR *emgr)
{
  GWEN_MSG *msg;

  DBG_DEBUG(GWEN_LOGDOMAIN, "Writing to endpoint %s", GWEN_MsgEndpoint_GetName(ep));
  msg=GWEN_MsgEndpoint_GetFirstSendMessage(ep);
  if (msg) {
    uint8_t pos;
    int remaining;
    int rv;

    pos=GWEN_Msg_GetCurrentPos(msg);
    remaining=GWEN_Msg_GetRemainingBytes(msg);
    if (remaining>0) {
      const uint8_t *buf;
      int fd;

      if (pos==0) {
        /* start new message */
        rv=_isLineBusy(ep);
        if (rv<0 || rv==1) {
          DBG_ERROR(AQH_LOGDOMAIN, "Line busy, not sending");
          usleep(100);
          return 0;
        }
        rv=_startMsg(ep);
        if (rv<0) {
          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
          return rv;
        }
      }

      fd=GWEN_MsgEndpoint_GetFd(ep);
      /* start new message */
      buf=GWEN_Msg_GetBuffer(msg)+pos;
      do {
        rv=write(fd, buf, remaining);
      } while(rv<0 && errno==EINTR);
      if (rv<0) {
        if (errno==EAGAIN || errno==EWOULDBLOCK)
          return GWEN_ERROR_TRY_AGAIN;
        DBG_ERROR(GWEN_LOGDOMAIN, "Error on write(): %s (%d)", strerror(errno), errno);
        return GWEN_ERROR_IO;
      }
      GWEN_Msg_IncCurrentPos(msg, rv);
      if (rv==remaining) {
        /* end current message */
        rv=_endMsg(ep);
        GWEN_Msg_List_Del(msg);
	GWEN_Msg_free(msg);
        if (rv<0) {
          DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
          return rv;
        }
      }
    }
  }
  return 0;
}



int _openDevice(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;
  int fd;
  int status;
  int i;
  struct termios options;
  int rv;
  int m;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  fd=open(xep->deviceName, O_NOCTTY | O_NDELAY | O_RDWR);
  if (fd<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on open(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  rv=tcgetattr(fd, &(xep->previousOptions));
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcgetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  memset(&options, 0, sizeof(options)); /* preset */

  options.c_cflag=CLOCAL | CREAD | CS8;
  options.c_iflag=IGNPAR | IGNBRK;
  options.c_oflag=0;
  options.c_lflag=0;
  cfmakeraw(&options);
  options.c_cc[VTIME]=0;  /* read timeout in deciseconds */
  options.c_cc[VMIN]=0;   /* no minimum number of receive bytes */

  rv=cfsetispeed(&options, AQH_MSG_ENDPOINT_NODE_BAUDRATE);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetispeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  rv=cfsetospeed(&options, AQH_MSG_ENDPOINT_NODE_BAUDRATE);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on cfsetospeed(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }

  rv=tcflush(fd, TCIOFLUSH);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcflush(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }

  rv=tcsetattr(fd, TCSANOW, &options);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on tcsetattr(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }

  return fd;
}



int _startMsg(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep->intendedAttnState==1) {
    int rv;

    rv=_attnLow(ep);
    if (rv<0) {
      DBG_ERROR(AQH_LOGDOMAIN, "here (%d)", rv);
      return rv;
    }
    usleep(AQH_MSG_ENDPOINT_NODE_BYTE_MICROSECS/5);
  }

  return 0;
}



int _endMsg(GWEN_MSG_ENDPOINT *ep)
{
  /* TODO: flush before releasing ATTN */
  _attnHigh(ep);
}



int _isLineBusy(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  if (xep->intendedAttnState==0) {
    /* we pulled the line low ourselves, and because of the circuitry nobody can pull it high */
    return 0;
  }
  return _isAttnLow(ep);
}



int _attnLow(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;
  int status;
  int rv;
  int fd;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  fd=GWEN_MsgEndpoint_GetFd(ep);
  rv=ioctl(fd, TIOCMGET, &status); /* GET the State of MODEM bits in Status */
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  status |= TIOCM_DTR | TIOCM_RTS;        /* clear the DTR pin (cave: signals inverted!) */
  rv=ioctl(fd, TIOCMSET, &status);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  xep->intendedAttnState=0;
  return 0;
}



int _attnHigh(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;
  int status;
  int rv;
  int fd;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  fd=GWEN_MsgEndpoint_GetFd(ep);
  rv=ioctl(fd, TIOCMGET, &status);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  status |= TIOCM_DTR;        /* Set the DTR pin */
  status &= ~ (TIOCM_DTR | TIOCM_RTS);      /* clear the DTR pin (cave: signals inverted!) */
  rv=ioctl(fd, TIOCMSET, &status);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }

  xep->intendedAttnState=1;
  return 0;
}



int _isAttnLow(GWEN_MSG_ENDPOINT *ep)
{
  AQH_MSG_ENDPOINT_NODE *xep;
  int status;
  int rv;
  int fd;

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, AQH_MSG_ENDPOINT_NODE, ep);
  fd=GWEN_MsgEndpoint_GetFd(ep);
  rv=ioctl(fd, TIOCMGET, &status);
  if (rv<0) {
    DBG_ERROR(AQH_LOGDOMAIN, "Error on ioctl(%s): %s (%d)", xep->deviceName, strerror(errno), errno);
    return GWEN_ERROR_IO;
  }
  return (status & TIOCM_CTS)?1:0;
}