Add initial MQTT client features

platformio.ini

@@ -22,3 +22,4 @@ src_filter = +<motor-esp32/*> +<common/*>
 
 [env:primary-esp32]
 src_filter = +<primary-esp32/*> +<common/*>
+lib_deps = plapointe6/EspMQTTClient@^1.13.3

src/primary-esp32/i2c_comms.cpp

@@ -21,11 +21,13 @@ static SemaphoreHandle_t motor_status_semaphore;
 static xQueueHandle motor_status_queue;
 
 static xQueueHandle motor_control_queue;
+static xQueueHandle motor_parameter_queue;
 
 
-// Ask the I2C task to get get a status update from the motor ESP32
+// Ask the I2C task to get a status update from the motor ESP32
 void request_motor_status_update() {
     xSemaphoreGive(motor_status_semaphore);
+    // Wake up task
     xTaskNotifyGive(i2c_comms_task_handle);
 }
 
@@ -34,8 +36,18 @@ bool get_motor_status_update(motor_status_packet_t *status_packet) {
     return (xQueueReceive(motor_status_queue, status_packet, 0) == pdTRUE);
 }
 
+// Ask the I2C task to send a control packet
 void set_motor_control(motor_control_group_t new_motor_control){
     xQueueOverwrite(motor_control_queue, &new_motor_control);
+    // Wake up task
+    xTaskNotifyGive(i2c_comms_task_handle);
+}
+
+// Ask the I2C task to send a motor parameter packet
+void set_motor_parameters(motor_parameters_group_t new_motor_parameters){
+    xQueueOverwrite(motor_parameter_queue, &new_motor_parameters);
+    // Wake up task
+    xTaskNotifyGive(i2c_comms_task_handle);
 }
 
 static void i2c_comms_task(void *pvParameters) {
@@ -59,6 +71,8 @@ static void i2c_comms_task(void *pvParameters) {
             }
         }
 
+        // TODO Get pipe pressure update
+
         // Send a motor control packet if requested
         if (xQueueReceive(motor_control_queue, &motor_write_packet.control, 0)==pdTRUE){
             motor_write_packet.packet_type = MOTOR_CONTROL_PACKET;
@@ -66,6 +80,14 @@ static void i2c_comms_task(void *pvParameters) {
             Wire.write((byte *)&motor_write_packet, sizeof(motor_write_packet));
             Wire.endTransmission();
         }
+
+        // Send a motor parameter packet if requested
+        if (xQueueReceive(motor_parameter_queue, &motor_write_packet.params, 0)==pdTRUE){
+            motor_write_packet.packet_type = MOTOR_PARAMETERS_PACKET;
+            Wire.beginTransmission(MOTOR_ESP32_SLAVE_I2C_ADDR);
+            Wire.write((byte *)&motor_write_packet, sizeof(motor_write_packet));
+            Wire.endTransmission();
+        }
     }
 }
 
@@ -76,7 +98,9 @@ void i2c_comms_setup(void) {
     // Initialise our queues and semaphores
     motor_status_semaphore = xSemaphoreCreateBinary();
     motor_status_queue = xQueueCreate(1, sizeof(motor_status_packet_t));
+
     motor_control_queue = xQueueCreate(1, sizeof(motor_control_group_t));
+    motor_parameter_queue = xQueueCreate(1, sizeof(motor_parameters_group_t));
 
     // Kick off our I2C comms task
     xTaskCreate(i2c_comms_task, "MotorCtrl", I2C_TASK_STACK_SIZE, NULL, 5, &i2c_comms_task_handle);

src/primary-esp32/i2c_comms.h

@@ -9,5 +9,6 @@ bool get_motor_status_update(motor_status_packet_t *status_packet);
 void request_motor_status_update();
 
 void set_motor_control(motor_control_group_t new_motor_control);
+void set_motor_parameters(motor_parameters_group_t new_motor_parameters);
 
 #endif

src/primary-esp32/main.cpp

@@ -1,40 +1,99 @@
-#include "i2c_comms.h"
 #include <Arduino.h>
 
+#include "EspMQTTClient.h"
+
+#include "i2c_comms.h"
+
+#define MOTOR_STATUS_UPDATE_PERIOD_MS 500
+#define HEARTBEAT_PERIOD_MS           1000
+
+EspMQTTClient mqtt_client("devtest", // Wifi SSID
+                          "87usc6rs", // WiFi password
+                          "10.9.2.101",  // MQTT Broker server ip
+                          "",            // Can be omitted if not needed
+                          "",            // Can be omitted if not needed
+                          "Airseeder",   // Client name that uniquely identify your device
+                          1883           // The MQTT port, default to 1883. this line can be omitted
+);
+
 void setup() {
     Serial.setTimeout(1);
     Serial.begin(57600);
     i2c_comms_setup();
+
+    // Hopefully this lets us get missed messages during a connection.
+    mqtt_client.enableMQTTPersistence();
+
+    mqtt_client.setMqttReconnectionAttemptDelay(10000);
+    mqtt_client.setWifiReconnectionAttemptDelay(25000);
+
+    Serial.println("Connected to MQTT broker");
+}
+
+char msg_buffer[100];
+
+
+void onRollerControlMessage(const String &message) {}
+
+// Builtin function name used by EspMQTTClient
+void onConnectionEstablished() {
+    Serial.println("Connected to MQTT broker");
+
+    mqtt_client.subscribe("airseeder/set/rollercontrol", onRollerControlMessage);
+
+    sprintf(msg_buffer, "%u", mqtt_client.getConnectionEstablishedCount());
+    mqtt_client.publish("airseeder/status/connect_count", msg_buffer, true);
 }
 
 motor_status_packet_t motor_status;
 
+motor_control_group_t motor_control_group;
+
 unsigned long now = 0;
 unsigned long last_motor_status = 0;
-
+unsigned long last_heartbeat = 0;
-motor_control_group_t motor_control_group;
 
 void loop() {
     now = millis();
 
-    if (now - last_motor_status > 500) {
+    if (now - last_motor_status > MOTOR_STATUS_UPDATE_PERIOD_MS) {
         last_motor_status = now;
         request_motor_status_update();
-        Serial.println("--Rq motor update--");
     }
 
     if (get_motor_status_update(&motor_status)) {
-        Serial.print("Dsr:");
+        // Message format is "[timestamp] [enabled] [desired_speed] [actual_speed] [revs] [motor_duty] [motor_current]".
-        Serial.print(motor_status.motor_1_control.desired_speed);
+        // [timestamp] is an unsigned integer, the number of milliseconds since power on
-        Serial.print(" Rsp:");
+        // [enabled] is a boolean, written as "0" or "1". Shows if the motor is active at all
-        Serial.print(motor_status.roller_1_speed);
+        // [desired_speed] and [actual_speed] are floats, in revs/second
-        Serial.print("   Err:");
+        // [revs] is a float, showing a count of total revolutions, wrapping at 10,000.
-        Serial.print(motor_status.motor_1_control.desired_speed - motor_status.roller_1_speed);
+        // [motor_duty] is a float from 0.00 to 1.00 representing how much of the available PWM duty cycle the motor is
-        Serial.print(" Mdty:");
+        // being driven with. [motor_current] is an estimated average current draw of the motor in amps (a float).
-        Serial.println(motor_status.motor_1_duty);
+        // Floats are given with 2 digits after the decimal (dynamic before the decimal)
+
+        sprintf(msg_buffer, "%lu %d %.2f %.2f %.2f %.2f 0.00", now, motor_status.motor_1_control.enabled,
+                motor_status.motor_1_control.desired_speed, motor_status.roller_1_speed, motor_status.roller_1_revs,
+                motor_status.motor_1_duty);
+        mqtt_client.publish("airseeder/status/roller1", msg_buffer, true);
+        sprintf(msg_buffer, "%lu %d %.2f %.2f %.2f %.2f 0.00", now, motor_status.motor_2_control.enabled,
+                motor_status.motor_2_control.desired_speed, motor_status.roller_2_speed, motor_status.roller_2_revs,
+                motor_status.motor_2_duty);
+        mqtt_client.publish("airseeder/status/roller2", msg_buffer, true);
+        sprintf(msg_buffer, "%lu %d %.2f %.2f %.2f %.2f 0.00", now, motor_status.motor_3_control.enabled,
+                motor_status.motor_3_control.desired_speed, motor_status.roller_3_speed, motor_status.roller_3_revs,
+                motor_status.motor_3_duty);
+        mqtt_client.publish("airseeder/status/roller3", msg_buffer, true);
+    }
+
+    if (now - last_heartbeat > HEARTBEAT_PERIOD_MS) {
+        last_heartbeat = now;
+        sprintf(msg_buffer, "%lu", now);
+        mqtt_client.publish("airseeder/status/heartbeat", msg_buffer, true);
     }
 
     motor_control_group.motor_1.enabled = true;
     motor_control_group.motor_1.desired_speed = ((millis() % 10000) / 10000.0) * 0.65;
     set_motor_control(motor_control_group);
+
+    mqtt_client.loop();
 }