bmsd/nodes.c

/**
 * @file nodes.c
 * @brief Handle nodes status
 * @author Michiel Broek <mbroek at mbse dot eu>
 *
 * Copyright (C) 2018-2022
 *
 * This file is part of the bms (Brewery Management System)
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2, or (at your option) any
 * later version.
 *
 * bms is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with ThermFerm; see the file COPYING.  If not, write to the Free
 * Software Foundation, 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 */


#include "bms.h"
#include "xutil.h"
#include "nodes.h"
#include "mqtt.h"
#include "mysql.h"
#include "websocket.h"


sys_node_list			*nodes = NULL;

extern int			debug;
extern sys_fermenter_list	*fermenters;
extern sys_co2meter_list	*co2meters;
extern sys_ispindel_list	*ispindels;


void node_birth_data(char *topic, char *payload)
{
    struct json_object	*jobj, *val, *metric, *metric2;
    sys_node_list	*node, *tmpp;
    struct tm		*mytime;
    char		*group_id, *message_type, *edge_node, *msg = NULL, buf[74];
    bool		new_node = true;

//    fprintf(stdout, "node_birth: %s %s\n", topic, payload);

    strtok(topic, "/");	// ignore namespace
    group_id = strtok(NULL, "/");
    message_type = strtok(NULL, "/");
    edge_node = strtok(NULL, "/\0");

    /*
     * Search node record in the memory array and use it if found.
     */
    if (nodes) {
	for (tmpp = nodes; tmpp; tmpp = tmpp->next) {
	    if ((strcmp(tmpp->group_id, group_id) == 0) && (strcmp(tmpp->node, edge_node) == 0)) {
		new_node = false;
		node = tmpp;
		node->lastseen = time(NULL);
		node->online = true;
		break;
	    }
	}
    }

    /*
     * Allocate new node if not yet known.
     */
    if (new_node) {
	node = (sys_node_list *)malloc(sizeof(sys_node_list));
    	node->uuid = NULL;
   	node->next = NULL;
    	node->group_id = xstrcpy(group_id);
    	node->node = xstrcpy(edge_node);
    	node->online = true;
    	node->hardwaremake = node->hardwaremodel = node->os = node->os_version = node->firmware = NULL;
    	node->firstseen = node->lastseen = time(NULL);
    	node->temperature = node->humidity = node->barometer = 0.0;
    	node->gps_latitude = node->gps_longitude = node->gps_altitude = 0.0;
	node->net_address = node->net_ifname = node->net_ssid = NULL;
	node->net_rssi = 0;
	node->interval = 300;
    }

    /*
     * Process the JSON formatted payload. 
     * Update only the fields that are found in the payload.
     */
    jobj = json_tokener_parse(payload);

    if (json_object_object_get_ex(jobj, "timestamp", &val)) {
	if (strcmp((char *)"NDATA", message_type)) {
	    node->firstseen = json_object_get_int(val);
	    syslog(LOG_NOTICE, "Online node `%s/%s'", node->group_id, node->node);
	} else {
	    node->lastseen = json_object_get_int(val);
	}
    }

    if (json_object_object_get_ex(jobj, "metric", &metric)) {
	if (json_object_object_get_ex(metric, "uuid", &val)) {
	    if (node->uuid)
		free(node->uuid);
	    node->uuid = xstrcpy((char *)json_object_get_string(val));
	}
	if (json_object_object_get_ex(metric, "interval", &val)) {
	    node->interval = json_object_get_int(val);
	}
	if (json_object_object_get_ex(metric, "properties", &metric2)) {
	    if (json_object_object_get_ex(metric2, "hardwaremake", &val)) {
		if (node->hardwaremake)
		    free(node->hardwaremake);
		node->hardwaremake = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "hardwaremodel", &val)) {
		if (node->hardwaremodel)
		    free(node->hardwaremodel);
		node->hardwaremodel = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "os", &val)) {
		if (node->os)
		    free(node->os);
		node->os = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "os_version", &val)) {
		if (node->os_version)
		    free(node->os_version);
		node->os_version = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "FW", &val)) {
		if (node->firmware)
		    free(node->firmware);
		node->firmware = xstrcpy((char *)json_object_get_string(val));
	    }
	}
	if (json_object_object_get_ex(metric, "THB", &metric2)) {
	    if (json_object_object_get_ex(metric2, "temperature", &val)) {
		node->temperature = json_object_get_double(val);
	    }
	    if (json_object_object_get_ex(metric2, "humidity", &val)) {
		node->humidity= json_object_get_double(val);
	    }
	    if (json_object_object_get_ex(metric2, "barometer", &val)) {
		node->barometer = json_object_get_double(val);
	    }
	}
	if (json_object_object_get_ex(metric, "GPS", &metric2)) {
	    if (json_object_object_get_ex(metric2, "latitude", &val)) {
		node->gps_latitude = json_object_get_double(val);
	    }
	    if (json_object_object_get_ex(metric2, "longitude", &val)) {
		node->gps_longitude = json_object_get_double(val);
	    }
	    if (json_object_object_get_ex(metric2, "altitude", &val)) {
		node->gps_altitude = json_object_get_double(val);
	    }
	}
	if (json_object_object_get_ex(metric, "net", &metric2)) {
	    if (json_object_object_get_ex(metric2, "address", &val)) {
		if (node->net_address)
		    free(node->net_address);
		node->net_address = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "ifname", &val)) {
		if (node->net_ifname)
		    free(node->net_ifname);
		node->net_ifname = xstrcpy((char *)json_object_get_string(val));
	    }
	    if (json_object_object_get_ex(metric2, "ssid", &val)) {
                if (node->net_ssid)
                    free(node->net_ssid);
                node->net_ssid = xstrcpy((char *)json_object_get_string(val));
            }
	    if (json_object_object_get_ex(metric2, "rssi", &val)) {
		node->net_rssi = json_object_get_int(val);
	    }
	}
    }
    json_object_put(jobj);

    msg = xstrcpy((char *)"{\"node\":\"");
    msg = xstrcat(msg, edge_node);
    msg = xstrcat(msg, (char *)"\",\"group_id\":\"");
    msg = xstrcat(msg, group_id);
    msg = xstrcat(msg, (char *)"\",\"online\":");
    msg = xstrcat(msg, node->online ? (char *)"1":(char *)"0");
    msg = xstrcat(msg, (char *)",\"interval\":");
    snprintf(buf, 73, "%d", node->interval);
    msg = xstrcat(msg, buf);
    if (node->hardwaremake) {
	msg = xstrcat(msg, (char *)",\"hardwaremake\":\"");
	msg = xstrcat(msg, node->hardwaremake);
	msg = xstrcat(msg, (char *)"\"");
    }
    if (node->hardwaremodel) {
	msg = xstrcat(msg, (char *)",\"hardwaremodel\":\"");
	msg = xstrcat(msg, node->hardwaremodel);
	msg = xstrcat(msg, (char *)"\"");
    }
    if (node->os) {
	msg = xstrcat(msg, (char *)",\"os\":\"");
	msg = xstrcat(msg, node->os);
	msg = xstrcat(msg, (char *)"\"");
    }
    if (node->os_version) {
	msg = xstrcat(msg, (char *)",\"os_version\":\"");
	msg = xstrcat(msg, node->os_version);
	msg = xstrcat(msg, (char *)"\"");
    }
    if (node->firmware) {
	msg = xstrcat(msg, (char *)",\"firmware\":\"");
	msg = xstrcat(msg, node->firmware);
	msg = xstrcat(msg, (char *)"\"");
    }
    msg = xstrcat(msg, (char *)",\"lastseen\":\"");
    mytime = localtime(&node->lastseen);
    snprintf(buf, 73, "%04d-%02d-%02d %02d:%02d:%02d",
            mytime->tm_year + 1900, mytime->tm_mon + 1, mytime->tm_mday, mytime->tm_hour, mytime->tm_min, mytime->tm_sec);
    msg = xstrcat(msg, buf);
    msg = xstrcat(msg, (char *)"\"");
    if (node->temperature) {
    	msg = xstrcat(msg, (char *)",\"temperature\":");
    	snprintf(buf, 64, "%.1f", node->temperature);
    	msg = xstrcat(msg, buf);
    }
    if (node->humidity) {
    	msg = xstrcat(msg, (char *)",\"humidity\":");
    	snprintf(buf, 64, "%.1f", node->humidity);
    	msg = xstrcat(msg, buf);
    }
    msg = xstrcat(msg, (char *)",\"net_ifname\":\"");
    msg = xstrcat(msg, node->net_ifname);
    msg = xstrcat(msg, (char *)"\",\"net_address\":\"");
    msg = xstrcat(msg, node->net_address);
    msg = xstrcat(msg, (char *)"\"");
    if (node->net_ssid) {
	msg = xstrcat(msg, (char *)",\"net_ssid\":\"");
	msg = xstrcat(msg, node->net_ssid);
	msg = xstrcat(msg, (char *)"\"");
    }
    if (node->net_rssi != 0) {
    	msg = xstrcat(msg, (char *)",\"net_rssi\":");
    	snprintf(buf, 64, "%d", node->net_rssi);
    	msg = xstrcat(msg, buf);
    }
    msg = xstrcat(msg, (char *)"}");
    ws_broadcast(msg);
    free(msg);
    msg = NULL;

