bsb/daemon/burrow-bsbd.c

   1 /*
   2  *      Boiler System Bus Interface Daemon
   3  *
   4  *      (c) 2020 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/string.h>
  11 #include <ucw/unaligned.h>
  12
  13 #include <stdarg.h>
  14 #include <stdio.h>
  15 #include <stdint.h>
  16 #include <stdlib.h>
  17 #include <string.h>
  18 #include <syslog.h>
  19 #include <time.h>
  20 #include <unistd.h>
  21
  22 #include <libusb.h>
  23 #include <mosquitto.h>
  24
  25 typedef unsigned char byte;
  26 typedef uint32_t u32;
  27 typedef unsigned int uint;
  28
  29 #include "../firmware/interface.h"
  30
  31 /*** MQTT ***/
  32
  33 static struct mosquitto *mosq;
  34 static bool mqtt_connected;
  35
  36 static void mqtt_publish(const char *topic, const char *fmt, ...)
  37 {
  38         va_list args;
  39         va_start(args, fmt);
  40
  41         if (mqtt_connected) {
  42                 char m[256];
  43                 int l = vsnprintf(m, sizeof(m), fmt, args);
  44                 int err = mosquitto_publish(mosq, NULL, topic, l, m, 0, true);
  45                 if (err != MOSQ_ERR_SUCCESS)
  46                         msg(L_ERROR, "Mosquitto: Publish failed, error=%d", err);
  47         }
  48
  49         va_end(args);
  50 }
  51
  52 static void mqtt_publish_ephemeral(const char *topic, const char *fmt, ...)
  53 {
  54         va_list args;
  55         va_start(args, fmt);
  56
  57         if (mqtt_connected) {
  58                 char m[256];
  59                 int l = vsnprintf(m, sizeof(m), fmt, args);
  60                 int err = mosquitto_publish(mosq, NULL, topic, l, m, 0, false);
  61                 if (err != MOSQ_ERR_SUCCESS)
  62                         msg(L_ERROR, "Mosquitto: Publish failed, error=%d", err);
  63         }
  64
  65         va_end(args);
  66 }
  67
  68 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
  69 {
  70         if (!status) {
  71                 msg(L_DEBUG, "MQTT: Connection established");
  72                 mqtt_connected = true;
  73                 mqtt_publish("status/bsb", "ok");
  74         } else if (mqtt_connected) {
  75                 msg(L_DEBUG, "MQTT: Connection lost");
  76                 mqtt_connected = false;
  77         }
  78 }
  79
  80 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
  81 {
  82         msg(L_DEBUG, "MQTT(%d): %s", level, message);
  83 }
  84
  85 static void mqtt_init(void)
  86 {
  87         mosquitto_lib_init();
  88
  89         mosq = mosquitto_new("bsbd", 1, NULL);
  90         if (!mosq)
  91                 die("Mosquitto: Initialization failed");
  92
  93         mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
  94         mosquitto_log_callback_set(mosq, mqtt_log_callback);
  95
  96         if (mosquitto_will_set(mosq, "status/bsb", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
  97                 die("Mosquitto: Unable to set will");
  98
  99         if (mosquitto_connect_async(mosq, "127.0.0.1", 1883, 60) != MOSQ_ERR_SUCCESS)
 100                 die("Mosquitto: Unable to connect");
 101
 102         if (mosquitto_loop_start(mosq))
 103                 die("Mosquitto: Cannot start service thread");
 104 }
 105
 106 /*** USB ***/
 107
 108 static struct libusb_context *usb_ctxt;
 109 static struct libusb_device_handle *devh;
 110
 111 static void usb_error(const char *msg, ...)
 112 {
 113         va_list args;
 114         va_start(args, msg);
 115         ucw_vmsg(L_ERROR, msg, args);
 116         fputc('\n', stderr);
 117         va_end(args);
 118
 119         if (devh) {
 120                 libusb_close(devh);
 121                 devh = NULL;
 122         }
 123 }
 124
 125 static void open_device(void)
 126 {
 127         int err;
 128         libusb_device **devlist;
 129         ssize_t devn = libusb_get_device_list(usb_ctxt, &devlist);
 130         if (devn < 0)
 131                 die("Cannot enumerate USB devices: error %d", (int) devn);
 132
 133         for (ssize_t i=0; i<devn; i++) {
 134                 struct libusb_device_descriptor desc;
 135                 libusb_device *dev = devlist[i];
 136                 if (!libusb_get_device_descriptor(dev, &desc)) {
 137                         if (desc.idVendor == BSB_USB_VENDOR && desc.idProduct == BSB_USB_PRODUCT) {
 138                                 msg(L_INFO, "Found device at usb%d.%d", libusb_get_bus_number(dev), libusb_get_device_address(dev));
 139
 140                                 if (err = libusb_open(dev, &devh)) {
 141                                         usb_error("Cannot open device: error %d", err);
 142                                         goto out;
 143                                 }
 144                                 libusb_reset_device(devh);
 145                                 if (err = libusb_claim_interface(devh, 0)) {
 146                                         usb_error("Cannot claim interface: error %d", err);
 147                                         goto out;
 148                                 }
 149
 150                                 goto out;
 151                         }
 152                 }
 153         }
 154
 155 out:
 156         libusb_free_device_list(devlist, 1);
 157 }
 158
 159 static void init_usb(void)
 160 {
 161         int err;
 162         if (err = libusb_init(&usb_ctxt))
 163                 die("Cannot initialize libusb: error %d", err);
 164         // libusb_set_debug(usb_ctxt, 3);
 165         open_device();
 166
 167 }
 168
 169 /*** Protocol ***/
 170
 171 static const char * const stat_names[] = {
 172 #define P(x) #x,
 173         BSB_STATS
 174 #undef P
 175 };
 176
 177 static int get_s16_be(const byte *x)
 178 {
 179         int val = get_u16_be(x);
 180         if (val >= 0x8000)
 181                 return val - 0x10000;
 182         else
 183                 return val;
 184 }
 185
 186 static void process_stats(time_t t, byte *resp, uint len)
 187 {
 188         for (uint i=0; i < ARRAY_SIZE(stat_names) && 4*i + 3 < (uint) len; i++) {
 189                 char item[64];
 190                 snprintf(item, sizeof(item), "bsb/stats/%s", stat_names[i]);
 191                 mqtt_publish(item, "%u %lld", (uint) get_u32_le(resp+4*i), (long long) t);
 192         }
 193 }
 194
 195 static void process_info(time_t t, byte *p, uint len)
 196 {
 197         if (len < 4)
 198                 return;
 199
 200         u32 addr = get_u32_be(p);
 201         p += 4;
 202         len -= 4;
 203
 204         switch (addr) {
 205                 case 0x0500006b:
 206                         if (len >= 2) {
 207                                 int err = get_u16_be(p);
 208                                 mqtt_publish("burrow/heating/error", "%d %lld", err, (long long) t);
 209                         }
 210                         break;
 211                 case 0x0500006c:
 212                         if (len >= 7) {
 213                                 mqtt_publish("burrow/heating/clock", "%04d-%02d-%02dT%02d:%02d %lld",
 214                                         get_u16_be(p) + 1900, p[2], p[3],
 215                                         p[5], p[6],
 216                                         (long long) t);
 217                         }
 218                         break;
 219                 case 0x05000219:
 220                         if (len >= 4) {
 221                                 int temp = get_s16_be(p);
 222                                 uint press = get_u16_be(p + 2);
 223                                 mqtt_publish("burrow/heating/outside-temp", "%.2f %lld", temp / 64., (long long) t);
 224                                 mqtt_publish("burrow/heating/water-pressure", "%.1f %lld", press / 10., (long long) t);
 225                         }
 226                         break;
 227                 case 0x05000229:
 228                         // AGU.2 status
 229                         if (len >= 2) {
 230                                 int temp = get_s16_be(p);
 231                                 mqtt_publish("burrow/heating/circuit1/mix-temp", "%.2f %lld", temp / 64., (long long) t);
 232                         }
 233                         break;
 234                 case 0x05040227:
 235                         // AGU.2 control
 236                         if (len >= 4) {
 237                                 uint m = get_u16_be(p);
 238                                 mqtt_publish("burrow/heating/circuit1/mix-valve", "%u %lld", m, (long long) t);
 239                                 mqtt_publish("burrow/heating/circuit1/pump", "%u %lld", (p[3] == 3), (long long) t);
 240                         }
 241                         break;
 242                 case 0x2d000211:
 243                         // Circuit 1 status
 244                         if (len >= 10) {
 245                                 mqtt_publish("burrow/heating/circuit1/active", "%u %lld", (p[8] != 0), (long long) t);
 246                         }
 247                         break;
 248                 case 0x2e000211:
 249                         // Circuit 2 status
 250                         if (len >= 10) {
 251                                 mqtt_publish("burrow/heating/circuit2/active", "%u %lld", (p[8] != 0), (long long) t);
 252                         }
 253                         break;
 254                 case 0x31000212:
 255                         // Hot water status
 256                         if (len >= 3) {
 257                                 mqtt_publish("burrow/heating/water/active", "%u %lld", (p[1] != 0), (long long) t);
 258                         }
 259                         break;
 260                 case 0x3d2d0215:
 261                         // Room 1 status
 262                         if (len >= 2) {
 263                                 int temp = get_s16_be(p);
 264                                 mqtt_publish("burrow/heating/circuit1/room-temp", "%.2f %lld", temp / 64., (long long) t);
 265                         }
 266                         break;
 267                 case 0x3e2e0215:
 268                         // Room 2 status
 269                         if (len >= 2) {
 270                                 int temp = get_s16_be(p);
 271                                 mqtt_publish("burrow/heating/circuit2/room-temp", "%.2f %lld", temp / 64., (long long) t);
 272                         }
 273                         break;
 274         }
 275 }
 276
 277 static void process_answer(time_t t, byte *p, uint len)
 278 {
 279         if (len < 4)
 280                 return;
 281
 282         u32 addr = get_u32_be(p);
 283         p += 4;
 284         len -= 4;
 285
 286         switch (addr) {
 287         }
 288 }
 289
 290 static void process_frame(time_t t, byte *pkt, uint len)
 291 {
 292         if (!len) {
 293                 msg(L_ERROR, "Received empty frame");
 294                 return;
 295         }
 296
 297         byte status = *pkt++;
 298         len--;
 299
 300         if (!len) {
 301                 msg(L_ERROR, "Frame transmit status: %u", status);
 302                 return;
 303         }
 304
 305         if (len < 6) {
 306                 msg(L_ERROR, "Received truncated frame");
 307                 return;
 308         }
 309
 310         char hex[3*len + 1];
 311         mem_to_hex(hex, pkt, len, ' ');
 312         mqtt_publish_ephemeral("bsb/frame", "%s", hex, t);
 313
 314         msg(L_DEBUG, "<< %s", hex);
 315
 316         byte *body = pkt + BF_BODY;
 317         uint body_len = len - BF_BODY - 2;      // 2 for CRC
 318
 319         switch (pkt[BF_OP]) {
 320                 case BSB_OP_INFO:
 321                         process_info(t, body, body_len);
 322                         break;
 323                 case BSB_OP_ANSWER:
 324                         process_answer(t, body, body_len);
 325                         break;
 326         }
 327 }
 328
 329 static int use_daemon;
 330 static int use_debug;
 331
 332 static struct opt_section options = {
 333         OPT_ITEMS {
 334                 OPT_HELP("A daemon for controlling the air conditioning controller via MQTT"),
 335                 OPT_HELP(""),
 336                 OPT_HELP("Options:"),
 337                 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
 338                 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
 339                 OPT_HELP_OPTION,
 340                 OPT_CONF_OPTIONS,
 341                 OPT_END
 342         }
 343 };
 344
 345 int main(int argc UNUSED, char **argv)
 346 {
 347         log_init(argv[0]);
 348         opt_parse(&options, argv+1);
 349
 350         if (use_daemon) {
 351                 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
 352                 log_set_default_stream(ls);
 353         }
 354         if (!use_debug)
 355                 log_default_stream()->levels &= ~(1U << L_DEBUG);
 356
 357         mqtt_init();
 358         init_usb();
 359
 360         time_t now = time(NULL);
 361         time_t last_stats = 0;
 362         // time_t last_query = now;
 363         int err, received;
 364         byte resp[64];
 365
 366         for (;;) {
 367                 if (!devh) {
 368                         msg(L_INFO, "Waiting for device to appear...");
 369                         while (!devh) {
 370                                 sleep(5);
 371                                 open_device();
 372                         }
 373                 }
 374
 375                 now = time(NULL);
 376                 if (last_stats + 60 < now) {
 377                         if ((received = libusb_control_transfer(devh, 0xc0, 0x00, 0, 0, resp, sizeof(resp), 1000)) < 0) {
 378                                 usb_error("Receive failed: error %d", received);
 379                                 continue;
 380                         }
 381
 382                         process_stats(now, resp, received);
 383                         last_stats = now;
 384                 }
 385
 386 #if 0
 387                 if (last_query + 10 < now) {
 388                         byte pkt[] = { 0xdc, 0xc2, 0x00, 0x0b, 0x06, 0x3d, 0x2e, 0x11, 0x25, 0x00, 0x00 };
 389                         if (err = libusb_bulk_transfer(devh, 0x01, pkt, sizeof(pkt), &received, 2000)) {
 390                                 usb_error("Send failed: error %d", err);
 391                                 continue;
 392                         } else {
 393                                 // msg(L_DEBUG"Send OK: %d bytes", received);
 394                         }
 395                         last_query = now;
 396                 }
 397 #endif
 398
 399                 if (err = libusb_interrupt_transfer(devh, 0x82, resp, 64, &received, 1000)) {
 400                         if (err != LIBUSB_ERROR_TIMEOUT)
 401                                 usb_error("Receive failed: error %d", err);
 402                         continue;
 403                 }
 404
 405                 process_frame(now, resp, received);
 406         }
 407 }