arexxd/arexxd.c

/*
 *	Linux Interfece for Arexx Data Loggers
 *
 *	(c) 2011-2012 Martin Mares <mj@ucw.cz>
 */

#include <stdio.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <math.h>
#include <time.h>
#include <getopt.h>
#include <syslog.h>
#include <signal.h>
#include <sys/stat.h>
#include <libusb-1.0/libusb.h>
#include <rrd.h>

#define DEFAULT_LOG_DIR "/var/log/arexxd"

typedef unsigned char byte;
static libusb_context *usb_ctxt;
static libusb_device_handle *devh;

static int use_syslog;
static int debug_mode;
static int debug_packets;
static int debug_raw_data;
static int debug_usb;
static char *log_dir = DEFAULT_LOG_DIR;

static void die(char *fmt, ...)
{
	va_list args;
	va_start(args, fmt);
	if (use_syslog)
		vsyslog(LOG_CRIT, fmt, args);
	else {
		vfprintf(stderr, fmt, args);
		fprintf(stderr, "\n");
	}
	va_end(args);
	exit(1);
}

static void log_error(char *fmt, ...)
{
	va_list args;
	va_start(args, fmt);
	if (use_syslog)
		vsyslog(LOG_ERR, fmt, args);
	else {
		vfprintf(stderr, fmt, args);
		fprintf(stderr, "\n");
	}
	va_end(args);
}

static void log_info(char *fmt, ...)
{
	va_list args;
	va_start(args, fmt);
	if (use_syslog)
		vsyslog(LOG_INFO, fmt, args);
	else {
		vfprintf(stderr, fmt, args);
		fprintf(stderr, "\n");
	}
	va_end(args);
}

static void log_pkt(char *fmt, ...)
{
	if (!debug_packets)
		return;
	va_list args;
	va_start(args, fmt);
	vprintf(fmt, args);
	va_end(args);
}

/*** RRD interface ***/

#define MAX_ARGS 20
#define MAX_ARG_SIZE 1024

static int arg_cnt;
static char *arg_ptr[MAX_ARGS+1];
static char arg_buf[MAX_ARG_SIZE];
static int arg_pos;

static void arg_new(void)
{
	arg_cnt = 1;
	arg_pos = 0;
	arg_ptr[0] = "rrdtool";
}

static void arg_push(const char *fmt, ...)
{
	if (arg_cnt >= MAX_ARGS)
		die("MAX_ARGS exceeded");
	va_list va;
	va_start(va, fmt);
	int len = 1 + vsnprintf(arg_buf + arg_pos, MAX_ARG_SIZE - arg_pos, fmt, va);
	if (arg_pos + len > MAX_ARG_SIZE)
		die("MAX_ARG_SIZE exceeded");
	arg_ptr[arg_cnt++] = arg_buf + arg_pos;
	arg_ptr[arg_cnt] = NULL;
	arg_pos += len;
}

static void rrd_point(time_t t, const char *name, double val, char *unit)
{
	char rr_name[256];
	snprintf(rr_name, sizeof(rr_name), "sensor-%s.rrd", name);

	struct stat st;
	if (stat(rr_name, &st) < 0 || !st.st_size) {
		// We have to create the RRD
		log_info("Creating %s", rr_name);
		arg_new();
		arg_push(rr_name);
		arg_push("--start");
		arg_push("%d", (int) time(NULL) - 28*86400);
		arg_push("--step");
		arg_push("60");
		if (!strcmp(unit, "%RH"))
			arg_push("DS:rh:GAUGE:300:0:100");
		else if (!strcmp(unit, "ppm"))
			arg_push("DS:ppm:GAUGE:300:0:1000000");
		else
			arg_push("DS:temp:GAUGE:300:-200:200");
		arg_push("RRA:AVERAGE:0.25:1:20160");		// Last 14 days with full resolution
		arg_push("RRA:AVERAGE:0.25:60:88800");		// Last 10 years with 1h resolution
		arg_push("RRA:MIN:0.25:60:88800");		// including minima and maxima
		arg_push("RRA:MAX:0.25:60:88800");
		rrd_create(arg_cnt, arg_ptr);
		if (rrd_test_error()) {
			log_error("rrd_create on %s failed: %s", rr_name, rrd_get_error());
			return;
		}
	}

	arg_new();
	arg_push(rr_name);
	arg_push("%d:%f", t, val);
	rrd_update(arg_cnt, arg_ptr);
	if (rrd_test_error())
		log_error("rrd_update on %s failed: %s", rr_name, rrd_get_error());
}