//    node_dump(node);

    if (new_node) {
    	if (nodes == NULL) {
	    nodes = node;
	} else {
	    for (tmpp = nodes; tmpp; tmpp = tmpp->next) {
		if (tmpp->next == NULL) {
		    tmpp->next = node;
		    break;
		}
	    }
	}
	node_mysql_insert(node);
    } else {
	node_mysql_update(node);
    }
}


void node_dump(sys_node_list *node)
{
    if (debug) {
	printf("online    %s\n", node->online ? "yes":"no");
    	printf("uuid      %s\n", node->uuid);
    	printf("node      %s\n", node->node);
    	printf("group     %s\n", node->group_id);
    	printf("hw make   %s  model %s\n", node->hardwaremake, node->hardwaremodel);
    	printf("os        %s  version %s\n", node->os, node->os_version);
    	printf("firmware  %s\n", node->firmware);
    	printf("first     %ld   last %ld\n", node->firstseen, node->lastseen);
    	printf("THB       %.2f  %.2f  %.2f\n", node->temperature, node->humidity, node->barometer);
    	printf("GPS       %.5f  %.5f  %.5f\n", node->gps_latitude, node->gps_longitude, node->gps_altitude);
	printf("net       %s:%s\n", node->net_ifname, node->net_address);
	printf("ssid rssi %s %d\n", node->net_ssid, node->net_rssi);
	printf("interval  %d\n", node->interval);
    }
}


