diff --git a/src/app_context.h b/src/app_context.h new file mode 100644 index 0000000..2d450de --- /dev/null +++ b/src/app_context.h @@ -0,0 +1,28 @@ +#pragma once + +#include + +#include "config.h" +#include "data_model.h" + +struct ReceiverSharedState { + SenderStatus sender_statuses[NUM_SENDERS]; + FaultCounters sender_faults_remote[NUM_SENDERS]; + FaultCounters sender_faults_remote_published[NUM_SENDERS]; + FaultType sender_last_error_remote[NUM_SENDERS]; + FaultType sender_last_error_remote_published[NUM_SENDERS]; + uint32_t sender_last_error_remote_utc[NUM_SENDERS]; + uint32_t sender_last_error_remote_ms[NUM_SENDERS]; + bool sender_discovery_sent[NUM_SENDERS]; + uint16_t last_batch_id_rx[NUM_SENDERS]; + + FaultCounters receiver_faults; + FaultCounters receiver_faults_published; + FaultType receiver_last_error; + FaultType receiver_last_error_published; + uint32_t receiver_last_error_utc; + uint32_t receiver_last_error_ms; + bool receiver_discovery_sent; + bool ap_mode; +}; + diff --git a/src/main.cpp b/src/main.cpp index 4f5f3ad..9306bd7 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,162 +1,36 @@ #include + +#include "app_context.h" #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 "mqtt_client.h" -#include "web_server.h" #include "display_ui.h" -#include "test_mode.h" +#include "lora_transport.h" +#include "mqtt_client.h" +#include "payload_codec.h" +#include "power_manager.h" +#include "receiver_pipeline.h" #include "sd_logger.h" #include "sender_state_machine.h" +#include "test_mode.h" +#include "time_manager.h" +#include "web_server.h" +#include "wifi_manager.h" + #include -#include + #ifdef ARDUINO_ARCH_ESP32 -#include #include -#include -#include -#include +#include #endif 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 ReceiverSharedState g_receiver_shared = {}; static SenderStateMachine g_sender_state_machine; - -static void watchdog_kick(); -static void finish_inflight_batch(); -static void invalidate_inflight_encode_cache(); +static ReceiverPipeline g_receiver_pipeline; static void serial_debug_printf(const char *fmt, ...) { if (!SERIAL_DEBUG_MODE) { @@ -170,515 +44,6 @@ static void serial_debug_printf(const char *fmt, ...) { 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 uint16_t g_last_batch_id_rx[NUM_SENDERS] = {}; - -struct BatchRxState { - bool active; - uint16_t batch_id; - uint8_t next_index; - uint8_t expected_chunks; - uint16_t total_len; - uint16_t received_len; - uint32_t last_rx_ms; - uint32_t timeout_ms; - uint8_t buffer[BATCH_MAX_COMPRESSED]; -}; - -static BatchRxState g_batch_rx = {}; - -static void init_sender_statuses() { - for (uint8_t i = 0; i < NUM_SENDERS; ++i) { - g_sender_statuses[i] = {}; - g_sender_statuses[i].has_data = false; - g_sender_statuses[i].last_update_ts_utc = 0; - g_sender_statuses[i].rx_batches_total = 0; - g_sender_statuses[i].rx_batches_duplicate = 0; - g_sender_statuses[i].rx_last_duplicate_ts_utc = 0; - g_sender_statuses[i].last_data.short_id = EXPECTED_SENDER_IDS[i]; - snprintf(g_sender_statuses[i].last_data.device_id, sizeof(g_sender_statuses[i].last_data.device_id), "dd3-%04X", EXPECTED_SENDER_IDS[i]); - g_sender_faults_remote[i] = {}; - g_sender_faults_remote_published[i] = {}; - g_sender_last_error_remote[i] = FaultType::None; - g_sender_last_error_remote_published[i] = FaultType::None; - g_sender_last_error_remote_utc[i] = 0; - g_sender_last_error_remote_ms[i] = 0; - g_sender_discovery_sent[i] = false; - } -} - -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; - } -} - -static uint32_t age_seconds(uint32_t ts_utc, uint32_t ts_ms) { - if (time_is_synced() && ts_utc > 0) { - uint32_t now = time_get_utc(); - return now > ts_utc ? now - ts_utc : 0; - } - return (millis() - ts_ms) / 1000; -} - -static bool counters_changed(const FaultCounters &a, const FaultCounters &b) { - return a.meter_read_fail != b.meter_read_fail || a.decode_fail != b.decode_fail || a.lora_tx_fail != b.lora_tx_fail; -} - -static void publish_faults_if_needed(const char *device_id, const FaultCounters &counters, FaultCounters &last_published, - FaultType last_error, FaultType &last_error_published, uint32_t last_error_utc, uint32_t last_error_ms) { - if (!mqtt_is_connected()) { - return; - } - if (!counters_changed(counters, last_published) && last_error == last_error_published) { - return; - } - uint32_t age = last_error != FaultType::None ? age_seconds(last_error_utc, last_error_ms) : 0; - if (mqtt_publish_faults(device_id, counters, last_error, age)) { - last_published = counters; - last_error_published = last_error; - } -} - #ifdef ARDUINO_ARCH_ESP32 static void watchdog_init() { esp_task_wdt_deinit(); @@ -698,566 +63,10 @@ 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 reset_batch_rx() { - g_batch_rx.active = false; - g_batch_rx.batch_id = 0; - g_batch_rx.next_index = 0; - g_batch_rx.expected_chunks = 0; - g_batch_rx.total_len = 0; - g_batch_rx.received_len = 0; - g_batch_rx.last_rx_ms = 0; - g_batch_rx.timeout_ms = 0; -} - -static bool process_batch_packet(const LoraPacket &pkt, BatchInput &out_batch, bool &decode_error, uint16_t &out_batch_id) { - decode_error = false; - if (pkt.payload_len < BATCH_HEADER_SIZE) { - return false; - } - uint16_t batch_id = read_u16_le(&pkt.payload[0]); - uint8_t chunk_index = pkt.payload[2]; - uint8_t chunk_count = pkt.payload[3]; - uint16_t total_len = read_u16_le(&pkt.payload[4]); - const uint8_t *chunk_data = &pkt.payload[BATCH_HEADER_SIZE]; - size_t chunk_len = pkt.payload_len - BATCH_HEADER_SIZE; - uint32_t now_ms = millis(); - - if (!g_batch_rx.active || batch_id != g_batch_rx.batch_id || (now_ms - g_batch_rx.last_rx_ms > g_batch_rx.timeout_ms)) { - if (chunk_index != 0) { - reset_batch_rx(); - return false; - } - if (total_len == 0 || total_len > BATCH_MAX_COMPRESSED || chunk_count == 0) { - reset_batch_rx(); - return false; - } - g_batch_rx.active = true; - g_batch_rx.batch_id = batch_id; - g_batch_rx.expected_chunks = chunk_count; - g_batch_rx.total_len = total_len; - g_batch_rx.received_len = 0; - g_batch_rx.next_index = 0; - g_batch_rx.timeout_ms = compute_batch_rx_timeout_ms(total_len, chunk_count); - } - - if (!g_batch_rx.active || chunk_index != g_batch_rx.next_index || chunk_count != g_batch_rx.expected_chunks) { - reset_batch_rx(); - return false; - } - - if (g_batch_rx.received_len + chunk_len > g_batch_rx.total_len || g_batch_rx.received_len + chunk_len > BATCH_MAX_COMPRESSED) { - reset_batch_rx(); - return false; - } - - memcpy(&g_batch_rx.buffer[g_batch_rx.received_len], chunk_data, chunk_len); - g_batch_rx.received_len += static_cast(chunk_len); - g_batch_rx.next_index++; - g_batch_rx.last_rx_ms = now_ms; - - if (g_batch_rx.next_index == g_batch_rx.expected_chunks && g_batch_rx.received_len == g_batch_rx.total_len) { - if (!decode_batch(g_batch_rx.buffer, g_batch_rx.received_len, &out_batch)) { - decode_error = true; - reset_batch_rx(); - return false; - } - out_batch_id = batch_id; - reset_batch_rx(); - return true; - } - - return false; -} - void setup() { Serial.begin(115200); delay(200); + #ifdef PAYLOAD_CODEC_TEST payload_codec_self_test(); #endif @@ -1265,6 +74,7 @@ void setup() { watchdog_init(); g_role = detect_role(); init_device_ids(g_short_id, g_device_id, sizeof(g_device_id)); + display_set_role(g_role); if (SERIAL_DEBUG_MODE) { #ifdef ARDUINO_ARCH_ESP32 @@ -1283,509 +93,48 @@ void setup() { sender_cfg.short_id = g_short_id; sender_cfg.device_id = g_device_id; g_sender_state_machine.begin(sender_cfg); + return; + } + + power_receiver_init(); + lora_receive_continuous(); + pinMode(PIN_ROLE, INPUT); // release pulldown before SD uses GPIO14 as SCK + sd_logger_init(); + wifi_manager_init(); + + ReceiverPipelineConfig receiver_cfg = {}; + receiver_cfg.short_id = g_short_id; + receiver_cfg.device_id = g_device_id; + receiver_cfg.shared = &g_receiver_shared; + g_receiver_pipeline.begin(receiver_cfg); + + display_set_sender_statuses(g_receiver_shared.sender_statuses, NUM_SENDERS); + + bool has_cfg = wifi_load_config(g_cfg); + if (has_cfg && wifi_connect_sta(g_cfg)) { + g_receiver_shared.ap_mode = false; + time_receiver_init(g_cfg.ntp_server_1.c_str(), g_cfg.ntp_server_2.c_str()); + mqtt_init(g_cfg, g_device_id); + web_server_set_config(g_cfg); + web_server_set_sender_faults(g_receiver_shared.sender_faults_remote, g_receiver_shared.sender_last_error_remote); + web_server_begin_sta(g_receiver_shared.sender_statuses, NUM_SENDERS); } else { - power_receiver_init(); - lora_receive_continuous(); - pinMode(PIN_ROLE, INPUT); // release pulldown before SD uses GPIO14 as SCK - sd_logger_init(); - wifi_manager_init(); - init_sender_statuses(); - display_set_sender_statuses(g_sender_statuses, NUM_SENDERS); - - bool has_cfg = wifi_load_config(g_cfg); - if (has_cfg && wifi_connect_sta(g_cfg)) { - g_ap_mode = false; - time_receiver_init(g_cfg.ntp_server_1.c_str(), g_cfg.ntp_server_2.c_str()); - mqtt_init(g_cfg, g_device_id); - web_server_set_config(g_cfg); - web_server_set_sender_faults(g_sender_faults_remote, g_sender_last_error_remote); - web_server_begin_sta(g_sender_statuses, NUM_SENDERS); - } else { - g_ap_mode = true; - char ap_ssid[32]; - snprintf(ap_ssid, sizeof(ap_ssid), "%s%04X", AP_SSID_PREFIX, g_short_id); - wifi_start_ap(ap_ssid, AP_PASSWORD); - if (g_cfg.ntp_server_1.isEmpty()) { - g_cfg.ntp_server_1 = "pool.ntp.org"; - } - if (g_cfg.ntp_server_2.isEmpty()) { - g_cfg.ntp_server_2 = "time.nist.gov"; - } - web_server_set_config(g_cfg); - web_server_set_sender_faults(g_sender_faults_remote, g_sender_last_error_remote); - web_server_begin_ap(g_sender_statuses, NUM_SENDERS); + g_receiver_shared.ap_mode = true; + char ap_ssid[32]; + snprintf(ap_ssid, sizeof(ap_ssid), "%s%04X", AP_SSID_PREFIX, g_short_id); + wifi_start_ap(ap_ssid, AP_PASSWORD); + if (g_cfg.ntp_server_1.isEmpty()) { + g_cfg.ntp_server_1 = "pool.ntp.org"; } + if (g_cfg.ntp_server_2.isEmpty()) { + g_cfg.ntp_server_2 = "time.nist.gov"; + } + web_server_set_config(g_cfg); + web_server_set_sender_faults(g_receiver_shared.sender_faults_remote, g_receiver_shared.sender_last_error_remote); + web_server_begin_ap(g_receiver_shared.sender_statuses, NUM_SENDERS); } } -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 void receiver_loop() { - watchdog_kick(); - LoraPacket pkt = {}; - if (lora_receive(pkt, 0)) { - if (pkt.msg_kind == LoraMsgKind::BatchUp) { - BatchInput batch = {}; - bool decode_error = false; - uint16_t batch_id = 0; - if (process_batch_packet(pkt, batch, decode_error, batch_id)) { - int8_t sender_idx = -1; - for (uint8_t i = 0; i < NUM_SENDERS; ++i) { - if (pkt.device_id_short == EXPECTED_SENDER_IDS[i]) { - sender_idx = static_cast(i); - break; - } - } - - if (sender_idx < 0) { - receiver_note_rx_reject(RxRejectReason::UnknownSender, "batch"); - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - serial_debug_printf("batch: reject unknown_sender short_id=%04X sender_id=%u batch_id=%u", - pkt.device_id_short, - static_cast(batch.sender_id), - static_cast(batch_id)); - goto receiver_loop_done; - } - - uint16_t expected_sender_id = static_cast(sender_idx + 1); - if (batch.sender_id != expected_sender_id) { - receiver_note_rx_reject(RxRejectReason::DeviceIdMismatch, "batch"); - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - serial_debug_printf("batch: reject device_id_mismatch short_id=%04X sender_id=%u expected=%u batch_id=%u", - pkt.device_id_short, - static_cast(batch.sender_id), - static_cast(expected_sender_id), - static_cast(batch_id)); - goto receiver_loop_done; - } - - bool duplicate = g_last_batch_id_rx[sender_idx] == batch_id; - SenderStatus &status = g_sender_statuses[sender_idx]; - if (status.rx_batches_total < UINT32_MAX) { - status.rx_batches_total++; - } - if (duplicate) { - if (status.rx_batches_duplicate < UINT32_MAX) { - status.rx_batches_duplicate++; - } - uint32_t duplicate_ts = time_get_utc(); - if (duplicate_ts == 0) { - duplicate_ts = batch.t_last; - } - status.rx_last_duplicate_ts_utc = duplicate_ts; - } - - send_batch_ack(batch_id, batch.n); - if (duplicate) { - goto receiver_loop_done; - } - g_last_batch_id_rx[sender_idx] = batch_id; - if (batch.n == 0) { - goto receiver_loop_done; - } - if (batch.n > METER_BATCH_MAX_SAMPLES) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - if (bit_count32(batch.present_mask) != batch.n) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - - size_t count = batch.n; - uint16_t short_id = pkt.device_id_short; - if (short_id == 0) { - short_id = short_id_from_sender_id(batch.sender_id); - } - if (batch.t_last < static_cast(METER_BATCH_MAX_SAMPLES - 1) || batch.t_last < MIN_ACCEPTED_EPOCH_UTC) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - const uint32_t window_start = batch.t_last - static_cast(METER_BATCH_MAX_SAMPLES - 1); - - MeterData samples[METER_BATCH_MAX_SAMPLES]; - float bat_v = batch.battery_mV > 0 ? static_cast(batch.battery_mV) / 1000.0f : NAN; - size_t s = 0; - for (uint8_t slot = 0; slot < METER_BATCH_MAX_SAMPLES; ++slot) { - if ((batch.present_mask & (1UL << slot)) == 0) { - continue; - } - if (s >= count) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - MeterData &data = samples[s]; - data = {}; - data.short_id = short_id; - if (short_id != 0) { - snprintf(data.device_id, sizeof(data.device_id), "dd3-%04X", short_id); - } else { - snprintf(data.device_id, sizeof(data.device_id), "dd3-0000"); - } - data.ts_utc = window_start + static_cast(slot); - if (data.ts_utc < MIN_ACCEPTED_EPOCH_UTC) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - data.energy_total_kwh = static_cast(batch.energy_wh[s]) / 1000.0f; - data.phase_power_w[0] = static_cast(batch.p1_w[s]); - data.phase_power_w[1] = static_cast(batch.p2_w[s]); - data.phase_power_w[2] = static_cast(batch.p3_w[s]); - data.total_power_w = data.phase_power_w[0] + data.phase_power_w[1] + data.phase_power_w[2]; - data.battery_voltage_v = bat_v; - data.battery_percent = !isnan(bat_v) ? battery_percent_from_voltage(bat_v) : 0; - data.valid = true; - data.link_valid = true; - data.link_rssi_dbm = pkt.rssi_dbm; - data.link_snr_db = pkt.snr_db; - data.err_meter_read = batch.err_m; - data.err_decode = batch.err_d; - data.err_lora_tx = batch.err_tx; - data.last_error = static_cast(batch.err_last); - data.rx_reject_reason = batch.err_rx_reject; - sd_logger_log_sample(data, (s + 1 == count) && data.last_error != FaultType::None); - s++; - } - if (s != count) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - goto receiver_loop_done; - } - - web_server_set_last_batch(static_cast(sender_idx), samples, count); - for (size_t s = 0; s < count; ++s) { - mqtt_publish_state(samples[s]); - } - g_sender_statuses[sender_idx].last_data = samples[count - 1]; - g_sender_statuses[sender_idx].last_update_ts_utc = samples[count - 1].ts_utc; - g_sender_statuses[sender_idx].has_data = true; - g_sender_faults_remote[sender_idx].meter_read_fail = samples[count - 1].err_meter_read; - g_sender_faults_remote[sender_idx].lora_tx_fail = samples[count - 1].err_lora_tx; - g_sender_last_error_remote[sender_idx] = samples[count - 1].last_error; - g_sender_last_error_remote_utc[sender_idx] = time_get_utc(); - g_sender_last_error_remote_ms[sender_idx] = millis(); - if (ENABLE_HA_DISCOVERY && !g_sender_discovery_sent[sender_idx]) { - g_sender_discovery_sent[sender_idx] = mqtt_publish_discovery(samples[count - 1].device_id); - } - publish_faults_if_needed(samples[count - 1].device_id, g_sender_faults_remote[sender_idx], g_sender_faults_remote_published[sender_idx], - g_sender_last_error_remote[sender_idx], g_sender_last_error_remote_published[sender_idx], - g_sender_last_error_remote_utc[sender_idx], g_sender_last_error_remote_ms[sender_idx]); - } else if (decode_error) { - note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); - display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); - } - } - } - -receiver_loop_done: - mqtt_loop(); - web_server_loop(); - if (ENABLE_HA_DISCOVERY && !g_receiver_discovery_sent) { - g_receiver_discovery_sent = mqtt_publish_discovery(g_device_id); - } - publish_faults_if_needed(g_device_id, g_receiver_faults, g_receiver_faults_published, - g_receiver_last_error, g_receiver_last_error_published, g_receiver_last_error_utc, g_receiver_last_error_ms); - display_set_receiver_status(g_ap_mode, wifi_is_connected() ? wifi_get_ssid().c_str() : "AP", mqtt_is_connected()); - display_tick(); - watchdog_kick(); -} - void loop() { #ifdef ENABLE_TEST_MODE if (g_role == DeviceRole::Sender) { @@ -1794,10 +143,10 @@ void loop() { watchdog_kick(); delay(50); } else { - test_receiver_loop(g_sender_statuses, NUM_SENDERS, g_short_id); + test_receiver_loop(g_receiver_shared.sender_statuses, NUM_SENDERS, g_short_id); mqtt_loop(); web_server_loop(); - display_set_receiver_status(g_ap_mode, wifi_is_connected() ? wifi_get_ssid().c_str() : "AP", mqtt_is_connected()); + display_set_receiver_status(g_receiver_shared.ap_mode, wifi_is_connected() ? wifi_get_ssid().c_str() : "AP", mqtt_is_connected()); display_tick(); watchdog_kick(); delay(50); @@ -1808,6 +157,6 @@ void loop() { if (g_role == DeviceRole::Sender) { g_sender_state_machine.loop(); } else { - receiver_loop(); + g_receiver_pipeline.loop(); } } diff --git a/src/receiver_pipeline.cpp b/src/receiver_pipeline.cpp new file mode 100644 index 0000000..be44c11 --- /dev/null +++ b/src/receiver_pipeline.cpp @@ -0,0 +1,549 @@ +#include "receiver_pipeline.h" + +#include +#include +#include + +#include "config.h" +#include "display_ui.h" +#include "lora_transport.h" +#include "mqtt_client.h" +#include "payload_codec.h" +#include "power_manager.h" +#include "sd_logger.h" +#include "time_manager.h" +#include "web_server.h" +#include "wifi_manager.h" + +#ifdef ARDUINO_ARCH_ESP32 +#include +#endif + +namespace { + +static uint16_t g_short_id = 0; +static char g_device_id[16] = ""; +static ReceiverSharedState *g_shared = nullptr; +static RxRejectReason g_receiver_rx_reject_reason = RxRejectReason::None; +static uint32_t g_receiver_rx_reject_log_ms = 0; + +#define g_sender_statuses (g_shared->sender_statuses) +#define g_sender_faults_remote (g_shared->sender_faults_remote) +#define g_sender_faults_remote_published (g_shared->sender_faults_remote_published) +#define g_sender_last_error_remote (g_shared->sender_last_error_remote) +#define g_sender_last_error_remote_published (g_shared->sender_last_error_remote_published) +#define g_sender_last_error_remote_utc (g_shared->sender_last_error_remote_utc) +#define g_sender_last_error_remote_ms (g_shared->sender_last_error_remote_ms) +#define g_sender_discovery_sent (g_shared->sender_discovery_sent) +#define g_last_batch_id_rx (g_shared->last_batch_id_rx) +#define g_receiver_faults (g_shared->receiver_faults) +#define g_receiver_faults_published (g_shared->receiver_faults_published) +#define g_receiver_last_error (g_shared->receiver_last_error) +#define g_receiver_last_error_published (g_shared->receiver_last_error_published) +#define g_receiver_last_error_utc (g_shared->receiver_last_error_utc) +#define g_receiver_last_error_ms (g_shared->receiver_last_error_ms) +#define g_receiver_discovery_sent (g_shared->receiver_discovery_sent) +#define g_ap_mode (g_shared->ap_mode) + +static void watchdog_kick() { +#ifdef ARDUINO_ARCH_ESP32 + esp_task_wdt_reset(); +#endif +} + +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; + +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); +} + +static uint8_t bit_count32(uint32_t value) { + uint8_t count = 0; + while (value != 0) { + value &= (value - 1); + count++; + } + return count; +} + +struct BatchRxState { + bool active; + uint16_t batch_id; + uint8_t next_index; + uint8_t expected_chunks; + uint16_t total_len; + uint16_t received_len; + uint32_t last_rx_ms; + uint32_t timeout_ms; + uint8_t buffer[BATCH_MAX_COMPRESSED]; +}; + +static BatchRxState g_batch_rx = {}; + +static void init_sender_statuses() { + for (uint8_t i = 0; i < NUM_SENDERS; ++i) { + g_sender_statuses[i] = {}; + g_sender_statuses[i].has_data = false; + g_sender_statuses[i].last_update_ts_utc = 0; + g_sender_statuses[i].rx_batches_total = 0; + g_sender_statuses[i].rx_batches_duplicate = 0; + g_sender_statuses[i].rx_last_duplicate_ts_utc = 0; + g_sender_statuses[i].last_data.short_id = EXPECTED_SENDER_IDS[i]; + snprintf(g_sender_statuses[i].last_data.device_id, sizeof(g_sender_statuses[i].last_data.device_id), "dd3-%04X", EXPECTED_SENDER_IDS[i]); + g_sender_faults_remote[i] = {}; + g_sender_faults_remote_published[i] = {}; + g_sender_last_error_remote[i] = FaultType::None; + g_sender_last_error_remote_published[i] = FaultType::None; + g_sender_last_error_remote_utc[i] = 0; + g_sender_last_error_remote_ms[i] = 0; + g_sender_discovery_sent[i] = false; + } +} + +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 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 uint32_t age_seconds(uint32_t ts_utc, uint32_t ts_ms) { + if (time_is_synced() && ts_utc > 0) { + uint32_t now = time_get_utc(); + return now > ts_utc ? now - ts_utc : 0; + } + return (millis() - ts_ms) / 1000; +} + +static bool counters_changed(const FaultCounters &a, const FaultCounters &b) { + return a.meter_read_fail != b.meter_read_fail || a.decode_fail != b.decode_fail || a.lora_tx_fail != b.lora_tx_fail; +} + +static void publish_faults_if_needed(const char *device_id, const FaultCounters &counters, FaultCounters &last_published, + FaultType last_error, FaultType &last_error_published, uint32_t last_error_utc, uint32_t last_error_ms) { + if (!mqtt_is_connected()) { + return; + } + if (!counters_changed(counters, last_published) && last_error == last_error_published) { + return; + } + uint32_t age = last_error != FaultType::None ? age_seconds(last_error_utc, last_error_ms) : 0; + if (mqtt_publish_faults(device_id, counters, last_error, age)) { + last_published = counters; + last_error_published = last_error; + } +} + +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 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 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 void reset_batch_rx() { + g_batch_rx.active = false; + g_batch_rx.batch_id = 0; + g_batch_rx.next_index = 0; + g_batch_rx.expected_chunks = 0; + g_batch_rx.total_len = 0; + g_batch_rx.received_len = 0; + g_batch_rx.last_rx_ms = 0; + g_batch_rx.timeout_ms = 0; +} + +static bool process_batch_packet(const LoraPacket &pkt, BatchInput &out_batch, bool &decode_error, uint16_t &out_batch_id) { + decode_error = false; + if (pkt.payload_len < BATCH_HEADER_SIZE) { + return false; + } + uint16_t batch_id = read_u16_le(&pkt.payload[0]); + uint8_t chunk_index = pkt.payload[2]; + uint8_t chunk_count = pkt.payload[3]; + uint16_t total_len = read_u16_le(&pkt.payload[4]); + const uint8_t *chunk_data = &pkt.payload[BATCH_HEADER_SIZE]; + size_t chunk_len = pkt.payload_len - BATCH_HEADER_SIZE; + uint32_t now_ms = millis(); + + if (!g_batch_rx.active || batch_id != g_batch_rx.batch_id || (now_ms - g_batch_rx.last_rx_ms > g_batch_rx.timeout_ms)) { + if (chunk_index != 0) { + reset_batch_rx(); + return false; + } + if (total_len == 0 || total_len > BATCH_MAX_COMPRESSED || chunk_count == 0) { + reset_batch_rx(); + return false; + } + g_batch_rx.active = true; + g_batch_rx.batch_id = batch_id; + g_batch_rx.expected_chunks = chunk_count; + g_batch_rx.total_len = total_len; + g_batch_rx.received_len = 0; + g_batch_rx.next_index = 0; + g_batch_rx.timeout_ms = compute_batch_rx_timeout_ms(total_len, chunk_count); + } + + if (!g_batch_rx.active || chunk_index != g_batch_rx.next_index || chunk_count != g_batch_rx.expected_chunks) { + reset_batch_rx(); + return false; + } + + if (g_batch_rx.received_len + chunk_len > g_batch_rx.total_len || g_batch_rx.received_len + chunk_len > BATCH_MAX_COMPRESSED) { + reset_batch_rx(); + return false; + } + + memcpy(&g_batch_rx.buffer[g_batch_rx.received_len], chunk_data, chunk_len); + g_batch_rx.received_len += static_cast(chunk_len); + g_batch_rx.next_index++; + g_batch_rx.last_rx_ms = now_ms; + + if (g_batch_rx.next_index == g_batch_rx.expected_chunks && g_batch_rx.received_len == g_batch_rx.total_len) { + if (!decode_batch(g_batch_rx.buffer, g_batch_rx.received_len, &out_batch)) { + decode_error = true; + reset_batch_rx(); + return false; + } + out_batch_id = batch_id; + reset_batch_rx(); + return true; + } + + return false; +} + +static void receiver_loop() { + watchdog_kick(); + LoraPacket pkt = {}; + if (lora_receive(pkt, 0)) { + if (pkt.msg_kind == LoraMsgKind::BatchUp) { + BatchInput batch = {}; + bool decode_error = false; + uint16_t batch_id = 0; + if (process_batch_packet(pkt, batch, decode_error, batch_id)) { + int8_t sender_idx = -1; + for (uint8_t i = 0; i < NUM_SENDERS; ++i) { + if (pkt.device_id_short == EXPECTED_SENDER_IDS[i]) { + sender_idx = static_cast(i); + break; + } + } + + if (sender_idx < 0) { + receiver_note_rx_reject(RxRejectReason::UnknownSender, "batch"); + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + serial_debug_printf("batch: reject unknown_sender short_id=%04X sender_id=%u batch_id=%u", + pkt.device_id_short, + static_cast(batch.sender_id), + static_cast(batch_id)); + goto receiver_loop_done; + } + + uint16_t expected_sender_id = static_cast(sender_idx + 1); + if (batch.sender_id != expected_sender_id) { + receiver_note_rx_reject(RxRejectReason::DeviceIdMismatch, "batch"); + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + serial_debug_printf("batch: reject device_id_mismatch short_id=%04X sender_id=%u expected=%u batch_id=%u", + pkt.device_id_short, + static_cast(batch.sender_id), + static_cast(expected_sender_id), + static_cast(batch_id)); + goto receiver_loop_done; + } + + bool duplicate = g_last_batch_id_rx[sender_idx] == batch_id; + SenderStatus &status = g_sender_statuses[sender_idx]; + if (status.rx_batches_total < UINT32_MAX) { + status.rx_batches_total++; + } + if (duplicate) { + if (status.rx_batches_duplicate < UINT32_MAX) { + status.rx_batches_duplicate++; + } + uint32_t duplicate_ts = time_get_utc(); + if (duplicate_ts == 0) { + duplicate_ts = batch.t_last; + } + status.rx_last_duplicate_ts_utc = duplicate_ts; + } + + send_batch_ack(batch_id, batch.n); + if (duplicate) { + goto receiver_loop_done; + } + g_last_batch_id_rx[sender_idx] = batch_id; + if (batch.n == 0) { + goto receiver_loop_done; + } + if (batch.n > METER_BATCH_MAX_SAMPLES) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + if (bit_count32(batch.present_mask) != batch.n) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + + size_t count = batch.n; + uint16_t short_id = pkt.device_id_short; + if (short_id == 0) { + short_id = short_id_from_sender_id(batch.sender_id); + } + if (batch.t_last < static_cast(METER_BATCH_MAX_SAMPLES - 1) || batch.t_last < MIN_ACCEPTED_EPOCH_UTC) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + const uint32_t window_start = batch.t_last - static_cast(METER_BATCH_MAX_SAMPLES - 1); + + MeterData samples[METER_BATCH_MAX_SAMPLES]; + float bat_v = batch.battery_mV > 0 ? static_cast(batch.battery_mV) / 1000.0f : NAN; + size_t s = 0; + for (uint8_t slot = 0; slot < METER_BATCH_MAX_SAMPLES; ++slot) { + if ((batch.present_mask & (1UL << slot)) == 0) { + continue; + } + if (s >= count) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + MeterData &data = samples[s]; + data = {}; + data.short_id = short_id; + if (short_id != 0) { + snprintf(data.device_id, sizeof(data.device_id), "dd3-%04X", short_id); + } else { + snprintf(data.device_id, sizeof(data.device_id), "dd3-0000"); + } + data.ts_utc = window_start + static_cast(slot); + if (data.ts_utc < MIN_ACCEPTED_EPOCH_UTC) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + data.energy_total_kwh = static_cast(batch.energy_wh[s]) / 1000.0f; + data.phase_power_w[0] = static_cast(batch.p1_w[s]); + data.phase_power_w[1] = static_cast(batch.p2_w[s]); + data.phase_power_w[2] = static_cast(batch.p3_w[s]); + data.total_power_w = data.phase_power_w[0] + data.phase_power_w[1] + data.phase_power_w[2]; + data.battery_voltage_v = bat_v; + data.battery_percent = !isnan(bat_v) ? battery_percent_from_voltage(bat_v) : 0; + data.valid = true; + data.link_valid = true; + data.link_rssi_dbm = pkt.rssi_dbm; + data.link_snr_db = pkt.snr_db; + data.err_meter_read = batch.err_m; + data.err_decode = batch.err_d; + data.err_lora_tx = batch.err_tx; + data.last_error = static_cast(batch.err_last); + data.rx_reject_reason = batch.err_rx_reject; + sd_logger_log_sample(data, (s + 1 == count) && data.last_error != FaultType::None); + s++; + } + if (s != count) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + goto receiver_loop_done; + } + + web_server_set_last_batch(static_cast(sender_idx), samples, count); + for (size_t s = 0; s < count; ++s) { + mqtt_publish_state(samples[s]); + } + g_sender_statuses[sender_idx].last_data = samples[count - 1]; + g_sender_statuses[sender_idx].last_update_ts_utc = samples[count - 1].ts_utc; + g_sender_statuses[sender_idx].has_data = true; + g_sender_faults_remote[sender_idx].meter_read_fail = samples[count - 1].err_meter_read; + g_sender_faults_remote[sender_idx].lora_tx_fail = samples[count - 1].err_lora_tx; + g_sender_last_error_remote[sender_idx] = samples[count - 1].last_error; + g_sender_last_error_remote_utc[sender_idx] = time_get_utc(); + g_sender_last_error_remote_ms[sender_idx] = millis(); + if (ENABLE_HA_DISCOVERY && !g_sender_discovery_sent[sender_idx]) { + g_sender_discovery_sent[sender_idx] = mqtt_publish_discovery(samples[count - 1].device_id); + } + publish_faults_if_needed(samples[count - 1].device_id, g_sender_faults_remote[sender_idx], g_sender_faults_remote_published[sender_idx], + g_sender_last_error_remote[sender_idx], g_sender_last_error_remote_published[sender_idx], + g_sender_last_error_remote_utc[sender_idx], g_sender_last_error_remote_ms[sender_idx]); + } else if (decode_error) { + note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::Decode); + display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms); + } + } + } + +receiver_loop_done: + mqtt_loop(); + web_server_loop(); + if (ENABLE_HA_DISCOVERY && !g_receiver_discovery_sent) { + g_receiver_discovery_sent = mqtt_publish_discovery(g_device_id); + } + publish_faults_if_needed(g_device_id, g_receiver_faults, g_receiver_faults_published, + g_receiver_last_error, g_receiver_last_error_published, g_receiver_last_error_utc, g_receiver_last_error_ms); + display_set_receiver_status(g_ap_mode, wifi_is_connected() ? wifi_get_ssid().c_str() : "AP", mqtt_is_connected()); + display_tick(); + watchdog_kick(); +} + +} // namespace + +bool ReceiverPipeline::begin(const ReceiverPipelineConfig &config) { + if (!config.shared) { + return false; + } + g_shared = config.shared; + *g_shared = {}; + 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'; + } + init_sender_statuses(); + reset_batch_rx(); + g_receiver_rx_reject_reason = RxRejectReason::None; + g_receiver_rx_reject_log_ms = 0; + return true; +} + +void ReceiverPipeline::loop() { + if (!g_shared) { + return; + } + receiver_loop(); +} + +ReceiverStats ReceiverPipeline::stats() const { + ReceiverStats stats = {}; + if (!g_shared) { + return stats; + } + stats.receiver_decode_fail = g_receiver_faults.decode_fail; + stats.receiver_lora_tx_fail = g_receiver_faults.lora_tx_fail; + stats.last_rx_reject = g_receiver_rx_reject_reason; + stats.receiver_discovery_sent = g_receiver_discovery_sent; + return stats; +} + diff --git a/src/receiver_pipeline.h b/src/receiver_pipeline.h new file mode 100644 index 0000000..1b65af4 --- /dev/null +++ b/src/receiver_pipeline.h @@ -0,0 +1,27 @@ +#pragma once + +#include + +#include "app_context.h" +#include "data_model.h" + +struct ReceiverPipelineConfig { + uint16_t short_id; + const char *device_id; + ReceiverSharedState *shared; +}; + +struct ReceiverStats { + uint32_t receiver_decode_fail; + uint32_t receiver_lora_tx_fail; + RxRejectReason last_rx_reject; + bool receiver_discovery_sent; +}; + +class ReceiverPipeline { +public: + bool begin(const ReceiverPipelineConfig &config); + void loop(); + ReceiverStats stats() const; +}; +