diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index c41ba038..74fcf0de 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -779,6 +779,13 @@ void MyMesh::begin(FILESYSTEM *fs) { // Set build date bridge.setBuildDate(getBuildDate()); +#ifdef WITH_MQTT_BRIDGE + // Set stats sources for automatic stats collection (optional - can be done in custom initialization) + // This enables stats to be included in status messages automatically + // this (Mesh*) inherits from Dispatcher, so it can be passed as Dispatcher* + bridge.setStatsSources(this, _radio, _cli.getBoard(), _ms); +#endif + bridge.begin(); } #endif diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index d50c10ca..766582d3 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -471,7 +471,9 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch } else if (memcmp(config, "mqtt.tx", 7) == 0) { sprintf(reply, "> %s", _prefs->mqtt_tx_enabled ? "on" : "off"); } else if (memcmp(config, "mqtt.interval", 13) == 0) { - sprintf(reply, "> %d", (uint32_t)_prefs->mqtt_status_interval); + // Display interval in minutes (rounded) + uint32_t minutes = (_prefs->mqtt_status_interval + 29999) / 60000; // Round up + sprintf(reply, "> %u minutes (%lu ms)", minutes, _prefs->mqtt_status_interval); } else if (memcmp(config, "mqtt.server", 11) == 0) { sprintf(reply, "> %s", _prefs->mqtt_server); } else if (memcmp(config, "mqtt.port", 9) == 0) { @@ -714,14 +716,16 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch _prefs->mqtt_tx_enabled = memcmp(&config[8], "on", 2) == 0; savePrefs(); strcpy(reply, "OK"); - } else if (memcmp(config, "mqtt.interval ", 15) == 0) { - uint32_t interval = _atoi(&config[15]); - if (interval >= 1000 && interval <= 3600000) { // 1 second to 1 hour - _prefs->mqtt_status_interval = interval; + } else if (memcmp(config, "mqtt.interval ", 14) == 0) { + uint32_t minutes = _atoi(&config[14]); + if (minutes >= 1 && minutes <= 60) { // 1 minute to 60 minutes + _prefs->mqtt_status_interval = minutes * 60000; // Convert minutes to milliseconds savePrefs(); - strcpy(reply, "OK"); + // Restart bridge to pick up new interval value + _callbacks->restartBridge(); + sprintf(reply, "OK - interval set to %u minutes (%lu ms), bridge restarted", minutes, _prefs->mqtt_status_interval); } else { - strcpy(reply, "Error: interval must be between 1000-3600000 ms"); + strcpy(reply, "Error: interval must be between 1-60 minutes"); } } else if (memcmp(config, "wifi.ssid ", 10) == 0) { StrHelper::strncpy(_prefs->wifi_ssid, &config[10], sizeof(_prefs->wifi_ssid)); diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index d23867b4..b85e4a68 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -14,9 +14,16 @@ int MQTTMessageBuilder::buildStatusMessage( const char* status, const char* timestamp, char* buffer, - size_t buffer_size + size_t buffer_size, + int battery_mv, + int uptime_secs, + int errors, + int queue_len, + int noise_floor, + int tx_air_secs, + int rx_air_secs ) { - DynamicJsonDocument doc(512); + DynamicJsonDocument doc(768); // Increased size to accommodate stats JsonObject root = doc.to(); root["status"] = status; @@ -28,6 +35,34 @@ int MQTTMessageBuilder::buildStatusMessage( root["radio"] = radio; root["client_version"] = client_version; + // Add stats object if any stats are provided + if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 || + noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0) { + JsonObject stats = root.createNestedObject("stats"); + + if (battery_mv >= 0) { + stats["battery_mv"] = battery_mv; + } + if (uptime_secs >= 0) { + stats["uptime_secs"] = uptime_secs; + } + if (errors >= 0) { + stats["errors"] = errors; + } + if (queue_len >= 0) { + stats["queue_len"] = queue_len; + } + if (noise_floor > -999) { + stats["noise_floor"] = noise_floor; + } + if (tx_air_secs >= 0) { + stats["tx_air_secs"] = tx_air_secs; + } + if (rx_air_secs >= 0) { + stats["rx_air_secs"] = rx_air_secs; + } + } + size_t len = serializeJson(root, buffer, buffer_size); return (len > 0 && len < buffer_size) ? len : 0; } diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 3c96f90b..9106e20c 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -30,6 +30,13 @@ public: * @param timestamp ISO 8601 timestamp * @param buffer Output buffer for JSON string * @param buffer_size Size of output buffer + * @param battery_mv Battery voltage in millivolts (optional, -1 to omit) + * @param uptime_secs Uptime in seconds (optional, -1 to omit) + * @param errors Error flags (optional, -1 to omit) + * @param queue_len Queue length (optional, -1 to omit) + * @param noise_floor Noise floor in dBm (optional, -999 to omit) + * @param tx_air_secs TX air time in seconds (optional, -1 to omit) + * @param rx_air_secs RX air time in seconds (optional, -1 to omit) * @return Length of JSON string, or 0 on error */ static int buildStatusMessage( @@ -42,7 +49,14 @@ public: const char* status, const char* timestamp, char* buffer, - size_t buffer_size + size_t buffer_size, + int battery_mv = -1, + int uptime_secs = -1, + int errors = -1, + int queue_len = -1, + int noise_floor = -999, + int tx_air_secs = -1, + int rx_air_secs = -1 ); /** diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index f54d5ee0..02f09e53 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -47,7 +47,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCCloc _timezone(nullptr), _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0), _analyzer_us_enabled(false), _analyzer_eu_enabled(false), _identity(identity), _analyzer_us_client(nullptr), _analyzer_eu_client(nullptr), _config_valid(false), - _last_no_broker_log(0) { + _last_no_broker_log(0), _dispatcher(nullptr), _radio(nullptr), _board(nullptr), _ms(nullptr) { // Initialize default values strncpy(_origin, "MeshCore-Repeater", sizeof(_origin) - 1); @@ -140,7 +140,26 @@ void MQTTBridge::begin() { _packets_enabled = _prefs->mqtt_packets_enabled; _raw_enabled = _prefs->mqtt_raw_enabled; _tx_enabled = _prefs->mqtt_tx_enabled; - _status_interval = _prefs->mqtt_status_interval; + // Set status interval to 5 minutes (300000 ms), or use preference if set and valid + // Sanity check: interval should be between 1 second (1000ms) and 1 hour (3600000ms) + // This field may be uninitialized if preferences were saved before this field was added + if (_prefs->mqtt_status_interval >= 1000 && _prefs->mqtt_status_interval <= 3600000) { + _status_interval = _prefs->mqtt_status_interval; + MQTT_DEBUG_PRINTLN("Using preference status interval: %lu ms", _status_interval); + } else { + // Invalid or uninitialized value - fix it in preferences and use default + if (_prefs->mqtt_status_interval > 0 && _prefs->mqtt_status_interval != 300000) { + MQTT_DEBUG_PRINTLN("Invalid preference status interval: %lu ms (fixing to default 300000 ms)", + _prefs->mqtt_status_interval); + } + _prefs->mqtt_status_interval = 300000; // Fix the preference value + _status_interval = 300000; // 5 minutes default + // Note: We don't save preferences here as that should be done by the caller if needed + // This ensures the correct value is used for this session + } + + MQTT_DEBUG_PRINTLN("Status publishing: enabled=%s, interval=%lu ms", + _status_enabled ? "true" : "false", _status_interval); MQTT_DEBUG_PRINTLN("Origin: %s, IATA: %s", _origin, _iata); MQTT_DEBUG_PRINTLN("Device ID: %s", _device_id); @@ -325,16 +344,37 @@ void MQTTBridge::loop() { syncTimeWithNTP(); } - // Publish status updates - if (_status_enabled && millis() - _last_status_publish > _status_interval) { - publishStatus(); - _last_status_publish = millis(); + // Publish status updates (handle millis() overflow correctly) + if (_status_enabled) { + unsigned long now = millis(); + unsigned long elapsed = (now >= _last_status_publish) ? + (now - _last_status_publish) : + (ULONG_MAX - _last_status_publish + now + 1); + + if (elapsed >= _status_interval) { + MQTT_DEBUG_PRINTLN("Status publish timer expired (elapsed: %lu ms, interval: %lu ms)", elapsed, _status_interval); + if (publishStatus()) { + _last_status_publish = now; // Only update timer on successful publication + MQTT_DEBUG_PRINTLN("Status published successfully, next publish in %lu ms", _status_interval); + } else { + MQTT_DEBUG_PRINTLN("Status publish failed, will retry next loop"); + // If publication failed (no brokers connected), don't update timer so we retry next loop + } + } } // Memory monitoring (every 5 minutes) static unsigned long last_memory_log = 0; if (millis() - last_memory_log > 300000) { // 5 minutes logMemoryStatus(); + // Debug: Log status timer state when memory check happens + if (_status_enabled) { + unsigned long elapsed = (millis() >= _last_status_publish) ? + (millis() - _last_status_publish) : + (ULONG_MAX - _last_status_publish + millis() + 1); + MQTT_DEBUG_PRINTLN("Memory check: Status timer - elapsed: %lu ms, interval: %lu ms, next: %lu ms", + elapsed, _status_interval, _status_interval - elapsed); + } last_memory_log = millis(); } @@ -538,11 +578,22 @@ void MQTTBridge::processPacketQueue() { } } -void MQTTBridge::publishStatus() { - if (!isAnyBrokerConnected() || !_config_valid) return; +bool MQTTBridge::publishStatus() { + // Check if we have any valid destinations (custom brokers or analyzer servers) + bool has_custom_brokers = isAnyBrokerConnected() && _config_valid; + bool has_analyzer_servers = (_analyzer_us_enabled && _analyzer_us_client && _analyzer_us_client->connected()) || + (_analyzer_eu_enabled && _analyzer_eu_client && _analyzer_eu_client->connected()); - // Status messages are smaller, but use consistent buffer size - char json_buffer[512]; + MQTT_DEBUG_PRINTLN("publishStatus() called - custom_brokers: %s, analyzer_servers: %s", + has_custom_brokers ? "yes" : "no", has_analyzer_servers ? "yes" : "no"); + + if (!has_custom_brokers && !has_analyzer_servers) { + MQTT_DEBUG_PRINTLN("No destinations available for status publish"); + return false; // No destinations available + } + + // Status messages with stats can be larger (~400-500 bytes), so increase buffer size + char json_buffer[768]; // Increased from 512 to accommodate stats object char origin_id[65]; char timestamp[32]; char radio_info[64]; @@ -567,7 +618,30 @@ void MQTTBridge::publishStatus() { char client_version[64]; snprintf(client_version, sizeof(client_version), "meshcore-custom-repeater/%s", _build_date); - // Build status message + // Collect stats on-demand if sources are available + int battery_mv = -1; + int uptime_secs = -1; + int errors = -1; + int noise_floor = -999; + int tx_air_secs = -1; + int rx_air_secs = -1; + + if (_board) { + battery_mv = _board->getBattMilliVolts(); + } + if (_ms) { + uptime_secs = _ms->getMillis() / 1000; + } + if (_dispatcher) { + errors = _dispatcher->getErrFlags(); + tx_air_secs = _dispatcher->getTotalAirTime() / 1000; + rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; + } + if (_radio) { + noise_floor = (int16_t)_radio->getNoiseFloor(); + } + + // Build status message with stats int len = MQTTMessageBuilder::buildStatusMessage( _origin, origin_id, @@ -578,30 +652,70 @@ void MQTTBridge::publishStatus() { "online", timestamp, json_buffer, - sizeof(json_buffer) + sizeof(json_buffer), + battery_mv, + uptime_secs, + errors, + _queue_count, // Use current queue length + noise_floor, + tx_air_secs, + rx_air_secs ); if (len > 0) { - // Publish to all connected brokers - for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) { - if (_brokers[i].enabled && _brokers[i].connected) { - char topic[128]; - snprintf(topic, sizeof(topic), "meshcore/%s/%s/status", _iata, _device_id); - MQTT_DEBUG_PRINTLN("Publishing status to topic: %s", topic); - - // Set broker for this connection (PsychicMqttClient uses URI format) - char broker_uri[128]; - snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port); - _mqtt_client->setServer(broker_uri); - _mqtt_client->publish(topic, 1, true, json_buffer, strlen(json_buffer)); // qos=1, retained=true + bool published = false; + + // Publish to all connected custom brokers + if (_config_valid) { + for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) { + if (_brokers[i].enabled && _brokers[i].connected) { + char topic[128]; + snprintf(topic, sizeof(topic), "meshcore/%s/%s/status", _iata, _device_id); + MQTT_DEBUG_PRINTLN("Publishing status to topic: %s", topic); + + // Set broker for this connection (PsychicMqttClient uses URI format) + char broker_uri[128]; + snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port); + _mqtt_client->setServer(broker_uri); + if (_mqtt_client->publish(topic, 1, true, json_buffer, strlen(json_buffer)) > 0) { + published = true; + } + } } } - // Also publish to Let's Mesh Analyzer servers - char analyzer_topic[128]; - snprintf(analyzer_topic, sizeof(analyzer_topic), "meshcore/%s/%s/status", _iata, _device_id); - publishToAnalyzerServers(analyzer_topic, json_buffer, true); + // Always publish to Let's Mesh Analyzer servers if enabled and connected + if (has_analyzer_servers) { + char analyzer_topic[128]; + snprintf(analyzer_topic, sizeof(analyzer_topic), "meshcore/%s/%s/status", _iata, _device_id); + + // Try to publish to analyzer servers + bool analyzer_published = false; + if (_analyzer_us_enabled && _analyzer_us_client && _analyzer_us_client->connected()) { + _analyzer_us_client->publish(analyzer_topic, 1, true, json_buffer, strlen(json_buffer)); + analyzer_published = true; + MQTT_DEBUG_PRINTLN("Published status to US analyzer server"); + } + if (_analyzer_eu_enabled && _analyzer_eu_client && _analyzer_eu_client->connected()) { + _analyzer_eu_client->publish(analyzer_topic, 1, true, json_buffer, strlen(json_buffer)); + analyzer_published = true; + MQTT_DEBUG_PRINTLN("Published status to EU analyzer server"); + } + + if (analyzer_published) { + published = true; + } + } + + // Return true if we successfully published to at least one destination + if (published) { + MQTT_DEBUG_PRINTLN("Status published successfully"); + return true; + } } + + MQTT_DEBUG_PRINTLN("Status publish failed - no destinations or build failed"); + return false; // Failed to build or publish message } void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, @@ -1102,7 +1216,8 @@ void MQTTBridge::publishStatusToAnalyzerClient(PsychicMqttClient* client, const snprintf(status_topic, sizeof(status_topic), "meshcore/%s/%s/status", _iata, _device_id); // Build proper status message using MQTTMessageBuilder - char json_buffer[512]; + // Status messages with stats can be larger (~400-500 bytes) + char json_buffer[768]; // Increased from 512 to accommodate stats object char origin_id[65]; char timestamp[32]; char radio_info[64]; @@ -1127,7 +1242,30 @@ void MQTTBridge::publishStatusToAnalyzerClient(PsychicMqttClient* client, const char client_version[64]; snprintf(client_version, sizeof(client_version), "meshcore-custom-repeater/%s", _build_date); - // Build status message using MQTTMessageBuilder + // Collect stats on-demand if sources are available + int battery_mv = -1; + int uptime_secs = -1; + int errors = -1; + int noise_floor = -999; + int tx_air_secs = -1; + int rx_air_secs = -1; + + if (_board) { + battery_mv = _board->getBattMilliVolts(); + } + if (_ms) { + uptime_secs = _ms->getMillis() / 1000; + } + if (_dispatcher) { + errors = _dispatcher->getErrFlags(); + tx_air_secs = _dispatcher->getTotalAirTime() / 1000; + rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; + } + if (_radio) { + noise_floor = (int16_t)_radio->getNoiseFloor(); + } + + // Build status message using MQTTMessageBuilder with stats int len = MQTTMessageBuilder::buildStatusMessage( _origin, origin_id, @@ -1138,7 +1276,14 @@ void MQTTBridge::publishStatusToAnalyzerClient(PsychicMqttClient* client, const "online", timestamp, json_buffer, - sizeof(json_buffer) + sizeof(json_buffer), + battery_mv, + uptime_secs, + errors, + _queue_count, // Use current queue length + noise_floor, + tx_air_secs, + rx_air_secs ); if (len > 0) { @@ -1182,6 +1327,14 @@ int MQTTBridge::getQueueSize() const { return _queue_count; } +void MQTTBridge::setStatsSources(mesh::Dispatcher* dispatcher, mesh::Radio* radio, + mesh::MainBoard* board, mesh::MillisecondClock* ms) { + _dispatcher = dispatcher; + _radio = radio; + _board = board; + _ms = ms; +} + void MQTTBridge::syncTimeWithNTP() { if (!WiFi.isConnected()) { MQTT_DEBUG_PRINTLN("Cannot sync time - WiFi not connected"); diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 2c9d629e..387ff0e3 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -140,10 +140,16 @@ private: unsigned long _last_analyzer_eu_log; static const unsigned long ANALYZER_LOG_INTERVAL = 30000; // Log every 30 seconds max + // Optional pointers for collecting stats internally (set by mesh if available) + mesh::Dispatcher* _dispatcher; // For air times and errors + mesh::Radio* _radio; // For noise floor + mesh::MainBoard* _board; // For battery voltage + mesh::MillisecondClock* _ms; // For uptime + // Internal methods void connectToBrokers(); void processPacketQueue(); - void publishStatus(); + bool publishStatus(); // Returns true if status was successfully published void publishPacket(mesh::Packet* packet, bool is_tx, const uint8_t* raw_data = nullptr, int raw_len = 0, float snr = 0.0f, float rssi = 0.0f); @@ -330,6 +336,18 @@ public: */ int getQueueSize() const; + /** + * Set optional pointers for stats collection. + * If these are set, stats will be collected automatically when publishing status. + * + * @param dispatcher Dispatcher (or Mesh*) for air times and errors + * @param radio Radio for noise floor + * @param board MainBoard for battery voltage + * @param ms MillisecondClock for uptime + */ + void setStatsSources(mesh::Dispatcher* dispatcher, mesh::Radio* radio, + mesh::MainBoard* board, mesh::MillisecondClock* ms); + private: /** * Log memory status for debugging