2 * Linux Interfece for Arexx Data Loggers
4 * (c) 2011-2018 Martin Mares <mj@ucw.cz>
19 #include <libusb-1.0/libusb.h>
21 #define DEFAULT_LOG_DIR "/var/log/arexxd"
24 * Data points received from the logger are sometimes corrupted by noise.
25 * This effects not only the measured values, but also sensor IDs and timestamps.
26 * Since rrdtool cannot skip back in time, a random timestamp in the future can
27 * cause all further measurements to be dropped. To minimize impact of these
28 * problems, we drop data points which are too far in the past or in the future.
30 * Furthermore, you can ignore data from unrecognized sensors, i.e., those
31 * which are not handled by correct_point().
33 #define MAX_PAST_TIME 30*86400
34 #define MAX_FUTURE_TIME 300
35 #define IGNORE_UNKNOWN_SENSORS
37 typedef unsigned char byte;
38 typedef unsigned int uint;
39 static libusb_context *usb_ctxt;
40 static libusb_device_handle *devh;
42 static int use_syslog;
43 static int debug_mode;
44 static int debug_packets;
45 static int debug_raw_data;
47 static char *log_dir = DEFAULT_LOG_DIR;
50 #define UNUSED __attribute__((unused))
52 static int data_point_counter; // Since last log message
53 static time_t packet_rx_time;
55 static void die(char *fmt, ...)
60 vsyslog(LOG_CRIT, fmt, args);
62 vfprintf(stderr, fmt, args);
63 fprintf(stderr, "\n");
69 static void log_error(char *fmt, ...)
74 vsyslog(LOG_ERR, fmt, args);
76 vfprintf(stderr, fmt, args);
77 fprintf(stderr, "\n");
82 static void log_info(char *fmt, ...)
87 vsyslog(LOG_INFO, fmt, args);
89 vfprintf(stderr, fmt, args);
90 fprintf(stderr, "\n");
95 static void log_pkt(char *fmt, ...)
105 /*** MQTT interface ***/
107 #include <mosquitto.h>
109 static struct mosquitto *mosq;
111 static void mqtt_publish(const char *topic, const char *fmt, ...)
116 int l = vsnprintf(m, sizeof(m), fmt, args);
117 if (mosquitto_publish(mosq, NULL, topic, l, m, 0, true) != MOSQ_ERR_SUCCESS)
118 log_error("Mosquitto: publish failed");
122 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
125 mqtt_publish("burrow/arexxd/status", "ok");
128 static void mqtt_init(void)
130 mosquitto_lib_init();
131 mosq = mosquitto_new("arexxd", 1, NULL);
133 die("Mosquitto: initialization failed");
135 if (mosquitto_will_set(mosq, "burrow/arexxd/status", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
136 die("Mosquitto: unable to set will");
138 mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
140 if (mosquitto_connect(mosq, "10.32.184.5", 1883, 60) != MOSQ_ERR_SUCCESS)
141 die("Mosquitto: connect failed");
143 if (mosquitto_loop_start(mosq))
144 die("Mosquitto: cannot start service thread");
147 static void mqtt_point(time_t t, const char *name, double val, char *unit UNUSED)
149 // We do not feed past data to MQTT (so MAX_PAST_TIME is stronger for us)
150 if (t < packet_rx_time - 30)
154 snprintf(topic, sizeof(topic), "burrow/arexxd/%s", name);
155 mqtt_publish(topic, "%.3f", val);
156 snprintf(topic, sizeof(topic), "burrow/arexxd/%s_timestamp", name);
157 mqtt_publish(topic, "%llu", (unsigned long long) t);
160 /*** RRD interface ***/
165 #define MAX_ARG_SIZE 1024
168 static char *arg_ptr[MAX_ARGS+1];
169 static char arg_buf[MAX_ARG_SIZE];
172 static void arg_new(void)
176 arg_ptr[0] = "rrdtool";
179 static void arg_push(const char *fmt, ...)
181 if (arg_cnt >= MAX_ARGS)
182 die("MAX_ARGS exceeded");
185 int len = 1 + vsnprintf(arg_buf + arg_pos, MAX_ARG_SIZE - arg_pos, fmt, va);
186 if (arg_pos + len > MAX_ARG_SIZE)
187 die("MAX_ARG_SIZE exceeded");
188 arg_ptr[arg_cnt++] = arg_buf + arg_pos;
189 arg_ptr[arg_cnt] = NULL;
193 static void rrd_point(time_t t, const char *name, double val, char *unit)
196 snprintf(rr_name, sizeof(rr_name), "sensor-%s.rrd", name);
199 if (stat(rr_name, &st) < 0 || !st.st_size) {
200 // We have to create the RRD
201 log_info("Creating %s", rr_name);
205 arg_push("%d", (int) time(NULL) - 28*86400);
208 if (!strcmp(unit, "%RH"))
209 arg_push("DS:rh:GAUGE:300:0:100");
210 else if (!strcmp(unit, "ppm"))
211 arg_push("DS:ppm:GAUGE:300:0:1000000");
213 arg_push("DS:temp:GAUGE:300:-200:200");
214 arg_push("RRA:AVERAGE:0.25:1:20160"); // Last 14 days with full resolution
215 arg_push("RRA:AVERAGE:0.25:60:88800"); // Last 10 years with 1h resolution
216 arg_push("RRA:MIN:0.25:60:88800"); // including minima and maxima
217 arg_push("RRA:MAX:0.25:60:88800");
218 rrd_create(arg_cnt, arg_ptr);
219 if (rrd_test_error()) {
220 log_error("rrd_create on %s failed: %s", rr_name, rrd_get_error());
228 arg_push("%d:%f", t, val);
229 rrd_update(arg_cnt, arg_ptr);
230 if (rrd_test_error()) {
231 log_error("rrd_update on %s failed: %s", rr_name, rrd_get_error());
238 #define TIME_OFFSET 946681200 // Timestamp of 2000-01-01 00:00:00
240 static double correct_point(uint id, double val, const char **name)
243 * Manually calculated corrections and renames for my sensors.
244 * Replace with your formulae.
254 *name = "machinarium";
266 #ifdef IGNORE_UNKNOWN_SENSORS
273 static void cooked_point(time_t t, uint id, double val, char *unit, int q)
276 snprintf(namebuf, sizeof(namebuf), "%u", id);
277 const char *name = namebuf;
279 double val2 = correct_point(id, val, &name);
281 if (debug_raw_data) {
283 localtime_r(&t, &tm);
285 strftime(tbuf, sizeof(tbuf), "%Y-%m-%d %H:%M:%S", &tm);
286 printf("== %s id=%d name=%s val=%.3f val2=%.3f unit=%s q=%d\n", tbuf, id, name, val, val2, unit, q);
290 log_error("Ignored data from unknown sensor %d", id);
293 if (t < packet_rx_time - MAX_PAST_TIME) {
294 log_error("Data point from sensor %d too far in the past (%d sec)", packet_rx_time - t);
297 if (t > packet_rx_time + MAX_FUTURE_TIME) {
298 log_error("Data point from sensor %d too far in the future (%d sec)", t - packet_rx_time);
302 data_point_counter++;
303 rrd_point(t, name, val2, unit);
304 mqtt_point(t, name, val2, unit);
307 static void raw_point(uint t, uint id, int raw, int q)
310 * The binary blob provided by Arexx contains an embedded XML fragment
311 * with descriptions of all known sensor types. If you want to see it,
312 * grep the blob for "<deviceinfo>". The meanings of the parameters are
315 * m1, m2 Device type matches if (raw_sensor_id & m1) == m2
316 * type Unit measured by the sensor (1=Celsius, 2=RH%, 3=CO2 ppm)
317 * dm User-visible sensor ID = raw_sensor_id & dm
318 * i 1 if the raw value is signed
319 * p[] Coefficients of transformation polynomial (x^0 first)
320 * vLo, vUp Upper and lower bound on the final value
321 * scale Scaling function:
322 * 0 = identity (default)
325 * 3 = (x < 0) ? 0 : log10(x)
326 * 4 = (x < 0) ? 0 : log(x)
328 * The raw values are transformed this way:
329 * - sign-extend if signed
330 * - apply the transformation polynomial
331 * - apply the scaling function
332 * - drop if outside the interval [vLo,vUp]
334 * This function applies the necessary transform for sensors we've
335 * seen in the wild. We deliberately ignore the "dm" parameter as we want
336 * to report different channels of a single sensor as multiple sensors.
342 uint idhi = id & 0xfffff000;
343 uint idext = id & 0xf0000000;
345 if (idhi == 0x1000) {
350 } else if (idhi == 0x2000 || idext == 0x20000000) {
357 } else if (idhi == 0x4000) {
364 z = -2.8e-6*z*z + 0.0405*z - 4;
369 } else if (idhi == 0x6000) {
379 z = (z + 1.9184e-7) * z;
380 z = (z - 1.0998e-3) * z;
388 log_error("Unknown sensor type 0x%08x", id);
392 if (z < lo || z > hi) {
393 log_error("Sensor %d: value %f out of range", id, z);
397 cooked_point(t + TIME_OFFSET, id & 0x0fffffff, z, unit, q);
400 /*** USB interface ***/
402 static int rx_endpoint, tx_endpoint;
404 static int parse_descriptors(libusb_device *dev)
407 struct libusb_config_descriptor *desc;
409 if (err = libusb_get_active_config_descriptor(dev, &desc)) {
410 log_error("libusb_get_config_descriptor failed: error %d", err);
413 if (desc->bNumInterfaces != 1) {
414 log_error("Unexpected number of interfaces: %d", desc->bNumInterfaces);
418 const struct libusb_interface *iface = &desc->interface[0];
419 if (iface->num_altsetting != 1) {
420 log_error("Unexpected number of alternate interface settings: %d", iface->num_altsetting);
424 const struct libusb_interface_descriptor *ifd = &iface->altsetting[0];
425 if (ifd->bNumEndpoints != 2) {
426 log_error("Unexpected number of endpoints: %d", ifd->bNumEndpoints);
430 rx_endpoint = tx_endpoint = -1;
431 for (int i=0; i<2; i++) {
432 const struct libusb_endpoint_descriptor *epd = &ifd->endpoint[i];
433 if (epd->bEndpointAddress & 0x80)
434 rx_endpoint = epd->bEndpointAddress;
436 tx_endpoint = epd->bEndpointAddress;
438 if (rx_endpoint < 0 || tx_endpoint < 0) {
439 log_error("Failed to identify endpoints");
443 log_pkt("Found endpoints: rx==%02x tx=%02x\n", rx_endpoint, tx_endpoint);
444 libusb_free_config_descriptor(desc);
448 libusb_free_config_descriptor(desc);
452 static int find_device(void)
454 libusb_device **devlist;
455 ssize_t devn = libusb_get_device_list(usb_ctxt, &devlist);
457 log_error("Cannot enumerate USB devices: error %d", (int) devn);
461 for (ssize_t i=0; i<devn; i++) {
462 struct libusb_device_descriptor desc;
463 libusb_device *dev = devlist[i];
464 if (!libusb_get_device_descriptor(dev, &desc)) {
465 if (desc.idVendor == 0x0451 && desc.idProduct == 0x3211) {
466 log_info("Arexx data logger found at usb%d.%d", libusb_get_bus_number(dev), libusb_get_device_address(dev));
467 if (!parse_descriptors(dev))
470 if (err = libusb_open(dev, &devh)) {
471 log_error("libusb_open() failed: error %d", err);
474 if (err = libusb_claim_interface(devh, 0)) {
475 log_error("libusb_claim_interface() failed: error %d", err);
479 libusb_free_device_list(devlist, 1);
486 libusb_free_device_list(devlist, 1);
490 static void release_device(void)
492 libusb_release_interface(devh, 0);
493 libusb_reset_device(devh);
498 static void dump_packet(byte *pkt)
500 for (int i=0; i<64; i++) {
502 log_pkt("\t%02x:", i);
503 log_pkt(" %02x", pkt[i]);
509 static void my_msleep(int ms)
511 struct timespec ts = { .tv_sec = ms/1000, .tv_nsec = (ms%1000) * 1000000 };
512 nanosleep(&ts, NULL);
515 static int send_and_receive(byte *req, byte *reply)
518 time_t t = time(NULL);
520 localtime_r(&t, &tm);
523 strftime(tbuf, sizeof(tbuf), "%Y-%m-%d %H:%M:%S", &tm);
524 log_pkt("## %s\n", tbuf);
527 int err, transferred;
528 if (err = libusb_bulk_transfer(devh, tx_endpoint, req, 64, &transferred, 200)) {
529 if (err == LIBUSB_ERROR_TIMEOUT) {
530 log_pkt(">> xmit timed out\n");
533 log_pkt(">> xmit error %d\n", err);
534 log_error("Transmit error: %d", err);
538 log_pkt(">> xmit %d bytes\n", transferred);
542 if (err = libusb_bulk_transfer(devh, rx_endpoint, reply, 64, &transferred, 200)) {
543 if (err == LIBUSB_ERROR_TIMEOUT) {
544 log_pkt("<< recv timed out\n");
547 log_pkt("<< recv error %d\n", err);
548 log_error("Receive error: %d", err);
551 packet_rx_time = time(NULL);
553 log_pkt("<< recv %d bytes\n", transferred);
554 while (transferred < 64)
555 reply[transferred++] = 0xff;
561 static unsigned int get_be16(byte *p)
563 return p[1] | (p[0] << 8);
566 static unsigned int get_le16(byte *p)
568 return p[0] | (p[1] << 8);
571 static unsigned int get_le32(byte *p)
573 return get_le16(p) | (get_le16(p+2) << 16);
576 static void put_le16(byte *p, unsigned int x)
582 static void put_le32(byte *p, unsigned int x)
585 put_le16(p+2, x>>16);
588 static int parse_packet(byte *reply)
591 log_error("Unknown packet type %02x", reply[0]);
598 byte *p = reply + pos;
600 if (!len || len == 0xff)
602 if (pos + len > 64) {
603 log_error("Tuple truncated");
615 q = (len > 9) ? p[9] : -1;
624 log_error("Unknown tuple length %02x", len);
628 if (debug_raw_data) {
629 printf("... %02x: id=%08x raw=%d t=%u", len, id, raw, t);
634 raw_point(t, id, raw, q);
643 static void set_clock(void)
645 byte req[64], reply[64];
648 time_t t = time(NULL);
649 put_le32(req+1, t-TIME_OFFSET);
650 send_and_receive(req, reply);
654 * Original software also sends a packet with type 3 and the timestamp,
655 * but it does not make any sense, especially as they ignore the sensor
656 * readings in the answer.
659 send_and_receive(req, reply);
666 static sigset_t term_sigs;
667 static volatile sig_atomic_t want_shutdown;
669 static void sigterm_handler(int sig UNUSED)
674 static void interruptible_msleep(int ms)
676 sigprocmask(SIG_UNBLOCK, &term_sigs, NULL);
678 sigprocmask(SIG_BLOCK, &term_sigs, NULL);
681 static const struct option long_options[] = {
682 { "debug", 0, NULL, 'd' },
683 { "log-dir", 1, NULL, 'l' },
684 { "no-fork", 0, NULL, 'n' },
685 { "debug-packets", 0, NULL, 'p' },
686 { "debug-raw", 0, NULL, 'r' },
687 { "version", 0, NULL, 'V' },
688 { NULL, 0, NULL, 0 },
691 static void usage(void)
694 Usage: arexxd <options>\n\
697 -d, --debug Debug mode (no chdir, no fork, no syslog)\n\
698 -l, --log-dir=<dir> Directory where all received data should be stored\n\
699 -n, --no-fork Do not fork\n\
700 -p, --debug-packets Log all packets sent and received\n\
701 -r, --debug-raw Log conversion from raw values\n\
702 -u, --debug-usb Enable libusb debug messages (to stdout/stderr)\n\
703 -V, --version Show daemon version\n\
708 int main(int argc, char **argv)
711 while ((opt = getopt_long(argc, argv, "dl:npruV", long_options, NULL)) >= 0)
732 printf("arexxd " AREXXD_VERSION "\n");
733 printf("(c) 2011-2018 Martin Mares <mj@ucw.cz>\n");
742 if (err = libusb_init(&usb_ctxt))
743 die("Cannot initialize libusb: error %d", err);
745 libusb_set_debug(usb_ctxt, 3);
748 if (chdir(log_dir) < 0)
749 die("Cannot change directory to %s: %m", log_dir);
750 if (debug_packets || debug_raw_data) {
752 if (open("debug", O_WRONLY | O_CREAT | O_APPEND, 0666) < 0)
753 die("Cannot open debug log: %m");
756 openlog("arexxd", LOG_NDELAY, LOG_DAEMON);
760 die("fork() failed: %m");
770 struct sigaction sa = { .sa_handler = sigterm_handler };
771 sigaction(SIGTERM, &sa, NULL);
772 sigaction(SIGINT, &sa, NULL);
774 sigemptyset(&term_sigs);
775 sigaddset(&term_sigs, SIGTERM);
776 sigaddset(&term_sigs, SIGINT);
777 sigprocmask(SIG_BLOCK, &term_sigs, NULL);
780 while (!want_shutdown) {
781 if (!find_device()) {
784 log_error("Data logger not connected, waiting until it appears");
786 interruptible_msleep(30000);
789 log_info("Listening");
791 time_t last_sync = 0;
792 time_t last_show = 0;
795 data_point_counter = 0;
796 while (!want_shutdown) {
797 time_t now = time(NULL);
798 if (now > last_sync + 900) {
799 log_info("Synchronizing data logger time");
803 if (want_stats && now > last_show + 300) {
804 log_info("Stats: received %d data points", data_point_counter);
805 data_point_counter = 0;
809 byte req[64], reply[64];
810 memset(req, 0, sizeof(req));
812 err = send_and_receive(req, reply);
816 if (err > 0 && parse_packet(reply))
819 interruptible_msleep(4000);
822 interruptible_msleep(5);
825 log_info("Disconnecting data logger");
828 interruptible_msleep(10000);
831 log_info("Terminated");