Record data in a RRD

diff --git a/Makefile b/Makefile
index e1364334233168d90afb0c67aba446911581bff3..04cfded65c779edee889e0d53f196c8c798e9ada 100644 (file)
--- a/Makefile
+++ b/Makefile
@@ -1,7 +1,7 @@
 CC=gcc
 LD=gcc
 CFLAGS=-O2 -Wall -W -Wno-parentheses -Wstrict-prototypes -Wmissing-prototypes -Wundef -Wredundant-decls -std=gnu99
-LDLIBS=-lusb-1.0 -lm
+LDLIBS=-lusb-1.0 -lm -lrrd
 
 all: arexx
 

diff --git a/arexx.c b/arexx.c
index 95c9c9389cf4c9f9b8168bc264c323fd35db94e1..796d13abcecb9877a432b110c67b5d550ec7009a 100644 (file)
--- a/arexx.c
+++ b/arexx.c
@@ -11,7 +11,9 @@
 #include <unistd.h>
 #include <math.h>
 #include <time.h>
+#include <sys/stat.h>
 #include <libusb-1.0/libusb.h>
+#include <rrd.h>
 
 typedef unsigned char byte;
 static libusb_context *usb_ctxt;
@@ -32,94 +34,71 @@ static void die(char *msg, ...)
        exit(1);
 }
 
-static void find_device(void)
-{
-       libusb_device **devlist;
-       ssize_t devn = libusb_get_device_list(usb_ctxt, &devlist);
-       if (devn < 0)
-               die("Cannot enumerate USB devices: error %d\n", (int) devn);
-
-       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) {
-                               printf("Arexx USB receiver found at usb%d.%d\n", libusb_get_bus_number(dev), libusb_get_device_address(dev));
-                               int err;
-                               if (err = libusb_open(dev, &devh))
-                                       die("libusb_open() failed: error %d\n", err);
-                               if (err = libusb_claim_interface(devh, 0))
-                                       die("libusb_claim_interface() failed: error %d\n", err);
-                               libusb_free_device_list(devlist, 1);
-                               return;
-                       }
-               }
-       }
+/*** RRD interface ***/
 
-       die("No Arexx USB receiver found");
-}
+#define RRD_NAME "/tmp/power.rrd"
+#define SLOT_SIZE 10                                   // 10 seconds per averaging slot
+#define MAX_ARGS 20
+#define MAX_ARG_SIZE 1024
 
-static void dump_packet(byte *pkt)
-{
-       for (int i=0; i<64; i++) {
-               if (!(i % 16))
-                       printf("\t%02x:", i);
-               printf(" %02x", pkt[i]);
-               if (i % 16 == 15)
-                       printf("\n");
-       }
-}
-
-static int send_and_receive(byte *req, byte *reply)
-{
-       int err, transferred;
-       if (err = libusb_bulk_transfer(devh, 0x01, req, 64, &transferred, 200)) {
-               fprintf(stderr, "Transmit error: %d\n", err);
-               return 0;
-       }
-       if (debug_packets) {
-               printf(">> xmit %d bytes\n", transferred);
-               dump_packet(req);
-       }
-       if (err = libusb_bulk_transfer(devh, 0x81, reply, 64, &transferred, 200)) {
-               fprintf(stderr, "Receive error: %d\n", err);
-               return 0;
-       }
-       if (debug_packets)
-               printf("<< recv %d bytes\n", transferred);
-       while (transferred < 64)
-               reply[transferred++] = 0xff;
-       if (debug_packets)
-               dump_packet(reply);
-       return 1;
-}
+static int arg_cnt;
+static char *arg_ptr[MAX_ARGS+1];
+static char arg_buf[MAX_ARG_SIZE];
+static int arg_pos;
 
-static unsigned int get_be16(byte *p)
+static void arg_new(void)
 {
-       return p[1] | (p[0] << 8);
+       arg_cnt = 1;
+       arg_pos = 0;
+       arg_ptr[0] = "rrdtool";
 }
 
