Agronode_feeder/feeder_lib/fiedler_dehtare/fiedler.c

#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <stdarg.h>

#include <curl/curl.h>

#include <confuse.h>


#include "../../status.h"
#include "../../feeder.h"


	cfg_opt_t type_opts[] =//todo: predelat na sensor id
	{
	    CFG_STR((char *)"lib", 0, CFGF_NONE),
	    CFG_INT((char *)"port", 0, CFGF_NONE),
	    CFG_STR((char *)"proto", 0, CFGF_NONE),
	    CFG_STR((char *)"server", 0, CFGF_NONE),
	    CFG_INT((char *)"timeout", 0, CFGF_NONE),
	    CFG_STR_LIST((char *)"devid", 0, CFGF_NONE),
	    CFG_END()
	};
	cfg_opt_t db_opts[] =
	{
	    CFG_STR((char *)"server", 0, CFGF_NONE),
	    CFG_INT((char *)"port", 0, CFGF_NONE),
	    CFG_STR((char *)"uri", 0, CFGF_NONE),
	    CFG_STR((char *)"timezone", 0, CFGF_NONE),
	    CFG_INT((char *)"max_msg", 0, CFGF_NONE),
	    CFG_END()
	};
	cfg_opt_t cfg_opts[] = 
	{
	    CFG_SEC((char *)"db", db_opts, CFGF_MULTI | CFGF_TITLE),
	    CFG_SEC((char *)"type", type_opts, CFGF_MULTI | CFGF_TITLE),
	    CFG_STR((char *)"logfile", 0, CFGF_NONE),
	    CFG_INT((char *)"loglevel", 7, CFGF_NONE),
	    CFG_INT((char *)"serverdelay", 0, CFGF_NONE),
	    CFG_END()
	};



typedef struct import_item_t
{
	int imported;
	char filename[256];
	struct import_item_t * next;
} import_item_t;

struct import_item_t * import_list;

typedef struct measurement_t
{
	unsigned long unit_id;
	unsigned int sensor_id;
	time_t time;
	double value;
	struct measurement_t * next;
} measurement_t;

struct measurement_t * measurement_list;

static CURL *curl_g;
static cfg_t * cfg;

static void (*feederLog) (int priority, const char *fmt, ...);

int setLog(void *func)
{
    feederLog = func;
    return 0;
}

void addToImport(char * filename)
{
	struct import_item_t * item, * last_item;
	
	item = malloc(sizeof(struct import_item_t));
	if (item)
	{
		strncpy(item->filename, filename, 255);
		item->imported = 0;
		item->next = NULL;
		if (import_list == NULL)
			import_list = item;
		else
		{
			last_item = import_list;
			while (last_item->next != NULL)
				last_item = last_item->next;
			last_item->next = item;
		}
	}
}

void freeImportList(void)
{
	struct import_item_t * delete_item, * last_item;

	if (import_list == NULL)
		return;

	last_item = import_list;
	while (last_item->next != NULL)
	{
		delete_item = last_item;
		free(delete_item);
		last_item = last_item->next;
	}
	free(last_item);
	import_list = NULL;
}

static void parseListLine(char * filename)
{
	if (strstr(filename, "imported") == NULL)
	{
		feederLog(LOG_DEBUG, "Adding file to be imported: %s\n", filename);
		addToImport(filename);
	}
}

static size_t ftpWriteList(void * buffer, size_t size, size_t nmemb, void * data)
{
	static char filename[256];
	static size_t pos = 0;
	size_t i;
	void (* line_func)(char *);

	line_func = data;
	for (i = 0; i < size * nmemb; i++)
	{
		if (((char *)buffer)[i] == '\n')
		{
			filename[pos] = '\0';
			if (data != NULL)
				line_func(filename);
			pos = 0;
		}
		else
			filename[pos++] = ((char *)buffer)[i];
	}
//	printf("ftpWrite()\n");
//	printf("Data readen: \n%s \n", (char *)buffer);
	return nmemb * size;
}