void node_death(char *topic)
{
    char		*group_id, *edge_node, *msg;
    sys_node_list	*tmpp;

    strtok(topic, "/"); // ignore namespace
    group_id = strtok(NULL, "/");
    strtok(NULL, "/");	// ignore message_type
    edge_node = strtok(NULL, "/\0");

    syslog(LOG_NOTICE, "Offline node `%s/%s'", group_id, edge_node);
    node_mysql_death(edge_node);

    for (tmpp = nodes; tmpp; tmpp = tmpp->next) {
	if (strcmp(tmpp->node, edge_node) == 0) {
	    tmpp->online = false;
	    msg = xstrcpy((char *)"{\"node\":\"");
    	    msg = xstrcat(msg, edge_node);
    	    msg = xstrcat(msg, (char *)"\",\"group_id\":\"");
    	    msg = xstrcat(msg, group_id);
    	    msg = xstrcat(msg, (char *)"\",\"online\":0}");
    	    ws_broadcast(msg);
    	    free(msg);
    	    msg = NULL;
	    break;
	}
    }
}


void nodes_check_online()
{
    sys_node_list	*tmpn;
    sys_fermenter_list  *tmpf;
    sys_co2meter_list	*tmpc;
    sys_ispindel_list	*tmpi;
    time_t		now = time(NULL);
    char		*msg = NULL;

    for (tmpn = nodes; tmpn; tmpn = tmpn->next) {
//	if (debug)
//	    printf("%-20s online %s  %ld  %d\n", tmpn->node, tmpn->online ? "yes":"no ", tmpn->lastseen, tmpn->interval);
	if (tmpn->online && ((now - tmpn->lastseen) > (tmpn->interval * 2 + 5))) { // 2 times interval + 5 seconds
	    syslog(LOG_NOTICE, "Timeout node `%s/%s' after %ld seconds", tmpn->group_id, tmpn->node, (now - tmpn->lastseen));
	    tmpn->online = false;
	    node_mysql_death(tmpn->node);
	    msg = xstrcpy((char *)"{\"node\":\"");
            msg = xstrcat(msg, tmpn->node);
            msg = xstrcat(msg, (char *)"\",\"group_id\":\"");
            msg = xstrcat(msg, tmpn->group_id);
            msg = xstrcat(msg, (char *)"\",\"online\":0}");
            ws_broadcast(msg);
            free(msg);
            msg = NULL;

	    for (tmpf = fermenters; tmpf; tmpf = tmpf->next) {
            	if (strcmp(tmpf->node, tmpn->node) == 0) {
                    if (tmpf->online) {
                    	syslog(LOG_NOTICE, "Timeout fermenter %s/%s", tmpf->node, tmpf->alias);
                	tmpf->online = false;
			fermenter_mysql_death(tmpf->node, tmpf->alias);
			msg = xstrcpy((char *)"{\"device\":\"fermenters\",\"node\":\"");
            		msg = xstrcat(msg, tmpf->node);
			msg = xstrcat(msg, (char *)"\",\"unit\":\"");
    			msg = xstrcat(msg, tmpf->alias);
            		msg = xstrcat(msg, (char *)"\",\"online\":0}");
            		ws_broadcast(msg);
syslog(LOG_NOTICE, msg);
            		free(msg);
            		msg = NULL;
		    }
            	}
            }

	    for (tmpc = co2meters; tmpc; tmpc = tmpc->next) {
		if (strcmp(tmpc->node, tmpn->node) == 0) {
		    if (tmpc->online) {
			syslog(LOG_NOTICE, "Timeout co2meter %s/%s", tmpc->node, tmpc->alias);
			tmpc->online = false;
			co2meter_mysql_death(tmpc->node, tmpc->alias);
			msg = xstrcpy((char *)"{\"device\":\"co2meters\",\"node\":\"");
                        msg = xstrcat(msg, tmpc->node);
                        msg = xstrcat(msg, (char *)"\",\"unit\":\"");
                        msg = xstrcat(msg, tmpc->alias);
                        msg = xstrcat(msg, (char *)"\",\"online\":0}");
                        ws_broadcast(msg);
syslog(LOG_NOTICE, msg);
                        free(msg);
                        msg = NULL;
		    }
		}
	    }

	    for (tmpi = ispindels; tmpi; tmpi = tmpi->next) {
		if (strcmp(tmpi->node, tmpn->node) == 0) {
		    if (tmpi->online) {
                        syslog(LOG_NOTICE, "Timeout ispindel %s", tmpi->node);
                        tmpi->online = false;
                        ispindel_mysql_death(tmpi->node);
			msg = xstrcpy((char *)"{\"device\":\"ispindels\",\"node\":\"");
                        msg = xstrcat(msg, tmpi->node);
                        msg = xstrcat(msg, (char *)"\",\"unit\":\"");
                        msg = xstrcat(msg, tmpi->alias);
                        msg = xstrcat(msg, (char *)"\",\"online\":0}");
                        ws_broadcast(msg);
syslog(LOG_NOTICE, msg);
                        free(msg);
                        msg = NULL;
                    }
		}
	    }
	}
    }
}