-static unsigned int get_le16(byte *p)
+static void arg_push(const char *fmt, ...)
 {
-       return p[0] | (p[1] << 8);
+       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 unsigned int get_le32(byte *p)
+static void rrd_point(time_t t, int id, double val)
 {
-       return get_le16(p) | (get_le16(p+2) << 16);
-}
+       char rr_name[256];
+       snprintf(rr_name, sizeof(rr_name), "sensor-%d.rrd", id);
+
+       struct stat st;
+       if (stat(rr_name, &st) < 0 || !st.st_size) {
+               // We have to create the RRD
+               printf("### Creating %s\n", 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");
+               arg_push("DS:temp:GAUGE:300:0:10000");          // Anything over 10 kW is considered a ghost
+               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
+               rrd_create(arg_cnt, arg_ptr);
+               if (rrd_test_error())
+                       die("rrd_create failed: %s", rrd_get_error());
+       }
 
-static void put_le16(byte *p, unsigned int x)
-{
-       p[0] = x;
-       p[1] = x >> 8;
+       arg_new();
+       arg_push(rr_name);
+       arg_push("%d:%f", t, val);
+       rrd_update(arg_cnt, arg_ptr);
+       if (rrd_test_error())
+               printf("### rrd_update failed: %s\n", rrd_get_error());
 }
 
-static void put_le32(byte *p, unsigned int x)
-{
-       put_le16(p, x);
-       put_le16(p+2, x>>16);
-}
+/*** Transforms ***/
 
 static void cooked_point(time_t t, int id, double val, char *unit, int q)
 {
@@ -130,6 +109,8 @@ static void cooked_point(time_t t, int id, double val, char *unit, int q)
        strftime(tbuf, sizeof(tbuf), "%Y-%m-%d %H:%M:%S", &tm);
 
        printf("== %s id=%d val=%.3f unit=%s q=%d\n", tbuf, id, val, unit, q);
+
+       rrd_point(t, id, val);
 }
 
 static void raw_point(int t, int id, int raw, int q)
@@ -224,6 +205,97 @@ static void raw_point(int t, int id, int raw, int q)
        cooked_point(t + TIME_OFFSET, id, z, unit, q);
 }
 
+/*** USB interface ***/
+
+static void find_device(void)
+{
+       libusb_device **devlist;
+       ssize_t devn = libusb_get_device_list(usb_ctxt, &devlist);
+       if (devn < 0)
+               die("Cannot enumerate USB devices: error %d\n", (int) devn);
+
+       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) {
+                               printf("Arexx USB receiver found at usb%d.%d\n", libusb_get_bus_number(dev), libusb_get_device_address(dev));
+                               int err;
+                               if (err = libusb_open(dev, &devh))
+                                       die("libusb_open() failed: error %d\n", err);
+                               if (err = libusb_claim_interface(devh, 0))
+                                       die("libusb_claim_interface() failed: error %d\n", err);
+                               libusb_free_device_list(devlist, 1);
+                               return;
+                       }
+               }
+       }
+
+       die("No Arexx USB receiver found");
+}
+
+static void dump_packet(byte *pkt)
+{
+       for (int i=0; i<64; i++) {
+               if (!(i % 16))
+                       printf("\t%02x:", i);
+               printf(" %02x", pkt[i]);
+               if (i % 16 == 15)
+                       printf("\n");
+       }
+}
+
+static int send_and_receive(byte *req, byte *reply)
+{
+       int err, transferred;
+       if (err = libusb_bulk_transfer(devh, 0x01, req, 64, &transferred, 200)) {
+               fprintf(stderr, "Transmit error: %d\n", err);
+               return 0;
+       }
+       if (debug_packets) {
+               printf(">> xmit %d bytes\n", transferred);
+               dump_packet(req);
+       }
+       if (err = libusb_bulk_transfer(devh, 0x81, reply, 64, &transferred, 200)) {
+               fprintf(stderr, "Receive error: %d\n", err);
+               return 0;
+       }
+       if (debug_packets)
+               printf("<< 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]) {