/*** Transforms ***/

#define TIME_OFFSET 946681200		// Timestamp of 2000-01-01 00:00:00

static int data_point_counter;		// Since last log message

static double correct_point(int id, double val, const char **name)
{
	/*
	 *  Manually calculated corrections and renames for my sensors.
	 *  Replace with your formulae.
	 */
	switch (id) {
		case 10415:
			*name = "ursarium";
			return val - 0.93;
		case 10707:
			*name = "balcony";
			return val - 0.71;
		case 19246:
			*name = "catarium";
			return val + 0.49;
		case 19247:
			*name = "catarium-rh";
			return val;
		case 12133:
			*name = "outside";
			return val + 0.44;
		default:
			return val;
	}
}

static void cooked_point(time_t t, int id, double val, char *unit, int q)
{
	char namebuf[16];
	snprintf(namebuf, sizeof(namebuf), "%d", id);
	const char *name = namebuf;

	double val2 = correct_point(id, val, &name);

	if (debug_raw_data) {
		struct tm tm;
		localtime_r(&t, &tm);
		char tbuf[64];
		strftime(tbuf, sizeof(tbuf), "%Y-%m-%d %H:%M:%S", &tm);
		printf("== %s id=%d name=%s val=%.3f val2=%.3f unit=%s q=%d\n", tbuf, id, name, val, val2, unit, q);
	}

	data_point_counter++;
	rrd_point(t, name, val2, unit);
}

static void raw_point(int t, int id, int raw, int q)
{
	/*
	 *  The binary blob provided by Arexx contains an embedded XML fragment
	 *  with descriptions of all known sensor types. If you want to see it,
	 *  grep the blob for "<deviceinfo>". The meanings of the parameters are
	 *  as follows:
	 *
	 *	m1, m2		Device type matches if (raw_sensor_id & m1) == m2
	 *	type		Unit measured by the sensor (1=Celsius, 2=RH%, 3=CO2 ppm)
	 *	dm		User-visible sensor ID = raw_sensor_id & dm
	 *	i		1 if the raw value is signed
	 *	p[]		Coefficients of transformation polynomial (x^0 first)
	 *	vLo, vUp	Upper and lower bound on the final value
	 *	scale		Scaling function:
	 *				0 = identity (default)
	 *				1 = 10^x
	 *				2 = exp(x)
	 *				3 = (x < 0) ? 0 : log10(x)
	 *				4 = (x < 0) ? 0 : log(x)
	 *
	 *  The raw values are transformed this way:
	 *	- sign-extend if signed
	 *	- apply the transformation polynomial
	 *	- apply the scaling function
	 *	- drop if outside the interval [vLo,vUp]
	 *
	 *  This function applies the necessary transform for sensors we've
	 *  seen in the wild. We deliberately ignore the "dm" parameter as we want
	 *  to report different channels of a single sensor as multiple sensors.
	 */

	double z = raw;
	double hi, lo;
	char *unit;
	int idhi = id & 0xf000;

	if (idhi == 0x1000) {
		z = 0.02*z - 273.15;
		lo = -200;
		hi = 600;
		unit = "C";
	} else if (idhi == 0x2000) {
		if (raw >= 0x8000)
			z -= 0x10000;
		z /= 128;
		lo = -60;
		hi = 125;
		unit = "C";
	} else if (idhi == 0x4000) {
		if (!(id & 1)) {
			z = z/100 - 39.6;
			lo = -60;
			hi = 125;
			unit = "C";
		} else {
			z = -2.8e-6*z*z + 0.0405*z - 4;
			lo = 0;
			hi = 100.1;
			unit = "%RH";
		}
	} else if (idhi == 0x6000) {
		if (!(id & 1)) {
			if (raw >= 0x8000)
				z -= 0x10000;
			z /= 128;
			lo = -60;
			hi = 125;
			unit = "C";
		} else {
			z = -3.8123e-11*z;
			z = (z + 1.9184e-7) * z;
			z = (z - 1.0998e-3) * z;
			z += 6.56;
			z = pow(10, z);
			lo = 0;
			hi = 1e6;
			unit = "ppm";
		}
	} else {
		log_error("Unknown sensor type 0x%04x", id);
		return;
	}

	if (z < lo || z > hi) {
		log_error("Sensor %d: value %f out of range", id, z);
		return;
	}

	cooked_point(t + TIME_OFFSET, id, z, unit, q);
}