void node_ws_receive(char *payload)
{
    struct json_object  *jobj, *val;
    char                *node = NULL, *group = NULL, *control = NULL, *topic = NULL, *pay = NULL, buf[33];

    syslog(LOG_NOTICE, "node_ws %s", payload);
    /*
     * Process the JSON formatted payload.
     */
    jobj = json_tokener_parse(payload);
    if (json_object_object_get_ex(jobj, "node", &val))
        node = xstrcpy((char *)json_object_get_string(val));
    if (json_object_object_get_ex(jobj, "group_id", &val))
        group = xstrcpy((char *)json_object_get_string(val));
    if (json_object_object_get_ex(jobj, "control", &val))
        control = xstrcpy((char *)json_object_get_string(val));

    if (node && group && control) {

	/*
	 * Prepare MQTT topic
	 */
	topic = xstrcpy((char *)"mbv1.0/");
	topic = xstrcat(topic, group);
	topic = xstrcat(topic, (char *)"/NCMD/");
	topic = xstrcat(topic, node);

	pay = xstrcpy((char *)"{\"timestamp\":");
    	snprintf(buf, 32, "%ld", time(NULL));
    	pay = xstrcat(pay, buf);
    	pay = xstrcat(pay, (char *)",\"metric\":");

	if (strcmp(control, (char *)"reboot") == 0) {
	    pay = xstrcat(pay, (char *)"{\"Node Control/Reboot\":true}}");
	} else if (strcmp(control, (char *)"rebirth") == 0) {
	    pay = xstrcat(pay, (char *)"{\"Node Control/Rebirth\":true}}");
	}
	syslog(LOG_NOTICE, "%s %s", topic, pay);
	mqtt_publish(topic, pay);
    }

    if (node)
        free(node);
    if (group)
	free(group);
    if (control)
    	free(control);
    if (topic)
	free(topic);
    if (pay)
	free(pay);
}