From 56960e05e2859a52a0bfe978ba5894851ca0e34d Mon Sep 17 00:00:00 2001 From: acidburns Date: Wed, 18 Feb 2026 02:17:12 +0100 Subject: [PATCH] refactor: move sender role logic into sender_state_machine --- src/main.cpp | 23 +- src/sender_state_machine.cpp | 1553 ++++++++++++++++++++++++++++++++++ src/sender_state_machine.h | 44 + 3 files changed, 1604 insertions(+), 16 deletions(-) create mode 100644 src/sender_state_machine.cpp create mode 100644 src/sender_state_machine.h diff --git a/src/main.cpp b/src/main.cpp index 30010aa..4f5f3ad 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -12,6 +12,7 @@ #include "display_ui.h" #include "test_mode.h" #include "sd_logger.h" +#include "sender_state_machine.h" #include #include #ifdef ARDUINO_ARCH_ESP32 @@ -151,6 +152,7 @@ enum class TxBuildError : uint8_t { }; static TxBuildError g_last_tx_build_error = TxBuildError::None; +static SenderStateMachine g_sender_state_machine; static void watchdog_kick(); static void finish_inflight_batch(); @@ -1277,21 +1279,10 @@ void setup() { display_set_self_ids(g_short_id, g_device_id); if (g_role == DeviceRole::Sender) { - power_sender_init(); - power_configure_unused_pins_sender(); - meter_init(); -#ifdef ARDUINO_ARCH_ESP32 - if (!meter_reader_start()) { - serial_debug_printf("meter: using inline polling fallback"); - } -#endif - g_last_sample_ms = millis() - METER_SAMPLE_INTERVAL_MS; - g_last_send_ms = millis(); - g_last_sync_request_ms = millis() - SYNC_REQUEST_INTERVAL_MS; - g_time_acquired = false; - g_sender_faults_reset_after_first_sync = false; - g_sender_faults_reset_hour_utc = UINT32_MAX; - update_battery_cache(); + SenderStateMachineConfig sender_cfg = {}; + sender_cfg.short_id = g_short_id; + sender_cfg.device_id = g_device_id; + g_sender_state_machine.begin(sender_cfg); } else { power_receiver_init(); lora_receive_continuous(); @@ -1815,7 +1806,7 @@ void loop() { #endif if (g_role == DeviceRole::Sender) { - sender_loop(); + g_sender_state_machine.loop(); } else { receiver_loop(); } diff --git a/src/sender_state_machine.cpp b/src/sender_state_machine.cpp new file mode 100644 index 0000000..232edc2 --- /dev/null +++ b/src/sender_state_machine.cpp @@ -0,0 +1,1553 @@ +#include "sender_state_machine.h" + +#include +#include "config.h" +#include "data_model.h" +#include "payload_codec.h" +#include "lora_transport.h" +#include "meter_driver.h" +#include "power_manager.h" +#include "time_manager.h" +#include "wifi_manager.h" +#include "display_ui.h" +#include +#include +#ifdef ARDUINO_ARCH_ESP32 +#include +#include +#include +#include +#endif + +namespace { + +#if defined(DD3_DEBUG) +static constexpr bool DD3_DEBUG_ENABLED = true; +#else +static constexpr bool DD3_DEBUG_ENABLED = false; +#endif + +enum class SenderPhase : uint8_t { + Syncing = 0, + Normal = 1, + Catchup = 2, + WaitAck = 3 +}; + +static SenderPhase g_sender_phase = SenderPhase::Syncing; + +static DeviceRole g_role = DeviceRole::Sender; +static uint16_t g_short_id = 0; +static char g_device_id[16] = ""; + +static SenderStatus g_sender_statuses[NUM_SENDERS]; +static bool g_ap_mode = false; +static WifiMqttConfig g_cfg; +static FaultCounters g_sender_faults = {}; +static FaultCounters g_receiver_faults = {}; +static FaultCounters g_receiver_faults_published = {}; +static FaultCounters g_sender_faults_remote[NUM_SENDERS] = {}; +static FaultCounters g_sender_faults_remote_published[NUM_SENDERS] = {}; +static FaultType g_sender_last_error = FaultType::None; +static FaultType g_receiver_last_error = FaultType::None; +static FaultType g_sender_last_error_remote[NUM_SENDERS] = {}; +static FaultType g_sender_last_error_remote_published[NUM_SENDERS] = {}; +static FaultType g_receiver_last_error_published = FaultType::None; +static uint32_t g_sender_last_error_utc = 0; +static uint32_t g_sender_last_error_ms = 0; +static uint32_t g_receiver_last_error_utc = 0; +static uint32_t g_receiver_last_error_ms = 0; +static uint32_t g_sender_last_error_remote_utc[NUM_SENDERS] = {}; +static uint32_t g_sender_last_error_remote_ms[NUM_SENDERS] = {}; +static bool g_sender_discovery_sent[NUM_SENDERS] = {}; +static bool g_receiver_discovery_sent = false; + +static constexpr size_t BATCH_HEADER_SIZE = 6; +static constexpr size_t BATCH_CHUNK_PAYLOAD = LORA_MAX_PAYLOAD - BATCH_HEADER_SIZE; +static constexpr size_t BATCH_MAX_COMPRESSED = 4096; +static constexpr uint32_t BATCH_RX_MARGIN_MS = 800; + +struct BatchBuffer { + uint16_t batch_id; + bool batch_id_valid; + uint8_t count; + uint16_t attempt_count; + uint16_t valid_count; + uint16_t invalid_count; + FaultType last_error; + MeterData samples[METER_BATCH_MAX_SAMPLES]; +}; + +static BatchBuffer g_batch_queue[BATCH_QUEUE_DEPTH]; +static uint8_t g_batch_head = 0; +static uint8_t g_batch_tail = 0; +static uint8_t g_batch_count = 0; + +static MeterData g_build_samples[METER_BATCH_MAX_SAMPLES]; +static uint8_t g_build_count = 0; + +static uint32_t g_last_sample_ms = 0; +static uint32_t g_last_sample_ts_utc = 0; +static uint32_t g_last_send_ms = 0; +static uint32_t g_last_batch_send_ms = 0; +static float g_last_battery_voltage_v = NAN; +static uint8_t g_last_battery_percent = 0; +static uint32_t g_last_battery_ms = 0; +static uint16_t g_batch_id = 1; +static uint16_t g_last_sent_batch_id = 0; +static uint16_t g_last_acked_batch_id = 0; +static uint8_t g_batch_retry_count = 0; +static bool g_batch_ack_pending = false; +static uint32_t g_batch_ack_timeout_ms = BATCH_ACK_TIMEOUT_MS; +static MeterData g_inflight_samples[METER_BATCH_MAX_SAMPLES]; +static uint8_t g_inflight_count = 0; +static uint16_t g_inflight_batch_id = 0; +static bool g_inflight_active = false; +static bool g_inflight_sync_request = false; +static uint8_t g_inflight_encoded_payload[BATCH_MAX_COMPRESSED]; +static size_t g_inflight_encoded_payload_len = 0; +static uint16_t g_inflight_encoded_batch_id = 0; +static bool g_inflight_encoded_sync_request = false; +static bool g_inflight_encoded_valid = false; +static uint32_t g_last_debug_log_ms = 0; +static uint32_t g_sender_rx_window_ms = 0; +static uint32_t g_sender_sleep_ms = 0; +static uint32_t g_sender_power_log_ms = 0; +static uint32_t g_meter_queue_high_water = 0; +static uint32_t g_meter_queue_drop_count = 0; +static uint32_t g_sender_ack_timeout_total = 0; +static uint32_t g_sender_ack_retry_total = 0; +static uint32_t g_sender_ack_rtt_last_ms = 0; +static uint32_t g_sender_ack_rtt_ewma_ms = 0; +static uint32_t g_sender_ack_miss_streak = 0; +static uint32_t g_last_ack_window_log_ms = 0; +static RxRejectReason g_sender_rx_reject_reason = RxRejectReason::None; +static uint32_t g_sender_rx_reject_log_ms = 0; +static RxRejectReason g_receiver_rx_reject_reason = RxRejectReason::None; +static uint32_t g_receiver_rx_reject_log_ms = 0; +static MeterData g_last_meter_data = {}; +static bool g_last_meter_valid = false; +static uint32_t g_last_meter_rx_ms = 0; +static uint32_t g_meter_stale_seconds = 0; +static bool g_meter_time_anchor_valid = false; +static int64_t g_meter_epoch_offset = 0; +static bool g_meter_time_prev_valid = false; +static uint32_t g_meter_time_prev_seconds = 0; +static uint32_t g_meter_time_prev_rx_ms = 0; +static bool g_meter_time_jump_pending = false; +static bool g_time_acquired = false; +static bool g_sender_faults_reset_after_first_sync = false; +static uint32_t g_sender_faults_reset_hour_utc = UINT32_MAX; +static uint32_t g_last_sync_request_ms = 0; +static uint32_t g_build_attempts = 0; +static uint32_t g_build_valid = 0; +static uint32_t g_build_invalid = 0; +static constexpr uint32_t METER_SAMPLE_MAX_AGE_MS = 15000; +static constexpr uint32_t METER_TIME_DELTA_TOLERANCE_S = 2; +static constexpr int64_t METER_TIME_ANCHOR_DRIFT_TOLERANCE_S = 2; + +struct MeterSampleEvent { + MeterData data; + uint32_t rx_ms; +}; +#ifdef ARDUINO_ARCH_ESP32 +static QueueHandle_t g_meter_sample_queue = nullptr; +static TaskHandle_t g_meter_reader_task = nullptr; +static bool g_meter_reader_task_running = false; +static constexpr UBaseType_t METER_SAMPLE_QUEUE_LEN = 8; +static constexpr uint32_t METER_READER_TASK_STACK_WORDS = 4096; +static constexpr UBaseType_t METER_READER_TASK_PRIORITY = 2; +static constexpr BaseType_t METER_READER_TASK_CORE = 0; +#endif + +enum class TxBuildError : uint8_t { + None = 0, + Encode = 1 +}; + +static TxBuildError g_last_tx_build_error = TxBuildError::None; + +static void watchdog_kick(); +static void finish_inflight_batch(); +static void invalidate_inflight_encode_cache(); + +static void serial_debug_printf(const char *fmt, ...) { + if (!SERIAL_DEBUG_MODE) { + return; + } + char buf[256]; + va_list args; + va_start(args, fmt); + vsnprintf(buf, sizeof(buf), fmt, args); + va_end(args); + Serial.println(buf); +} + +#ifdef ARDUINO_ARCH_ESP32 +static void update_meter_queue_high_water() { + if (!g_meter_sample_queue) { + return; + } + uint32_t depth = static_cast(uxQueueMessagesWaiting(g_meter_sample_queue)); + if (depth > g_meter_queue_high_water) { + g_meter_queue_high_water = depth; + } +} +#endif + +static void sender_log_diagnostics(uint32_t now_ms) { + if (!SERIAL_DEBUG_MODE) { + return; + } + if (now_ms - g_last_debug_log_ms < 5000) { + return; + } + g_last_debug_log_ms = now_ms; + + MeterDriverStats meter_stats = {}; + meter_get_stats(meter_stats); + + uint32_t queue_depth = 0; +#ifdef ARDUINO_ARCH_ESP32 + if (g_meter_sample_queue) { + queue_depth = static_cast(uxQueueMessagesWaiting(g_meter_sample_queue)); + } +#endif + + uint32_t meter_age_ms = 0; + if (meter_stats.last_good_frame_ms > 0 && now_ms >= meter_stats.last_good_frame_ms) { + meter_age_ms = now_ms - meter_stats.last_good_frame_ms; + } else if (meter_stats.last_good_frame_ms == 0) { + meter_age_ms = UINT32_MAX; + } + + serial_debug_printf( + "diag: q_depth=%lu q_hi=%lu q_drop=%lu batch_q=%u build=%u ack_pending=%u ack_retry_cur=%u ack_retry_total=%lu ack_timeout_total=%lu ack_rtt_last_ms=%lu ack_rtt_ewma_ms=%lu ack_miss_streak=%lu meter_ok=%lu meter_fail=%lu meter_ovf=%lu meter_timeout=%lu meter_age_ms=%lu rx_win_ms=%lu sleep_ms=%lu", + static_cast(queue_depth), + static_cast(g_meter_queue_high_water), + static_cast(g_meter_queue_drop_count), + static_cast(g_batch_count), + static_cast(g_build_count), + g_batch_ack_pending ? 1U : 0U, + static_cast(g_batch_retry_count), + static_cast(g_sender_ack_retry_total), + static_cast(g_sender_ack_timeout_total), + static_cast(g_sender_ack_rtt_last_ms), + static_cast(g_sender_ack_rtt_ewma_ms), + static_cast(g_sender_ack_miss_streak), + static_cast(meter_stats.frames_ok), + static_cast(meter_stats.frames_parse_fail), + static_cast(meter_stats.rx_overflow), + static_cast(meter_stats.rx_timeout), + static_cast(meter_age_ms), + static_cast(g_sender_rx_window_ms), + static_cast(g_sender_sleep_ms)); +} + +static void invalidate_inflight_encode_cache() { + g_inflight_encoded_payload_len = 0; + g_inflight_encoded_batch_id = 0; + g_inflight_encoded_sync_request = false; + g_inflight_encoded_valid = false; +} + +static uint8_t bit_count32(uint32_t value) { + uint8_t count = 0; + while (value != 0) { + value &= (value - 1); + count++; + } + return count; +} + +static uint32_t abs_diff_u32(uint32_t a, uint32_t b) { + return a >= b ? (a - b) : (b - a); +} + +static void meter_time_update_snapshot(MeterData &parsed, uint32_t rx_ms) { + if (!parsed.meter_seconds_valid) { + return; + } + + bool jump = false; + const char *jump_reason = nullptr; + uint32_t delta_meter_s = 0; + uint32_t delta_wall_s = 0; + if (g_meter_time_prev_valid) { + if (parsed.meter_seconds < g_meter_time_prev_seconds) { + jump = true; + jump_reason = "rollback"; + } else { + delta_meter_s = parsed.meter_seconds - g_meter_time_prev_seconds; + uint32_t delta_wall_ms = rx_ms - g_meter_time_prev_rx_ms; + delta_wall_s = (delta_wall_ms + 500) / 1000; + if (abs_diff_u32(delta_meter_s, delta_wall_s) > METER_TIME_DELTA_TOLERANCE_S) { + jump = true; + jump_reason = "delta"; + } + } + } + + if (time_is_synced()) { + uint32_t epoch_now = time_get_utc(); + if (epoch_now >= MIN_ACCEPTED_EPOCH_UTC) { + int64_t new_offset = static_cast(epoch_now) - static_cast(parsed.meter_seconds); + if (!g_meter_time_anchor_valid || jump) { + g_meter_epoch_offset = new_offset; + g_meter_time_anchor_valid = true; + } else { + int64_t drift_s = new_offset - g_meter_epoch_offset; + if (drift_s > METER_TIME_ANCHOR_DRIFT_TOLERANCE_S || drift_s < -METER_TIME_ANCHOR_DRIFT_TOLERANCE_S) { + jump = true; + jump_reason = jump_reason ? jump_reason : "anchor"; + g_meter_epoch_offset = new_offset; + } + } + } + } + + if (g_meter_time_anchor_valid) { + int64_t epoch64 = static_cast(parsed.meter_seconds) + g_meter_epoch_offset; + if (epoch64 > 0 && epoch64 <= static_cast(UINT32_MAX)) { + parsed.ts_utc = static_cast(epoch64); + } + } + + if (jump) { + g_meter_time_jump_pending = true; + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("meter_time: jump reason=%s sec=%lu prev=%lu d_meter=%lu d_wall=%lu", + jump_reason ? jump_reason : "unknown", + static_cast(parsed.meter_seconds), + static_cast(g_meter_time_prev_seconds), + static_cast(delta_meter_s), + static_cast(delta_wall_s)); + } + } + + g_meter_time_prev_seconds = parsed.meter_seconds; + g_meter_time_prev_rx_ms = rx_ms; + g_meter_time_prev_valid = true; +} + +static void set_last_meter_sample(const MeterData &parsed_in, uint32_t rx_ms) { + MeterData parsed = parsed_in; + meter_time_update_snapshot(parsed, rx_ms); + g_last_meter_data = parsed; + g_last_meter_valid = true; + g_last_meter_rx_ms = rx_ms; + g_meter_stale_seconds = 0; +} + +static bool parse_meter_frame_sample(const char *frame, size_t frame_len, MeterData &parsed) { + parsed = {}; + parsed.energy_total_kwh = NAN; + parsed.total_power_w = NAN; + parsed.phase_power_w[0] = NAN; + parsed.phase_power_w[1] = NAN; + parsed.phase_power_w[2] = NAN; + parsed.valid = false; + return meter_parse_frame(frame, frame_len, parsed); +} + +#ifdef ARDUINO_ARCH_ESP32 +static void meter_queue_push_latest(const MeterSampleEvent &event) { + if (!g_meter_sample_queue) { + return; + } + if (xQueueSend(g_meter_sample_queue, &event, 0) == pdTRUE) { + update_meter_queue_high_water(); + return; + } + g_meter_queue_drop_count++; + MeterSampleEvent dropped = {}; + xQueueReceive(g_meter_sample_queue, &dropped, 0); + if (xQueueSend(g_meter_sample_queue, &event, 0) == pdTRUE) { + update_meter_queue_high_water(); + return; + } + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("meter: queue push failed"); + } +} + +static void meter_reader_task_entry(void *arg) { + (void)arg; + for (;;) { + const char *frame = nullptr; + size_t frame_len = 0; + if (!meter_poll_frame(frame, frame_len)) { + vTaskDelay(pdMS_TO_TICKS(5)); + continue; + } + + MeterData parsed = {}; + if (parse_meter_frame_sample(frame, frame_len, parsed)) { + MeterSampleEvent event = {}; + event.data = parsed; + event.rx_ms = millis(); + meter_queue_push_latest(event); + } + } +} + +static bool meter_reader_start() { + if (g_meter_reader_task_running) { + return true; + } + if (!g_meter_sample_queue) { + g_meter_sample_queue = xQueueCreate(METER_SAMPLE_QUEUE_LEN, sizeof(MeterSampleEvent)); + if (!g_meter_sample_queue) { + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("meter: queue alloc failed"); + } + return false; + } + } + + BaseType_t rc = xTaskCreatePinnedToCore( + meter_reader_task_entry, + "meter_reader", + METER_READER_TASK_STACK_WORDS, + nullptr, + METER_READER_TASK_PRIORITY, + &g_meter_reader_task, + METER_READER_TASK_CORE); + if (rc != pdPASS) { + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("meter: task start failed rc=%ld", static_cast(rc)); + } + return false; + } + g_meter_reader_task_running = true; + serial_debug_printf("meter: reader task core=%ld queue=%u", + static_cast(METER_READER_TASK_CORE), + static_cast(METER_SAMPLE_QUEUE_LEN)); + return true; +} +#endif + +static void meter_reader_pump(uint32_t now_ms) { +#ifdef ARDUINO_ARCH_ESP32 + if (g_meter_reader_task_running && g_meter_sample_queue) { + MeterSampleEvent event = {}; + while (xQueueReceive(g_meter_sample_queue, &event, 0) == pdTRUE) { + set_last_meter_sample(event.data, event.rx_ms); + } + return; + } +#endif + + const char *frame = nullptr; + size_t frame_len = 0; + if (!meter_poll_frame(frame, frame_len)) { + return; + } + MeterData parsed = {}; + if (parse_meter_frame_sample(frame, frame_len, parsed)) { + set_last_meter_sample(parsed, now_ms); + } +} + + +static void update_battery_cache() { + MeterData tmp = {}; + read_battery(tmp); + g_last_battery_voltage_v = tmp.battery_voltage_v; + g_last_battery_percent = tmp.battery_percent; + g_last_battery_ms = millis(); +} + +static bool battery_sample_due(uint32_t now_ms) { + return g_last_battery_ms == 0 || now_ms - g_last_battery_ms >= BATTERY_SAMPLE_INTERVAL_MS; +} + +static bool batch_queue_drop_oldest() { + if (g_batch_count == 0) { + return false; + } + bool dropped_inflight = g_inflight_active && g_batch_queue[g_batch_tail].batch_id_valid && + g_inflight_batch_id == g_batch_queue[g_batch_tail].batch_id; + if (dropped_inflight) { + g_batch_ack_pending = false; + g_batch_retry_count = 0; + g_inflight_active = false; + g_inflight_count = 0; + g_inflight_batch_id = 0; + g_inflight_sync_request = false; + invalidate_inflight_encode_cache(); + } + g_batch_tail = (g_batch_tail + 1) % BATCH_QUEUE_DEPTH; + g_batch_count--; + return dropped_inflight; +} + +static void sender_note_rx_reject(RxRejectReason reason, const char *context) { + if (reason == RxRejectReason::None) { + return; + } + g_sender_rx_reject_reason = reason; + uint32_t now_ms = millis(); + if (SERIAL_DEBUG_MODE && now_ms - g_sender_rx_reject_log_ms >= 1000) { + g_sender_rx_reject_log_ms = now_ms; + serial_debug_printf("rx_reject: %s reason=%s", context, rx_reject_reason_text(reason)); + } +} + +static void receiver_note_rx_reject(RxRejectReason reason, const char *context) { + if (reason == RxRejectReason::None) { + return; + } + g_receiver_rx_reject_reason = reason; + uint32_t now_ms = millis(); + if (SERIAL_DEBUG_MODE && now_ms - g_receiver_rx_reject_log_ms >= 1000) { + g_receiver_rx_reject_log_ms = now_ms; + serial_debug_printf("rx_reject: %s reason=%s", context, rx_reject_reason_text(reason)); + } +} + +static BatchBuffer *batch_queue_peek() { + if (g_batch_count == 0) { + return nullptr; + } + return &g_batch_queue[g_batch_tail]; +} + +static void batch_queue_enqueue(const MeterData *samples, uint8_t count) { + if (!samples || count == 0) { + return; + } + if (g_batch_count >= BATCH_QUEUE_DEPTH) { + if (batch_queue_drop_oldest()) { + g_batch_id++; + } + } + BatchBuffer &slot = g_batch_queue[g_batch_head]; + slot.batch_id = 0; + slot.batch_id_valid = false; + slot.count = count; + slot.attempt_count = static_cast(g_build_attempts); + slot.valid_count = static_cast(g_build_valid); + slot.invalid_count = static_cast(g_build_invalid); + slot.last_error = g_sender_last_error; + for (uint8_t i = 0; i < count; ++i) { + slot.samples[i] = samples[i]; + } + g_batch_head = (g_batch_head + 1) % BATCH_QUEUE_DEPTH; + g_batch_count++; +} + +static void reset_build_counters() { + g_build_attempts = 0; + g_build_valid = 0; + g_build_invalid = 0; +} + +static bool append_meter_sample(const MeterData &data, bool meter_ok, bool has_snapshot) { + if (!has_snapshot) { + g_build_invalid++; + return false; + } + g_last_sample_ts_utc = data.ts_utc; + g_build_samples[g_build_count++] = data; + if (meter_ok) { + g_build_valid++; + } else { + g_build_invalid++; + } + if (g_build_count >= METER_BATCH_MAX_SAMPLES) { + batch_queue_enqueue(g_build_samples, g_build_count); + g_build_count = 0; + reset_build_counters(); + } + return true; +} + +static uint32_t last_sample_ts() { + if (g_last_sample_ts_utc == 0) { + uint32_t now_utc = time_get_utc(); + return now_utc > 0 ? now_utc : millis() / 1000; + } + return g_last_sample_ts_utc; +} + +static void note_fault(FaultCounters &counters, FaultType &last_type, uint32_t &last_ts_utc, uint32_t &last_ts_ms, FaultType type) { + if (type == FaultType::MeterRead) { + counters.meter_read_fail++; + } else if (type == FaultType::Decode) { + counters.decode_fail++; + } else if (type == FaultType::LoraTx) { + counters.lora_tx_fail++; + } + last_type = type; + last_ts_utc = time_get_utc(); + last_ts_ms = millis(); +} + +static void clear_faults(FaultCounters &counters, FaultType &last_type, uint32_t &last_ts_utc, uint32_t &last_ts_ms) { + counters = {}; + last_type = FaultType::None; + last_ts_utc = 0; + last_ts_ms = 0; +} + +static void sender_reset_fault_stats(const char *reason, uint32_t now_utc) { + clear_faults(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + g_sender_rx_reject_reason = RxRejectReason::None; + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("faults: reset scope=sender reason=%s ts_utc=%lu", + reason ? reason : "unknown", + static_cast(now_utc)); + } +} + +static void sender_reset_fault_stats_on_first_sync(uint32_t synced_utc) { + if (g_sender_faults_reset_after_first_sync || synced_utc < MIN_ACCEPTED_EPOCH_UTC) { + return; + } + sender_reset_fault_stats("first_sync", synced_utc); + g_sender_faults_reset_after_first_sync = true; + g_sender_faults_reset_hour_utc = synced_utc / 3600U; +} + +static void sender_reset_fault_stats_on_hour_boundary() { + if (!g_time_acquired || !g_sender_faults_reset_after_first_sync) { + return; + } + uint32_t now_utc = time_get_utc(); + if (now_utc < MIN_ACCEPTED_EPOCH_UTC) { + return; + } + uint32_t now_hour_utc = now_utc / 3600U; + if (g_sender_faults_reset_hour_utc == UINT32_MAX) { + g_sender_faults_reset_hour_utc = now_hour_utc; + return; + } + if (now_hour_utc > g_sender_faults_reset_hour_utc) { + sender_reset_fault_stats("hourly", now_utc); + g_sender_faults_reset_hour_utc = now_hour_utc; + } +} + + +#ifdef ARDUINO_ARCH_ESP32 +static void watchdog_init() { + esp_task_wdt_deinit(); + esp_task_wdt_config_t config = {}; + config.timeout_ms = WATCHDOG_TIMEOUT_SEC * 1000; + config.idle_core_mask = 0; + config.trigger_panic = true; + esp_task_wdt_init(&config); + esp_task_wdt_add(nullptr); +} + +static void watchdog_kick() { + esp_task_wdt_reset(); +} +#else +static void watchdog_init() {} +static void watchdog_kick() {} +#endif + +static void write_u16_le(uint8_t *dst, uint16_t value) { + dst[0] = static_cast(value & 0xFF); + dst[1] = static_cast((value >> 8) & 0xFF); +} + +static uint16_t read_u16_le(const uint8_t *src) { + return static_cast(src[0]) | (static_cast(src[1]) << 8); +} + +static void write_u16_be(uint8_t *dst, uint16_t value) { + dst[0] = static_cast((value >> 8) & 0xFF); + dst[1] = static_cast(value & 0xFF); +} + +static uint16_t read_u16_be(const uint8_t *src) { + return static_cast(src[0] << 8) | static_cast(src[1]); +} + +static void write_u32_be(uint8_t *dst, uint32_t value) { + dst[0] = static_cast((value >> 24) & 0xFF); + dst[1] = static_cast((value >> 16) & 0xFF); + dst[2] = static_cast((value >> 8) & 0xFF); + dst[3] = static_cast(value & 0xFF); +} + +static uint32_t read_u32_be(const uint8_t *src) { + return (static_cast(src[0]) << 24) | + (static_cast(src[1]) << 16) | + (static_cast(src[2]) << 8) | + static_cast(src[3]); +} + +static uint16_t sender_id_from_short_id(uint16_t short_id) { + for (uint8_t i = 0; i < NUM_SENDERS; ++i) { + if (EXPECTED_SENDER_IDS[i] == short_id) { + return static_cast(i + 1); + } + } + return 0; +} + +static uint16_t short_id_from_sender_id(uint16_t sender_id) { + if (sender_id == 0 || sender_id > NUM_SENDERS) { + return 0; + } + return EXPECTED_SENDER_IDS[sender_id - 1]; +} + +static uint32_t kwh_to_wh_from_float(float value) { + if (isnan(value)) { + return 0; + } + double wh = static_cast(value) * 1000.0; + if (wh < 0.0) { + wh = 0.0; + } + if (wh > static_cast(UINT32_MAX)) { + wh = static_cast(UINT32_MAX); + } + return static_cast(llround(wh)); +} + +static bool float_to_i16_w(float value, int16_t &out) { + if (isnan(value)) { + out = 0; + return true; + } + long rounded = lroundf(value); + if (rounded < INT16_MIN || rounded > INT16_MAX) { + return false; + } + out = static_cast(rounded); + return true; +} + +static int16_t float_to_i16_w_clamped(float value, bool &clamped) { + clamped = false; + if (isnan(value)) { + return 0; + } + long rounded = lroundf(value); + if (rounded < INT16_MIN) { + clamped = true; + return INT16_MIN; + } + if (rounded > INT16_MAX) { + clamped = true; + return INT16_MAX; + } + return static_cast(rounded); +} + +static uint16_t battery_mv_from_voltage(float value) { + if (isnan(value) || value <= 0.0f) { + return 0; + } + long mv = lroundf(value * 1000.0f); + if (mv < 0) { + mv = 0; + } + if (mv > UINT16_MAX) { + mv = UINT16_MAX; + } + return static_cast(mv); +} + +static uint32_t compute_batch_rx_timeout_ms(uint16_t total_len, uint8_t chunk_count) { + if (total_len == 0 || chunk_count == 0) { + return 10000; + } + size_t max_chunk_payload = total_len > BATCH_CHUNK_PAYLOAD ? BATCH_CHUNK_PAYLOAD : total_len; + size_t payload_len = BATCH_HEADER_SIZE + max_chunk_payload; + size_t packet_len = 3 + payload_len + 2; + uint32_t per_chunk_toa_ms = lora_airtime_ms(packet_len); + uint32_t timeout_ms = static_cast(chunk_count) * per_chunk_toa_ms + BATCH_RX_MARGIN_MS; + return timeout_ms < 10000 ? 10000 : timeout_ms; +} + +static uint32_t compute_batch_ack_timeout_ms(size_t payload_len) { + if (payload_len == 0) { + return 10000; + } + uint8_t chunk_count = static_cast((payload_len + BATCH_CHUNK_PAYLOAD - 1) / BATCH_CHUNK_PAYLOAD); + size_t packet_len = 3 + BATCH_HEADER_SIZE + (payload_len > BATCH_CHUNK_PAYLOAD ? BATCH_CHUNK_PAYLOAD : payload_len) + 2; + uint32_t per_chunk_toa_ms = lora_airtime_ms(packet_len); + uint32_t timeout_ms = static_cast(chunk_count) * per_chunk_toa_ms + BATCH_RX_MARGIN_MS; + return timeout_ms < 10000 ? 10000 : timeout_ms; +} + +static bool send_batch_payload(const uint8_t *data, size_t len, uint32_t ts_for_display, uint16_t batch_id) { + if (!data || len == 0 || len > BATCH_MAX_COMPRESSED) { + return false; + } + uint8_t chunk_count = static_cast((len + BATCH_CHUNK_PAYLOAD - 1) / BATCH_CHUNK_PAYLOAD); + if (chunk_count == 0) { + return false; + } + + bool all_ok = true; + size_t offset = 0; + for (uint8_t i = 0; i < chunk_count; ++i) { + size_t chunk_len = len - offset; + if (chunk_len > BATCH_CHUNK_PAYLOAD) { + chunk_len = BATCH_CHUNK_PAYLOAD; + } + LoraPacket pkt = {}; + pkt.msg_kind = LoraMsgKind::BatchUp; + pkt.device_id_short = g_short_id; + pkt.payload_len = chunk_len + BATCH_HEADER_SIZE; + + uint8_t *payload = pkt.payload; + write_u16_le(&payload[0], batch_id); + payload[2] = i; + payload[3] = chunk_count; + write_u16_le(&payload[4], static_cast(len)); + memcpy(&payload[BATCH_HEADER_SIZE], data + offset, chunk_len); + + watchdog_kick(); + uint32_t tx_start = millis(); + bool ok = lora_send(pkt); + uint32_t tx_ms = millis() - tx_start; + all_ok = all_ok && ok; + if (!ok) { + note_fault(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms, FaultType::LoraTx); + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + } + if (SERIAL_DEBUG_MODE && (!ok || tx_ms > 2000)) { + serial_debug_printf("tx: chunk %u/%u took %lums ok=%u", static_cast(i + 1), + static_cast(chunk_count), static_cast(tx_ms), ok ? 1 : 0); + } + offset += chunk_len; + delay(10); + } + + display_set_last_tx(all_ok, ts_for_display); + return all_ok; +} + +static void send_batch_ack(uint16_t batch_id, uint8_t sample_count) { + uint32_t epoch = time_get_utc(); + uint8_t time_valid = (time_is_synced() && epoch >= MIN_ACCEPTED_EPOCH_UTC) ? 1 : 0; + if (!time_valid) { + epoch = 0; + } + LoraPacket ack = {}; + ack.msg_kind = LoraMsgKind::AckDown; + ack.device_id_short = g_short_id; + ack.payload_len = LORA_ACK_DOWN_PAYLOAD_LEN; + ack.payload[0] = time_valid; + write_u16_be(&ack.payload[1], batch_id); + write_u32_be(&ack.payload[3], epoch); + uint8_t repeats = ACK_REPEAT_COUNT == 0 ? 1 : ACK_REPEAT_COUNT; + for (uint8_t i = 0; i < repeats; ++i) { + lora_send(ack); + if (i + 1 < repeats && ACK_REPEAT_DELAY_MS > 0) { + delay(ACK_REPEAT_DELAY_MS); + } + } + serial_debug_printf("ack: tx batch_id=%u time_valid=%u epoch=%lu samples=%u", + batch_id, + static_cast(time_valid), + static_cast(epoch), + static_cast(sample_count)); + lora_receive_continuous(); +} + +static bool prepare_inflight_from_queue() { + if (g_inflight_active) { + return true; + } + BatchBuffer *batch = batch_queue_peek(); + if (!batch || batch->count == 0) { + return false; + } + if (!batch->batch_id_valid) { + batch->batch_id = g_batch_id; + batch->batch_id_valid = true; + } + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("batch: id=%u desired=%u attempts=%u valid=%u invalid=%u err_last=%u", + batch->batch_id, + static_cast(METER_BATCH_MAX_SAMPLES), + static_cast(batch->attempt_count), + static_cast(batch->valid_count), + static_cast(batch->invalid_count), + static_cast(batch->last_error)); + } + g_inflight_count = batch->count; + g_inflight_batch_id = batch->batch_id; + for (uint8_t i = 0; i < g_inflight_count; ++i) { + g_inflight_samples[i] = batch->samples[i]; + } + g_inflight_active = true; + return true; +} + +static bool send_inflight_batch(uint32_t ts_for_display) { + g_last_tx_build_error = TxBuildError::None; + if (!g_inflight_active) { + return false; + } + + bool cache_match = g_inflight_encoded_valid && + g_inflight_encoded_batch_id == g_inflight_batch_id && + g_inflight_encoded_sync_request == g_inflight_sync_request; + if (cache_match) { + g_batch_ack_timeout_ms = compute_batch_ack_timeout_ms(g_inflight_encoded_payload_len); + uint32_t send_start = millis(); + bool ok = send_batch_payload(g_inflight_encoded_payload, g_inflight_encoded_payload_len, ts_for_display, g_inflight_batch_id); + uint32_t send_ms = millis() - send_start; + if (SERIAL_DEBUG_MODE && send_ms > 1000) { + serial_debug_printf("tx: resend batch took %lums", static_cast(send_ms)); + } + if (ok) { + g_last_batch_send_ms = millis(); + if (g_inflight_sync_request) { + serial_debug_printf("sync: request tx batch_id=%u", g_inflight_batch_id); + } else { + serial_debug_printf("tx: resent batch_id=%u len=%u", g_inflight_batch_id, static_cast(g_inflight_encoded_payload_len)); + } + } else if (g_inflight_sync_request) { + serial_debug_printf("sync: request tx failed batch_id=%u", g_inflight_batch_id); + } else { + serial_debug_printf("tx: resend failed batch_id=%u", g_inflight_batch_id); + } + return ok; + } + + BatchInput input = {}; + input.sender_id = sender_id_from_short_id(g_short_id); + input.batch_id = g_inflight_batch_id; + input.t_last = g_inflight_sync_request ? time_get_utc() : + g_inflight_samples[g_inflight_count - 1].ts_utc; + input.present_mask = 0; + input.n = 0; + input.battery_mV = g_inflight_sync_request ? battery_mv_from_voltage(g_last_battery_voltage_v) : + battery_mv_from_voltage(g_inflight_samples[g_inflight_count - 1].battery_voltage_v); + input.err_m = g_sender_faults.meter_read_fail > 255 ? 255 : static_cast(g_sender_faults.meter_read_fail); + input.err_d = g_sender_faults.decode_fail > 255 ? 255 : static_cast(g_sender_faults.decode_fail); + input.err_tx = g_sender_faults.lora_tx_fail > 255 ? 255 : static_cast(g_sender_faults.lora_tx_fail); + input.err_last = static_cast(g_sender_last_error); + input.err_rx_reject = static_cast(g_sender_rx_reject_reason); + uint8_t energy_regressions = 0; + uint8_t phase_clamps = 0; + uint8_t ts_dropped = 0; + uint8_t ts_collapsed = 0; + + if (!g_inflight_sync_request) { + if (input.t_last < static_cast(METER_BATCH_MAX_SAMPLES - 1)) { + g_last_tx_build_error = TxBuildError::Encode; + return false; + } + const uint32_t window_start = input.t_last - static_cast(METER_BATCH_MAX_SAMPLES - 1); + MeterData slot_samples[METER_BATCH_MAX_SAMPLES]; + bool slot_used[METER_BATCH_MAX_SAMPLES] = {}; + for (uint8_t i = 0; i < g_inflight_count; ++i) { + const MeterData &sample = g_inflight_samples[i]; + if (sample.ts_utc < window_start || sample.ts_utc > input.t_last) { + if (ts_dropped < 255) { + ts_dropped++; + } + continue; + } + uint8_t slot = static_cast(sample.ts_utc - window_start); + if (slot_used[slot] && ts_collapsed < 255) { + ts_collapsed++; + } + slot_used[slot] = true; + slot_samples[slot] = sample; + } + for (uint8_t slot = 0; slot < METER_BATCH_MAX_SAMPLES; ++slot) { + if (!slot_used[slot]) { + continue; + } + const uint8_t out_idx = input.n; + if (out_idx >= METER_BATCH_MAX_SAMPLES) { + g_last_tx_build_error = TxBuildError::Encode; + return false; + } + input.present_mask |= (1UL << slot); + input.n++; + input.energy_wh[out_idx] = kwh_to_wh_from_float(slot_samples[slot].energy_total_kwh); + bool c1 = false; + bool c2 = false; + bool c3 = false; + input.p1_w[out_idx] = float_to_i16_w_clamped(slot_samples[slot].phase_power_w[0], c1); + input.p2_w[out_idx] = float_to_i16_w_clamped(slot_samples[slot].phase_power_w[1], c2); + input.p3_w[out_idx] = float_to_i16_w_clamped(slot_samples[slot].phase_power_w[2], c3); + if (c1 && phase_clamps < 255) { + phase_clamps++; + } + if (c2 && phase_clamps < 255) { + phase_clamps++; + } + if (c3 && phase_clamps < 255) { + phase_clamps++; + } + } + } + + for (uint8_t i = 0; i < input.n; ++i) { + if (i > 0 && input.energy_wh[i] < input.energy_wh[i - 1]) { + input.energy_wh[i] = input.energy_wh[i - 1]; + if (energy_regressions < 255) { + energy_regressions++; + } + } + } + if (SERIAL_DEBUG_MODE && (energy_regressions > 0 || phase_clamps > 0 || ts_dropped > 0 || ts_collapsed > 0)) { + serial_debug_printf("tx: sanitize batch_id=%u energy_regress=%u phase_clamps=%u ts_drop=%u ts_dup=%u", + g_inflight_batch_id, + static_cast(energy_regressions), + static_cast(phase_clamps), + static_cast(ts_dropped), + static_cast(ts_collapsed)); + } + + static uint8_t encoded[BATCH_MAX_COMPRESSED]; + size_t encoded_len = 0; + uint32_t encode_start = millis(); + if (!encode_batch(input, encoded, sizeof(encoded), &encoded_len)) { + g_last_tx_build_error = TxBuildError::Encode; + invalidate_inflight_encode_cache(); + return false; + } + memcpy(g_inflight_encoded_payload, encoded, encoded_len); + g_inflight_encoded_payload_len = encoded_len; + g_inflight_encoded_batch_id = g_inflight_batch_id; + g_inflight_encoded_sync_request = g_inflight_sync_request; + g_inflight_encoded_valid = true; + uint32_t encode_ms = millis() - encode_start; + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("tx: batch_id=%u count=%u mask=%08lX bin_len=%u", + g_inflight_batch_id, + static_cast(input.n), + static_cast(input.present_mask), + static_cast(encoded_len)); + if (encode_ms > 200) { + serial_debug_printf("tx: encode took %lums", static_cast(encode_ms)); + } + } + g_batch_ack_timeout_ms = compute_batch_ack_timeout_ms(encoded_len); + + uint32_t send_start = millis(); + bool ok = send_batch_payload(encoded, encoded_len, ts_for_display, g_inflight_batch_id); + uint32_t send_ms = millis() - send_start; + if (SERIAL_DEBUG_MODE && send_ms > 1000) { + serial_debug_printf("tx: send batch took %lums", static_cast(send_ms)); + } + if (ok) { + g_last_batch_send_ms = millis(); + if (g_inflight_sync_request) { + serial_debug_printf("sync: request tx batch_id=%u", g_inflight_batch_id); + } else { + serial_debug_printf("tx: sent batch_id=%u len=%u", g_inflight_batch_id, static_cast(encoded_len)); + } + } else { + if (g_inflight_sync_request) { + serial_debug_printf("sync: request tx failed batch_id=%u", g_inflight_batch_id); + } else { + serial_debug_printf("tx: send failed batch_id=%u", g_inflight_batch_id); + } + } + return ok; +} + +static bool send_meter_batch(uint32_t ts_for_display) { + if (!prepare_inflight_from_queue()) { + return false; + } + g_inflight_sync_request = false; + bool ok = send_inflight_batch(ts_for_display); + if (ok) { + g_last_sent_batch_id = g_inflight_batch_id; + g_batch_ack_pending = true; + } else { + if (g_last_tx_build_error == TxBuildError::Encode) { + serial_debug_printf("tx: encode failed batch_id=%u dropped", g_inflight_batch_id); + note_fault(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms, FaultType::Decode); + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + finish_inflight_batch(); + return false; + } + g_inflight_active = false; + g_inflight_count = 0; + g_inflight_batch_id = 0; + g_inflight_sync_request = false; + invalidate_inflight_encode_cache(); + } + return ok; +} + +static bool send_sync_request() { + if (g_batch_ack_pending) { + return false; + } + if (battery_sample_due(millis())) { + update_battery_cache(); + } + g_inflight_active = true; + g_inflight_sync_request = true; + g_inflight_count = 0; + g_inflight_batch_id = g_batch_id; + if (SERIAL_DEBUG_MODE && g_build_attempts > 0) { + serial_debug_printf("batch: id=%u desired=%u attempts=%u valid=%u invalid=%u err_last=%u sync=1", + g_inflight_batch_id, + static_cast(METER_BATCH_MAX_SAMPLES), + static_cast(g_build_attempts), + static_cast(g_build_valid), + static_cast(g_build_invalid), + static_cast(g_sender_last_error)); + } + bool ok = send_inflight_batch(time_get_utc()); + if (ok) { + g_last_sent_batch_id = g_inflight_batch_id; + g_batch_ack_pending = true; + } else { + g_inflight_active = false; + g_inflight_sync_request = false; + g_inflight_batch_id = 0; + invalidate_inflight_encode_cache(); + } + return ok; +} + +static bool resend_inflight_batch(uint32_t ts_for_display) { + if (!g_batch_ack_pending || !g_inflight_active || (!g_inflight_sync_request && g_inflight_count == 0)) { + return false; + } + return send_inflight_batch(ts_for_display); +} + +static void finish_inflight_batch() { + if (g_batch_count > 0) { + batch_queue_drop_oldest(); + } + g_batch_ack_pending = false; + g_batch_retry_count = 0; + g_inflight_active = false; + g_inflight_count = 0; + g_inflight_batch_id = 0; + g_inflight_sync_request = false; + invalidate_inflight_encode_cache(); + g_batch_id++; +} + +static void sender_loop() { + watchdog_kick(); + uint32_t now_ms = millis(); + display_set_sender_queue(g_batch_count, g_build_count > 0); + display_set_sender_batches(g_last_acked_batch_id, g_batch_id); + sender_log_diagnostics(now_ms); + + meter_reader_pump(now_ms); + + if (g_time_acquired) { + sender_reset_fault_stats_on_hour_boundary(); + while (now_ms - g_last_sample_ms >= METER_SAMPLE_INTERVAL_MS) { + g_last_sample_ms += METER_SAMPLE_INTERVAL_MS; + MeterData data = {}; + data.short_id = g_short_id; + strncpy(data.device_id, g_device_id, sizeof(data.device_id)); + data.energy_total_kwh = NAN; + data.total_power_w = NAN; + data.phase_power_w[0] = NAN; + data.phase_power_w[1] = NAN; + data.phase_power_w[2] = NAN; + + g_build_attempts++; + uint32_t meter_age_ms = g_last_meter_valid ? (now_ms - g_last_meter_rx_ms) : UINT32_MAX; + // Reuse recent good samples to bridge short parser gaps without accepting stale data forever. + bool has_snapshot = g_last_meter_valid; + bool meter_ok = has_snapshot && meter_age_ms <= METER_SAMPLE_MAX_AGE_MS; + if (has_snapshot) { + data.meter_seconds = g_last_meter_data.meter_seconds; + data.meter_seconds_valid = g_last_meter_data.meter_seconds_valid; + data.energy_total_kwh = g_last_meter_data.energy_total_kwh; + data.total_power_w = g_last_meter_data.total_power_w; + data.phase_power_w[0] = g_last_meter_data.phase_power_w[0]; + data.phase_power_w[1] = g_last_meter_data.phase_power_w[1]; + data.phase_power_w[2] = g_last_meter_data.phase_power_w[2]; + g_meter_stale_seconds = meter_age_ms >= 1000 ? (meter_age_ms / 1000) : 0; + } else { + g_meter_stale_seconds = g_last_meter_valid ? (meter_age_ms / 1000) : (g_meter_stale_seconds + 1); + } + if (!meter_ok) { + note_fault(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms, FaultType::MeterRead); + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + } + if (g_meter_time_jump_pending) { + g_meter_time_jump_pending = false; + note_fault(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms, FaultType::MeterRead); + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + } + if (g_build_count == 0 && battery_sample_due(now_ms)) { + update_battery_cache(); + } + data.battery_voltage_v = g_last_battery_voltage_v; + data.battery_percent = g_last_battery_percent; + data.rx_reject_reason = static_cast(g_sender_rx_reject_reason); + uint32_t sample_ts_utc = 0; + if (has_snapshot && g_last_meter_data.meter_seconds_valid && g_last_meter_data.ts_utc >= MIN_ACCEPTED_EPOCH_UTC) { + sample_ts_utc = g_last_meter_data.ts_utc; + } else { + sample_ts_utc = time_get_utc(); + if (sample_ts_utc > 0 && now_ms > g_last_sample_ms) { + uint32_t lag_s = (now_ms - g_last_sample_ms) / 1000; + if (sample_ts_utc > lag_s) { + sample_ts_utc -= lag_s; + } + } + } + data.ts_utc = sample_ts_utc; + data.valid = has_snapshot; + + bool appended = append_meter_sample(data, meter_ok, has_snapshot); + (void)appended; + display_set_last_meter(data); + display_set_last_read(meter_ok, data.ts_utc); + } + + if (!g_batch_ack_pending && now_ms - g_last_send_ms >= METER_SEND_INTERVAL_MS) { + g_last_send_ms = now_ms; + if (g_build_count > 0) { + batch_queue_enqueue(g_build_samples, g_build_count); + g_build_count = 0; + reset_build_counters(); + } + if (g_batch_count > 0) { + send_meter_batch(last_sample_ts()); + } else if (g_build_attempts > 0) { + if (send_sync_request()) { + reset_build_counters(); + } + } + } + + // Catch-up mode: when backlog exists, send next queued batch without waiting + // for the regular 30s cadence. + if (!g_batch_ack_pending && g_batch_count > 1) { + send_meter_batch(last_sample_ts()); + } + } else { + if (!g_batch_ack_pending && now_ms - g_last_sync_request_ms >= SYNC_REQUEST_INTERVAL_MS) { + g_last_sync_request_ms = now_ms; + send_sync_request(); + } + } + + if (g_batch_ack_pending) { + LoraPacket ack_pkt = {}; + constexpr size_t ack_len = lora_frame_size(LORA_ACK_DOWN_PAYLOAD_LEN); + uint32_t ack_air_ms = lora_airtime_ms(ack_len); + uint32_t ack_window_first_ms = ack_air_ms + 200; + if (g_sender_ack_rtt_ewma_ms > 0) { + uint32_t rtt_based_ms = g_sender_ack_rtt_ewma_ms + 150; + if (rtt_based_ms > ack_window_first_ms) { + ack_window_first_ms = rtt_based_ms; + } + } + uint32_t miss_boost_ms = g_sender_ack_miss_streak * 150; + if (miss_boost_ms > 1200) { + miss_boost_ms = 1200; + } + ack_window_first_ms += miss_boost_ms; + if (ack_window_first_ms < 600) { + ack_window_first_ms = 600; + } + if (ack_window_first_ms > 2500) { + ack_window_first_ms = 2500; + } + uint32_t ack_window_second_ms = ack_window_first_ms + (ack_window_first_ms / 2); + uint32_t min_second_ms = ack_air_ms + 400; + if (ack_window_second_ms < min_second_ms) { + ack_window_second_ms = min_second_ms; + } + if (ack_window_second_ms > 5000) { + ack_window_second_ms = 5000; + } + if (SERIAL_DEBUG_MODE && (g_sender_ack_miss_streak > 0 || now_ms - g_last_ack_window_log_ms >= 10000)) { + g_last_ack_window_log_ms = now_ms; + serial_debug_printf("ack: rx windows=%lu/%lu airtime=%lu miss_streak=%lu", + static_cast(ack_window_first_ms), + static_cast(ack_window_second_ms), + static_cast(ack_air_ms), + static_cast(g_sender_ack_miss_streak)); + } + uint32_t rx_start = millis(); + bool got_ack = lora_receive_window(ack_pkt, ack_window_first_ms); + if (!got_ack) { + got_ack = lora_receive_window(ack_pkt, ack_window_second_ms); + } + uint32_t rx_elapsed = millis() - rx_start; + if (SERIAL_DEBUG_MODE) { + g_sender_rx_window_ms += rx_elapsed; + } + bool ack_accepted = false; + if (!got_ack) { + RxRejectReason reason = lora_get_last_rx_reject_reason(); + sender_note_rx_reject(reason, "ack"); + if (SERIAL_DEBUG_MODE) { + int16_t rssi_dbm = 0; + float snr_db = 0.0f; + bool has_signal = lora_get_last_rx_signal(rssi_dbm, snr_db); + const char *reason_text = reason == RxRejectReason::None ? "timeout" : rx_reject_reason_text(reason); + if (has_signal) { + serial_debug_printf("ack: rx miss reason=%s rssi=%d snr=%.1f", + reason_text, + static_cast(rssi_dbm), + static_cast(snr_db)); + } else { + serial_debug_printf("ack: rx miss reason=%s", reason_text); + } + } + } else if (ack_pkt.msg_kind != LoraMsgKind::AckDown) { + sender_note_rx_reject(RxRejectReason::InvalidMsgKind, "ack"); + if (SERIAL_DEBUG_MODE) { + uint16_t ack_id = ack_pkt.payload_len >= 3 ? read_u16_be(&ack_pkt.payload[1]) : 0; + serial_debug_printf("ack: reject msg_kind=%u payload_len=%u ack_id=%u", + static_cast(ack_pkt.msg_kind), + static_cast(ack_pkt.payload_len), + ack_id); + } + } else if (ack_pkt.payload_len < LORA_ACK_DOWN_PAYLOAD_LEN) { + sender_note_rx_reject(RxRejectReason::LengthMismatch, "ack"); + if (SERIAL_DEBUG_MODE) { + uint16_t ack_id = ack_pkt.payload_len >= 3 ? read_u16_be(&ack_pkt.payload[1]) : 0; + serial_debug_printf("ack: reject msg_kind=%u payload_len=%u ack_id=%u", + static_cast(ack_pkt.msg_kind), + static_cast(ack_pkt.payload_len), + ack_id); + } + } else { + uint8_t time_valid = ack_pkt.payload[0] & 0x01; + uint16_t ack_id = read_u16_be(&ack_pkt.payload[1]); + uint32_t ack_epoch = read_u32_be(&ack_pkt.payload[3]); + bool set_time = false; + if (g_batch_ack_pending && ack_id == g_last_sent_batch_id) { + ack_accepted = true; + g_sender_ack_rtt_last_ms = rx_elapsed; + if (g_sender_ack_rtt_ewma_ms == 0) { + g_sender_ack_rtt_ewma_ms = rx_elapsed; + } else { + g_sender_ack_rtt_ewma_ms = (g_sender_ack_rtt_ewma_ms * 3U + rx_elapsed + 1U) / 4U; + } + if (time_valid == 1 && ack_epoch >= MIN_ACCEPTED_EPOCH_UTC) { + time_set_utc(ack_epoch); + g_time_acquired = true; + sender_reset_fault_stats_on_first_sync(ack_epoch); + set_time = true; + } + g_last_acked_batch_id = ack_id; + serial_debug_printf("ack: rx ok batch_id=%u time_valid=%u epoch=%lu set=%u", + ack_id, + static_cast(time_valid), + static_cast(ack_epoch), + set_time ? 1 : 0); + finish_inflight_batch(); + } else { + if (ack_id != g_last_sent_batch_id) { + sender_note_rx_reject(RxRejectReason::BatchIdMismatch, "ack"); + if (SERIAL_DEBUG_MODE) { + serial_debug_printf("ack: reject msg_kind=%u payload_len=%u ack_id=%u", + static_cast(ack_pkt.msg_kind), + static_cast(ack_pkt.payload_len), + ack_id); + } + } + } + } + if (ack_accepted) { + g_sender_ack_miss_streak = 0; + } else if (g_sender_ack_miss_streak < UINT32_MAX) { + g_sender_ack_miss_streak++; + } + } + if (!g_batch_ack_pending) { + lora_sleep(); + } + + if (g_batch_ack_pending && (now_ms - g_last_batch_send_ms >= g_batch_ack_timeout_ms)) { + g_sender_ack_timeout_total++; + if (g_batch_retry_count < BATCH_MAX_RETRIES) { + g_batch_retry_count++; + g_sender_ack_retry_total++; + serial_debug_printf("ack: timeout batch_id=%u retry=%u", g_inflight_batch_id, g_batch_retry_count); + resend_inflight_batch(last_sample_ts()); + } else { + serial_debug_printf("ack: failed batch_id=%u policy=%s", g_inflight_batch_id, + BATCH_RETRY_POLICY == BatchRetryPolicy::Drop ? "drop" : "keep"); + if (BATCH_RETRY_POLICY == BatchRetryPolicy::Drop) { + finish_inflight_batch(); + } else { + g_batch_ack_pending = false; + g_batch_retry_count = 0; + g_inflight_active = false; + g_inflight_count = 0; + g_inflight_batch_id = 0; + g_inflight_sync_request = false; + invalidate_inflight_encode_cache(); + } + note_fault(g_sender_faults, g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms, FaultType::LoraTx); + display_set_last_error(g_sender_last_error, g_sender_last_error_utc, g_sender_last_error_ms); + } + } + + display_tick(); + + uint32_t next_due = g_time_acquired ? (g_last_sample_ms + METER_SAMPLE_INTERVAL_MS) : + (g_last_sync_request_ms + SYNC_REQUEST_INTERVAL_MS); + if (g_time_acquired) { + uint32_t next_send_due = g_last_send_ms + METER_SEND_INTERVAL_MS; + if (next_send_due < next_due) { + next_due = next_send_due; + } + } + if (!g_batch_ack_pending && next_due > now_ms) { + watchdog_kick(); + uint32_t idle_ms = next_due - now_ms; + if (SERIAL_DEBUG_MODE) { + g_sender_sleep_ms += idle_ms; + if (now_ms - g_sender_power_log_ms >= 10000) { + g_sender_power_log_ms = now_ms; + serial_debug_printf("power: rx_ms=%lu sleep_ms=%lu", static_cast(g_sender_rx_window_ms), + static_cast(g_sender_sleep_ms)); + } + } + lora_sleep(); + if (g_time_acquired) { + // Keep the meter reader task running while metering is active. + delay(idle_ms); + } else { + light_sleep_ms(idle_ms); + } + } +} + +static const char *sender_phase_text(SenderPhase phase) { + switch (phase) { + case SenderPhase::Syncing: return "SYNCING"; + case SenderPhase::Normal: return "NORMAL"; + case SenderPhase::Catchup: return "CATCHUP"; + case SenderPhase::WaitAck: return "WAIT_ACK"; + default: return "UNKNOWN"; + } +} + +static void sender_transition(SenderPhase next, const char *reason) { + if (next == g_sender_phase) { + return; + } + if (DD3_DEBUG_ENABLED && SERIAL_DEBUG_MODE) { + serial_debug_printf("state: %s -> %s reason=%s", sender_phase_text(g_sender_phase), sender_phase_text(next), reason ? reason : "none"); + } + g_sender_phase = next; +} + +static void sender_update_phase() { + if (g_batch_ack_pending) { + sender_transition(SenderPhase::WaitAck, "ack_pending"); + } else if (!g_time_acquired) { + sender_transition(SenderPhase::Syncing, "time_unsynced"); + } else if (g_batch_count > 1) { + sender_transition(SenderPhase::Catchup, "backlog"); + } else { + sender_transition(SenderPhase::Normal, "steady"); + } +} + +static void sender_validate_invariants() { + if (g_batch_count > BATCH_QUEUE_DEPTH) { + serial_debug_printf("inv: queue overflow count=%u max=%u", static_cast(g_batch_count), static_cast(BATCH_QUEUE_DEPTH)); + g_batch_count = BATCH_QUEUE_DEPTH; + } + if (g_batch_retry_count > BATCH_MAX_RETRIES) { + serial_debug_printf("inv: retry overflow retry=%u max=%u", static_cast(g_batch_retry_count), static_cast(BATCH_MAX_RETRIES)); + g_batch_retry_count = BATCH_MAX_RETRIES; + } + if (g_batch_ack_pending && !g_inflight_active && SERIAL_DEBUG_MODE) { + serial_debug_printf("inv: ack pending without inflight"); + } +} + +} // namespace + +bool SenderStateMachine::begin(const SenderStateMachineConfig &config) { + g_short_id = config.short_id; + if (config.device_id) { + strncpy(g_device_id, config.device_id, sizeof(g_device_id)); + g_device_id[sizeof(g_device_id) - 1] = '\0'; + } else { + g_device_id[0] = '\0'; + } + + power_sender_init(); + power_configure_unused_pins_sender(); + meter_init(); +#ifdef ARDUINO_ARCH_ESP32 + if (!meter_reader_start()) { + serial_debug_printf("meter: using inline polling fallback"); + } +#endif + g_last_sample_ms = millis() - METER_SAMPLE_INTERVAL_MS; + g_last_send_ms = millis(); + g_last_sync_request_ms = millis() - SYNC_REQUEST_INTERVAL_MS; + g_time_acquired = false; + g_sender_faults_reset_after_first_sync = false; + g_sender_faults_reset_hour_utc = UINT32_MAX; + update_battery_cache(); + sender_transition(SenderPhase::Syncing, "begin"); + return true; +} + +void SenderStateMachine::loop() { + sender_update_phase(); + sender_loop(); + sender_validate_invariants(); +} + +SenderStats SenderStateMachine::stats() const { + SenderStats stats = {}; + stats.queue_depth = g_batch_count; + stats.build_count = g_build_count; + stats.inflight_batch_id = g_inflight_batch_id; + stats.last_sent_batch_id = g_last_sent_batch_id; + stats.last_acked_batch_id = g_last_acked_batch_id; + stats.retry_count = g_batch_retry_count; + stats.ack_pending = g_batch_ack_pending; + stats.ack_timeout_total = g_sender_ack_timeout_total; + stats.ack_retry_total = g_sender_ack_retry_total; + stats.ack_miss_streak = g_sender_ack_miss_streak; + stats.rx_window_ms = g_sender_rx_window_ms; + stats.sleep_ms = g_sender_sleep_ms; + return stats; +} + +void SenderStateMachine::handleMeterRead(uint32_t now_ms) { + meter_reader_pump(now_ms); +} + +void SenderStateMachine::maybeSendBatch(uint32_t now_ms) { + (void)now_ms; +} + +void SenderStateMachine::handleAckWindow(uint32_t now_ms) { + (void)now_ms; +} + +bool SenderStateMachine::applyTimeFromAck(uint8_t time_valid, uint32_t ack_epoch) { + if (time_valid == 1 && ack_epoch >= MIN_ACCEPTED_EPOCH_UTC) { + time_set_utc(ack_epoch); + g_time_acquired = true; + sender_reset_fault_stats_on_first_sync(ack_epoch); + return true; + } + return false; +} + +void SenderStateMachine::validateInvariants() { + sender_validate_invariants(); +} diff --git a/src/sender_state_machine.h b/src/sender_state_machine.h new file mode 100644 index 0000000..a2816da --- /dev/null +++ b/src/sender_state_machine.h @@ -0,0 +1,44 @@ +#pragma once + +#include + +struct SenderStateMachineConfig { + uint16_t short_id; + const char *device_id; +}; + +struct SenderStats { + uint8_t queue_depth; + uint8_t build_count; + uint16_t inflight_batch_id; + uint16_t last_sent_batch_id; + uint16_t last_acked_batch_id; + uint8_t retry_count; + bool ack_pending; + uint32_t ack_timeout_total; + uint32_t ack_retry_total; + uint32_t ack_miss_streak; + uint32_t rx_window_ms; + uint32_t sleep_ms; +}; + +class SenderStateMachine { +public: + bool begin(const SenderStateMachineConfig &config); + void loop(); + SenderStats stats() const; + +private: + enum class State : uint8_t { + Syncing = 0, + Normal = 1, + Catchup = 2, + WaitAck = 3 + }; + + void handleMeterRead(uint32_t now_ms); + void maybeSendBatch(uint32_t now_ms); + void handleAckWindow(uint32_t now_ms); + bool applyTimeFromAck(uint8_t time_valid, uint32_t ack_epoch); + void validateInvariants(); +};