diff -r 5cf6f099c897 -r 18ace27338e5 thermferm/thermferm.c --- a/thermferm/thermferm.c Sat Mar 26 14:43:31 2016 +0100 +++ b/thermferm/thermferm.c Sat Apr 30 21:06:20 2016 +0200 @@ -67,6 +67,26 @@ unsigned char RevHeatONOFF[8] = { 0b11111, 0b10101, 0b10101, 0b10001, 0b10001, 0b10101, 0b10101, 0b11111 }; // [6] reverse HEAT symbol +#ifdef HAVE_MOSQUITTO_H + +#define STATUS_CONNECTING 0 +#define STATUS_CONNACK_RECVD 1 +#define STATUS_WAITING 2 + +/* Global variables for use in callbacks. */ +static int mqtt_qos = 0; +static int mqtt_status = STATUS_CONNECTING; +static int mqtt_mid_sent = 0; +static int mqtt_last_mid = -1; +static int mqtt_last_mid_sent = -1; +static int mqtt_connected = TRUE; +static int mqtt_disconnect_sent = FALSE; +static int mqtt_connect_lost = FALSE; +static int mqtt_my_shutdown = FALSE; +static int mqtt_use = FALSE; + +#endif + int server(void); void help(void); void die(int); @@ -942,6 +962,56 @@ } +#ifdef HAVE_MOSQUITTO_H + +void my_connect_callback(struct mosquitto *mosq, void *obj, int result) +{ + if (mqtt_connect_lost) { + mqtt_connect_lost = FALSE; + syslog(LOG_NOTICE, "Reconnect: %s", mosquitto_connack_string(result)); + } + + if (!result) { + mqtt_status = STATUS_CONNACK_RECVD; + } else { + syslog(LOG_NOTICE, "my_connect_callback: %s\n", mosquitto_connack_string(result)); + } +} + + + +void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc) +{ + if (mqtt_my_shutdown) { + syslog(LOG_NOTICE, "Acknowledged DISCONNECT from %s", Config.mosq_host); + mqtt_connected = FALSE; + } else { + /* + * The remote server was brought down. We must keep running + */ + syslog(LOG_NOTICE, "Received DISCONNECT from %s, connection lost", Config.mosq_host); + mqtt_connect_lost = TRUE; + } +} + + +void my_publish_callback(struct mosquitto *mosq, void *obj, int mid) +{ + mqtt_last_mid_sent = mid; +} + + + +void my_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str) +{ + syslog(LOG_NOTICE, "MQTT: %s", str); + printf("MQTT: %s\n", str); + mqtt_my_shutdown = TRUE; +} + + +#endif + int server(void) { @@ -962,13 +1032,98 @@ float LCDair, LCDbeer, LCDspL, LCDspH; unsigned char LCDstatC, LCDstatH; int LCDunit; - +#ifdef HAVE_MOSQUITTO_H + char *id = NULL, *state = NULL, hostname[256], topic[1024], err[1024]; + struct mosquitto *mosq = NULL; + int keepalive = 60; + unsigned int max_inflight = 20; +#endif if (lockprog((char *)"thermferm")) { syslog(LOG_NOTICE, "Can't lock"); return 1; } + +#ifdef HAVE_MOSQUITTO_H + /* + * Initialize mosquitto communication + */ + gethostname(hostname, 255); + mosquitto_lib_init(); + id = xstrcpy((char *)"thermferm/"); + id = xstrcat(id, 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, "mosquitto_new: Out of memory"); + break; + case EINVAL: + syslog(LOG_NOTICE, "mosquitto_new: Invalid id"); + break; + } + mosquitto_lib_cleanup(); + return 1; + } + + if (debug) { + mosquitto_log_callback_set(mosq, my_log_callback); + } + + /* + * Set our will + */ + state = xstrcpy((char *)"clients/"); + state = xstrcat(state, hostname); + state = xstrcat(state, (char *)"/thermferm/state"); + sprintf(buf, "0"); + if ((rc = mosquitto_will_set(mosq, state, strlen(buf), buf, mqtt_qos, TRUE))) { + if (rc == MOSQ_ERR_INVAL) { + syslog(LOG_NOTICE, "mosquitto_will_set: input parameters invalid"); + } else if (rc == MOSQ_ERR_NOMEM) { + syslog(LOG_NOTICE, "mosquitto_will_set: Out of Memory"); + } else if (rc == MOSQ_ERR_PAYLOAD_SIZE) { + syslog(LOG_NOTICE, "mosquitto_will_set: invalid payload size"); + } + mosquitto_lib_cleanup(); + return rc; + } + + 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); + + if ((rc = mosquitto_connect(mosq, Config.mosq_host, Config.mosq_port, keepalive))) { + if (rc == MOSQ_ERR_ERRNO) { + strerror_r(errno, err, 1024); + syslog(LOG_NOTICE, "mosquitto_connect: error: %s", err); + } else { + syslog(LOG_NOTICE, "mosquitto_connect: unable to connect (%d)", rc); + } + mosquitto_lib_cleanup(); + } else { + mqtt_use = TRUE; + syslog(LOG_NOTICE, "MQTT connected with %s:%d", Config.mosq_host, Config.mosq_port); + + /* + * Initialise is complete, report our presence state + */ + mosquitto_loop_start(mosq); + sprintf(buf, "1"); + rc = mosquitto_publish(mosq, &mqtt_mid_sent, state, strlen(buf), buf, mqtt_qos, 1); + } +#endif + + if ((rc = devices_detect())) { syslog(LOG_NOTICE, "Detected %d new devices", rc); wrconfig(); @@ -1032,6 +1187,7 @@ } #endif + /* * Initialize units for processing */ @@ -1041,6 +1197,32 @@ */ unit->heater_state = unit->cooler_state = unit->fan_state = unit->door_state = unit->light_state = 0; unit->heater_wait = unit->cooler_wait = unit->fan_wait = unit->light_wait = 0; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (unit->mode != UNITMODE_OFF)) { + sprintf(buf, "1"); + snprintf(topic, 1023, "fermenter/%s/%s/state", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + sprintf(buf, "%s", unit->name); + snprintf(topic, 1023, "fermenter/%s/%s/name", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + sprintf(buf, "0"); + if (unit->heater_address) { + snprintf(topic, 1023, "fermenter/%s/%s/heater", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + if (unit->cooler_address) { + snprintf(topic, 1023, "fermenter/%s/%s/cooler", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + if (unit->fan_address) { + snprintf(topic, 1023, "fermenter/%s/%s/fan", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + snprintf(topic, 1023, "fermenter/%s/%s/mode", hostname, unit->uuid); + sprintf(buf, "%s", UNITMODE[unit->mode]); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif if (unit->mode == UNITMODE_PROFILE) { if (!unit->profile) syslog(LOG_NOTICE, "Starting unit `%s' in profile mode, no profile defined.", unit->name); @@ -1055,6 +1237,13 @@ } else { syslog(LOG_NOTICE, "Starting unit `%s' in off state", unit->name); } + + /* + * Initialize logfile + */ + if (unit->mode != UNITMODE_OFF) { + initlog(unit->name); + } } #ifdef HAVE_WIRINGPI_H @@ -1066,14 +1255,6 @@ piUnlock(LOCK_LCD); #endif - /* - * Initialize logfiles for each unit - */ - for (unit = Config.units; unit; unit = unit->next) { - if (unit->mode != UNITMODE_OFF) { - initlog(unit->name); - } - } do { if (my_shutdown) @@ -1128,13 +1309,20 @@ #ifdef HAVE_WIRINGPI_H piLock(LOCK_LCD); #endif - lcd_buf_write(row, "Room temp N/A ", Config.temp_value / 1000.0, 0x01); + lcd_buf_write(row, "Room temp N/A "); #ifdef HAVE_WIRINGPI_H piUnlock(LOCK_LCD); #endif if (Config.temp_address) { rc = device_in(Config.temp_address, &temp); if (rc == DEVPRESENT_YES) { +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (Config.temp_value != temp)) { + sprintf(buf, "%.1f", temp / 1000.0); + snprintf(topic, 1023, "fermenter/%s/room/temperature", hostname); + rc = mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif Config.temp_value = temp; Config.temp_state = 0; #ifdef HAVE_WIRINGPI_H @@ -1155,7 +1343,7 @@ #ifdef HAVE_WIRINGPI_H piLock(LOCK_LCD); #endif - lcd_buf_write(row, " Humidity N/A ", Config.hum_value / 1000.0); + lcd_buf_write(row, " Humidity N/A "); #ifdef HAVE_WIRINGPI_H piUnlock(LOCK_LCD); #endif @@ -1163,6 +1351,13 @@ if (Config.hum_address) { rc = device_in(Config.hum_address, &temp); if (rc == DEVPRESENT_YES) { +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (Config.hum_value != temp)) { + sprintf(buf, "%.1f", temp / 1000.0); + snprintf(topic, 1023, "fermenter/%s/room/humidity", hostname); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif Config.hum_value = temp; Config.hum_state = 0; #ifdef HAVE_WIRINGPI_H @@ -1197,6 +1392,13 @@ deviation = 40000; if ((unit->air_temperature == 0) || (unit->air_temperature && (temp > (int)unit->air_temperature - deviation) && (temp < ((int)unit->air_temperature + deviation)))) { +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (unit->air_temperature != temp)) { + sprintf(buf, "%.3f", temp / 1000.0); + snprintf(topic, 1023, "fermenter/%s/%s/air/temperature", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif unit->air_temperature = temp; unit->air_state = 0; } else { @@ -1218,7 +1420,14 @@ deviation = 40000; if ((unit->beer_temperature == 0) || (unit->beer_temperature && (temp > (int)unit->beer_temperature - deviation) && (temp < ((int)unit->beer_temperature + deviation)))) { - unit->beer_temperature = temp; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (unit->beer_temperature != temp)) { + sprintf(buf, "%.3f", temp / 1000.0); + snprintf(topic, 1023, "fermenter/%s/%s/beer/temperature", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif + unit->beer_temperature = temp; unit->beer_state = 0; } else { syslog(LOG_NOTICE, "deviation error deviation=%d, old=%d new=%d", deviation, unit->beer_temperature, temp); @@ -1240,15 +1449,30 @@ if (unit->door_address) { rc = device_in(unit->door_address, &temp); if (rc == DEVPRESENT_YES) { +#ifdef HAVE_MOSQUITTO_H + snprintf(topic, 1023, "fermenter/%s/%s/door", hostname, unit->uuid); +#endif if (temp) { if (unit->door_state == 0) { syslog(LOG_NOTICE, "Unit `%s' door closed", unit->name); unit->door_state = 1; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use) { + sprintf(buf, "closed"); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } else { if (unit->door_state) { syslog(LOG_NOTICE, "Unit `%s' door opened", unit->name); unit->door_state = 0; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use) { + sprintf(buf, "open"); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } @@ -1600,6 +1824,13 @@ if (unit->heater_state != power) { syslog(LOG_NOTICE, "Unit `%s' heater %d%% => %d%%", unit->name, unit->heater_state, power); unit->heater_state = power; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->heater_address) { + sprintf(buf, "%d", unit->heater_state); + snprintf(topic, 1023, "fermenter/%s/%s/heater", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } else { @@ -1609,6 +1840,13 @@ if (unit->heater_state) { syslog(LOG_NOTICE, "Unit `%s' heater On => Off", unit->name); unit->heater_state = 0; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->heater_address) { + sprintf(buf, "0"); + snprintf(topic, 1023, "fermenter/%s/%s/heater", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } @@ -1627,6 +1865,13 @@ if (unit->cooler_state != power) { syslog(LOG_NOTICE, "Unit `%s' cooler %d%% => %d%%", unit->name, unit->cooler_state, power); unit->cooler_state = power; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->cooler_address) { + sprintf(buf, "%d", unit->cooler_state); + snprintf(topic, 1023, "fermenter/%s/%s/cooler", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } else { @@ -1636,6 +1881,13 @@ if (unit->cooler_state) { syslog(LOG_NOTICE, "Unit `%s' cooler On => Off", unit->name); unit->cooler_state = 0; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->cooler_address) { + sprintf(buf, "0"); + snprintf(topic, 1023, "fermenter/%s/%s/cooler", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } @@ -1660,6 +1912,13 @@ if (! unit->fan_state) { syslog(LOG_NOTICE, "Unit `%s' Fan Off => On", unit->name); unit->fan_state = 100; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->fan_address) { + sprintf(buf, "100"); + snprintf(topic, 1023, "fermenter/%s/%s/fan", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } else { @@ -1669,6 +1928,13 @@ if (unit->fan_state) { syslog(LOG_NOTICE, "Unit `%s' Fan On => Off", unit->name); unit->fan_state = 0; +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && unit->fan_address) { + sprintf(buf, "0"); + snprintf(topic, 1023, "fermenter/%s/%s/fan", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif } } } @@ -1708,6 +1974,16 @@ LCDspH = unit->prof_target_hi; } } +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (seconds == 60)) { + sprintf(buf, "%.1f", LCDspH); + snprintf(topic, 1023, "fermenter/%s/%s/setpoint/high", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + sprintf(buf, "%.1f", LCDspL); + snprintf(topic, 1023, "fermenter/%s/%s/setpoint/low", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif #ifdef HAVE_WIRINGPI_H piLock(LOCK_LCD); #endif @@ -1833,6 +2109,27 @@ * Stop units processing in a neat way */ for (unit = Config.units; unit; unit = unit->next) { + +#ifdef HAVE_MOSQUITTO_H + if (mqtt_use && (unit->mode != UNITMODE_OFF)) { + sprintf(buf, "0"); + if (unit->heater_address) { + snprintf(topic, 1023, "fermenter/%s/%s/heater", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + if (unit->cooler_address) { + snprintf(topic, 1023, "fermenter/%s/%s/cooler", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + if (unit->fan_address) { + snprintf(topic, 1023, "fermenter/%s/%s/fan", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } + snprintf(topic, 1023, "fermenter/%s/%s/state", hostname, unit->uuid); + mosquitto_publish(mosq, &mqtt_mid_sent, topic, strlen(buf), buf, mqtt_qos, 1); + } +#endif + /* * Turn everything off */ @@ -1845,6 +2142,37 @@ syslog(LOG_NOTICE, "Unit `%s' stopped in mode %s", unit->name, UNITMODE[unit->mode]); } +#ifdef HAVE_MOSQUITTO_H + + if (mqtt_use) { + /* + * Final publish 0 to clients//thermferm/state + */ + syslog(LOG_NOTICE, "MQTT disconnecting"); + sprintf(buf, "0"); + mosquitto_publish(mosq, &mqtt_mid_sent, state, strlen(buf), buf, mqtt_qos, true); + mqtt_last_mid = mqtt_mid_sent; + mqtt_status = STATUS_WAITING; + + do { + if (mqtt_status == STATUS_WAITING) { + if (debug) + fprintf(stdout, (char *)"Waiting\n"); + if (mqtt_last_mid_sent == mqtt_last_mid && mqtt_disconnect_sent == FALSE) { + mosquitto_disconnect(mosq); + mqtt_disconnect_sent = TRUE; + } + usleep(100000); + } + rc = MOSQ_ERR_SUCCESS; + } while (rc == MOSQ_ERR_SUCCESS && mqtt_connected); + + mosquitto_loop_stop(mosq, FALSE); + mosquitto_destroy(mosq); + mosquitto_lib_cleanup(); + } +#endif + syslog(LOG_NOTICE, "Out of loop"); if (debug) fprintf(stdout, (char *)"Out of loop\n");