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

added tool aqhome-mqttlog.

Browse Source
This commit is contained in:
Martin Preuss
2023-05-24 23:08:28 +02:00
parent f0917064af
commit b4175f4a89
6 changed files with 1107 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/0BUILD
View File

@@ -5,6 +5,7 @@
<subdirs> <subdirs>
aqhomed aqhomed
aqhome-tool aqhome-tool
aqhome-mqttlog
</subdirs> </subdirs>
</gwbuild> </gwbuild>

69
apps/aqhome-mqttlog/0BUILD Normal file
View File

@@ -0,0 +1,69 @@
<?xml?>
<gwbuild>
<target type="Program" name="aqhome-mqttlog" install="$(bindir)" >
<includes type="c" >
$(gwenhywfar_cflags)
-I$(topsrcdir)
-I$(topbuilddir)
-I$(builddir)
-I$(srcdir)
</includes>
<includes type="tm2" >
--include=$(builddir)
--include=$(srcdir)
</includes>
<setVar name="local/cflags">$(visibility_cflags)</setVar>
<setVar name="tm2flags" >
</setVar>
<setVar name="local/typefiles" >
itemvar.t2d
item.t2d
</setVar>
<setVar name="local/built_sources" >
</setVar>
<setVar name="local/built_headers_pub">
</setVar>
<setVar name="local/built_headers_priv" >
</setVar>
<headers dist="true" >
</headers>
<sources>
$(local/typefiles)
main.c
</sources>
<useTargets>
aqhome
</useTargets>
<libraries>
$(gwenhywfar_libs)
</libraries>
<subdirs>
</subdirs>
<extradist>
</extradist>
</target>
</gwbuild>

75
apps/aqhome-mqttlog/item.t2d Normal file
View File

@@ -0,0 +1,75 @@
<?xml?>
<tm2>
<type id="ITEM" type="pointer">
<descr>
</descr>
<lang id="c">
<identifier>ITEM</identifier>
<prefix>Item</prefix>
<baseFileName>item</baseFileName>
<flags>
with_db
with_list1
</flags>
<headers>
<header type="sys" loc="pre">aqhome/api.h</header>
<header type="local" loc="pre">itemvar.h</header>
</headers>
<inlines>
</inlines>
</lang>
<members>
<member name="id" type="char_ptr" maxlen="64">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
<member name="title" type="char_ptr" maxlen="64">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
<member name="topic" type="char_ptr" maxlen="64">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
<member name="dataType" type="char_ptr" maxlen="16">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
<member name="varList" type="ITEMVAR_LIST" elementName="var">
<default>NULL</default>
<preset>ItemVar_List_new()</preset>
<access>public</access>
<flags>own</flags>
<setflags>nodup</setflags>
<getflags>none</getflags>
</member>
</members>
</type>
</tm2>

50
apps/aqhome-mqttlog/itemvar.t2d Normal file
View File

@@ -0,0 +1,50 @@
<?xml?>
<tm2>
<type id="ITEMVAR" type="pointer">
<descr>
</descr>
<lang id="c">
<identifier>ITEMVAR</identifier>
<prefix>ItemVar</prefix>
<baseFileName>itemvar</baseFileName>
<flags>
with_db
with_list1
</flags>
<headers>
<header type="sys" loc="pre">aqhome/api.h</header>
</headers>
<inlines>
</inlines>
</lang>
<members>
<member name="name" type="char_ptr" maxlen="64">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
<member name="path" type="char_ptr" maxlen="64">
<access>public</access>
<flags>own</flags>
<setflags>const dup</setflags>
<getflags>const</getflags>
</member>
</members>
</type>
</tm2>

875
apps/aqhome-mqttlog/main.c Normal file
View File