/*** USB interface ***/

static int rx_endpoint, tx_endpoint;

static int parse_descriptors(libusb_device *dev)
{
	int err;
	struct libusb_config_descriptor *desc;

	if (err = libusb_get_active_config_descriptor(dev, &desc)) {
		log_error("libusb_get_config_descriptor failed: error %d", err);
		return 0;
	}
	if (desc->bNumInterfaces != 1) {
		log_error("Unexpected number of interfaces: %d", desc->bNumInterfaces);
		goto failed;
	}

	const struct libusb_interface *iface = &desc->interface[0];
	if (iface->num_altsetting != 1) {
		log_error("Unexpected number of alternate interface settings: %d", iface->num_altsetting);
		goto failed;
	}

	const struct libusb_interface_descriptor *ifd = &iface->altsetting[0];
	if (ifd->bNumEndpoints != 2) {
		log_error("Unexpected number of endpoints: %d", ifd->bNumEndpoints);
		goto failed;
	}

	rx_endpoint = tx_endpoint = -1;
	for (int i=0; i<2; i++) {
		const struct libusb_endpoint_descriptor *epd = &ifd->endpoint[i];
		if (epd->bEndpointAddress & 0x80)
			rx_endpoint = epd->bEndpointAddress;
		else
			tx_endpoint = epd->bEndpointAddress;
	}
	if (rx_endpoint < 0 || tx_endpoint < 0) {
		log_error("Failed to identify endpoints");
		goto failed;
	}

	log_pkt("Found endpoints: rx==%02x tx=%02x\n", rx_endpoint, tx_endpoint);
	libusb_free_config_descriptor(desc);
	return 1;

failed:
	libusb_free_config_descriptor(desc);
	return 0;
}

static int find_device(void)
{
	libusb_device **devlist;
	ssize_t devn = libusb_get_device_list(usb_ctxt, &devlist);
	if (devn < 0) {
		log_error("Cannot enumerate USB devices: error %d", (int) devn);
		return 0;
	}

	for (ssize_t i=0; i<devn; i++) {
		struct libusb_device_descriptor desc;
		libusb_device *dev = devlist[i];
		if (!libusb_get_device_descriptor(dev, &desc)) {
			if (desc.idVendor == 0x0451 && desc.idProduct == 0x3211) {
				log_info("Arexx data logger found at usb%d.%d", libusb_get_bus_number(dev), libusb_get_device_address(dev));
				if (!parse_descriptors(dev))
					continue;
				int err;
				if (err = libusb_open(dev, &devh)) {
					log_error("libusb_open() failed: error %d", err);
					goto failed;
				}
				if (err = libusb_claim_interface(devh, 0)) {
					log_error("libusb_claim_interface() failed: error %d", err);
					libusb_close(devh);
					goto failed;
				}
				libusb_free_device_list(devlist, 1);
				return 1;
			}
		}
	}

failed:
	libusb_free_device_list(devlist, 1);
	return 0;
}

static void release_device(void)
{
	libusb_release_interface(devh, 0);
	libusb_reset_device(devh);
	libusb_close(devh);
	devh = NULL;
}

static void dump_packet(byte *pkt)
{
	for (int i=0; i<64; i++) {
		if (!(i % 16))
			log_pkt("\t%02x:", i);
		log_pkt(" %02x", pkt[i]);
		if (i % 16 == 15)
			log_pkt("\n");
	}
}

