Mercurial > mbsePi-apps / file revision
thermferm/mqtt.c@13555c27b592
thermferm/mqtt.c
Thu, 25 Apr 2024 14:26:47 +0200
- author
- Michiel Broek <mbroek@mbse.eu>
- date
- Thu, 25 Apr 2024 14:26:47 +0200
- changeset 708
- 13555c27b592
- parent 703
- 344470c6bb1c
- permissions
- -rw-r--r--
Version 0.9.19a6. Fixes after a short trial on the production controller. Fixed json for alternate beer termperature sensor. Fixed division by 1000 for the room temperature and humidity values. The dropdown list for devices shows the address instead of description in the list.
/***************************************************************************** * Copyright (C) 2016-2024 * * Michiel Broek <mbroek at mbse dot eu> * * This file is part of the mbsePi-apps * * This is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the * Free Software Foundation; either version 2, or (at your option) any * later version. * * mbsePi-apps is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * General Public License for more details. * * You should have received a copy of the GNU General Public License * along with ThermFerm; see the file COPYING. If not, write to the Free * Software Foundation, 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. *****************************************************************************/ #include "thermferm.h" #include <sys/ioctl.h> #include <linux/wireless.h> #include "rdconfig.h" #include "devices.h" #include "xutil.h" #include "delay.h" #include "mqtt.h" #include "websocket.h" extern sys_config Config; extern int debug; extern int my_shutdown; extern int my_reboot; extern const char UNITMODE[5][8]; extern const char UNITSTAGE[4][12]; extern const char PROFSTATE[5][6]; extern const char DEVPRESENT[4][6]; int Sequence = 0; /* Global variables for use in callbacks. */ int mqtt_qos = 0; int mqtt_status = STATUS_CONNECTING; int mqtt_mid_sent = 0; int mqtt_last_mid = -1; int mqtt_last_mid_sent = -1; int mqtt_connected = TRUE; int mqtt_disconnect_sent = FALSE; int mqtt_connect_lost = FALSE; int mqtt_my_shutdown = FALSE; int mqtt_use = FALSE; int keepalive = 60; unsigned int max_inflight = 20; struct mosquitto *mosq = NULL; char *state = NULL; char my_hostname[256]; #ifndef HAVE_WIRINGPI_H /* * Code from wiringPi written by Gordon Henderson. * Copied here to have some sort of hardware detection without wiringPi. */ const char *piModelNames [20] = { "Model A", // 0 "Model B", // 1 "Model A+", // 2 "Model B+", // 3 "Pi 2", // 4 "Alpha", // 5 "CM", // 6 "Unknown07", // 07 "Pi 3", // 08 "Pi Zero", // 09 "CM3", // 10 "Unknown11", // 11 "Pi Zero-W", // 12 "Pi 3B+", // 13 "Pi 3A+", // 14 "Unknown15", // 15 "CM3+", // 16 "Unknown17", // 17 "Unknown18", // 18 "Unknown19", // 19 }; const char *piRevisionNames [16] = { "00", "01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12", "13", "14", "15", } ; const char *piMakerNames [16] = { "Sony", // 0 "Egoman", // 1 "Embest", // 2 "Unknown", // 3 "Embest", // 4 "Unknown05", // 5 "Unknown06", // 6 "Unknown07", // 7 "Unknown08", // 8 "Unknown09", // 9 "Unknown10", // 10 "Unknown11", // 11 "Unknown12", // 12 "Unknown13", // 13 "Unknown14", // 14 "Unknown15", // 15 } ; const int piMemorySize [8] = { 256, // 0 512, // 1 1024, // 2 0, // 3 0, // 4 0, // 5 0, // 6 0, // 7 } ; void piBoardId (int *model, int *rev, int *mem, int *maker, int *warranty) { FILE *cpuFd ; char line [120] ; char *c ; unsigned int revision ; int bRev, bType, bProc, bMfg, bMem, bWarranty ; *model = -1; // Mark no info if ((cpuFd = fopen ("/proc/cpuinfo", "r")) == NULL) { syslog(LOG_NOTICE, "Unable to open /proc/cpuinfo") ; return; } while (fgets (line, 120, cpuFd) != NULL) if (strncmp (line, "Revision", 8) == 0) break ; fclose (cpuFd) ; if (strncmp (line, "Revision", 8) != 0) { syslog(LOG_NOTICE, "No \"Revision\" line"); return; } // Chomp trailing CR/NL for (c = &line [strlen (line) - 1] ; (*c == '\n') || (*c == '\r') ; --c) *c = 0 ; if (debug) syslog(LOG_NOTICE, "piBoardId: Revision string: %s", line) ; // Need to work out if it's using the new or old encoding scheme: // Scan to the first character of the revision number for (c = line ; *c ; ++c) if (*c == ':') break ; if (*c != ':') { syslog(LOG_NOTICE, "Bogus \"Revision\" line (no colon)") ; return; } // Chomp spaces ++c ; while (isspace (*c)) ++c ; if (!isxdigit (*c)) syslog(LOG_NOTICE, "Bogus \"Revision\" line (no hex digit at start of revision)") ; revision = (unsigned int)strtol (c, NULL, 16) ; // Hex number with no leading 0x // Check for new way: if ((revision & (1 << 23)) != 0) { // New way if (debug) syslog(LOG_NOTICE, "piBoardId: New Way: revision is: %08X", revision) ; bRev = (revision & (0x0F << 0)) >> 0 ; bType = (revision & (0xFF << 4)) >> 4 ; bProc = (revision & (0x0F << 12)) >> 12 ; // Not used for now. bMfg = (revision & (0x0F << 16)) >> 16 ; bMem = (revision & (0x07 << 20)) >> 20 ; bWarranty = (revision & (0x03 << 24)) != 0 ; *model = bType ; *rev = bRev ; *mem = bMem ; *maker = bMfg ; *warranty = bWarranty ; if (debug) syslog(LOG_NOTICE, "piBoardId: rev: %d, type: %d, proc: %d, mfg: %d, mem: %d, warranty: %d", bRev, bType, bProc, bMfg, bMem, bWarranty) ; } else { // Old way if (debug) syslog(LOG_NOTICE, "piBoardId: Old Way: revision is: %s", c) ; if (!isdigit (*c)) { syslog(LOG_NOTICE, "Bogus \"Revision\" line (no digit at start of revision)") ; return; } // Make sure its long enough if (strlen (c) < 4) { syslog(LOG_NOTICE, "Bogus \"Revision\" line (not long enough)") ; return; } // If longer than 4, we'll assume it's been overvolted *warranty = strlen (c) > 4 ; // Extract last 4 characters: c = c + strlen (c) - 4 ; // Fill out the replys as appropriate /**/ if (strcmp (c, "0002") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0003") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_1 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0004") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "0005") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0006") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0007") == 0) { *model = PI_MODEL_A ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0008") == 0) { *model = PI_MODEL_A ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_SONY ; ; } else if (strcmp (c, "0009") == 0) { *model = PI_MODEL_A ; *rev = PI_VERSION_1_2 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "000d") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "000e") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "000f") == 0) { *model = PI_MODEL_B ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0010") == 0) { *model = PI_MODEL_BP ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "0013") == 0) { *model = PI_MODEL_BP ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_EMBEST ; } else if (strcmp (c, "0016") == 0) { *model = PI_MODEL_BP ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "0019") == 0) { *model = PI_MODEL_BP ; *rev = PI_VERSION_1_2 ; *mem = 1 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0011") == 0) { *model = PI_MODEL_CM ; *rev = PI_VERSION_1_1 ; *mem = 1 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "0014") == 0) { *model = PI_MODEL_CM ; *rev = PI_VERSION_1_1 ; *mem = 1 ; *maker = PI_MAKER_EMBEST ; } else if (strcmp (c, "0017") == 0) { *model = PI_MODEL_CM ; *rev = PI_VERSION_1_1 ; *mem = 1 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "001a") == 0) { *model = PI_MODEL_CM ; *rev = PI_VERSION_1_1 ; *mem = 1 ; *maker = PI_MAKER_EGOMAN ; } else if (strcmp (c, "0012") == 0) { *model = PI_MODEL_AP ; *rev = PI_VERSION_1_1 ; *mem = 0 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "0015") == 0) { *model = PI_MODEL_AP ; *rev = PI_VERSION_1_1 ; *mem = 1 ; *maker = PI_MAKER_EMBEST ; } else if (strcmp (c, "0018") == 0) { *model = PI_MODEL_AP ; *rev = PI_VERSION_1_1 ; *mem = 0 ; *maker = PI_MAKER_SONY ; } else if (strcmp (c, "001b") == 0) { *model = PI_MODEL_AP ; *rev = PI_VERSION_1_1 ; *mem = 0 ; *maker = PI_MAKER_EGOMAN ; } else { *model = -1 ; *rev = 0 ; *mem = 0 ; *maker = 0 ; } } } #endif char *payload_header(void) { char *tmp, buf[128]; tmp = xstrcpy((char *)"{\"timestamp\":"); sprintf(buf, "%ld", time(NULL)); tmp = xstrcat(tmp, buf); tmp = xstrcat(tmp, (char *)",\"seq\":"); sprintf(buf, "%d", Sequence++); tmp = xstrcat(tmp, buf); tmp = xstrcat(tmp, (char *)",\"metric\":"); return tmp; } char *topic_base(char *msgtype) { char *tmp; tmp = xstrcpy((char *)"mbv1.0/fermenters/"); tmp = xstrcat(tmp, msgtype); tmp = xstrcat(tmp, (char *)"/"); tmp = xstrcat(tmp, my_hostname); return tmp; } void my_connect_callback(struct mosquitto *my_mosq, void *obj, int result) { char *topic = NULL; if (mqtt_connect_lost) { mqtt_connect_lost = FALSE; syslog(LOG_NOTICE, "MQTT: reconnect: %s", mosquitto_connack_string(result)); } if (!result) { topic = topic_base((char *)"NCMD"); topic = xstrcat(topic, (char *)"/#"); mosquitto_subscribe(mosq, NULL, topic, 0); free(topic); topic = topic_base((char *)"DCMD"); topic = xstrcat(topic, (char *)"/#"); mosquitto_subscribe(mosq, NULL, topic, 0); free(topic); topic = NULL; mqtt_status = STATUS_CONNACK_RECVD; } else { syslog(LOG_NOTICE, "MQTT: my_connect_callback: %s\n", mosquitto_connack_string(result)); } } void my_disconnect_callback(struct mosquitto *my_mosq, void *obj, int rc) { if (mqtt_my_shutdown) { syslog(LOG_NOTICE, "MQTT: acknowledged DISCONNECT from %s", Config.mqtt_host); mqtt_connected = FALSE; } else { /* * The remote server was brought down. We must keep running */ syslog(LOG_NOTICE, "MQTT: received DISCONNECT from %s, connection lost", Config.mqtt_host); mqtt_connect_lost = TRUE; } } void my_publish_callback(struct mosquitto *my_mosq, void *obj, int mid) { mqtt_last_mid_sent = mid; } void my_subscribe_callback(struct mosquitto *my_mosq, void *userdata, int mid, int qos_count, const int *granted_qos) { int i; syslog(LOG_NOTICE, "Subscribed (mid: %d): %d", mid, granted_qos[0]); for (i = 1; i < qos_count; i++) { syslog(LOG_NOTICE, " %d", granted_qos[i]); } } void my_log_callback(struct mosquitto *my_mosq, void *obj, int level, const char *str) { // if (debug) // syslog(LOG_NOTICE, "MQTT: %s", str); } void my_message_callback(struct mosquitto *my_mosq, void *userdata, const struct mosquitto_message *message) { char *message_type, *message_node, *message_alias; units_list *unit; prof_step *step, *oldstep; struct json_object *jobj, *metric, *val, *setpoint, *profile, *profile1, *steps, *step1; time_t timestamp; int timediff; if (message->payloadlen) { /* * Process received commands */ strtok(message->topic, "/"); // Ignore mbv1.0 strtok(NULL, "/"); // Ignore group_id message_type = strtok(NULL, "/"); message_node = strtok(NULL, "/\0"); message_alias = strtok(NULL, "\0"); jobj = json_tokener_parse(message->payload); if (json_object_object_get_ex(jobj, "timestamp", &val)) { timestamp = json_object_get_int(val); timediff = (int)timestamp - time(NULL); if ((timediff < 61) && (timediff > -61)) { if (json_object_object_get_ex(jobj, "metric", &metric)) { if ((json_object_object_get_ex(metric, "Node Control/Reboot", &val)) && (strcmp(message_type, "NCMD") == 0)) { if (json_object_get_boolean(val) == true) { syslog(LOG_NOTICE, "MQTT: `Node Control/Reboot' command"); /* * Restart. The server process will restart which is handled * in the main thread loop. */ my_reboot = my_shutdown = TRUE; return; } } if ((json_object_object_get_ex(metric, "Node Control/Rebirth", &val)) && (strcmp(message_type, "NCMD") == 0)) { if (json_object_get_boolean(val) == true) { /* * Resend all birth certificates. */ syslog(LOG_NOTICE, "MQTT: `Node Control/Rebirth' command"); publishNData(true, 0); publishDBirthAll(); return; } } /* * DCMD, commands and configuration changes for a single fermenter. */ if ((strcmp(message_type, "DCMD") == 0) && message_node && message_alias) { syslog(LOG_NOTICE, "%s", (char *)json_object_get_string(metric)); for (unit = Config.units ; unit; unit = unit->next) { if (strcmp(unit->alias, message_alias) == 0) { syslog(LOG_NOTICE, "MQTT: DCMD for %s/%s", (char *)message_node, (char *)message_alias); if (json_object_object_get_ex(metric, "stage", &val)) { for (int i = 0; i < 4; i++) { if (strcmp((char *)json_object_get_string(val), UNITSTAGE[i]) == 0) { if (unit->stage != i) { syslog(LOG_NOTICE, "DCMD change fermenter %s: stage to %s", message_alias, UNITSTAGE[i]); unit->mqtt_flag |= MQTT_FLAG_DATA; unit->stage = i; if ((unit->mode != UNITMODE_OFF) && ! unit->event_msg) unit->event_msg = xstrcpy((char *)UNITSTAGE[i]); } break; } } } if (json_object_object_get_ex(metric, "mode", &val)) { for (int i = 0; i < 5; i++) { if (strcmp((char *)json_object_get_string(val), UNITMODE[i]) == 0) { if (unit->mode != i) { unit->mqtt_flag |= MQTT_FLAG_DATA; /* Initialize log if the unit is turned on */ if ((unit->mode == UNITMODE_OFF) && (i != UNITMODE_OFF)) { unit->mqtt_flag |= MQTT_FLAG_BIRTH; } if (i == UNITMODE_PROFILE) { /* Do some checks and refuse profile mode cannot be set */ if (unit->profile_uuid == NULL) { syslog(LOG_NOTICE, "Fermenter unit %s refuse profile, not loaded", message_alias); break; } } syslog(LOG_NOTICE, "DCMD change fermenter %s: mode to %s", message_alias, UNITMODE[i]); unit->mode = i; if ((unit->mode != UNITMODE_OFF) && ! unit->event_msg) unit->event_msg = xstrcpy((char *)UNITMODE[i]); /* Allways turn everything off after a mode change */ unit->PID_cool->OutP = unit->PID_heat->OutP = 0.0; unit->PID_cool->Mode = unit->PID_heat->Mode = PID_MODE_NONE; unit->heater_state = unit->cooler_state = unit->fan_state = unit->light_state = unit->light_timer = 0; unit->heater_wait = unit->cooler_wait = unit->fan_wait = unit->light_wait = 0; device_out(unit->heater_address, unit->heater_state); device_out(unit->cooler_address, unit->cooler_state); device_out(unit->fan_address, unit->fan_state); device_out(unit->light_address, unit->light_state); if (unit->mode == UNITMODE_PROFILE) { /* * Set a sane default until it will be overruled by the * main processing loop. */ unit->prof_target_lo = unit->profile_inittemp_lo; unit->prof_target_hi = unit->profile_inittemp_hi;; unit->prof_fridge_mode = 0; unit->prof_state = PROFILE_OFF; unit->prof_started = unit->prof_paused = unit->prof_primary_done = 0; unit->prof_peak_abs = unit->prof_peak_rel = 0.0; } } break; } } } if (json_object_object_get_ex(metric, "setpoint", &setpoint)) { if ((unit->mode == UNITMODE_FRIDGE) || (unit->mode == UNITMODE_BEER)) { /* * Only set new setpoints if running in FRIDGE or in BEER mode. */ if (json_object_object_get_ex(setpoint, "low", &val)) unit->PID_heat->SetP = json_object_get_double(val); if (json_object_object_get_ex(setpoint, "high", &val)) unit->PID_cool->SetP = json_object_get_double(val); if (unit->mode == UNITMODE_FRIDGE) { unit->fridge_set_lo = unit->PID_heat->SetP; unit->fridge_set_hi = unit->PID_cool->SetP; } else { unit->beer_set_lo = unit->PID_heat->SetP; unit->beer_set_hi = unit->PID_cool->SetP; } unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: setpoints %.1f %.1f", message_alias, unit->PID_heat->SetP, unit->PID_cool->SetP); } } if ((json_object_object_get_ex(metric, "heater", &setpoint)) && (unit->mode == UNITMODE_NONE)) { if (json_object_object_get_ex(setpoint, "state", &val)) { if (json_object_get_int(val) != unit->heater_state) { unit->heater_state = json_object_get_int(val); if (unit->heater_state) // Safety unit->cooler_state = 0; unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: heater_state to %d", message_alias, unit->heater_state); } } } if ((json_object_object_get_ex(metric, "cooler", &setpoint)) && (unit->mode == UNITMODE_NONE)) { if (json_object_object_get_ex(setpoint, "state", &val)) { if (json_object_get_int(val) != unit->cooler_state) { unit->cooler_state = json_object_get_int(val); if (unit->cooler_state) unit->heater_state = 0; unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: cooler_state to %d", message_alias, unit->cooler_state); } } } if ((json_object_object_get_ex(metric, "fan", &setpoint)) && (unit->mode == UNITMODE_NONE)) { if (json_object_object_get_ex(setpoint, "state", &val)) { if (json_object_get_int(val) != unit->fan_state) { unit->fan_state = json_object_get_int(val); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: fan_state to %d", message_alias, unit->fan_state); } } } if (json_object_object_get_ex(metric, "light", &setpoint)) { if (json_object_object_get_ex(setpoint, "state", &val)) { if (json_object_get_int(val) > 0) { unit->light_timer = 300; // 5 minutes syslog(LOG_NOTICE, "DCMD set fermenter %s: light_timer 300", message_alias); if (!unit->light_state) { unit->light_state = 1; unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: light_state to %d", message_alias, unit->light_state); } } } } if ((json_object_object_get_ex(metric, "product", &setpoint)) && (unit->mode == UNITMODE_OFF)) { if (json_object_object_get_ex(setpoint, "code", &val)) { if (strcmp((char *)json_object_get_string(val), unit->product_code)) { free(unit->product_code); unit->product_code = xstrcpy((char *)json_object_get_string(val)); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: product_code to `%s'", message_alias, unit->product_code); } } if (json_object_object_get_ex(setpoint, "name", &val)) { if (strcmp((char *)json_object_get_string(val), unit->product_name)) { free(unit->product_name); unit->product_name = xstrcpy((char *)json_object_get_string(val)); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: product_name to `%s'", message_alias, unit->product_name); } } if (json_object_object_get_ex(setpoint, "uuid", &val)) { if (strcmp((char *)json_object_get_string(val), unit->product_uuid)) { free(unit->product_uuid); unit->product_uuid = xstrcpy((char *)json_object_get_string(val)); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: product_uuid to `%s'", message_alias, unit->product_uuid); } } if (json_object_object_get_ex(setpoint, "yeast_lo", &val)) { unit->yeast_lo = json_object_get_double(val); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: yeast_lo to `%.1f'", message_alias, unit->yeast_lo); } if (json_object_object_get_ex(setpoint, "yeast_hi", &val)) { unit->yeast_hi = json_object_get_double(val); unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: yeast_hi to `%.1f'", message_alias, unit->yeast_hi); } } if (json_object_object_get_ex(metric, "profile", &profile)) { if (json_object_object_get_ex(profile, "command", &profile1)) { syslog(LOG_NOTICE, "profile command"); if (unit->mode == UNITMODE_PROFILE) { char *cmd = xstrcpy((char *)json_object_get_string(profile1)); if (! strcmp(cmd, (char *)"off")) { if (unit->prof_state == PROFILE_DONE) { unit->prof_state = PROFILE_OFF; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile to OFF", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; } } else if (! strcmp(cmd, (char *)"pause")) { if (unit->prof_state == PROFILE_RUN) { unit->prof_state = PROFILE_PAUSE; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile to PAUSE", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; } else if (unit->prof_state == PROFILE_PAUSE) { unit->prof_state = PROFILE_RUN; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile resume RUN", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; } } else if (! strcmp(cmd, (char *)"start")) { if (unit->prof_state == PROFILE_OFF) { unit->prof_state = PROFILE_RUN; unit->prof_started = time(NULL); unit->prof_paused = unit->prof_primary_done = 0; unit->prof_peak_abs = unit->prof_peak_rel = 0.0; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile start RUN", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; if (! unit->event_msg) unit->event_msg = xstrcpy((char *)"Profile start"); } } else if (! strcmp(cmd, (char *)"abort")) { if ((unit->prof_state == PROFILE_RUN) || (unit->prof_state == PROFILE_PAUSE)) { unit->prof_state = PROFILE_OFF; unit->prof_started = unit->prof_paused = unit->prof_primary_done = 0; unit->prof_peak_abs = unit->prof_peak_rel = 0.0; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile ABORT", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; if (! unit->event_msg) unit->event_msg = xstrcpy((char *)"Profile abort"); } } else if (! strcmp(cmd, (char *)"done")) { if (unit->prof_state == PROFILE_DONE) { unit->prof_state = PROFILE_OFF; unit->prof_started = unit->prof_paused = unit->prof_primary_done = 0; unit->prof_peak_abs = unit->prof_peak_rel = 0.0; syslog(LOG_NOTICE, "DCMD change fermenter `%s' profile OFF", message_alias); unit->mqtt_flag |= MQTT_FLAG_DATA; } } free(cmd); cmd = NULL; } } else if (json_object_object_get_ex(profile, "uuid", &profile1)) { if ((unit->prof_state == PROFILE_OFF) || (unit->prof_state == PROFILE_DONE) || (unit->prof_state == PROFILE_ABORT)) { if (unit->profile_uuid) free(unit->profile_uuid); if (unit->profile_name) free(unit->profile_name); if (unit->profile_steps) { for (step = unit->profile_steps; step; step = oldstep) { if (step->name) free(step->name); oldstep = step->next; free(step); } } unit->profile_steps = NULL; unit->profile_duration = unit->profile_totalsteps = 0; unit->profile_uuid = xstrcpy((char *)json_object_get_string(profile1)); if (json_object_object_get_ex(profile, "name", &val)) { unit->profile_name = xstrcpy((char *)json_object_get_string(val)); } if (json_object_object_get_ex(profile, "inittemp", &setpoint)) { if (json_object_object_get_ex(setpoint, "low", &val)) { unit->profile_inittemp_lo = json_object_get_double(val); } if (json_object_object_get_ex(setpoint, "high", &val)) { unit->profile_inittemp_hi = json_object_get_double(val); } } if (json_object_object_get_ex(profile, "fridgemode", &val)) { unit->profile_fridge_mode = json_object_get_int(val); if (unit->profile_fridge_mode) unit->profile_fridge_mode = 100; } if (json_object_object_get_ex(profile, "steps", &steps)) { int arraylen = json_object_array_length(steps); syslog(LOG_NOTICE, "profile new profile: start %d steps", arraylen); for (int i = 0; i < arraylen; i++) { /* * Parse the array of steps */ step1 = json_object_array_get_idx(steps, i); unit->profile_totalsteps++; step = (prof_step *)malloc(sizeof(prof_step)); step->next = NULL; step->name = NULL; step->steptime = step->resttime = step->fridge_mode = 0; step->target_lo = step->target_hi = 20.0; if (json_object_object_get_ex(step1, "name", &val)) { step->name = xstrcpy((char *)json_object_get_string(val)); } if (json_object_object_get_ex(step1, "steptime", &val)) { step->steptime = json_object_get_int(val); unit->profile_duration += step->steptime; } if (json_object_object_get_ex(step1, "resttime", &val)) { step->resttime = json_object_get_int(val); unit->profile_duration += step->resttime; } if (json_object_object_get_ex(step1, "fridgemode", &val)) { step->fridge_mode = json_object_get_int(val); if (step->fridge_mode) step->fridge_mode = 100; } if (json_object_object_get_ex(step1, "target_lo", &val)) { step->target_lo = json_object_get_double(val); } if (json_object_object_get_ex(step1, "target_hi", &val)) { step->target_hi = json_object_get_double(val); } syslog(LOG_NOTICE, "profile new profile: add step %d", unit->profile_totalsteps); if (unit->profile_steps == NULL) { unit->profile_steps = step; } else { for (oldstep = unit->profile_steps; oldstep; oldstep = oldstep->next) { if (oldstep->next == NULL) { oldstep->next = step; break; } } } } } unit->mqtt_flag |= MQTT_FLAG_DATA; syslog(LOG_NOTICE, "DCMD change fermenter %s: install profile `%s'", message_alias, unit->profile_name); wrconfig(); } } else { if ((unit->prof_state == PROFILE_OFF) || (unit->prof_state == PROFILE_DONE) || (unit->prof_state == PROFILE_ABORT)) { syslog(LOG_NOTICE, "DCMD change fermenter %s: delete profile `%s'", message_alias, unit->profile_name); if (unit->profile_uuid) free(unit->profile_uuid); if (unit->profile_name) free(unit->profile_name); unit->profile_uuid = unit->profile_name = NULL; if (unit->profile_steps) { for (step = unit->profile_steps; step; step = oldstep) { if (step->name) free(step->name); oldstep = step->next; free(step); } } unit->profile_steps = NULL; unit->profile_inittemp_lo = unit->profile_inittemp_hi = 20.0; unit->prof_percent = unit->profile_fridge_mode = 0; unit->prof_state = PROFILE_OFF; unit->profile_duration = unit->profile_totalsteps = 0; unit->prof_started = unit->prof_paused = unit->prof_primary_done = (time_t)0; unit->prof_peak_abs = unit->prof_peak_rel = 0.0; unit->mqtt_flag |= MQTT_FLAG_DATA; wrconfig(); } } } } if (unit->mqtt_flag) { if (unit->mqtt_flag & MQTT_FLAG_BIRTH) { publishDBirth(unit); } else { publishDData(unit); } if (unit->mqtt_flag & MQTT_FLAG_DEATH) { publishDDeath(unit); } } } return; } syslog(LOG_NOTICE, "MQTT: %s payload not understood\n", (char *)message->payload); return; } } else { syslog(LOG_NOTICE, "MQTT: got payload with timestamp %d seconds error", timediff); return; } } syslog(LOG_NOTICE, "MQTT: message callback %s :: %d", message->topic, message->payloadlen); } else { syslog(LOG_NOTICE, "MQTT: message callback %s (null)", message->topic); } } void publisher(struct mosquitto *my_mosq, char *topic, char *payload, bool retain) { // publish the data if (payload) mosquitto_publish(my_mosq, &mqtt_mid_sent, topic, strlen(payload), payload, mqtt_qos, retain); else mosquitto_publish(my_mosq, &mqtt_mid_sent, topic, 0, NULL, mqtt_qos, retain); } void pub_domoticz_temp(int idx, int value) { char *dload = NULL; char sidx[10], sval[20]; if (idx == 0) return; sprintf(sidx, "%d", idx); sprintf(sval, "%.3f", value / 1000.0); dload = xstrcpy((char *)"{\"command\":\"udevice\",\"idx\":"); dload = xstrcat(dload, sidx); dload = xstrcat(dload, (char *)",\"nvalue\":0,\"svalue\":\""); dload = xstrcat(dload, sval); dload = xstrcat(dload, (char *)"\"}"); publisher(mosq, (char *)"domoticz/in", dload, false); free(dload); dload = NULL; } void pub_domoticz_output(int idx, int value) { char *dload = NULL; char sidx[10], sval[10]; if (idx == 0) return; sprintf(sidx, "%d", idx); sprintf(sval, "%d", value); dload = xstrcpy((char *)"{\"command\":\"udevice\",\"idx\":"); dload = xstrcat(dload, sidx); dload = xstrcat(dload, (char *)",\"nvalue\":"); if (value >= 50) dload = xstrcat(dload, (char *)"1"); else dload = xstrcat(dload, (char *)"0"); dload = xstrcat(dload, (char *)",\"svalue\":\""); dload = xstrcat(dload, sval); dload = xstrcat(dload, (char *)"\"}"); publisher(mosq, (char *)"domoticz/in", dload, false); free(dload); dload = NULL; } char *unit_data(units_list *unit, bool birth) { char *payload = NULL; char buf[128]; bool comma = false; prof_step *pstep; payload = xstrcpy((char *)"{"); /* * Fixed unit values, never change these! */ if (birth) { payload = xstrcat(payload, (char *)"\"uuid\":\""); payload = xstrcat(payload, unit->uuid); payload = xstrcat(payload, (char *)"\",\"alias\":\""); payload = xstrcat(payload, unit->alias); payload = xstrcat(payload, (char *)"\","); } /* * Product (beer) loaded information. */ if ((unit->product_name && strlen(unit->product_name)) || (unit->product_code && strlen(unit->product_code)) || (unit->product_uuid && strlen(unit->product_uuid))) { comma = false; payload = xstrcat(payload, (char *)"\"product\":{"); if (unit->product_uuid && strlen(unit->product_uuid) && strcmp((char *)"(null)", unit->product_uuid)) { payload = xstrcat(payload, (char *)"\"uuid\":\""); payload = xstrcat(payload, unit->product_uuid); payload = xstrcat(payload, (char *)"\""); comma = true; } if (unit->product_code && strlen(unit->product_code)) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"code\":\""); payload = xstrcat(payload, unit->product_code); payload = xstrcat(payload, (char *)"\""); comma = true; } if (unit->product_name && strlen(unit->product_name)) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"name\":\""); payload = xstrcat(payload, unit->product_name); payload = xstrcat(payload, (char *)"\""); } sprintf(buf, "%.1f", unit->yeast_lo); payload = xstrcat(payload, (char *)",\"yeast_lo\":"); payload = xstrcat(payload, buf); sprintf(buf, "%.1f", unit->yeast_hi); payload = xstrcat(payload, (char *)",\"yeast_hi\":"); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } /* * Air temperature sensor */ if (unit->air_address) { payload = xstrcat(payload, (char *)",\"air\":{\"address\":\""); payload = xstrcat(payload, unit->air_address); payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)DEVPRESENT[unit->air_state]); payload = xstrcat(payload, (char *)"\",\"temperature\":"); sprintf(buf, "%.3f", unit->air_temperature / 1000.0); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->air_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_temp(unit->air_idx, unit->air_temperature); } else { payload = xstrcat(payload, (char *)",\"air\":null"); } /* * Beer temperature sensor */ if (unit->beer_address) { payload = xstrcat(payload, (char *)",\"beer\":{\"address\":\""); payload = xstrcat(payload, unit->beer_address); if (unit->beer_address2) { payload = xstrcat(payload, (char *)"\",\"address2\":\""); payload = xstrcat(payload, unit->beer_address2); } payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)DEVPRESENT[unit->beer_state]); payload = xstrcat(payload, (char *)"\",\"temperature\":"); sprintf(buf, "%.3f", unit->beer_temperature / 1000.0); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->beer_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_temp(unit->beer_idx, unit->beer_temperature); } else { payload = xstrcat(payload, (char *)",\"beer\":null"); } /* * External chiller temperature sensor */ if (unit->chiller_address) { payload = xstrcat(payload, (char *)",\"chiller\":{\"address\":\""); payload = xstrcat(payload, unit->chiller_address); payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)DEVPRESENT[unit->chiller_state]); payload = xstrcat(payload, (char *)"\",\"temperature\":"); sprintf(buf, "%.3f", unit->chiller_temperature / 1000.0); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->chiller_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_temp(unit->chiller_idx, unit->chiller_temperature); } else { payload = xstrcat(payload, (char *)",\"chiller\":null"); } /* * Heater control, power 0..100% and usage count. */ if (unit->heater_address) { payload = xstrcat(payload, (char *)",\"heater\":{\"address\":\""); payload = xstrcat(payload, unit->heater_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", unit->heater_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->heater_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->heater_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"delay\":"); sprintf(buf, "%d", unit->heater_delay); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->heater_idx, unit->heater_state); } else { payload = xstrcat(payload, (char *)",\"heater\":null"); } /* * Cooler control, power 0..100% and usage counter. */ if (unit->cooler_address) { payload = xstrcat(payload, (char *)",\"cooler\":{\"address\":\""); payload = xstrcat(payload, unit->cooler_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", unit->cooler_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->cooler_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->cooler_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"delay\":"); sprintf(buf, "%d", unit->cooler_delay); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->cooler_idx, unit->cooler_state); } else { payload = xstrcat(payload, (char *)",\"cooler\":null"); } /* * Fan control, 0..100% and usage counter. */ if (unit->fan_address) { payload = xstrcat(payload, (char *)",\"fan\":{\"address\":\""); payload = xstrcat(payload, unit->fan_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", unit->fan_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->fan_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->fan_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"delay\":"); sprintf(buf, "%d", unit->fan_delay); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->fan_idx, unit->fan_state); } else { payload = xstrcat(payload, (char *)",\"fan\":null"); } /* * Interior lights control, 0..100% and usage counter. */ if (unit->light_address) { payload = xstrcat(payload, (char *)",\"light\":{\"address\":\""); payload = xstrcat(payload, unit->light_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", unit->light_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->light_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->light_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"delay\":"); sprintf(buf, "%d", unit->light_delay); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->light_idx, unit->light_state); } else { payload = xstrcat(payload, (char *)",\"light\":null"); } /* * Door sensor. */ if (unit->door_address) { payload = xstrcat(payload, (char *)",\"door\":{\"address\":\""); payload = xstrcat(payload, unit->door_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", (unit->door_state) ? 0:1); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->door_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->door_idx, unit->door_state); } else { payload = xstrcat(payload, (char *)",\"door\":null"); } /* * PSU status */ if (unit->psu_address) { payload = xstrcat(payload, (char *)",\"psu\":{\"address\":\""); payload = xstrcat(payload, unit->psu_address); payload = xstrcat(payload, (char *)"\",\"state\":"); sprintf(buf, "%d", unit->psu_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idx\":"); sprintf(buf, "%d", unit->psu_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); pub_domoticz_output(unit->psu_idx, unit->psu_state); } else { payload = xstrcat(payload, (char *)",\"psu\":null"); } /* * Working mode and setpoints */ payload = xstrcat(payload, (char *)",\"stage\":\""); payload = xstrcat(payload, (char *)UNITSTAGE[unit->stage]); payload = xstrcat(payload, (char *)"\",\"mode\":\""); payload = xstrcat(payload, (char *)UNITMODE[unit->mode]); payload = xstrcat(payload, (char *)"\",\"setpoint\":{\"low\":"); sprintf(buf, "%.1f", unit->PID_heat->SetP); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"high\":"); sprintf(buf, "%.1f", unit->PID_cool->SetP); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"},\"alarm\":"); sprintf(buf, "%d", unit->alarm_flag); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"pidc\":{\"P\":"); sprintf(buf, "%.3f", unit->PID_cool->pGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"I\":"); sprintf(buf, "%.3f", unit->PID_cool->iGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"D\":"); sprintf(buf, "%.3f", unit->PID_cool->dGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"imax\":"); sprintf(buf, "%.1f", unit->PID_cool->iMax); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idle\":"); sprintf(buf, "%.2f", unit->PID_cool->idleRange); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"},\"pidh\":{\"P\":"); sprintf(buf, "%.3f", unit->PID_heat->pGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"I\":"); sprintf(buf, "%.3f", unit->PID_heat->iGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"D\":"); sprintf(buf, "%.3f", unit->PID_heat->dGain); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"imax\":"); sprintf(buf, "%.1f", unit->PID_heat->iMax); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"idle\":"); sprintf(buf, "%.2f", unit->PID_heat->idleRange); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); /* * Loaded profile and state */ if (unit->profile_uuid) { payload = xstrcat(payload, (char *)",\"profile\":{\"uuid\":\""); payload = xstrcat(payload, unit->profile_uuid); payload = xstrcat(payload, (char *)"\",\"name\":\""); payload = xstrcat(payload, unit->profile_name); payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)PROFSTATE[unit->prof_state]); payload = xstrcat(payload, (char *)"\",\"percent\":"); sprintf(buf, "%d", unit->prof_percent); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"inittemp\":{\"low\":"); sprintf(buf, "%.1f", unit->profile_inittemp_lo); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"high\":"); sprintf(buf, "%.1f", unit->profile_inittemp_hi); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"},\"fridgemode\":"); sprintf(buf, "%d", unit->profile_fridge_mode); payload = xstrcat(payload, buf); comma = false; if (unit->profile_steps) { payload = xstrcat(payload, (char *)",\"steps\":["); for (pstep = unit->profile_steps; pstep; pstep = pstep->next) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"{\"resttime\":"); sprintf(buf, "%d", pstep->resttime); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"steptime\":"); sprintf(buf, "%d", pstep->steptime); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"target\":{\"low\":"); sprintf(buf, "%.1f", pstep->target_lo); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"high\":"); sprintf(buf, "%.1f", pstep->target_hi); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"},\"fridgemode\":"); sprintf(buf, "%d", pstep->fridge_mode); payload = xstrcat(payload, buf); if (pstep->name) { payload = xstrcat(payload, (char *)",\"name\":\""); payload = xstrcat(payload, pstep->name); payload = xstrcat(payload, (char *)"\""); } payload = xstrcat(payload, (char *)"}"); comma = true; } payload = xstrcat(payload, (char *)"]"); } else { payload = xstrcat(payload, (char *)",\"steps\":null"); } payload = xstrcat(payload, (char *)"}"); } else { payload = xstrcat(payload, (char *)",\"profile\":null"); } payload = xstrcat(payload, (char *)"}"); return payload; } /** * @brief Publish DBIRTH for all active units. If there are no active units, don't * publish anything. This function should be called at program start. */ void publishDBirthAll(void) { char *topic = NULL, *payload = NULL, *payloadu = NULL; units_list *unit; int comma = FALSE; payload = payload_header(); payload = xstrcat(payload, (char *)"{\"units\":["); for (unit = Config.units; unit; unit = unit->next) { // if (unit->mode != UNITMODE_OFF) { if (comma) payload = xstrcat(payload, (char *)","); payloadu = unit_data(unit, true); payload = xstrcat(payload, payloadu); comma = TRUE; free(payloadu); payloadu = NULL; // } } if (comma) { // Only publish if there is at least one unit active. payload = xstrcat(payload, (char *)"]}}"); topic = topic_base((char *)"DBIRTH"); publisher(mosq, topic, payload, true); free(topic); topic = NULL; } free(payload); payload = NULL; } void publishDData(units_list *unit) { char *payload = NULL, *payloadu = NULL, *topic = NULL; if (mqtt_use) { payload = payload_header(); payloadu = unit_data(unit, false); payload = xstrcat(payload, payloadu); payload = xstrcat(payload, (char *)"}"); topic = xstrcat(topic_base((char *)"DDATA"), (char *)"/"); topic = xstrcat(topic, unit->alias); publisher(mosq, topic, payload, false); free(payload); payload = NULL; free(payloadu); payloadu = NULL; free(topic); topic = NULL; } /* * Build and send websocket message. */ payload = xstrcpy((char *)"{\"type\":\"fermenter\",\"unit\":\""); payload = xstrcat(payload, unit->alias); payload = xstrcat(payload, (char *)"\",\"metric\":"); payloadu = unit_data(unit, false); payload = xstrcat(payload, payloadu); payload = xstrcat(payload, (char *)"}"); ws_broadcast(payload); free(payload); payload = NULL; free(payloadu); payloadu = NULL; unit->mqtt_flag &= ~MQTT_FLAG_DATA; } void publishDBirth(units_list *unit) { char *payload = NULL, *topic = NULL; if (mqtt_use) { payload = payload_header(); payload = xstrcat(payload, unit_data(unit, true)); payload = xstrcat(payload, (char *)"}"); topic = xstrcat(topic_base((char *)"DBIRTH"), (char *)"/"); topic = xstrcat(topic, unit->alias); publisher(mosq, topic, payload, true); free(payload); payload = NULL; free(topic); topic = NULL; } } void publishDDeath(units_list *unit) { char *topic = NULL; if (mqtt_use) { // First delete presistent DBIRTH topic topic = xstrcat(topic_base((char *)"DBIRTH"), (char *)"/"); topic = xstrcat(topic, unit->alias); publisher(mosq, topic, NULL, true); free(topic); topic = NULL; topic = xstrcat(topic_base((char *)"DDEATH"), (char *)"/"); topic = xstrcat(topic, unit->alias); publisher(mosq, topic, NULL, true); free(topic); topic = NULL; } } void publishDLog(units_list *unit) { char buf[32], *payload = NULL, *topic = NULL; bool comma = false; if (mqtt_use) { payload = payload_header(); payload = xstrcat(payload, (char *)"{"); if ((unit->product_name && strlen(unit->product_name)) || (unit->product_code && strlen(unit->product_code)) || (unit->product_uuid && strlen(unit->product_uuid))) { comma = false; payload = xstrcat(payload, (char *)"\"product\":{"); if (unit->product_uuid && strlen(unit->product_uuid) && strcmp((char *)"(null)", unit->product_uuid)) { payload = xstrcat(payload, (char *)"\"uuid\":\""); payload = xstrcat(payload, unit->product_uuid); payload = xstrcat(payload, (char *)"\""); comma = true; } if (unit->product_code && strlen(unit->product_code)) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"code\":\""); payload = xstrcat(payload, unit->product_code); payload = xstrcat(payload, (char *)"\""); comma = true; } if (unit->product_name && strlen(unit->product_name)) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"name\":\""); payload = xstrcat(payload, unit->product_name); payload = xstrcat(payload, (char *)"\""); } payload = xstrcat(payload, (char *)"}"); comma = true; } if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"stage\":\""); payload = xstrcat(payload, (char *)UNITSTAGE[unit->stage]); payload = xstrcat(payload, (char *)"\",\"mode\":\""); payload = xstrcat(payload, (char *)UNITMODE[unit->mode]); payload = xstrcat(payload, (char *)"\",\"temperature\":{"); comma = false; if (unit->air_address) { payload = xstrcat(payload, (char *)"\"air\":"); sprintf(buf, "%.3f", unit->air_temperature / 1000.0); payload = xstrcat(payload, buf); comma = true; } if (unit->beer_address) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"beer\":"); sprintf(buf, "%.3f", unit->beer_temperature / 1000.0); payload = xstrcat(payload, buf); comma = true; } if (unit->chiller_address) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"chiller\":"); sprintf(buf, "%.3f", unit->chiller_temperature / 1000.0); payload = xstrcat(payload, buf); comma = true; } if (Config.temp_uuid) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"room\":"); sprintf(buf, "%.1f", Config.temp_value / 1000.0); payload = xstrcat(payload, buf); } payload = xstrcat(payload, (char *)"},\"setpoint\":{\"low\":"); sprintf(buf, "%.1f", unit->PID_heat->SetP); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"high\":"); sprintf(buf, "%.1f", unit->PID_cool->SetP); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); if (unit->heater_address) { payload = xstrcat(payload, (char *)",\"heater\":{\"power\":"); sprintf(buf, "%d", unit->heater_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->heater_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } if (unit->cooler_address) { payload = xstrcat(payload, (char *)",\"cooler\":{\"power\":"); sprintf(buf, "%d", unit->cooler_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->cooler_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } if (unit->fan_address) { payload = xstrcat(payload, (char *)",\"fan\":{\"power\":"); sprintf(buf, "%d", unit->fan_state); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"usage\":"); sprintf(buf, "%d", unit->fan_usage); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } // sg if (unit->event_msg) { payload = xstrcat(payload, (char *)",\"event\":\""); payload = xstrcat(payload, unit->event_msg); payload = xstrcat(payload, (char *)"\""); } payload = xstrcat(payload, (char *)",\"fermenter_uuid\":\""); payload = xstrcat(payload, unit->uuid); payload = xstrcat(payload, (char *)"\"}}"); topic = xstrcat(topic_base((char *)"DLOG"), (char *)"/"); topic = xstrcat(topic, unit->alias); publisher(mosq, topic, payload, false); free(payload); payload = NULL; free(topic); topic = NULL; } } char *global_json(void) { char *payload, buf[64]; struct utsname ubuf; payload = xstrcpy((char *)"{\"type\":\"global\",\"name\":\""); payload = xstrcat(payload, Config.name); payload = xstrcat(payload, (char *)"\",\"uuid\":\""); payload = xstrcat(payload, Config.uuid); payload = xstrcat(payload, (char *)"\",\"node\":\""); if (uname(&ubuf) == 0) { payload = xstrcat(payload, ubuf.nodename); payload = xstrcat(payload, (char *)"\",\"os\":\""); payload = xstrcat(payload, ubuf.sysname); payload = xstrcat(payload, (char *)"\",\"os_version\":\""); payload = xstrcat(payload, ubuf.release); } else { payload = xstrcat(payload, (char *)"Unknown\",\"os\":\"Unknown\",\"os_version\":\"Unknown"); } payload = xstrcat(payload, (char *)"\",\"FW\":\""); payload = xstrcat(payload, (char *)VERSION); payload = xstrcat(payload, (char *)"\",\"server_port\":"); sprintf(buf, "%d", Config.server_port); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"websocket_port\":"); sprintf(buf, "%d", Config.websocket_port); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"nextunit\":"); sprintf(buf, "%d", Config.next_unit); payload = xstrcat(payload, buf); if (Config.temp_uuid || Config.hum_uuid) { payload = xstrcat(payload, (char *)",\"THB\":{"); if (Config.temp_uuid) { payload = xstrcat(payload, (char *)"\"temperature\":{\"uuid\":\""); payload = xstrcat(payload, Config.temp_uuid); payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)DEVPRESENT[Config.temp_state]); payload = xstrcat(payload, (char *)"\",\"value\":"); sprintf(buf, "%.1f", Config.temp_value / 1000.0); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } if (Config.temp_uuid && Config.hum_uuid) payload = xstrcat(payload, (char *)","); if (Config.hum_uuid) { payload = xstrcat(payload, (char *)"\"humidity\":{\"uuid\":\""); payload = xstrcat(payload, Config.hum_uuid); payload = xstrcat(payload, (char *)"\",\"state\":\""); payload = xstrcat(payload, (char *)DEVPRESENT[Config.hum_state]); payload = xstrcat(payload, (char *)"\",\"value\":"); sprintf(buf, "%.1f", Config.hum_value / 1000.0); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } if (Config.temp_uuid || Config.hum_uuid) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"index\":"); sprintf(buf, "%d", Config.temp_hum_idx); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"}"); } payload = xstrcat(payload, (char *)",\"LCD\":{\"address\":"); sprintf(buf, "%d", Config.lcd_address); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"cols\":"); sprintf(buf, "%d", Config.lcd_cols); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)",\"rows\":"); sprintf(buf, "%d", Config.lcd_rows); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"},\"MQTT\":{\"host\":\""); payload = xstrcat(payload, Config.mqtt_host); payload = xstrcat(payload, (char *)"\",\"port\":"); sprintf(buf, "%d", Config.mqtt_port); payload = xstrcat(payload, buf); if (Config.mqtt_username) { payload = xstrcat(payload, (char *)",\"username\":\""); payload = xstrcat(payload, Config.mqtt_username); payload = xstrcat(payload, (char *)"\""); if (Config.mqtt_password) { payload = xstrcat(payload, (char *)",\"password\":\""); payload = xstrcat(payload, Config.mqtt_password); payload = xstrcat(payload, (char *)"\""); } } payload = xstrcat(payload, (char *)"}}"); return payload; } void node_ws(void) { char *payload = NULL; payload = global_json(); ws_broadcast(payload); free(payload); payload = NULL; } void publishNData(bool birth, int flag) { char *topic = NULL, *payload = NULL, sidx[10], buf[64]; struct utsname ubuf; bool comma = false; int model, rev, mem, maker, warranty; payload = payload_header(); payload = xstrcat(payload, (char *)"{"); if (birth) { payload = xstrcat(payload, (char *)"\"uuid\":\""); payload = xstrcat(payload, Config.uuid); payload = xstrcat(payload, (char *)"\","); #ifdef HAVE_WIRINGPI_H /* * Get the info from the WiringPi libary */ piBoardId (&model, &rev, &mem, &maker, &warranty); payload = xstrcat(payload, (char *)"\"properties\":{\"hardwaremake\":\"Raspberry Pi "); payload = xstrcat(payload, (char *)piMakerNames[maker]); payload = xstrcat(payload, (char *)"\",\"hardwaremodel\":\""); payload = xstrcat(payload, (char *)piModelNames[model]); payload = xstrcat(payload, (char *)" rev "); payload = xstrcat(payload, (char *)piRevisionNames[rev]); payload = xstrcat(payload, (char *)"\""); #else /* * Get the info from the internal function */ piBoardId (&model, &rev, &mem, &maker, &warranty); if (model != -1) { payload = xstrcat(payload, (char *)"\"properties\":{\"hardwaremake\":\"Raspberry Pi "); payload = xstrcat(payload, (char *)piMakerNames[maker]); payload = xstrcat(payload, (char *)"\",\"hardwaremodel\":\""); payload = xstrcat(payload, (char *)piModelNames[model]); payload = xstrcat(payload, (char *)" rev "); payload = xstrcat(payload, (char *)piRevisionNames[rev]); payload = xstrcat(payload, (char *)"\""); } else if (uname(&ubuf) == 0) { payload = xstrcat(payload, (char *)"\"properties\":{\"hardwaremake\":\""); payload = xstrcat(payload, ubuf.machine); payload = xstrcat(payload, (char *)"\",\"hardwaremodel\":\"Unknown\""); } else { payload = xstrcat(payload, (char *)"\"properties\":{\"hardwaremake\":\"Unknown\",\"hardwaremodel\":\"Unknown\""); } #endif if (uname(&ubuf) == 0) { payload = xstrcat(payload, (char *)",\"os\":\""); payload = xstrcat(payload, ubuf.sysname); payload = xstrcat(payload, (char *)"\",\"os_version\":\""); payload = xstrcat(payload, ubuf.release); payload = xstrcat(payload, (char *)"\""); } else { payload = xstrcat(payload, (char *)",\"os\":\"Unknown\",\"os_version\":\"Unknown\""); } payload = xstrcat(payload, (char *)",\"FW\":\""); payload = xstrcat(payload, (char *)VERSION); payload = xstrcat(payload, (char *)"\"}"); comma = true; } if (Config.temp_uuid || Config.hum_uuid) { if (comma) payload = xstrcat(payload, (char *)","); payload = xstrcat(payload, (char *)"\"THB\":{"); if (Config.temp_uuid) { payload = xstrcat(payload, (char *)"\"temperature\":"); sprintf(buf, "%.1f", Config.temp_value / 1000.0); payload = xstrcat(payload, buf); } if (Config.temp_uuid && Config.hum_uuid) payload = xstrcat(payload, (char *)","); if (Config.hum_uuid) { payload = xstrcat(payload, (char *)"\"humidity\":"); sprintf(buf, "%.1f", Config.hum_value / 1000.0); payload = xstrcat(payload, buf); } payload = xstrcat(payload, (char *)"}"); } /* * Find our network information */ FILE *f; char line[100], *ifname, *c, ip[NI_MAXHOST]; struct ifaddrs *ifaddr, *ifa; int family, s, sock, rssi; struct iwreq wrq; struct iw_statistics iwstats; char essid[IW_ESSID_MAX_SIZE+1]; if ((f = fopen("/proc/net/route" , "r"))) { while (fgets(line, 100, f)) { ifname = strtok(line , " \t"); c = strtok(NULL , " \t"); // Take the entry with destination '00000000' if (ifname != NULL && c != NULL && (strcmp(c , "00000000") == 0)) { if (getifaddrs(&ifaddr) == -1) { syslog(LOG_NOTICE, "error getifaddrs error %d", errno); goto neterr; } //Walk through linked list, maintaining head pointer so we can free list later for (ifa = ifaddr; ifa != NULL; ifa = ifa->ifa_next) { if (ifa->ifa_addr == NULL) { continue; } family = ifa->ifa_addr->sa_family; if ((strcmp( ifa->ifa_name, ifname) == 0) && (family == AF_INET)) { s = getnameinfo(ifa->ifa_addr, (family == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6), ip, NI_MAXHOST, NULL, 0, NI_NUMERICHOST); if (s != 0) { syslog(LOG_NOTICE, "getnameinfo() error=%d for %s", errno, ifname); goto neterr; } payload = xstrcat(payload, (char *)",\"net\":{\"address\":\""); payload = xstrcat(payload, ip); payload = xstrcat(payload, (char *)"\",\"ifname\":\""); payload = xstrcat(payload, ifname); payload = xstrcat(payload, (char *)"\""); /* * Now detect WiFi on this interface. */ strncpy(wrq.ifr_name, ifa->ifa_name, IFNAMSIZ-1); if ((sock = socket(AF_INET, SOCK_DGRAM, 0)) == -1) { syslog(LOG_NOTICE, "Error %d opening INET socket", errno); } else { wrq.u.essid.pointer = essid; if (ioctl(sock, SIOCGIWESSID, &wrq) != -1) { payload = xstrcat(payload, (char *)",\"ssid\":\""); payload = xstrcat(payload, wrq.u.essid.pointer); payload = xstrcat(payload, (char *)"\""); } memset(&iwstats, 0, sizeof(iwstats)); wrq.u.data.pointer = &iwstats; wrq.u.data.length = sizeof(struct iw_statistics); wrq.u.data.flags = 1; if (ioctl(sock, SIOCGIWSTATS, &wrq) != -1) { if ((iwstats.qual.updated & (IW_QUAL_QUAL_INVALID | IW_QUAL_LEVEL_INVALID)) == 0) { /* iwstats.qual.level is __u8 */ if (iwstats.qual.updated & IW_QUAL_DBM) { rssi = (iwstats.qual.level > 63) ? iwstats.qual.level - 0x100 : iwstats.qual.level; } else { /* level seems 0..100% */ if (iwstats.qual.level > 90) rssi = -10; else rssi = (iwstats.qual.level - 100); } payload = xstrcat(payload, (char *)",\"rssi\":"); sprintf(buf, "%d", rssi); payload = xstrcat(payload, buf); } else { syslog(LOG_NOTICE, "Signal level is invalid"); } } close(sock); } payload = xstrcat(payload, (char *)"}"); } } freeifaddrs(ifaddr); } } fclose(f); } neterr: payload = xstrcat(payload, (char *)"}}"); if (birth) { topic = topic_base((char *)"NBIRTH"); publisher(mosq, topic, payload, true); } else { topic = topic_base((char *)"NDATA"); publisher(mosq, topic, payload, false); } free(topic); topic = NULL; free(payload); payload = NULL; if ((Config.temp_uuid || Config.hum_uuid) && Config.temp_hum_idx) { sprintf(sidx, "%d", Config.temp_hum_idx); sprintf(buf, "%.1f;%.1f;0", Config.temp_value / 1000.0, Config.hum_value / 1000.0); payload = xstrcpy((char *)"{\"command\":\"udevice\",\"idx\":"); payload = xstrcat(payload, sidx); payload = xstrcat(payload, (char *)",\"nvalue\":0,\"svalue\":\""); payload = xstrcat(payload, buf); payload = xstrcat(payload, (char *)"\"}"); publisher(mosq, (char *)"domoticz/in", payload, false); free(payload); payload = NULL; } node_ws(); } void mqtt_connect(void) { char *id = NULL, *topic; char err[1024]; int rc; /* * Initialize mosquitto communication */ gethostname(my_hostname, 255); if (strchr(my_hostname, '.')) { char *p = strchr(my_hostname, '.'); *p = '\0'; } mosquitto_lib_init(); id = xstrcpy((char *)"thermferm/"); id = xstrcat(id, my_hostname); if (strlen(id) > MOSQ_MQTT_ID_MAX_LENGTH) { /* * Enforce maximum client id length of 23 characters */ id[MOSQ_MQTT_ID_MAX_LENGTH] = '\0'; } mosq = mosquitto_new(id, TRUE, NULL); if (!mosq) { switch(errno) { case ENOMEM: syslog(LOG_NOTICE, "MQTT: mosquitto_new: Out of memory"); break; case EINVAL: syslog(LOG_NOTICE, "MQTT: mosquitto_new: Invalid id"); break; } mosquitto_lib_cleanup(); return; } free(id); id = NULL; /* * Set our will */ topic = topic_base((char *)"NDEATH"); if ((rc = mosquitto_will_set(mosq, topic, 0, NULL, mqtt_qos, false))) { if (rc > MOSQ_ERR_SUCCESS) syslog(LOG_NOTICE, "MQTT: mosquitto_will_set: %s", mosquitto_strerror(rc)); mosquitto_lib_cleanup(); return; } free(topic); if (debug) mosquitto_log_callback_set(mosq, my_log_callback); mosquitto_max_inflight_messages_set(mosq, max_inflight); mosquitto_connect_callback_set(mosq, my_connect_callback); mosquitto_disconnect_callback_set(mosq, my_disconnect_callback); mosquitto_publish_callback_set(mosq, my_publish_callback); mosquitto_message_callback_set(mosq, my_message_callback); mosquitto_subscribe_callback_set(mosq, my_subscribe_callback); if (Config.mqtt_username) { if (Config.mqtt_password) { rc = mosquitto_username_pw_set(mosq, Config.mqtt_username, Config.mqtt_password); } else { rc = mosquitto_username_pw_set(mosq, Config.mqtt_username, NULL); } if (rc == MOSQ_ERR_INVAL) { syslog(LOG_NOTICE, "MQTT: mosquitto_username_pw_set: Invalid id"); } else if (rc == MOSQ_ERR_NOMEM) { syslog(LOG_NOTICE, "MQTT: mosquitto_username_pw_set: Out of memory"); } if (rc != MOSQ_ERR_SUCCESS) { mosquitto_lib_cleanup(); return; } } if ((rc = mosquitto_connect(mosq, Config.mqtt_host, Config.mqtt_port, keepalive))) { if (rc == MOSQ_ERR_ERRNO) { strerror_r(errno, err, 1024); syslog(LOG_NOTICE, "MQTT: mosquitto_connect: error: %s", err); } else { syslog(LOG_NOTICE, "MQTT: mosquitto_connect: unable to connect (%d)", rc); } mosquitto_lib_cleanup(); syslog(LOG_NOTICE, "MQTT: will run without an MQTT broker."); } else { mqtt_use = TRUE; syslog(LOG_NOTICE, "MQTT: connected with %s:%d", Config.mqtt_host, Config.mqtt_port); /* * Initialise is complete, report our presence state */ mosquitto_loop_start(mosq); publishNData(true, 0); } } void mqtt_disconnect(void) { int rc; char *topic; if (mqtt_use) { /* * Final publish 0 to clients/<hostname>/thermferm/state * After that, remove the retained topic. */ syslog(LOG_NOTICE, "MQTT disconnecting"); topic = topic_base((char *)"DBIRTH"); publisher(mosq, topic, NULL, true); // Not always needed, but ... free(topic); topic = topic_base((char *)"DDEATH"); publisher(mosq, topic, NULL, true); free(topic); topic = topic_base((char *)"NBIRTH"); publisher(mosq, topic, NULL, true); free(topic); topic = topic_base((char *)"NDEATH"); publisher(mosq, topic, NULL, true); free(topic); mqtt_last_mid = mqtt_mid_sent; mqtt_status = STATUS_WAITING; mqtt_my_shutdown = TRUE; do { if (mqtt_status == STATUS_WAITING) { if (mqtt_last_mid_sent == mqtt_last_mid && mqtt_disconnect_sent == FALSE) { mosquitto_disconnect(mosq); mqtt_disconnect_sent = TRUE; } mDelay(100); } rc = MOSQ_ERR_SUCCESS; } while (rc == MOSQ_ERR_SUCCESS && mqtt_connected); mosquitto_loop_stop(mosq, FALSE); mosquitto_destroy(mosq); mosquitto_lib_cleanup(); mqtt_use = FALSE; mqtt_status = STATUS_CONNECTING; mqtt_mid_sent = 0; mqtt_last_mid = -1; mqtt_last_mid_sent = -1; mqtt_connected = TRUE; mqtt_disconnect_sent = FALSE; mqtt_connect_lost = FALSE; mqtt_my_shutdown = FALSE; syslog(LOG_NOTICE, "MQTT disconnected"); } }
