power/daemon/burrow-powerd.c

   1 /*
   2  *      A MQTT Gateway Daemon for the Power Meter
   3  *
   4  *      (c) 2019 Martin Mares <mj@ucw.cz>
   5  */
   6
   7 #include <ucw/lib.h>
   8 #include <ucw/log.h>
   9 #include <ucw/opt.h>
  10 #include <ucw/strtonum.h>
  11 #include <ucw/stkstring.h>
  12
  13 #include <errno.h>
  14 #include <stdio.h>
  15 #include <stdarg.h>
  16 #include <stdlib.h>
  17 #include <string.h>
  18 #include <syslog.h>
  19 #include <time.h>
  20 #include <unistd.h>
  21
  22 #include <modbus.h>
  23 #include <mosquitto.h>
  24
  25 /*** MQTT ***/
  26
  27 static struct mosquitto *mosq;
  28 static bool mqtt_connected;
  29
  30 static void mqtt_error(const char *operation, int err, bool teardown)
  31 {
  32         msg(L_ERROR, "Mosquitto: %s failed: error %d", operation, err);
  33
  34         if (teardown) {
  35                 mosquitto_destroy(mosq);
  36                 mosq = NULL;
  37                 mqtt_connected = false;
  38         } else if (err == MOSQ_ERR_NO_CONN || err == MOSQ_ERR_CONN_REFUSED || err == MOSQ_ERR_CONN_LOST) {
  39                 mqtt_connected = false;
  40         }
  41 }
  42
  43 static void mqtt_publish(const char *topic, const char *fmt, ...)
  44 {
  45         va_list args;
  46         va_start(args, fmt);
  47
  48         if (mqtt_connected) {
  49                 char m[256];
  50                 int l = vsnprintf(m, sizeof(m), fmt, args);
  51                 msg(L_DEBUG, "MQTT > %s %s", topic, m);
  52                 int err = mosquitto_publish(mosq, NULL, topic, l, m, 0, true);
  53                 if (err != MOSQ_ERR_SUCCESS)
  54                         mqtt_error("publish", err, false);
  55         }
  56
  57         va_end(args);
  58 }
  59
  60 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level UNUSED, const char *message UNUSED)
  61 {
  62         // msg(L_INFO, "MQTT(%d): %s", level, message);
  63 }
  64
  65 static bool mqtt_connect(void)
  66 {
  67         int err;
  68
  69         if (mqtt_connected)
  70                 return true;
  71
  72         if (!mosq) {
  73                 mosq = mosquitto_new("powerd", 1, NULL);
  74                 if (!mosq)
  75                         die("Mosquitto: initialization failed");
  76
  77                 mosquitto_log_callback_set(mosq, mqtt_log_callback);
  78
  79                 err = mosquitto_will_set(mosq, "status/power-meter", 4, "dead", 0, true);
  80                 if (err != MOSQ_ERR_SUCCESS) {
  81                         mqtt_error("will_set", err, true);
  82                         return false;
  83                 }
  84
  85                 err = mosquitto_connect(mosq, "127.0.0.1", 1883, 60);
  86                 if (err != MOSQ_ERR_SUCCESS) {
  87                         mqtt_error("connect", err, true);
  88                         return false;
  89                 }
  90         } else {
  91                 err = mosquitto_reconnect(mosq);
  92                 if (err != MOSQ_ERR_SUCCESS) {
  93                         mqtt_error("reconnect", err, false);
  94                         return false;
  95                 }
  96         }
  97
  98         mqtt_connected = true;
  99
 100         mqtt_publish("status/power-meter", "ok");
 101
 102         return mqtt_connected;
 103 }
 104
 105 /*** MODBUS ***/
 106
 107 static modbus_t *modbus;
 108 static bool mb_is_open;
 109
 110 static void mb_error(const char *operation)
 111 {
 112         msg(L_ERROR, "MODBUS: %s failed: %s", operation, modbus_strerror(errno));
 113
 114         if (modbus) {
 115                 if (mb_is_open) {
 116                         modbus_close(modbus);
 117                         mb_is_open = false;
 118                 }
 119                 modbus_free(modbus);
 120                 modbus = NULL;
 121         }
 122 }
 123
 124 static bool mb_connect(void)
 125 {
 126         if (modbus)
 127                 return true;
 128
 129         modbus = modbus_new_rtu("/dev/modbus-power", 9600, 'N', 8, 1);
 130         if (!modbus) {
 131                 mb_error("open");
 132                 return false;
 133         }
 134
 135         modbus_set_slave(modbus, 42);
 136
 137         if (modbus_connect(modbus) < 0) {
 138                 mb_error("connect");
 139                 return false;
 140         }
 141
 142         mb_is_open = true;
 143         return true;
 144 }
 145
 146 /*** Main loop ***/
 147
 148 static u16 mb_regs[22];
 149
 150 static s32 get_s32(uint i)
 151 {
 152         if (mb_regs[i+1] < 0x8000)
 153                 return (mb_regs[i+1] << 16) | mb_regs[i];
 154         else
 155                 return ((mb_regs[i+1] - 65536) * 65536) + mb_regs[i];
 156 }
 157
 158 static void scan_power_meter(time_t now)
 159 {
 160         long int lnow = now;
 161
 162         if (modbus_read_registers(modbus, 0, 22, mb_regs) < 0) {
 163                 mb_error("read");
 164                 return;
 165         }
 166
 167         mqtt_publish("burrow/power/voltage/l1n", "%.1f %ld", get_s32(0) / 10., lnow);
 168         mqtt_publish("burrow/power/voltage/l2n", "%.1f %ld", get_s32(2) / 10., lnow);
 169         mqtt_publish("burrow/power/voltage/l3n", "%.1f %ld", get_s32(4) / 10., lnow);
 170
 171         mqtt_publish("burrow/power/current/l1", "%.3f %ld", get_s32(6) / 1000., lnow);
 172         mqtt_publish("burrow/power/current/l2", "%.3f %ld", get_s32(8) / 1000., lnow);
 173         mqtt_publish("burrow/power/current/l3", "%.3f %ld", get_s32(10) / 1000., lnow);
 174
 175         mqtt_publish("burrow/power/power", "%.1f %ld", get_s32(12) / 10., lnow);
 176         mqtt_publish("burrow/power/energy", "%.1f %ld", get_s32(14) / 10., lnow);
 177
 178         mqtt_publish("burrow/power/reactive/power", "%.1f %ld", get_s32(18) / 10., lnow);
 179         mqtt_publish("burrow/power/reactive/energy", "%.1f %ld", get_s32(20) / 10., lnow);
 180 }
 181
 182 static int use_daemon;
 183 static int use_debug;
 184
 185 static struct opt_section options = {
 186         OPT_ITEMS {
 187                 OPT_HELP("A daemon for sending power meter readings to MQTT"),
 188                 OPT_HELP(""),
 189                 OPT_HELP("Options:"),
 190                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
 191                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
 192                 OPT_HELP_OPTION,
 193                 OPT_CONF_OPTIONS,
 194                 OPT_END
 195         }
 196 };
 197
 198 int main(int argc UNUSED, char **argv)
 199 {
 200         log_init(argv[0]);
 201         opt_parse(&options, argv+1);
 202
 203         if (use_daemon) {
 204                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
 205                 log_set_default_stream(ls);
 206         }
 207         if (!use_debug)
 208                 log_default_stream()->levels &= ~(1U << L_DEBUG);
 209
 210         mosquitto_lib_init();
 211
 212         time_t next_run = 0;
 213         for (;;) {
 214                 if (!mqtt_connect() || !mb_connect()) {
 215                         sleep(5);
 216                         continue;
 217                 }
 218
 219                 time_t now = time(NULL);
 220                 if (now < next_run) {
 221                         int err = mosquitto_loop(mosq, (next_run - now) * 1000, 1);
 222                         if (err != MOSQ_ERR_SUCCESS)
 223                                 mqtt_error("loop", err, false);
 224                         continue;
 225                 }
 226
 227                 next_run = now + 10;
 228                 scan_power_meter(now);
 229         }
 230 }