static int send_and_receive(byte *req, byte *reply)
{
	if (debug_packets) {
		time_t t = time(NULL);
		struct tm tm;
		localtime_r(&t, &tm);

		char tbuf[64];
		strftime(tbuf, sizeof(tbuf), "%Y-%m-%d %H:%M:%S", &tm);
		log_pkt("## %s\n", tbuf);
	}

	int err, transferred;
	if (err = libusb_bulk_transfer(devh, tx_endpoint, req, 64, &transferred, 200)) {
		if (err == LIBUSB_ERROR_TIMEOUT) {
			log_pkt(">> xmit timed out\n");
			return 0;
		}
		log_pkt(">> xmit error %d\n", err);
		log_error("Transmit error: %d", err);
		return err;
	}
	if (debug_packets) {
		log_pkt(">> xmit %d bytes\n", transferred);
		dump_packet(req);
	}
	if (err = libusb_bulk_transfer(devh, rx_endpoint, reply, 64, &transferred, 200)) {
		if (err == LIBUSB_ERROR_TIMEOUT) {
			log_pkt("<< recv timed out\n");
			return 0;
		}
		log_pkt("<< recv error %d\n", err);
		log_error("Receive error: %d", err);
		return err;
	}
	if (debug_packets)
		log_pkt("<< recv %d bytes\n", transferred);
	while (transferred < 64)
		reply[transferred++] = 0xff;
	if (debug_packets)
		dump_packet(reply);
	return 1;
}

static unsigned int get_be16(byte *p)
{
	return p[1] | (p[0] << 8);
}

static unsigned int get_le16(byte *p)
{
	return p[0] | (p[1] << 8);
}

static unsigned int get_le32(byte *p)
{
	return get_le16(p) | (get_le16(p+2) << 16);
}

static void put_le16(byte *p, unsigned int x)
{
	p[0] = x;
	p[1] = x >> 8;
}

static void put_le32(byte *p, unsigned int x)
{
	put_le16(p, x);
	put_le16(p+2, x>>16);
}

static int parse_packet(byte *reply)
{
	if (reply[0]) {
		log_error("Unknown packet type %02x", reply[0]);
		return 0;
	}

	int pos = 1;
	int points = 0;
	while (pos < 64) {
		byte *p = reply + pos;
		int len = p[0];
		if (!len || len == 0xff)
			break;
		if (len < 9 || len > 10) {
			log_error("Unknown tuple length %02x", len);
			break;
		}
		if (pos + len > 64) {
			log_error("Tuple truncated");
			break;
		}
		int id = get_le16(p+1);
		int raw = get_be16(p+3);
		int t = get_le32(p+5);
		int q = (len > 9) ? p[9] : -1;
		if (debug_raw_data) {
			printf("... %02x: id=%d raw=%d t=%d", len, id, raw, t);
			if (len > 9)
				printf(" q=%d", q);
			printf("\n");
		}
		raw_point(t, id, raw, q);
		pos += len;
		points++;
	}

	return points;
}

static void set_clock(void)
{
	byte req[64], reply[64];
	memset(req, 0, 64);
	req[0] = 4;
	time_t t = time(NULL);
	put_le32(req+1, t-TIME_OFFSET);
	send_and_receive(req, reply);

#if 0
	/*
	 *  Original software also sends a packet with type 3 and the timestamp,
	 *  but it does not make any sense, especially as they ignore the sensor
	 *  readings in the answer.
	 */
	req[0] = 3;
	send_and_receive(req, reply);
	parse_packet(reply);
#endif
}

/*** Main ***/

static sigset_t term_sigs;
static volatile sig_atomic_t want_shutdown;

static void sigterm_handler(int sig __attribute__((unused)))
{
	want_shutdown = 1;
}

static void interruptible_msleep(int ms)
{
	sigprocmask(SIG_UNBLOCK, &term_sigs, NULL);
	struct timespec ts = { .tv_sec = ms/1000, .tv_nsec = (ms%1000) * 1000000 };
	nanosleep(&ts, NULL);
	sigprocmask(SIG_BLOCK, &term_sigs, NULL);
}

