#include "receiver_pipeline.h" #include #include #include #include "config.h" #include "display_ui.h" #include "json_codec.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; } static bool mqtt_publish_sample(const MeterData &data) { #ifdef ENABLE_TEST_MODE String payload; if (!meterDataToJson(data, payload)) { return false; } return mqtt_publish_test(data.device_id, payload); #else return mqtt_publish_state(data); #endif } 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_sample(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; }