void addMeasurement(unsigned long unit_id, unsigned long sensor_id, time_t tp, double value)
{
	struct measurement_t * measurement, * last_measurement;
	
	measurement = malloc(sizeof(struct measurement_t));
	if (measurement)
	{
		measurement->unit_id = unit_id;
		measurement->sensor_id = sensor_id;
		measurement->time = tp;
		measurement->value = value;
		measurement->next = NULL;

		if (measurement_list == NULL)
			measurement_list = measurement;
		else
		{
			last_measurement = measurement_list;	//todo: optimalizaci (pamatovat si posledniho clena), aby nemusel vzdycky prochazet celej list
			while (last_measurement->next != NULL)
				last_measurement = last_measurement->next;
			last_measurement->next = measurement;
		}
	}
}

typedef struct sensor_id_table_t
{
    unsigned long unit_id;
    unsigned long sensor_id[32];
} sensor_id_table_t;

struct sensor_id_table_t sensor_id_table[] =		//todo: ono se to meni podle jednotky a zapojeni, takze by to chtelo do konfiguraku, ktery si tenhle precte
{
    {
	20458,
	{000000000, 540070001, 450100001, 340290000, 
	 540070002, 450100002, 340300001, 340300002,

	 540080001, 540080002, 540080003, 000000000,
	 000000000, 450110001, 450110002, 340310000, 
	 360190000}
    },
    {
	20454,
	{000000000, 540070000, 450100000, 340290000, 
	 340320000, 480050000, 480060000, 000000000,
	 
	 000000000, 000000000, 000000000, 000000000,
	 000000000, 000000000, 000000000, 000000000, 
	 360190000}
    },
    {
	0,
	{00000}
    }
};

static void parseFileLine(char * line)
{
	char * sep, * s;
	int i;

	unsigned long id;
	unsigned int ack_channel;
	time_t ack_time;
	double value;
	struct tm ack_time_tm;
	
	sensor_id_table_t * sid;

	if (line[0] == '#')	//this is comment line, skip it
		return;

	s = line;
	i = 0;
	ack_time = 0;
	ack_channel = 0;
	id = 0;
	while (sep = strchr(s, '\t'), sep != NULL)
	{
		*sep = 0;
		switch (i)
		{
			case 0:
				id = atol(s);
			break;
			case 1:
				ack_channel = atoi(s);
			break;
			case 2:
				sscanf(s, "%d-%d-%d %d:%d:%d", &ack_time_tm.tm_year, &ack_time_tm.tm_mon, &ack_time_tm.tm_mday, &ack_time_tm.tm_hour, &ack_time_tm.tm_min, &ack_time_tm.tm_sec);
				ack_time_tm.tm_year -= 1900;
				ack_time_tm.tm_mon -= 1;
				ack_time = mktime(&ack_time_tm);
			break;
		}
		s = sep + 1;
		i++;
	}
	sep = strchr(s, ',');
	if (sep != NULL)
		*sep = '.';
	value = atof(s);

	sid = &sensor_id_table[0];
	while (sid->unit_id != 0)
	{
	    if (sid->unit_id == id)
	    {
		if (sid->sensor_id[ack_channel] != 0)
		    addMeasurement(id, sid->sensor_id[ack_channel], ack_time, value);
	    }
	    sid++;
	}
	
/*	if (sensor_id_table[ack_channel] != 0)
		addMeasurement(id, sensor_id_table[ack_channel] * 10000, ack_time, value);
*/		
//	printf("Line: id: %ld, channel: %d, time: %ld, value: %0.2f\n", id, ack_channel, ack_time, value);
}

CURLcode ftpImport(CURL * curl, char * path)
{
	CURLcode res;
	struct import_item_t * item;
	char * full_filename;

	res = 0;
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)parseFileLine);
	item = import_list;
	while (item != NULL)
	{
		full_filename = malloc(strlen(path) + strlen(item->filename) + 1);
		strcpy(full_filename, path);
		strcpy(full_filename + strlen(full_filename), item->filename);
		curl_easy_setopt(curl, CURLOPT_URL, full_filename);
		res = curl_easy_perform(curl);
		
		free(full_filename);
		item->imported = 1;
		item = item->next;
	}
	curl_easy_setopt(curl, CURLOPT_URL, path);
	return res;
}