@@ -0,0 +1,875 @@
/****************************************************************************
* 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 "./item.h"
#include <aqhome/api.h>
#include <aqhome/aqhome.h>
#include <aqhome/msgmanager.h>
#include <aqhome/msg/endpoint_log.h>
#include <aqhome/mqtt/endpoint_mqttc.h>
#include "aqhome/mqtt/endpoint_mqttc.h"
#include <aqhome/mqtt/msg_mqtt_connect.h>
#include <aqhome/mqtt/msg_mqtt_connack.h>
#include <aqhome/mqtt/msg_mqtt_publish.h>
#include <aqhome/mqtt/msg_mqtt_subscribe.h>
#include <aqhome/mqtt/msg_mqtt_suback.h>
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/endpoint_tcpc.h>
#include <gwenhywfar/endpoint_connectable.h>
#include <gwenhywfar/args.h>
#include <gwenhywfar/logger.h>
#include <gwenhywfar/db.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/cgui.h>
#include <gwenhywfar/text.h>
#include <gwenhywfar/json_read.h>
#include <gwenhywfar/directory.h>
#ifdef HAVE_SIGNAL_H
# include <signal.h>
#endif
#ifdef HAVE_SYS_TYPES_H
# include <sys/types.h>
#endif
#ifdef HAVE_SYS_STAT_H
# include <sys/stat.h>
#endif
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <time.h>
#define I18N(msg) msg
#define I18S(msg) msg
//#define FULL_DEBUG
static int _serve(GWEN_DB_NODE *dbArgs);
static GWEN_MSG_ENDPOINT *_createMqttEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_DB_NODE *dbArgs);
static int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs);
static int _createPidFile(const char *pidFilename);
static ITEM_LIST *_readItems(GWEN_DB_NODE *dbArgs);
static const ITEM *_getItemForTopic(const ITEM_LIST *itemList, const char *topic);
int _subscribe(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp, const char *topicFilter);
int _mqttConnect(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp);
GWEN_MSG *_awaitPacket(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp, uint8_t expectedPacketType);
void _handlePublishMsg(const char *baseFolder, const ITEM_LIST *itemList, const GWEN_MSG *msg);
void _handlePublish(const char *baseFolder, const ITEM_LIST *itemList, const char *topic, const char *value);
void _handleRawMsgForItem(const char *baseFolder, const ITEM *item, const char *value);
void _handleJsonMsgForItem(const char *baseFolder, const ITEM *item, const char *value);
void _handleJsonItemVar(const char *baseFolder, const ITEM *item, const ITEMVAR *itemVar, GWEN_JSON_ELEM *jeRoot);
void _writeValueAccordingToItem(const char *baseFolder, const ITEM *item, const ITEMVAR *itemVar, const char *value);
void _writeToFile(const char *filename, const char *txt);
#ifdef HAVE_SIGNAL_H
static int _setSignalHandlers(void);
static int _setupSigAction(struct sigaction *sa, int sig);
static void _signalHandler(int s);
static struct sigaction saINT,saTERM, saHUP, saTSTP, saCONT;
#endif
static int stopService=0;
int main(int argc, char **argv)
{
GWEN_DB_NODE *dbArgs;
int rv;
GWEN_GUI *gui;
const char *s;
rv=GWEN_Init();
if (rv) {
fprintf(stderr, "ERROR: Unable to init Gwen.\n");
return 2;
}
GWEN_Logger_Open(0, "aqhome-mqttlog", 0, GWEN_LoggerType_Console, GWEN_LoggerFacility_User);
GWEN_Logger_SetLevel(0, GWEN_LoggerLevel_Warning);
/*GWEN_Logger_SetLevel(0, GWEN_LoggerLevel_Info);*/
rv=AQH_Init();
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return 2;
}
dbArgs=GWEN_DB_Group_new("arguments");
rv=_readArgs(argc, argv, dbArgs);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return 2;
}
else if (rv==1) {
DBG_INFO(NULL, "Help printed, done");
return 0;
}
gui=GWEN_Gui_CGui_new();
s=GWEN_DB_GetCharValue(dbArgs, "charset", 0, NULL);
if (s && *s)
GWEN_Gui_SetCharSet(gui, s);
GWEN_Gui_SetGui(gui);
rv=_serve(dbArgs);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return 2;
}
GWEN_DB_Group_free(dbArgs);
GWEN_Gui_SetGui(NULL);
GWEN_Gui_free(gui);
return 0;
}
int _serve(GWEN_DB_NODE *dbArgs)
{
const char *pidFile;
GWEN_MSG_ENDPOINT_MGR *emgr;
GWEN_MSG_ENDPOINT *epTcp;
ITEM_LIST *itemList;
int rv;
int timeout;
time_t startTime;
const char *baseFolder;
startTime=time(NULL);
itemList=_readItems(dbArgs);
if (itemList==NULL) {
DBG_ERROR(NULL, "No items to listen to, aborting.");
return GWEN_ERROR_GENERIC;
}
rv=_setSignalHandlers();
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
baseFolder=GWEN_DB_GetCharValue(dbArgs, "writeToFolder", 0, "/tmp/aqhome");
timeout=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 0);
pidFile=GWEN_DB_GetCharValue(dbArgs, "pidfile", 0, "aqhome-mqttlog.pid");
if (pidFile && *pidFile) {
rv=_createPidFile(pidFile);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
}
emgr=GWEN_MsgEndpointMgr_new();
epTcp=_createMqttEndpoint(emgr, dbArgs);
if (epTcp==NULL) {
DBG_INFO(NULL, "here");
GWEN_MsgEndpointMgr_free(emgr);
return GWEN_ERROR_GENERIC;
}
rv=_mqttConnect(emgr, epTcp);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
GWEN_MsgEndpointMgr_free(emgr);
return rv;
}
rv=_subscribe(emgr, epTcp, "#");
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
GWEN_MsgEndpointMgr_free(emgr);
return rv;
}
while(!stopService) {
DBG_DEBUG(NULL, "Next loop");
GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
GWEN_MsgEndpointMgr_RunAllEndpoints(emgr);
if (GWEN_ConnectableMsgEndpoint_GetState(epTcp)<GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED) {
DBG_INFO(NULL, "Not connected...");
}
else {
GWEN_MSG *msg;
msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp);
if (msg) {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(AQH_MQTTMSG_MSGTYPE_PUBLISH & 0xf0)) {
#ifdef FULL_DEBUG
GWEN_BUFFER *buf;
buf=GWEN_Buffer_new(0, 256, 0, 1);
AQH_PublishMqttMsg_DumpToBuffer(msg, buf, "received");
fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(buf));
GWEN_Buffer_free(buf);
#endif
_handlePublishMsg(baseFolder, itemList, msg);
}
else {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
}
GWEN_Msg_free(msg);
}
}
if (timeout) {
time_t now;
now=time(NULL);
if ((now-startTime)>timeout) {
DBG_INFO(NULL, "Timeout, stopping service");
break;
}
}
}
if (pidFile && *pidFile)
remove(pidFile);
GWEN_MsgEndpointMgr_free(emgr);
return 0;
}
void _handlePublishMsg(const char *baseFolder, const ITEM_LIST *itemList, const GWEN_MSG *msg)
{
char *topic;
char *value;
topic=AQH_PublishMqttMsg_ExtractTopic(msg);
value=AQH_PublishMqttMsg_ExtractValue(msg);
if (topic && value)
_handlePublish(baseFolder, itemList, topic, value);
else {
DBG_ERROR(NULL, "Either topic or value missing in PUBLISH msg");
}
free(value);
free(topic);
}
void _handlePublish(const char *baseFolder, const ITEM_LIST *itemList, const char *topic, const char *value)
{
const ITEM *item;
item=_getItemForTopic(itemList, topic);
if (item) {
const char *t;
DBG_INFO(NULL, "HANDLING topic \"%s\"", topic);
t=Item_GetDataType(item);
if (t && strcasecmp(t, "json")==0)
_handleJsonMsgForItem(baseFolder, item, value);
else
_handleRawMsgForItem(baseFolder, item, value);
}
else {
DBG_INFO(NULL, "ignoring topic \"%s\"", topic);
}
}
void _handleJsonMsgForItem(const char *baseFolder, const ITEM *item, const char *value)
{
GWEN_JSON_ELEM *jeRoot;
jeRoot=GWEN_JsonElement_fromString(value);
if (jeRoot==NULL) {
DBG_INFO(NULL, "Could not parse JSON value: %s", value);
}
else {
const ITEMVAR_LIST *itemVarList;
itemVarList=Item_GetVarList(item);
if (itemVarList) {
ITEMVAR *itemVar;
itemVar=ItemVar_List_First(itemVarList);
while(itemVar) {
_handleJsonItemVar(baseFolder, item, itemVar, jeRoot);
itemVar=ItemVar_List_Next(itemVar);
}
}
GWEN_JsonElement_free(jeRoot);
}
}
void _handleJsonItemVar(const char *baseFolder, const ITEM *item, const ITEMVAR *itemVar, GWEN_JSON_ELEM *jeRoot)
{
const char *path;
path=ItemVar_GetPath(itemVar);
if (path) {
GWEN_JSON_ELEM *je;
je=GWEN_JsonElement_GetElementByPath(jeRoot, path, 0);
if (je) {
const char *s;
s=GWEN_JsonElement_GetData(je);
if (s && *s)
_writeValueAccordingToItem(baseFolder, item, itemVar, s);
else {
DBG_ERROR(NULL, "Path \"%s\" in JSON data contains no data", path);
}
}
else {
DBG_ERROR(NULL, "Path \"%s\" not found in JSON data", path);
}
}
}
void _handleRawMsgForItem(const char *baseFolder, const ITEM *item, const char *value)
{
const ITEMVAR_LIST *itemVarList;
itemVarList=Item_GetVarList(item);
if (itemVarList) {
ITEMVAR *itemVar;
itemVar=ItemVar_List_First(itemVarList);
while(itemVar) {
_writeValueAccordingToItem(baseFolder, item, itemVar, value);
itemVar=ItemVar_List_Next(itemVar);
}
}
}
void _writeValueAccordingToItem(const char *baseFolder, const ITEM *item, const ITEMVAR *itemVar, const char *value)
{
const char *id;
const char *name;
id=Item_GetId(item);
name=ItemVar_GetName(itemVar);
if (id && *id && name && *name) {
GWEN_BUFFER *fnameBuf;
fnameBuf=GWEN_Buffer_new(0, 256, 0, 1);
GWEN_Buffer_AppendString(fnameBuf, baseFolder);
GWEN_Buffer_AppendString(fnameBuf, GWEN_DIR_SEPARATOR_S);
GWEN_Buffer_AppendArgs(fnameBuf, "%s_%s", id, name);
_writeToFile(GWEN_Buffer_GetStart(fnameBuf), value);
GWEN_Buffer_free(fnameBuf);
}
else {
DBG_ERROR(NULL, "Either id or name missing in item list file");
}
}
void _writeToFile(const char *filename, const char *txt)
{
if (txt && *txt) {
GWEN_BUFFER *tmpNameBuf;
int rv;
tmpNameBuf=GWEN_Buffer_new(0, 256, 0, 1);
GWEN_Buffer_AppendString(tmpNameBuf, filename);
GWEN_Buffer_AppendString(tmpNameBuf, ".tmp");
rv=GWEN_Directory_GetPath(GWEN_Buffer_GetStart(tmpNameBuf), GWEN_PATH_FLAGS_VARIABLE);
if (rv<0) {
DBG_INFO(AQH_LOGDOMAIN, "Error getting path for %s (%d)", GWEN_Buffer_GetStart(tmpNameBuf), rv);
}
else {
FILE *f;
f=fopen(GWEN_Buffer_GetStart(tmpNameBuf), "w");
if (f) {
if (1!=fwrite(txt, strlen(txt), 1, f)) {
DBG_ERROR(AQH_LOGDOMAIN, "Error writing.");
fclose(f);
}
else {
fclose(f);
rename(GWEN_Buffer_GetStart(tmpNameBuf), filename);
}
}
}
GWEN_Buffer_free(tmpNameBuf);
}
}
const ITEM *_getItemForTopic(const ITEM_LIST *itemList, const char *topic)
{
const ITEM *item;
item=Item_List_First(itemList);
while(item) {
const char *s;
s=Item_GetTopic(item);
if (s && GWEN_Text_ComparePattern(topic, s, 0)!=-1)
return item;
item=Item_List_Next(item);
}
return NULL;
}
int _subscribe(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp, const char *topicFilter)
{
uint16_t pckId;
GWEN_MSG *msgOut;
GWEN_MSG *msgIn;
DBG_INFO(NULL, "Sending SUBSCRIBE %s", topicFilter);
pckId=AQH_MqttClientEndpoint_GetNextPacketId(epTcp);
msgOut=GWEN_SubscribeMqttMsg_new(AQH_MQTTMSG_MSGTYPE_SUBSCRIBE, pckId, topicFilter, 0);
if (msgOut==NULL) {
DBG_ERROR(NULL, "Error creating message");
return GWEN_ERROR_INTERNAL;
}
DBG_ERROR(NULL, "Sending this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msgOut), GWEN_Msg_GetBytesInBuffer(msgOut), 2);
GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut);
DBG_INFO(NULL, "Waiting for response");
msgIn=_awaitPacket(emgr, epTcp, AQH_MQTTMSG_MSGTYPE_SUBACK);
if (msgIn) {
GWEN_BUFFER *buf;
buf=GWEN_Buffer_new(0, 256, 0, 1);
AQH_SubAckMqttMsg_DumpToBuffer(msgIn, buf, "received");
DBG_INFO(NULL, "%s", GWEN_Buffer_GetStart(buf));
GWEN_Buffer_free(buf);
GWEN_Msg_free(msgIn);
}
return 0;
}
GWEN_MSG_ENDPOINT *_createMqttEndpoint(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_DB_NODE *dbArgs)
{
const char *mqttAddress;
int mqttPort;
const char *mqttClientId;
const char *mqttTopicPrefix;
int mqttKeepAlive;
mqttAddress=GWEN_DB_GetCharValue(dbArgs, "mqttAddress", 0, NULL);
mqttPort=GWEN_DB_GetIntValue(dbArgs, "mqttPort", 0, 1883);
mqttClientId=GWEN_DB_GetCharValue(dbArgs, "mqttClientId", 0, "aqhome-mqttlog");
mqttTopicPrefix=GWEN_DB_GetCharValue(dbArgs, "mqttTopicPrefix", 0, "aqhome/sensors");
mqttKeepAlive=GWEN_DB_GetIntValue(dbArgs, "mqttKeepAlive", 0, 600);
if (mqttAddress && *mqttAddress && mqttPort) {
GWEN_MSG_ENDPOINT *epMqtt;
DBG_INFO(AQH_LOGDOMAIN, "Connecting to %s (port %d)", mqttAddress, mqttPort);
epMqtt=AQH_MqttClientEndpoint_new(mqttAddress, mqttPort, NULL, AQH_MSGMGR_ENDPOINTGROUP_MQTT);
if (epMqtt==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP");
return NULL;
}
GWEN_MsgEndpoint_SetAcceptedGroupIds(epMqtt, AQH_MSGMGR_ENDPOINTGROUP_NODE | AQH_MSGMGR_ENDPOINTGROUP_MQTT);
if (mqttClientId && *mqttClientId)
AQH_MqttClientEndpoint_SetClientId(epMqtt, mqttClientId);
if (mqttTopicPrefix && *mqttTopicPrefix)
AQH_MqttClientEndpoint_SetTopicPrefix(epMqtt, mqttTopicPrefix);
AQH_MqttClientEndpoint_SetKeepAliveTime(epMqtt, mqttKeepAlive);
GWEN_MsgEndpointMgr_AddEndpoint(emgr, epMqtt);
return epMqtt;
}
return NULL;
}
int _setSignalHandlers(void)
{
#ifdef HAVE_SIGNAL_H
int rv;
rv=_setupSigAction(&saINT, SIGINT);
if (rv)
return rv;
rv=_setupSigAction(&saTERM, SIGTERM);
if (rv)
return rv;
rv=_setupSigAction(&saHUP, SIGHUP);
if (rv)
return rv;
# ifdef SIGTSTP
rv=_setupSigAction(&saTSTP, SIGTSTP);
if (rv)
return rv;
# endif
# ifdef SIGCONT
rv=_setupSigAction(&saCONT, SIGCONT);
if (rv)
return rv;
# endif
#endif
return 0;
}
int _setupSigAction(struct sigaction *sa, int sig)
{
sa->sa_handler=_signalHandler;
sigemptyset(&sa->sa_mask);
sa->sa_flags=0;
if (sigaction(sig, sa, 0)) {
DBG_ERROR(NULL, "Could not setup signal handler for signal %d", sig);
return GWEN_ERROR_IO;
}
return 0;
}
void _signalHandler(int s)
{
switch(s) {
case SIGINT:
case SIGTERM:
case SIGHUP:
DBG_WARN(0, "Received signal %d, stopping service in next loop.",s);
stopService=1;
break;
default:
DBG_WARN(0, "Unknown signal %d",s);
break;
}
}
int _createPidFile(const char *pidFilename)
{
FILE *f;
int pidfd;
if (remove(pidFilename)==0) {
DBG_ERROR(0, "Old PID file existed, removed. (Unclean shutdown?)");
}
#ifdef HAVE_SYS_STAT_H
pidfd = open(pidFilename, O_EXCL|O_CREAT|O_WRONLY, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
if (pidfd < 0) {
DBG_ERROR(NULL, "Could not create PID file \"%s\" (%s), aborting.", pidFilename, strerror(errno));
return GWEN_ERROR_IO;
}
f = fdopen(pidfd, "w");
#else /* HAVE_STAT_H */
f=fopen(pidFilename,"w+");
#endif /* HAVE_STAT_H */
/* write pid */
#ifdef HAVE_GETPID
fprintf(f,"%d\n",getpid());
#else
fprintf(f,"-1\n");
#endif
if (fclose(f)) {
DBG_ERROR(0, "Could not close PID file \"%s\" (%s), aborting.", pidFilename, strerror(errno));
return GWEN_ERROR_IO;
}
return 0;
}
int _mqttConnect(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp)
{
while(GWEN_ConnectableMsgEndpoint_GetState(epTcp)<GWEN_ENDPOINT_MQTTC_STATE_ESTABLISHED) {
DBG_DEBUG(AQH_LOGDOMAIN, "Next loop");
GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
GWEN_MsgEndpointMgr_RunAllEndpoints(emgr);
}
return 0;
}
GWEN_MSG *_awaitPacket(GWEN_MSG_ENDPOINT_MGR *emgr, GWEN_MSG_ENDPOINT *epTcp, uint8_t expectedPacketType)
{
while(!stopService) {
GWEN_MSG *msg;
DBG_DEBUG(AQH_LOGDOMAIN, "Next loop");
//GWEN_MsgManager_LoopOnce(emgr);
GWEN_MsgEndpointMgr_IoLoopOnce(emgr);
msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp);
if (msg) {
if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(expectedPacketType & 0xf0)) {
return msg;
}
else {
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
}
}
} /* for */
return NULL;
}
ITEM_LIST *_readItems(GWEN_DB_NODE *dbArgs)
{
const char *itemFile;
itemFile=GWEN_DB_GetCharValue(dbArgs, "itemfile", 0, NULL);
if (itemFile && *itemFile) {
ITEM_LIST *itemList;
GWEN_DB_NODE *dbItemList;
GWEN_DB_NODE *dbItem;
int rv;
dbItemList=GWEN_DB_Group_new("items");
rv=GWEN_DB_ReadFile(dbItemList, itemFile,
GWEN_DB_FLAGS_DEFAULT |
GWEN_PATH_FLAGS_CREATE_GROUP |
GWEN_DB_FLAGS_ALLOW_EMPTY_STREAM);
if (rv<0) {
DBG_ERROR(NULL, "Error reading item file \"%s\" (%d)", itemFile, rv);
GWEN_DB_Group_free(dbItemList);
return NULL;
}
itemList=Item_List_new();
dbItem=GWEN_DB_FindFirstGroup(dbItemList, "item");
while(dbItem) {
ITEM *item;
item=Item_fromDb(dbItem);
Item_List_Add(item, itemList);
dbItem=GWEN_DB_FindNextGroup(dbItem, "item");
}
GWEN_DB_Group_free(dbItemList);
if (Item_List_GetCount(itemList)==0) {
DBG_INFO(NULL, "No items in file");
Item_List_free(itemList);
return NULL;
}
return itemList;
}
return NULL;
}
int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs)
{
int rv;
const GWEN_ARGS args[]= {
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"cfgdir", /* name */
0, /* minnum */
1, /* maxnum */
"D", /* short option */
"cfgdir", /* long option */
I18S("Specify the configuration folder"),
I18S("Specify the configuration folder")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"charset", /* name */
0, /* minnum */
1, /* maxnum */
0, /* short option */
"charset", /* long option */
I18S("Specify the output character set"), /* short description */
I18S("Specify the output character set") /* long description */
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttAddress", /* name */
0, /* minnum */
1, /* maxnum */
"ma", /* short option */
"mqttaddress", /* long option */
I18S("Specify the address of the MQTT server to connect to (disabled if missing)"),
I18S("Specify the address of the MQTT server to connect to (disabled if missing)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttPort", /* name */
0, /* minnum */
1, /* maxnum */
"mp", /* short option */
"mqttport", /* long option */
I18S("Specify the port of the MQTT server (default: 1883)"),
I18S("Specify the port of the MQTT server (default: 1883)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttClientId", /* name */
0, /* minnum */
1, /* maxnum */
NULL, /* short option */
"mqttclientid", /* long option */
I18S("Specify client id for the MQTT server (default: \"aqhomed\")"),
I18S("Specify client id for the MQTT server (default: \"aqhomed\")")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttKeepAlive", /* name */
0, /* minnum */
1, /* maxnum */
"mk", /* short option */
"mqttkeepalive", /* long option */
I18S("Specify keepalive time in seconds (defaults: 600)"),
I18S("Specify keepalive time in seconds (defaults: 600)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"writeToFolder", /* name */
0, /* minnum */
1, /* maxnum */
"W", /* short option */
NULL, /* long option */
I18S("Specify folder to write received values to"),
I18S("Specify folder to write received values to")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"pidfile", /* name */
0, /* minnum */
1, /* maxnum */
"p", /* short option */
"pidfile", /* long option */
I18S("Specify the PID file"),
I18S("Specify the PID file")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"itemfile", /* name */
0, /* minnum */
1, /* maxnum */
"i", /* short option */
"itemfile", /* long option */
I18S("Specify the item definitions file"),
I18S("Specify the item definitions file")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"timeout", /* name */
0, /* minnum */
1, /* maxnum */
"T", /* short option */
"timeout", /* long option */
I18S("Specify timeout in second (default: no timeout)"),
I18S("Specify timeout in second (default: no timeout)")
},
{
GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST, /* flags */
GWEN_ArgsType_Int, /* type */
"help", /* name */
0, /* minnum */
0, /* maxnum */
"h", /* short option */
"help",
I18S("Show this help screen."),
I18S("Show this help screen.")
}
};
rv=GWEN_Args_Check(argc, argv, 1, 0, args, dbArgs);
if (rv==GWEN_ARGS_RESULT_ERROR) {
fprintf(stderr, "ERROR: Could not parse arguments main\n");
return GWEN_ERROR_INVALID;
}
else if (rv==GWEN_ARGS_RESULT_HELP) {
GWEN_BUFFER *ubuf;
ubuf=GWEN_Buffer_new(0, 1024, 0, 1);
GWEN_Buffer_AppendArgs(ubuf,
I18N("This is version %s.\nUsage: %s [OPTIONS]\n\nOptions:\n"),
AQHOME_VERSION_STRING,
argv[0]);
if (GWEN_Args_Usage(args, ubuf, GWEN_ArgsOutType_Txt)) {
fprintf(stderr, "ERROR: Could not create help string\n");
return 1;
}
GWEN_Buffer_AppendString(ubuf, "\n");
fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(ubuf));
GWEN_Buffer_free(ubuf);
return GWEN_ERROR_CLOSE;
}
return 0;
}

37
apps/aqhome-mqttlog/mqttlog.conf Normal file
View File

@@ -0,0 +1,37 @@
item {
id="stube_plug01"
title="Stube Heimkinosystem"
topic="tele/tasmota/plug01/SENSOR"
dataType="json"
varList {
var {
name="powerusage"
path="ENERGY/POWER"
}
var {
name="totalenergy"
path="ENERGY/TOTAL"
}
}
}
item {
id="buero_rack_temp"
title="Serverrack im Buero"
topic="aqhome/sensors/75766d68/1/temperature"
dataType="simple"
varList {
var {
name="temp"
path="value"
}
}
}