diff --git a/include/config.h b/include/config.h index a8191d7..c592ff0 100644 --- a/include/config.h +++ b/include/config.h @@ -37,14 +37,14 @@ // Device Identity // ============================================================================ #define CLIENT_ID "tanksensor" -#define DEVICE_VERSION "v1.0.0" +#define DEVICE_VERSION "v2.0.0" +#define MDNS_HOSTNAME "dusa" +#define OTA_PORT 80 // ============================================================================ // ADC Configuration // ============================================================================ #define ADC_RESOLUTION 12 // 12-bit ADC (0-4095) -#define ADC_MAX 4095 -#define ADC_VREF 3.3f // ADC reading parameters #define ADC_SAMPLES 64 // Number of samples to average @@ -63,7 +63,6 @@ // ============================================================================ #define READ_INTERVAL_MS 5000 // Read ADC every 5 seconds #define PUBLISH_INTERVAL_MS 30000 // Publish to MQTT every 30 seconds -#define HEARTBEAT_INTERVAL_MS 60000 // Keepalive every 60 seconds // Connection timeouts #define WIFI_CONNECT_TIMEOUT_MS 30000 // WiFi connection timeout @@ -74,6 +73,9 @@ #define WIFI_RECONNECT_DELAY_MS 5000 // Delay between WiFi reconnect attempts #define MQTT_RECONNECT_DELAY_MS 2000 // Delay between MQTT reconnect attempts +// Watchdog +#define WDT_TIMEOUT_S 30 // Hardware watchdog timeout (seconds) + // ============================================================================ // Pin Assignments (XIAO ESP32-S3) // ============================================================================ diff --git a/platformio.ini b/platformio.ini index 449c525..7897b36 100644 --- a/platformio.ini +++ b/platformio.ini @@ -1,9 +1,12 @@ ; PlatformIO configuration for ESP32-S3 Tank Sensor -; +; ; To use: pio run -t upload ; Monitor: pio device monitor -[env:xiao_esp32s3] +[platformio] +default_envs = xiao_esp32s3 + +[env] platform = espressif32 board = seeed_xiao_esp32s3 framework = arduino @@ -12,15 +15,29 @@ framework = arduino monitor_speed = 115200 ; Library dependencies -lib_deps = +lib_deps = knolleary/PubSubClient@^2.8 bblanchon/ArduinoJson@^7.0.0 + GxProjectorClient=https://github.com/nuketownada/gx-projector-client.git#v0.1.0 ; Build flags ; Set credentials via environment variables: ; WIFI_SSID=MyNetwork WIFI_PASS=secret pio run build_flags = -DARDUINO_USB_CDC_ON_BOOT=1 +; ============================================================================ +; USB Upload (default) - for initial flash or when device is local +; ============================================================================ +[env:xiao_esp32s3] ; Upload settings (adjust port as needed) ; upload_port = /dev/ttyACM0 ; Linux ; upload_port = COM3 ; Windows + +; ============================================================================ +; HTTP OTA Upload - for remote updates (board must run OTA-capable firmware) +; Usage: pio run -e ota -t upload +; ============================================================================ +[env:ota] +; Custom upload via HTTP POST (mDNS name from MDNS_HOSTNAME, port 80) +upload_protocol = custom +upload_command = curl --fail --show-error --progress-bar -X POST -F "firmware=@$SOURCE" http://dusa.local/update diff --git a/src/main.cpp b/src/main.cpp index 40226bd..ae74ea0 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -15,8 +15,13 @@ #include #endif #include +#include +#include +#include #include #include +#include +#include // ============================================================================ // Data Structures @@ -34,18 +39,14 @@ struct TankConfig { struct TankState { int rawADC; // Raw ADC reading int level; // Level percentage (0-100) - int lastPublishedLevel; - bool registered; - char topicPath[64]; // W/portalId/tank/N path from registration - int deviceInstance; }; -// State machine states +// State machine states (registration itself rides gxClient.connect(); WAIT polls +// the async DBus binding) enum DeviceState { STATE_INIT, STATE_WIFI_CONNECT, STATE_MQTT_CONNECT, - STATE_REGISTER, STATE_WAIT_REGISTRATION, STATE_RUNNING }; @@ -64,6 +65,14 @@ const TankConfig tanks[] = { }; const int TANK_COUNT = sizeof(tanks) / sizeof(tanks[0]); +// gx-projector-client service table: one Victron "tank" dbus service per tank +// (bus names become tank.mqtt_tanksensor_) +const gx::ServiceDef gxServices[] = { + { TANK_FRESH_SERVICE_ID, "tank" }, + { TANK_GREY_SERVICE_ID, "tank" }, + { TANK_BLACK_SERVICE_ID, "tank" }, +}; + // ============================================================================ // Global State // ============================================================================ @@ -77,16 +86,21 @@ PubSubClient mqttClient(wifiClient); TankState tankStates[TANK_COUNT]; DeviceState deviceState = STATE_INIT; +WebServer otaServer(OTA_PORT); +bool otaInitialized = false; + +// The gx-device-projector contract (registration/binding, online liveness, cookie +// re-announce, Proxy relay) lives in the shared library. dusa has no non-contract +// topics, so every inbound MQTT message is the library's to consume. +gx::GxSession gxSession(CLIENT_ID, DEVICE_VERSION, gxServices, TANK_COUNT); +gx::GxClient gxClient(mqttClient, gxSession); -char portalId[32] = ""; -bool allTanksRegistered = false; int vrefADC = 0; // Voltage reference reading from 5V rail IPAddress mqttServerIP; // Timing unsigned long lastReadTime = 0; unsigned long lastPublishTime = 0; -unsigned long lastHeartbeatTime = 0; unsigned long stateEnteredTime = 0; unsigned long lastWiFiAttempt = 0; unsigned long lastMQTTAttempt = 0; @@ -152,7 +166,6 @@ void updateLedStatus() { case STATE_MQTT_CONNECT: ledBlink(125); // 4Hz - fast blink break; - case STATE_REGISTER: case STATE_WAIT_REGISTRATION: ledDoubleBlink(); break; @@ -278,6 +291,69 @@ bool resolveMqttServer() { return false; } +// ============================================================================ +// OTA Functions +// ============================================================================ + +void otaSetup() { + if (otaInitialized) return; + + MDNS.begin(MDNS_HOSTNAME); + + otaServer.on("/", HTTP_GET, []() { + otaServer.send(200, "text/html", + "Dusa OTA" + "