CURLcode ftpMarkImported(CURL * curl)
{
	struct import_item_t * item;
	char req[512];
	char * temp;
	CURLcode res;

	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, NULL);
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, NULL);

	item = import_list;
	while (item != NULL)
	{
		if (item->imported)
		{
			sprintf(req, "RNFR %s", item->filename);
 			curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, req);
			res = curl_easy_perform(curl);
			if (res != 0)
			    feederLog(LOG_ERR, "fiedler: curl RNFR error %d\n", res);
			else
			    feederLog(LOG_DEBUG, "fiedler: Renamed from %s\n", item->filename);

			temp = strchr(item->filename, '.');
			temp[0] = 0;
			strcpy(item->filename + strlen(item->filename), "_imported.txt");
			sprintf(req, "RNTO %s", item->filename);
			curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, req);
			res = curl_easy_perform(curl);
			if (res != 0)
			    feederLog(LOG_ERR, "fiedler: curl RNTO error %d\n", res);
			else
			    feederLog(LOG_DEBUG, "fiedler: Renamed to %s\n", item->filename);
		}
		item = item->next;
	}
	freeImportList();
	return 0;
}

CURLcode ftpList(CURL * curl)
{
	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, ftpWriteList);
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)parseListLine);
	curl_easy_setopt(curl, CURLOPT_VERBOSE, 1L);
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "NLST");
 
	return curl_easy_perform(curl);
}

unsigned int timeout(void * lib_data, int sock, unsigned char * data)
{
	CURLcode res;
	char path[1024];

	if (curl_g)
	{
		strcpy(path, "ftp://fiedler:Yeen0iFa@test.sensors.lesprojekt.cz/DATA_DEHTARE/");
//		strcpy(path, "ftp://fiedler:Yeen0iFa@test.sensors.lesprojekt.cz/temp/");
		curl_easy_setopt(curl_g, CURLOPT_URL, path);

		res = ftpList(curl_g);
		if (import_list == NULL)
		    feederLog(LOG_DEBUG, "fiedler: Nothing to import\n");
		res = ftpImport(curl_g, path);
		res = ftpMarkImported(curl_g);
	}
	return 0;
}



unsigned int
process(void *lib_data, int sock, unsigned char *data,
        unsigned int length, unsigned long long int *id, time_t * tm,
        double *result_array, uint64_t * sensors, unsigned int *type)
{
	struct measurement_t * measurement;

	if (measurement_list == NULL)
		return 0;

	measurement = measurement_list;
	
	result_array[0] = measurement->value;
	sensors[0] = measurement->sensor_id;
	*id = measurement->unit_id;
	*tm = measurement->time;
	*type = VALUES_TYPE_OBS;

	measurement_list = measurement->next;
	free(measurement);

        return 1;
}

int curlDebug(CURL * curl, curl_infotype info, char * string, size_t size, void * data)
{
    if (feederLog != NULL)
    {
	if (info == CURLINFO_TEXT)
	{
	    string[size] = 0;
	    feederLog(LOG_DEBUG, "fiedler: %s", string);
	}
    }
    return 0;
}

int init(void *param)
{
    feederLog = NULL;

    param = param;

    cfg = cfg_init(cfg_opts, CFGF_NONE);
    if (cfg_parse(cfg, "feeder.conf") != CFG_SUCCESS)
    {
        return -1;
    }
    

    curl_global_init(CURL_GLOBAL_DEFAULT);

//    printf("Curl version: %s \n", curl_version());

    curl_g = curl_easy_init();

    curl_easy_setopt(curl_g, CURLOPT_DEBUGFUNCTION, curlDebug);

    import_list = 0;
    measurement_list = 0;

    return 0;
}


/*
to test only:
int main(void)
{
	unsigned long long int id;
	time_t tp;
	double values[64];
	uint64_t sensors[64];
	unsigned int data_type;

//	timeout(NULL, 0, NULL);
	while (process(NULL, 0, NULL, 0, &id, &tp, values, sensors, &data_type) > 0)
	{
		printf("Zaznam: id %lld, %lld, %f\n", id, sensors[0], values[0]);
	}
	
	return 0;
}
*/