static const struct option long_options[] = {
	{ "debug",		0, NULL, 'd' },
	{ "log-dir",		1, NULL, 'l' },
	{ "debug-packets",	0, NULL, 'p' },
	{ "debug-raw",		0, NULL, 'r' },
	{ NULL,			0, NULL, 0 },
};

static void usage(void)
{
	fprintf(stderr, "\n\
Usage: arexxd <options>\n\
\n\
Options:\n\
-d, --debug		Debug mode (no chdir, no fork, no syslog)\n\
-l, --log-dir=<dir>	Directory where all received data should be stored\n\
-p, --debug-packets	Log all packets sent and received\n\
-r, --debug-raw		Log conversion from raw values\n\
-u, --debug-usb		Enable libusb debug messages (to stdout/stderr)\n\
");
	exit(1);
}

int main(int argc, char **argv)
{
	int opt;
	while ((opt = getopt_long(argc, argv, "dl:pru", long_options, NULL)) >= 0)
		switch (opt) {
			case 'd':
				debug_mode++;
				break;
			case 'l':
				log_dir = optarg;
				break;
			case 'p':
				debug_packets++;
				break;
			case 'r':
				debug_raw_data++;
				break;
			case 'u':
				debug_usb++;
				break;
			default:
				usage();
		}
	if (optind < argc)
		usage();

	int err;
	if (err = libusb_init(&usb_ctxt))
		die("Cannot initialize libusb: error %d", err);
	if (debug_usb)
		libusb_set_debug(usb_ctxt, 3);

	if (!debug_mode) {
		if (chdir(log_dir) < 0)
			die("Cannot change directory to %s: %m", log_dir);
		if (debug_packets || debug_raw_data) {
			close(1);
			if (open("debug", O_WRONLY | O_CREAT | O_APPEND, 0666) < 0)
				die("Cannot open debug log: %m");
			setlinebuf(stdout);
		}
		openlog("arexxd", LOG_NDELAY, LOG_DAEMON);
		pid_t pid = fork();
		if (pid < 0)
			die("fork() failed: %m");
		if (pid)
			return 0;
		setsid();
		use_syslog = 1;
	}

	struct sigaction sa = { .sa_handler = sigterm_handler };
	sigaction(SIGTERM, &sa, NULL);
	sigaction(SIGINT, &sa, NULL);

	sigemptyset(&term_sigs);
	sigaddset(&term_sigs, SIGTERM);
	sigaddset(&term_sigs, SIGINT);
	sigprocmask(SIG_BLOCK, &term_sigs, NULL);

	int inited = 0;
	while (!want_shutdown) {
		if (!find_device()) {
			if (!inited) {
				inited = 1;
				log_error("Data logger not connected, waiting until it appears");
			}
			interruptible_msleep(30000);
			continue;
		}
		log_info("Listening");

		time_t last_sync = 0;
		time_t last_show = 0;
		int want_stats = 0;
		int want_sleep = 0;
		data_point_counter = 0;
		while (!want_shutdown) {
			time_t now = time(NULL);
			if (now > last_sync + 900) {
				log_info("Synchronizing data logger time");
				set_clock();
				last_sync = now;
			}
			if (want_stats && now > last_show + 300) {
				log_info("Stats: received %d data points", data_point_counter);
				data_point_counter = 0;
				last_show = now;
			}

			byte req[64], reply[64];
			memset(req, 0, sizeof(req));
			req[0] = 3;
			err = send_and_receive(req, reply);
			if (err < 0)
				break;
			want_sleep = 1;
			if (err > 0 && parse_packet(reply))
				want_sleep = 0;
			if (want_sleep) {
				interruptible_msleep(4000);
				want_stats = 1;
			} else
				interruptible_msleep(5);
		}

		log_info("Disconnecting data logger");
		release_device();
		inited = 0;
		interruptible_msleep(10000);
	}

	log_info("Terminated");
	return 0;
}