C3MA/DD3-LoRa-Bridge-MultiSender
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor LoRa protocol to batch+ack with ACK-based time bootstrap

Browse Source
This commit is contained in:
acidburns
2026-02-04 11:57:49 +01:00
parent f08d9a34d3
commit f0503af8c7
20 changed files with 326 additions and 1048 deletions
Download Patch File Download Diff File Expand all files Collapse all files

672
src/main.cpp
View File

@@ -1,9 +1,7 @@
#include <Arduino.h>
#include "config.h"
#include "data_model.h"
#include "json_codec.h"
#include "payload_codec.h"
#include "compressor.h"
#include "lora_transport.h"
#include "meter_driver.h"
#include "power_manager.h"
@@ -28,8 +26,6 @@ static char g_device_id[16] = "";
static SenderStatus g_sender_statuses[NUM_SENDERS];
static bool g_ap_mode = false;
static WifiMqttConfig g_cfg;
static uint32_t g_last_timesync_ms = 0;
static constexpr uint32_t TIME_SYNC_OFFSET_MS = 15000;
static uint32_t g_boot_ms = 0;
static FaultCounters g_sender_faults = {};
static FaultCounters g_receiver_faults = {};
@@ -50,18 +46,6 @@ 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;
struct TimeSyncBurstState {
bool active;
uint32_t start_ms;
uint32_t last_send_ms;
uint32_t last_drift_check_ms;
bool last_drift_ok;
};
static TimeSyncBurstState g_timesync_burst[NUM_SENDERS] = {};
static uint32_t g_sender_last_timesync_rx_ms = 0;
static bool g_sender_timesync_error = 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;
@@ -99,21 +83,19 @@ 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 uint32_t g_last_debug_log_ms = 0;
static uint32_t g_sender_last_timesync_check_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 uint8_t g_sender_timesync_mode = 0;
static RxRejectReason g_sender_rx_reject_reason = RxRejectReason::None;
static uint32_t g_sender_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 constexpr uint32_t SENDER_TIMESYNC_ACQUIRE_MS = 10UL * 60UL * 1000UL;
static constexpr uint32_t SENDER_TIMESYNC_ACQUIRE_INTERVAL_SEC = 20;
static constexpr uint32_t SENDER_TIMESYNC_ACQUIRE_WINDOW_MS = 3000;
static bool g_time_acquired = false;
static uint32_t g_last_sync_request_ms = 0;
static void watchdog_kick();
@@ -129,17 +111,6 @@ static void serial_debug_printf(const char *fmt, ...) {
Serial.println(buf);
}
static void sender_set_timesync_mode(uint8_t mode) {
if (g_sender_timesync_mode == mode) {
return;
}
g_sender_timesync_mode = mode;
if (SERIAL_DEBUG_MODE) {
const char *label = mode == 2 ? "acquire" : (mode == 1 ? "slow" : "fast");
serial_debug_printf("timesync: mode=%s", label);
}
}
static uint16_t g_last_batch_id_rx[NUM_SENDERS] = {};
struct BatchRxState {
@@ -185,37 +156,6 @@ 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 sender_timesync_window_due() {
uint32_t interval_sec = SENDER_TIMESYNC_CHECK_SEC_FAST;
bool in_acquire = (g_sender_last_timesync_rx_ms == 0) && (millis() - g_boot_ms < SENDER_TIMESYNC_ACQUIRE_MS);
bool allow_slow = (millis() - g_boot_ms >= 60000UL) && time_is_synced() && time_rtc_present() &&
(g_sender_last_timesync_rx_ms > 0);
// RTC boot time is not evidence of receiving network TimeSync.
if (in_acquire) {
interval_sec = SENDER_TIMESYNC_ACQUIRE_INTERVAL_SEC;
sender_set_timesync_mode(2);
} else if (allow_slow) {
interval_sec = SENDER_TIMESYNC_CHECK_SEC_SLOW;
sender_set_timesync_mode(1);
} else {
sender_set_timesync_mode(0);
}
static uint32_t last_interval_sec = 0;
if (last_interval_sec != interval_sec) {
last_interval_sec = interval_sec;
g_sender_last_timesync_check_ms = millis();
}
if (g_sender_last_timesync_check_ms == 0) {
g_sender_last_timesync_check_ms = millis() - interval_sec * 1000UL;
}
uint32_t now_ms = millis();
if (now_ms - g_sender_last_timesync_check_ms >= interval_sec * 1000UL) {
g_sender_last_timesync_check_ms = now_ms;
return true;
}
return false;
}
static bool batch_queue_drop_oldest() {
if (g_batch_count == 0) {
return false;
@@ -228,33 +168,13 @@ static bool batch_queue_drop_oldest() {
g_inflight_active = false;
g_inflight_count = 0;
g_inflight_batch_id = 0;
g_inflight_sync_request = false;
}
g_batch_tail = (g_batch_tail + 1) % BATCH_QUEUE_DEPTH;
g_batch_count--;
return dropped_inflight;
}
static void receiver_note_timesync_drift(uint8_t sender_idx, uint32_t sender_ts_utc) {
if (sender_idx >= NUM_SENDERS) {
return;
}
if (!time_is_synced() || sender_ts_utc == 0) {
return;
}
uint32_t now_utc = time_get_utc();
uint32_t diff = now_utc > sender_ts_utc ? now_utc - sender_ts_utc : sender_ts_utc - now_utc;
TimeSyncBurstState &state = g_timesync_burst[sender_idx];
state.last_drift_check_ms = millis();
state.last_drift_ok = diff <= TIME_SYNC_DRIFT_THRESHOLD_SEC;
if (!state.last_drift_ok) {
if (!state.active) {
state.active = true;
state.start_ms = millis();
state.last_send_ms = 0;
}
}
}
static void sender_note_rx_reject(RxRejectReason reason, const char *context) {
if (reason == RxRejectReason::None) {
return;
@@ -370,6 +290,29 @@ static uint16_t read_u16_le(const uint8_t *src) {
return static_cast<uint16_t>(src[0]) | (static_cast<uint16_t>(src[1]) << 8);
}
static void write_u16_be(uint8_t *dst, uint16_t value) {
dst[0] = static_cast<uint8_t>((value >> 8) & 0xFF);
dst[1] = static_cast<uint8_t>(value & 0xFF);
}
static uint16_t read_u16_be(const uint8_t *src) {
return static_cast<uint16_t>(src[0] << 8) | static_cast<uint16_t>(src[1]);
}
static void write_u32_be(uint8_t *dst, uint32_t value) {
dst[0] = static_cast<uint8_t>((value >> 24) & 0xFF);
dst[1] = static_cast<uint8_t>((value >> 16) & 0xFF);
dst[2] = static_cast<uint8_t>((value >> 8) & 0xFF);
dst[3] = static_cast<uint8_t>(value & 0xFF);
}
static uint32_t read_u32_be(const uint8_t *src) {
return (static_cast<uint32_t>(src[0]) << 24) |
(static_cast<uint32_t>(src[1]) << 16) |
(static_cast<uint32_t>(src[2]) << 8) |
static_cast<uint32_t>(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) {
@@ -433,7 +376,7 @@ static uint32_t compute_batch_rx_timeout_ms(uint16_t total_len, uint8_t chunk_co
}
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 = 5 + payload_len + 2;
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<uint32_t>(chunk_count) * per_chunk_toa_ms + BATCH_RX_MARGIN_MS;
return timeout_ms < 10000 ? 10000 : timeout_ms;
@@ -444,7 +387,7 @@ static uint32_t compute_batch_ack_timeout_ms(size_t payload_len) {
return 10000;
}
uint8_t chunk_count = static_cast<uint8_t>((payload_len + BATCH_CHUNK_PAYLOAD - 1) / BATCH_CHUNK_PAYLOAD);
size_t packet_len = 5 + BATCH_HEADER_SIZE + (payload_len > BATCH_CHUNK_PAYLOAD ? BATCH_CHUNK_PAYLOAD : payload_len) + 2;
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<uint32_t>(chunk_count) * per_chunk_toa_ms + BATCH_RX_MARGIN_MS;
return timeout_ms < 10000 ? 10000 : timeout_ms;
@@ -467,10 +410,8 @@ static bool send_batch_payload(const uint8_t *data, size_t len, uint32_t ts_for_
chunk_len = BATCH_CHUNK_PAYLOAD;
}
LoraPacket pkt = {};
pkt.protocol_version = PROTOCOL_VERSION;
pkt.role = DeviceRole::Sender;
pkt.msg_kind = LoraMsgKind::BatchUp;
pkt.device_id_short = g_short_id;
pkt.payload_type = PayloadType::MeterBatch;
pkt.payload_len = chunk_len + BATCH_HEADER_SIZE;
uint8_t *payload = pkt.payload;
@@ -501,27 +442,31 @@ static bool send_batch_payload(const uint8_t *data, size_t len, uint32_t ts_for_
return all_ok;
}
static void send_batch_ack(uint16_t batch_id, uint16_t sender_id) {
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.protocol_version = PROTOCOL_VERSION;
ack.role = DeviceRole::Receiver;
ack.msg_kind = LoraMsgKind::AckDown;
ack.device_id_short = g_short_id;
ack.payload_type = PayloadType::Ack;
ack.payload_len = 6;
write_u16_le(&ack.payload[0], batch_id);
write_u16_le(&ack.payload[2], sender_id);
write_u16_le(&ack.payload[4], g_short_id);
ack.payload_len = 7;
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) {
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("ack: tx repeat %u/%u batch_id=%u", static_cast<unsigned>(i + 1),
static_cast<unsigned>(repeats), batch_id);
}
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<unsigned>(time_valid),
static_cast<unsigned long>(epoch),
static_cast<unsigned>(sample_count));
lora_receive_continuous();
}
@@ -547,23 +492,25 @@ static bool prepare_inflight_from_queue() {
}
static bool send_inflight_batch(uint32_t ts_for_display) {
if (!g_inflight_active || g_inflight_count == 0) {
if (!g_inflight_active) {
return false;
}
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_samples[g_inflight_count - 1].ts_utc;
input.t_last = g_inflight_sync_request ? time_get_utc() :
g_inflight_samples[g_inflight_count - 1].ts_utc;
uint32_t dt_s = METER_SAMPLE_INTERVAL_MS / 1000;
input.dt_s = dt_s > 0 ? static_cast<uint8_t>(dt_s) : 1;
input.n = g_inflight_count;
input.battery_mV = battery_mv_from_voltage(g_inflight_samples[g_inflight_count - 1].battery_voltage_v);
input.n = g_inflight_sync_request ? 0 : g_inflight_count;
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<uint8_t>(g_sender_faults.meter_read_fail);
input.err_d = g_sender_faults.decode_fail > 255 ? 255 : static_cast<uint8_t>(g_sender_faults.decode_fail);
input.err_tx = g_sender_faults.lora_tx_fail > 255 ? 255 : static_cast<uint8_t>(g_sender_faults.lora_tx_fail);
input.err_last = static_cast<uint8_t>(g_sender_last_error);
input.err_rx_reject = static_cast<uint8_t>(g_sender_rx_reject_reason);
for (uint8_t i = 0; i < g_inflight_count; ++i) {
for (uint8_t i = 0; i < input.n; ++i) {
input.energy_wh[i] = kwh_to_wh_from_float(g_inflight_samples[i].energy_total_kwh);
if (!float_to_i16_w(g_inflight_samples[i].phase_power_w[0], input.p1_w[i]) ||
!float_to_i16_w(g_inflight_samples[i].phase_power_w[1], input.p2_w[i]) ||
@@ -580,7 +527,7 @@ static bool send_inflight_batch(uint32_t ts_for_display) {
}
uint32_t encode_ms = millis() - encode_start;
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("tx: batch_id=%u count=%u bin_len=%u", g_inflight_batch_id, g_inflight_count,
serial_debug_printf("tx: batch_id=%u count=%u bin_len=%u", g_inflight_batch_id, input.n,
static_cast<unsigned>(encoded_len));
if (encode_ms > 200) {
serial_debug_printf("tx: encode took %lums", static_cast<unsigned long>(encode_ms));
@@ -596,9 +543,17 @@ static bool send_inflight_batch(uint32_t ts_for_display) {
}
if (ok) {
g_last_batch_send_ms = millis();
serial_debug_printf("tx: sent batch_id=%u len=%u", g_inflight_batch_id, static_cast<unsigned>(encoded_len));
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<unsigned>(encoded_len));
}
} else {
serial_debug_printf("tx: send failed batch_id=%u", g_inflight_batch_id);
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;
}
@@ -607,6 +562,7 @@ 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;
@@ -615,12 +571,36 @@ static bool send_meter_batch(uint32_t ts_for_display) {
g_inflight_active = false;
g_inflight_count = 0;
g_inflight_batch_id = 0;
g_inflight_sync_request = false;
}
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;
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;
}
return ok;
}
static bool resend_inflight_batch(uint32_t ts_for_display) {
if (!g_batch_ack_pending || !g_inflight_active || g_inflight_count == 0) {
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);
@@ -635,6 +615,7 @@ static void finish_inflight_batch() {
g_inflight_active = false;
g_inflight_count = 0;
g_inflight_batch_id = 0;
g_inflight_sync_request = false;
g_batch_id++;
}
@@ -731,8 +712,6 @@ void setup() {
lora_init();
display_init();
time_rtc_init();
time_try_load_from_rtc();
display_set_self_ids(g_short_id, g_device_id);
if (g_role == DeviceRole::Sender) {
@@ -741,6 +720,8 @@ void setup() {
meter_init();
g_last_sample_ms = millis() - METER_SAMPLE_INTERVAL_MS;
g_last_send_ms = millis();
g_last_sync_request_ms = millis() - SYNC_REQUEST_INTERVAL_MS;
g_time_acquired = false;
update_battery_cache();
} else {
power_receiver_init();
@@ -794,75 +775,80 @@ static void sender_loop() {
g_batch_retry_count);
}
const char *frame = nullptr;
size_t frame_len = 0;
if (meter_poll_frame(frame, frame_len)) {
MeterData 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;
if (meter_parse_frame(frame, frame_len, parsed)) {
g_last_meter_data = parsed;
g_last_meter_valid = true;
g_last_meter_rx_ms = now_ms;
g_meter_stale_seconds = 0;
if (g_time_acquired) {
const char *frame = nullptr;
size_t frame_len = 0;
if (meter_poll_frame(frame, frame_len)) {
MeterData 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;
if (meter_parse_frame(frame, frame_len, parsed)) {
g_last_meter_data = parsed;
g_last_meter_valid = true;
g_last_meter_rx_ms = now_ms;
g_meter_stale_seconds = 0;
}
}
}
if (now_ms - g_last_sample_ms >= METER_SAMPLE_INTERVAL_MS) {
g_last_sample_ms = now_ms;
MeterData data = {};
data.short_id = g_short_id;
strncpy(data.device_id, g_device_id, sizeof(data.device_id));
if (now_ms - g_last_sample_ms >= METER_SAMPLE_INTERVAL_MS) {
g_last_sample_ms = now_ms;
MeterData data = {};
data.short_id = g_short_id;
strncpy(data.device_id, g_device_id, sizeof(data.device_id));
bool meter_ok = g_last_meter_valid;
if (meter_ok) {
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];
uint32_t age_ms = now_ms - g_last_meter_rx_ms;
g_meter_stale_seconds = age_ms >= 1000 ? (age_ms / 1000) : 0;
} else {
g_meter_stale_seconds++;
bool meter_ok = g_last_meter_valid;
if (meter_ok) {
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];
uint32_t age_ms = now_ms - g_last_meter_rx_ms;
g_meter_stale_seconds = age_ms >= 1000 ? (age_ms / 1000) : 0;
} else {
g_meter_stale_seconds++;
}
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_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<uint8_t>(g_sender_rx_reject_reason);
data.ts_utc = time_get_utc();
data.valid = meter_ok;
g_last_sample_ts_utc = data.ts_utc;
g_build_samples[g_build_count++] = data;
if (g_build_count >= METER_BATCH_MAX_SAMPLES) {
batch_queue_enqueue(g_build_samples, g_build_count);
g_build_count = 0;
}
display_set_last_meter(data);
display_set_last_read(meter_ok, data.ts_utc);
}
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_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<uint8_t>(g_sender_rx_reject_reason);
uint32_t now_utc = time_get_utc();
data.ts_utc = now_utc > 0 ? now_utc : millis() / 1000;
data.valid = meter_ok;
g_last_sample_ts_utc = data.ts_utc;
g_build_samples[g_build_count++] = data;
if (g_build_count >= METER_BATCH_MAX_SAMPLES) {
batch_queue_enqueue(g_build_samples, g_build_count);
g_build_count = 0;
if (!g_batch_ack_pending && now_ms - g_last_send_ms >= METER_SEND_INTERVAL_MS) {
g_last_send_ms = now_ms;
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();
}
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;
send_meter_batch(last_sample_ts());
}
if (g_batch_ack_pending) {
LoraPacket ack_pkt = {};
const uint32_t ack_len = 5 + 6 + 2;
const uint32_t ack_len = 3 + 7 + 2;
uint32_t ack_air_ms = lora_airtime_ms(ack_len);
uint32_t ack_window_ms = ack_air_ms + 300;
if (ack_window_ms < 1200) {
@@ -886,74 +872,35 @@ static void sender_loop() {
}
if (!got_ack) {
sender_note_rx_reject(lora_get_last_rx_reject_reason(), "ack");
} else if (ack_pkt.role != DeviceRole::Receiver) {
sender_note_rx_reject(RxRejectReason::WrongRole, "ack");
} else if (ack_pkt.payload_type != PayloadType::Ack) {
sender_note_rx_reject(RxRejectReason::WrongPayloadType, "ack");
} else if (ack_pkt.payload_len < 6) {
} else if (ack_pkt.msg_kind != LoraMsgKind::AckDown) {
sender_note_rx_reject(RxRejectReason::InvalidMsgKind, "ack");
} else if (ack_pkt.payload_len < 7) {
sender_note_rx_reject(RxRejectReason::LengthMismatch, "ack");
} else {
uint16_t ack_id = read_u16_le(ack_pkt.payload);
uint16_t ack_sender = read_u16_le(&ack_pkt.payload[2]);
uint16_t ack_receiver = read_u16_le(&ack_pkt.payload[4]);
if (ack_sender == g_short_id && ack_receiver == ack_pkt.device_id_short &&
g_batch_ack_pending && ack_id == g_last_sent_batch_id) {
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) {
if (time_valid == 1 && ack_epoch >= MIN_ACCEPTED_EPOCH_UTC) {
time_set_utc(ack_epoch);
g_time_acquired = true;
set_time = true;
}
g_last_acked_batch_id = ack_id;
serial_debug_printf("ack: rx ok batch_id=%u", ack_id);
serial_debug_printf("ack: rx ok batch_id=%u time_valid=%u epoch=%lu set=%u",
ack_id,
static_cast<unsigned>(time_valid),
static_cast<unsigned long>(ack_epoch),
set_time ? 1 : 0);
finish_inflight_batch();
} else {
if (ack_sender != g_short_id || ack_receiver != ack_pkt.device_id_short) {
sender_note_rx_reject(RxRejectReason::DeviceIdMismatch, "ack");
} else if (ack_id != g_last_sent_batch_id) {
if (ack_id != g_last_sent_batch_id) {
sender_note_rx_reject(RxRejectReason::BatchIdMismatch, "ack");
}
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("ack: reject batch_id=%u sender=%u receiver=%u exp_batch=%u exp_sender=%u",
ack_id, ack_sender, ack_receiver, g_last_sent_batch_id, g_short_id);
}
}
}
}
bool timesync_due = (!g_batch_ack_pending && sender_timesync_window_due());
if (timesync_due) {
LoraPacket rx = {};
uint32_t rx_start = millis();
uint32_t window_ms = (g_sender_timesync_mode == 2) ? SENDER_TIMESYNC_ACQUIRE_WINDOW_MS : SENDER_TIMESYNC_WINDOW_MS;
bool got = lora_receive_window(rx, window_ms);
uint32_t rx_elapsed = millis() - rx_start;
if (SERIAL_DEBUG_MODE) {
g_sender_rx_window_ms += rx_elapsed;
}
if (!got) {
sender_note_rx_reject(lora_get_last_rx_reject_reason(), "timesync");
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("timesync: rx miss window_ms=%lu", static_cast<unsigned long>(window_ms));
}
} else if (rx.role != DeviceRole::Receiver) {
sender_note_rx_reject(RxRejectReason::WrongRole, "timesync");
} else if (rx.payload_type != PayloadType::TimeSync) {
sender_note_rx_reject(RxRejectReason::WrongPayloadType, "timesync");
} else if (time_handle_timesync_payload(rx.payload, rx.payload_len)) {
g_sender_last_timesync_rx_ms = now_ms;
if (g_sender_timesync_error) {
g_sender_timesync_error = false;
display_set_last_error(FaultType::None, 0, 0);
}
serial_debug_printf("timesync: rx ok window_ms=%lu", static_cast<unsigned long>(window_ms));
} else {
sender_note_rx_reject(RxRejectReason::LengthMismatch, "timesync");
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("timesync: rx miss window_ms=%lu", static_cast<unsigned long>(window_ms));
}
}
}
uint32_t timesync_age_ms = (g_sender_last_timesync_rx_ms > 0) ? (now_ms - g_sender_last_timesync_rx_ms)
: (now_ms - g_boot_ms);
if (!g_sender_timesync_error && timesync_age_ms > TIME_SYNC_ERROR_TIMEOUT_MS) {
g_sender_timesync_error = true;
display_set_last_error(FaultType::TimeSync, time_get_utc(), now_ms);
}
if (!g_batch_ack_pending) {
lora_sleep();
}
@@ -974,6 +921,7 @@ static void sender_loop() {
g_inflight_active = false;
g_inflight_count = 0;
g_inflight_batch_id = 0;
g_inflight_sync_request = false;
}
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);
@@ -982,9 +930,14 @@ static void sender_loop() {
display_tick();
uint32_t next_sample_due = g_last_sample_ms + METER_SAMPLE_INTERVAL_MS;
uint32_t next_send_due = g_last_send_ms + METER_SEND_INTERVAL_MS;
uint32_t next_due = next_sample_due < next_send_due ? next_sample_due : next_send_due;
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();
if (SERIAL_DEBUG_MODE) {
@@ -1002,64 +955,13 @@ static void sender_loop() {
static void receiver_loop() {
watchdog_kick();
if (g_last_timesync_ms == 0) {
g_last_timesync_ms = millis() - (TIME_SYNC_INTERVAL_SEC * 1000UL - TIME_SYNC_OFFSET_MS);
}
LoraPacket pkt = {};
if (lora_receive(pkt, 0) && pkt.protocol_version == PROTOCOL_VERSION) {
if (pkt.payload_type == PayloadType::MeterData) {
uint8_t decompressed[256];
size_t decompressed_len = 0;
if (!decompressBuffer(pkt.payload, pkt.payload_len, decompressed, sizeof(decompressed) - 1, decompressed_len)) {
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);
} else {
if (decompressed_len >= sizeof(decompressed)) {
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);
return;
}
decompressed[decompressed_len] = '\0';
MeterData data = {};
if (jsonToMeterData(String(reinterpret_cast<const char *>(decompressed)), data)) {
data.link_valid = true;
data.link_rssi_dbm = pkt.rssi_dbm;
data.link_snr_db = pkt.snr_db;
sd_logger_log_sample(data, data.last_error != FaultType::None);
for (uint8_t i = 0; i < NUM_SENDERS; ++i) {
if (pkt.device_id_short == EXPECTED_SENDER_IDS[i]) {
data.short_id = pkt.device_id_short;
g_sender_statuses[i].last_data = data;
g_sender_statuses[i].last_update_ts_utc = data.ts_utc;
g_sender_statuses[i].has_data = true;
g_sender_faults_remote[i].meter_read_fail = data.err_meter_read;
g_sender_faults_remote[i].lora_tx_fail = data.err_lora_tx;
receiver_note_timesync_drift(i, data.ts_utc);
g_sender_last_error_remote[i] = data.last_error;
g_sender_last_error_remote_utc[i] = time_get_utc();
g_sender_last_error_remote_ms[i] = millis();
mqtt_publish_state(data);
if (ENABLE_HA_DISCOVERY && !g_sender_discovery_sent[i]) {
g_sender_discovery_sent[i] = mqtt_publish_discovery(data.device_id);
}
publish_faults_if_needed(data.device_id, g_sender_faults_remote[i], g_sender_faults_remote_published[i],
g_sender_last_error_remote[i], g_sender_last_error_remote_published[i],
g_sender_last_error_remote_utc[i], g_sender_last_error_remote_ms[i]);
break;
}
}
} else {
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);
}
}
} else if (pkt.payload_type == PayloadType::MeterBatch) {
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)) {
MeterData samples[METER_BATCH_MAX_SAMPLES];
size_t count = 0;
int8_t sender_idx = -1;
for (uint8_t i = 0; i < NUM_SENDERS; ++i) {
if (pkt.device_id_short == EXPECTED_SENDER_IDS[i]) {
@@ -1067,87 +969,92 @@ static void receiver_loop() {
break;
}
}
bool duplicate = sender_idx >= 0 && g_last_batch_id_rx[sender_idx] == batch_id;
if (duplicate) {
send_batch_ack(batch_id, pkt.device_id_short);
} else {
g_last_batch_id_rx[sender_idx] = batch_id;
send_batch_ack(batch_id, pkt.device_id_short);
count = batch.n;
if (count == 0 || count > 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);
return;
}
uint16_t short_id = pkt.device_id_short;
if (short_id == 0) {
short_id = short_id_from_sender_id(batch.sender_id);
}
uint64_t span = static_cast<uint64_t>(batch.dt_s) * static_cast<uint64_t>(count - 1);
if (batch.t_last < span) {
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);
return;
}
uint32_t t_first = batch.t_last - static_cast<uint32_t>(span);
float bat_v = batch.battery_mV > 0 ? static_cast<float>(batch.battery_mV) / 1000.0f : NAN;
for (size_t s = 0; s < count; ++s) {
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 = t_first + static_cast<uint32_t>(s) * batch.dt_s;
data.energy_total_kwh = static_cast<float>(batch.energy_wh[s]) / 1000.0f;
data.phase_power_w[0] = static_cast<float>(batch.p1_w[s]);
data.phase_power_w[1] = static_cast<float>(batch.p2_w[s]);
data.phase_power_w[2] = static_cast<float>(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;
if (!isnan(bat_v)) {
data.battery_percent = battery_percent_from_voltage(bat_v);
} else {
data.battery_percent = 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<FaultType>(batch.err_last);
data.rx_reject_reason = batch.err_rx_reject;
sd_logger_log_sample(data, (s + 1 == count) && data.last_error != FaultType::None);
}
if (sender_idx >= 0) {
web_server_set_last_batch(static_cast<uint8_t>(sender_idx), samples, count);
for (size_t s = 0; s < count; ++s) {
samples[s].short_id = pkt.device_id_short;
mqtt_publish_state(samples[s]);
}
if (count > 0) {
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;
receiver_note_timesync_drift(static_cast<uint8_t>(sender_idx), samples[count - 1].ts_utc);
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]);
}
bool duplicate = sender_idx >= 0 && g_last_batch_id_rx[sender_idx] == batch_id;
send_batch_ack(batch_id, batch.n);
if (duplicate) {
goto receiver_loop_done;
}
if (sender_idx >= 0) {
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;
}
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);
}
uint64_t span = static_cast<uint64_t>(batch.dt_s) * static_cast<uint64_t>(count - 1);
if (batch.t_last < span || 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;
}
uint32_t t_first = batch.t_last - static_cast<uint32_t>(span);
if (t_first < 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;
}
MeterData samples[METER_BATCH_MAX_SAMPLES];
float bat_v = batch.battery_mV > 0 ? static_cast<float>(batch.battery_mV) / 1000.0f : NAN;
for (size_t s = 0; s < count; ++s) {
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 = t_first + static_cast<uint32_t>(s) * batch.dt_s;
data.energy_total_kwh = static_cast<float>(batch.energy_wh[s]) / 1000.0f;
data.phase_power_w[0] = static_cast<float>(batch.p1_w[s]);
data.phase_power_w[1] = static_cast<float>(batch.p2_w[s]);
data.phase_power_w[2] = static_cast<float>(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<FaultType>(batch.err_last);
data.rx_reject_reason = batch.err_rx_reject;
sd_logger_log_sample(data, (s + 1 == count) && data.last_error != FaultType::None);
}
if (sender_idx >= 0) {
web_server_set_last_batch(static_cast<uint8_t>(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);
@@ -1156,42 +1063,7 @@ static void receiver_loop() {
}
}
uint32_t interval_sec = TIME_SYNC_INTERVAL_SEC;
uint32_t now_ms = millis();
if (!g_ap_mode) {
bool burst_sent = false;
for (uint8_t i = 0; i < NUM_SENDERS; ++i) {
TimeSyncBurstState &state = g_timesync_burst[i];
if (state.active) {
if (now_ms - state.start_ms >= TIME_SYNC_BURST_DURATION_MS) {
state.active = false;
} else if (state.last_send_ms == 0 || now_ms - state.last_send_ms >= TIME_SYNC_BURST_INTERVAL_MS) {
state.last_send_ms = now_ms;
burst_sent = true;
}
}
}
if (burst_sent) {
if (!time_send_timesync(g_short_id)) {
note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::LoraTx);
display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms);
}
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("timesync: tx burst");
}
g_last_timesync_ms = now_ms;
} else if (now_ms - g_last_timesync_ms > interval_sec * 1000UL) {
g_last_timesync_ms = now_ms;
if (!time_send_timesync(g_short_id)) {
note_fault(g_receiver_faults, g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms, FaultType::LoraTx);
display_set_last_error(g_receiver_last_error, g_receiver_last_error_utc, g_receiver_last_error_ms);
}
if (SERIAL_DEBUG_MODE) {
serial_debug_printf("timesync: tx normal");
}
}
}
receiver_loop_done:
mqtt_loop();
web_server_loop();
if (ENABLE_HA_DISCOVERY && !g_receiver_discovery_sent) {