Dusa OTA Update

" + "

Version: " DEVICE_VERSION "

" + "
" + "

" + "" + "
" + ); + }); + + otaServer.on("/version", HTTP_GET, []() { + otaServer.send(200, "application/json", + "{\"version\":\"" DEVICE_VERSION "\",\"client_id\":\"" CLIENT_ID "\"}"); + }); + + otaServer.on("/update", HTTP_POST, []() { + bool ok = !Update.hasError(); + otaServer.sendHeader("Connection", "close"); + otaServer.send(ok ? 200 : 500, "text/plain", ok ? "OK\n" : "FAIL\n"); + if (ok) { + delay(500); + ESP.restart(); + } + }, []() { + HTTPUpload& upload = otaServer.upload(); + if (upload.status == UPLOAD_FILE_START) { + DEBUG_PRINTF("OTA update: %s\n", upload.filename.c_str()); + if (!Update.begin(UPDATE_SIZE_UNKNOWN)) { + DEBUG_PRINTF("OTA begin failed: %s\n", Update.errorString()); + } + } else if (upload.status == UPLOAD_FILE_WRITE) { + esp_task_wdt_reset(); + if (Update.write(upload.buf, upload.currentSize) != upload.currentSize) { + DEBUG_PRINTF("OTA write failed: %s\n", Update.errorString()); + } + } else if (upload.status == UPLOAD_FILE_END) { + if (Update.end(true)) { + DEBUG_PRINTF("OTA complete: %u bytes\n", upload.totalSize); + } else { + DEBUG_PRINTF("OTA end failed: %s\n", Update.errorString()); + } + } else if (upload.status == UPLOAD_FILE_ABORTED) { + Update.abort(); + DEBUG_PRINTLN("OTA aborted"); + } + }); + + otaServer.begin(); + otaInitialized = true; + DEBUG_PRINTF("OTA ready: http://%s.local/update\n", MDNS_HOSTNAME); +} + // ============================================================================ // MQTT Functions // ============================================================================ @@ -290,46 +366,45 @@ void mqttSetup() { wifiClient.setInsecure(); #endif mqttClient.setCallback(mqttCallback); - mqttClient.setBufferSize(512); // Increase for JSON payloads + // (No setBufferSize needed: gxClient.connect() floors the buffer at 1024 — + // the old 512 was borderline for a 3-tank DBus reply.) } -String buildLastWillPayload() { - JsonDocument doc; - doc["clientId"] = CLIENT_ID; - doc["connected"] = 0; - doc["version"] = DEVICE_VERSION; - doc["services"].to(); // Empty services - - String payload; - serializeJson(doc, payload); - return payload; +// Board-authored initial values for one tank's registration init, rebuilt from the +// tank table on every (re-)announce. Under v2 these seed the dbus paths once at +// device build (freakent applies init only while the device isn't already online), +// so the static identity leaves the periodic Proxy publish. +void fillTankInit(const char* tag, JsonObject init, void* ctx) { + (void)ctx; + for (int i = 0; i < TANK_COUNT; i++) { + if (strcmp(tanks[i].serviceId, tag) != 0) continue; + init["CustomName"] = tanks[i].name; + init["FluidType"] = tanks[i].fluidType; + init["Capacity"] = tanks[i].capacity / 1000.0f; // Convert L to m³ + return; + } } bool mqttConnect() { if (mqttClient.connected()) { return true; } - + unsigned long now = millis(); if (now - lastMQTTAttempt < MQTT_RECONNECT_DELAY_MS) { return false; } lastMQTTAttempt = now; - + DEBUG_PRINTF("Connecting to MQTT: %s:%d\n", mqttServerIP.toString().c_str(), MQTT_PORT); - - // Build last will message - String willTopic = String("device/") + CLIENT_ID + "/Status"; - String willPayload = buildLastWillPayload(); - - if (mqttClient.connect(CLIENT_ID, MQTT_USER, MQTT_PASS, willTopic.c_str(), 0, false, willPayload.c_str())) { + + // The lib runs the whole contract prologue: CONNECT with will = RETAINED "0" on + // device//online (a dead sensor can no longer leave ghost tanks — the old + // non-retained /Status will only reached live subscribers), retained online=1, + // subscribe DBus, publish the NON-retained v2 registration. Binding completes + // asynchronously in STATE_WAIT_REGISTRATION. + if (gxClient.connect(MQTT_USER, MQTT_PASS)) { DEBUG_PRINTLN("MQTT connected"); - - // Subscribe to DBus response topic - String subTopic = String("device/") + CLIENT_ID + "/DBus"; - mqttClient.subscribe(subTopic.c_str()); - DEBUG_PRINTF("Subscribed to: %s\n", subTopic.c_str()); - return true; } else { DEBUG_PRINTF("MQTT connect failed, rc=%d\n", mqttClient.state()); @@ -338,107 +413,9 @@ bool mqttConnect() { } void mqttCallback(char* topic, byte* payload, unsigned int length) { - // Null-terminate the payload - char jsonBuffer[512]; - if (length >= sizeof(jsonBuffer)) { - DEBUG_PRINTLN("MQTT payload too large"); - return; - } - memcpy(jsonBuffer, payload, length); - jsonBuffer[length] = '\0'; - - DEBUG_PRINTF("MQTT received [%s]: %s\n", topic, jsonBuffer); - - // Check if this is our DBus response - String expectedTopic = String("device/") + CLIENT_ID + "/DBus"; - if (String(topic) != expectedTopic) { - return; - } - - // Parse the registration response - JsonDocument doc; - DeserializationError error = deserializeJson(doc, jsonBuffer); - - if (error) { - DEBUG_PRINTF("JSON parse error: %s\n", error.c_str()); - return; - } - - // Extract portalId - if (doc["portalId"].is()) { - strlcpy(portalId, doc["portalId"] | "", sizeof(portalId)); - DEBUG_PRINTF("Portal ID: %s\n", portalId); - } - - // Extract device instances and topic paths for each tank - JsonObject deviceInstances = doc["deviceInstance"]; - JsonObject topicPaths = doc["topicPath"]; - - for (int i = 0; i < TANK_COUNT; i++) { - const char* serviceId = tanks[i].serviceId; - - if (deviceInstances[serviceId].is()) { - tankStates[i].deviceInstance = deviceInstances[serviceId]; - DEBUG_PRINTF("Tank %s instance: %d\n", serviceId, tankStates[i].deviceInstance); - } - - if (topicPaths[serviceId].is()) { - JsonObject paths = topicPaths[serviceId]; - if (paths["W"].is()) { - strlcpy(tankStates[i].topicPath, paths["W"] | "", sizeof(tankStates[i].topicPath)); - tankStates[i].registered = true; - DEBUG_PRINTF("Tank %s topic: %s\n", serviceId, tankStates[i].topicPath); - } - } - } - - // Check if all tanks are registered - allTanksRegistered = true; - for (int i = 0; i < TANK_COUNT; i++) { - if (!tankStates[i].registered) { - allTanksRegistered = false; - break; - } - } - - if (allTanksRegistered) { - DEBUG_PRINTLN("All tanks registered successfully"); - } -} - -// ============================================================================ -// Registration Functions -// ============================================================================ - -void sendRegistration() { - JsonDocument doc; - - doc["clientId"] = CLIENT_ID; - doc["connected"] = 1; - doc["version"] = DEVICE_VERSION; - - JsonObject services = doc["services"].to(); - for (int i = 0; i < TANK_COUNT; i++) { - services[tanks[i].serviceId] = "tank"; - } - - char payload[384]; - serializeJson(doc, payload, sizeof(payload)); - - String topic = String("device/") + CLIENT_ID + "/Status"; - - DEBUG_PRINTF("Publishing registration to %s: %s\n", topic.c_str(), payload); - mqttClient.publish(topic.c_str(), payload, true); // Retained -} - -void clearRegistration() { - portalId[0] = '\0'; - allTanksRegistered = false; - for (int i = 0; i < TANK_COUNT; i++) { - tankStates[i].registered = false; - tankStates[i].topicPath[0] = '\0'; - tankStates[i].deviceInstance = 0; - } + // Everything dusa receives is projector-contract traffic (DBus binding, online + // self-heal, cookie re-announce) — all consumed inside the library. + gxClient.handleMessage(topic, payload, length); } // ============================================================================ @@ -448,32 +425,22 @@ void clearRegistration() { void publishTankViaProxy(int tankIndex) { TankConfig const& tank = tanks[tankIndex]; TankState& state = tankStates[tankIndex]; - - if (!state.registered || state.topicPath[0] == '\0') { - return; - } - + float remaining = tank.capacity * state.level / 100.0f; - + + // {"topicPath":"W//tank/","values":{...}} — null when not bound. + // Static identity (CustomName/FluidType/Capacity) rides the registration init + // now; only the measurements flow per publish. JsonDocument doc; - doc["topicPath"] = state.topicPath; - - JsonObject values = doc["values"].to(); + JsonObject values = gxSession.proxyValues(doc, tank.serviceId); + if (values.isNull()) return; + values["Level"] = state.level; - values["FluidType"] = tank.fluidType; - values["Capacity"] = tank.capacity / 1000.0f; // Convert L to m³ - values["Remaining"] = remaining / 1000.0f; // Convert L to m³ - - char payload[256]; - serializeJson(doc, payload, sizeof(payload)); - - char topic[64]; - snprintf(topic, sizeof(topic), "device/%s/Proxy", CLIENT_ID); - - DEBUG_PRINTF("Publishing %s: %s\n", topic, payload); - mqttClient.publish(topic, payload); - - state.lastPublishedLevel = state.level; + values["Remaining"] = remaining / 1000.0f; // Convert L to m³ + values["Status"] = 0; // 0 = OK + + DEBUG_PRINTF("Publishing %s: %s=%d%%\n", gxSession.proxyTopic(), tank.serviceId, state.level); + gxClient.publishProxy(doc); // streamed, non-retained } void publishAllTanks() { @@ -500,14 +467,14 @@ void runStateMachine() { switch (deviceState) { case STATE_INIT: - clearRegistration(); changeState(STATE_WIFI_CONNECT); break; case STATE_WIFI_CONNECT: if (wifiIsConnected()) { DEBUG_PRINTF("WiFi connected, IP: %s\n", WiFi.localIP().toString().c_str()); - + otaSetup(); + // Resolve MQTT server hostname (supports mDNS .local names) if (resolveMqttServer()) { mqttClient.setServer(mqttServerIP, MQTT_PORT); @@ -534,52 +501,57 @@ void runStateMachine() { } if (mqttConnect()) { - changeState(STATE_REGISTER); + // gxClient.connect() already published the registration; wait for + // the async DBus binding. + changeState(STATE_WAIT_REGISTRATION); } else if (stateTime > MQTT_CONNECT_TIMEOUT_MS) { DEBUG_PRINTLN("MQTT timeout, retrying..."); stateEnteredTime = now; } break; - - case STATE_REGISTER: - if (!mqttClient.connected()) { - changeState(STATE_MQTT_CONNECT); - break; - } - - sendRegistration(); - changeState(STATE_WAIT_REGISTRATION); - break; - + case STATE_WAIT_REGISTRATION: if (!mqttClient.connected()) { - clearRegistration(); changeState(STATE_MQTT_CONNECT); break; } - - if (allTanksRegistered) { + + if (gxSession.bound()) { + DEBUG_PRINTLN("All tanks registered successfully"); changeState(STATE_RUNNING); // Force immediate publish lastPublishTime = 0; } else if (stateTime > REGISTRATION_TIMEOUT_MS) { - DEBUG_PRINTLN("Registration timeout, retrying..."); - changeState(STATE_REGISTER); + // Registration and binding are per-connection: retry means redoing + // the whole handshake on a fresh connection. + DEBUG_PRINTLN("Registration timeout, reconnecting..."); + mqttClient.disconnect(); + changeState(STATE_MQTT_CONNECT); } break; - + case STATE_RUNNING: if (!wifiIsConnected()) { changeState(STATE_WIFI_CONNECT); break; } - + if (!mqttClient.connected()) { - clearRegistration(); + resolveMqttServer(); + mqttClient.setServer(mqttServerIP, MQTT_PORT); changeState(STATE_MQTT_CONNECT); break; } - + + if (gxClient.needsRebind()) { + // A re-announce reply carried different instances: our topic bases + // are stale. Reconnect to rebind cleanly (connect() clears the flag). + DEBUG_PRINTLN("Instance rebind required, reconnecting..."); + mqttClient.disconnect(); + changeState(STATE_MQTT_CONNECT); + break; + } + // Normal operation handled in main loop break; } @@ -617,16 +589,16 @@ void processSerialCommand(String cmd) { wifiIsConnected() ? "Connected" : "Disconnected", WiFi.localIP().toString().c_str()); Serial.printf("MQTT: %s\n", mqttClient.connected() ? "Connected" : "Disconnected"); - Serial.printf("Portal ID: %s\n", portalId); + Serial.printf("Portal ID: %s\n", gxSession.portalId()); + Serial.printf("Registered: %s\n", gxSession.bound() ? "yes" : "no"); Serial.printf("Vref ADC: %d\n", vrefADC); Serial.println("\nTanks:"); for (int i = 0; i < TANK_COUNT; i++) { - Serial.printf(" %s: %d%% (ADC=%d, registered=%s, instance=%d)\n", + Serial.printf(" %s: %d%% (ADC=%d, instance=%d)\n", tanks[i].name, tankStates[i].level, tankStates[i].rawADC, - tankStates[i].registered ? "yes" : "no", - tankStates[i].deviceInstance); + gxSession.instance(tanks[i].serviceId)); } } else if (cmd == "HELP") { @@ -679,23 +651,39 @@ void setup() { for (int i = 0; i < TANK_COUNT; i++) { tankStates[i].rawADC = 0; tankStates[i].level = 0; - tankStates[i].lastPublishedLevel = -1; - tankStates[i].registered = false; - tankStates[i].topicPath[0] = '\0'; - tankStates[i].deviceInstance = 0; } - + + // Projector contract wiring: init values per tank (no message handler — dusa + // has no non-contract topics) + gxClient.setInitFiller(fillTankInit, nullptr); + ledSetup(); adcSetup(); wifiSetup(); mqttSetup(); - + + // Hardware watchdog — resets device if loop() stalls +#if ESP_ARDUINO_VERSION_MAJOR >= 3 + esp_task_wdt_config_t wdtConfig = { + .timeout_ms = WDT_TIMEOUT_S * 1000, + .idle_core_mask = 0, + .trigger_panic = true, + }; + esp_task_wdt_reconfigure(&wdtConfig); +#else + // arduino-esp32 2.x (IDF 4.4) API + esp_task_wdt_init(WDT_TIMEOUT_S, true); +#endif + esp_task_wdt_add(NULL); + DEBUG_PRINTLN("Setup complete, starting state machine"); } void loop() { unsigned long now = millis(); - + esp_task_wdt_reset(); + otaServer.handleClient(); + // Always process MQTT messages if (mqttClient.connected()) { mqttClient.loop();