Wi-Fi 上传太耗电？比较每次上传与攒一批再发

1,<jid16hex>,<seq>,<boot_uptime_ms>,0,0,<valid>,<t_milli>,<rh_raw_milli>,<crc32>\n
ACK: <jid16hex>,<seq>  # 不retain，无换行

// ---- main/BatchModel.h ----
// SPDX-License-Identifier: MIT
#pragma once
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <limits>
extern "C" {
#include "station_model.h"
}
struct BatchRecord {
 uint64_t journal,lastAllocated,generated,dropped;uint32_t count,head,wakesWaiting;
 char rows[8][STATION_LINE_MAX];uint32_t checksum;
};
inline void seal(BatchRecord& b){b.checksum=station_crc32(&b,offsetof(BatchRecord,checksum));}
inline bool intact(const BatchRecord& b){
 if(!b.journal||b.count>8||b.head>7||b.lastAllocated<b.count||b.checksum!=station_crc32(&b,offsetof(BatchRecord,checksum)))return false;
 for(uint32_t i=0;i<b.count;i++){const char* p=b.rows[(b.head+i)%8];if(!memchr(p,0,STATION_LINE_MAX))return false;station_record_t row{};if(!station_record_decode(p,&row)||row.journal!=b.journal||row.sequence!=b.lastAllocated-b.count+i+1)return false;}return true;
}
inline bool resume(BatchRecord& b,bool timerDeepWake,uint64_t newJournal){bool continued=timerDeepWake&&intact(b);if(!continued){memset(&b,0,sizeof b);b.journal=newJournal?newJournal:1;seal(b);}return continued;}
inline void wakeAge(BatchRecord& b){if(b.count&&b.wakesWaiting<UINT32_MAX)++b.wakesWaiting;}
inline bool append(BatchRecord& b,uint64_t bootUptime,bool valid,int32_t t,int32_t rh){
 if(b.generated<UINT64_MAX)++b.generated;
 if(b.count==8||b.lastAllocated==UINT64_MAX){if(b.dropped<UINT64_MAX)++b.dropped;seal(b);return false;}
 station_record_t row{};row.journal=b.journal;row.sequence=b.lastAllocated+1;row.uptime=bootUptime;row.sample_valid=valid;row.t_milli=valid?t:0;row.rh_milli=valid?rh:0;
 char* dst=b.rows[(b.head+b.count)%8];memset(dst,0,STATION_LINE_MAX);if(!station_record_encode(&row,dst,STATION_LINE_MAX))return false;
 ++b.lastAllocated;++b.count;seal(b);return true;
}
inline bool uploadDue(const BatchRecord& b,unsigned threshold,bool finalTail){return b.count&&(b.count>=threshold||b.wakesWaiting>=8||finalTail);}
inline uint64_t headSequence(const BatchRecord& b){return b.count?b.lastAllocated-b.count+1:0;}
inline bool applicationAck(BatchRecord& b,uint64_t journal,uint64_t sequence,uint64_t inflight){
 if(!b.count||!inflight||journal!=b.journal||sequence!=inflight||sequence!=headSequence(b))return false;
 memset(b.rows[b.head],0,STATION_LINE_MAX);b.head=(b.head+1)%8;--b.count;if(!b.count)b.wakesWaiting=0;seal(b);return true;
}

// ---- main/main.cpp ----
// SPDX-License-Identifier: MIT
#include <cstdio>
#include <cstring>
#include <atomic>
#include <inttypes.h>
#include "sdkconfig.h"
#include "esp_sleep.h"
#include "esp_system.h"
#include "esp_timer.h"
#include "esp_random.h"
#include "esp_mac.h"
#include "esp_wifi.h"
#include "esp_event.h"
#include "esp_netif.h"
#include "nvs_flash.h"
#include "mqtt_client.h"
#include "driver/uart.h"
#include "SensorBus.h"
#include "BatchModel.h"
#include "freertos/event_groups.h"
#include "freertos/queue.h"
RTC_DATA_ATTR static BatchRecord batch{};
static EventGroupHandle_t flags;static QueueHandle_t acks;static constexpr int IP_OK=1,FAIL=2,MQTT_READY=4;
static char rowTopic[80],ackTopic[80];static std::atomic<int> subscriptionId{-1};
struct Ack {uint64_t journal,sequence;};
static void wifiEvent(void*,esp_event_base_t base,int32_t id,void*){
 if(base==WIFI_EVENT&&id==WIFI_EVENT_STA_START)esp_wifi_connect();
 if(base==WIFI_EVENT&&id==WIFI_EVENT_STA_DISCONNECTED)xEventGroupSetBits(flags,FAIL);
 if(base==IP_EVENT&&id==IP_EVENT_STA_GOT_IP)xEventGroupSetBits(flags,IP_OK);
}
static void mqttEvent(void*,esp_event_base_t,int32_t id,void* data){
 auto e=static_cast<esp_mqtt_event_handle_t>(data);
 if(id==MQTT_EVENT_CONNECTED){int n=esp_mqtt_client_subscribe(e->client,ackTopic,1);subscriptionId.store(n);if(n<0)xEventGroupSetBits(flags,FAIL);}
 if(id==MQTT_EVENT_SUBSCRIBED&&e->msg_id==subscriptionId.load())xEventGroupSetBits(flags,MQTT_READY);
 if(id==MQTT_EVENT_DISCONNECTED){xEventGroupClearBits(flags,MQTT_READY);xEventGroupSetBits(flags,FAIL);}
 if(id==MQTT_EVENT_DATA&&station_ack_frame(e->topic,e->topic_len,ackTopic,e->current_data_offset,e->total_data_len,e->data_len,e->retain)){
  Ack a{};if(station_ack_decode(e->data,e->data_len,&a.journal,&a.sequence))xQueueSend(acks,&a,0); // full queue: no deletion; same row can be retried next wake
 }
}
static void upload(){
 if(!CONFIG_HX_SSID[0]||!CONFIG_HX_BROKER[0]){puts("CONFIG_DISABLED: RTC cache only, not connected or delivered");return;}
 if(strlen(CONFIG_HX_SSID)>31||strlen(CONFIG_HX_PASSWORD)>63||strncmp(CONFIG_HX_BROKER,"mqtt://",7)){puts("CONFIG_INVALID: SSID<=31, password<=63, trusted LAN mqtt:// required; cache retained");return;}
 flags=xEventGroupCreate();acks=xQueueCreate(4,sizeof(Ack));if(!flags||!acks){puts("ALLOC_ERROR cache retained");return;}
 ESP_ERROR_CHECK(nvs_flash_init()); // no automatic erase on NVS error
 ESP_ERROR_CHECK(esp_netif_init());ESP_ERROR_CHECK(esp_event_loop_create_default());esp_netif_create_default_wifi_sta();
 ESP_ERROR_CHECK(esp_event_handler_register(WIFI_EVENT,ESP_EVENT_ANY_ID,wifiEvent,nullptr));ESP_ERROR_CHECK(esp_event_handler_register(IP_EVENT,IP_EVENT_STA_GOT_IP,wifiEvent,nullptr));
 wifi_init_config_t init=WIFI_INIT_CONFIG_DEFAULT();ESP_ERROR_CHECK(esp_wifi_init(&init));ESP_ERROR_CHECK(esp_wifi_set_storage(WIFI_STORAGE_RAM));ESP_ERROR_CHECK(esp_wifi_set_mode(WIFI_MODE_STA));wifi_config_t cfg{};
 strlcpy(reinterpret_cast<char*>(cfg.sta.ssid),CONFIG_HX_SSID,sizeof cfg.sta.ssid);strlcpy(reinterpret_cast<char*>(cfg.sta.password),CONFIG_HX_PASSWORD,sizeof cfg.sta.password);ESP_ERROR_CHECK(esp_wifi_set_config(WIFI_IF_STA,&cfg));ESP_ERROR_CHECK(esp_wifi_start());
 auto bits=xEventGroupWaitBits(flags,IP_OK|FAIL,pdFALSE,pdFALSE,pdMS_TO_TICKS(8000));
 if((bits&IP_OK)&&!(bits&FAIL)){
  esp_mqtt_client_config_t m{};m.broker.address.uri=CONFIG_HX_BROKER;m.network.disable_auto_reconnect=true;m.network.timeout_ms=2000;m.outbox.limit=4096;
  auto client=esp_mqtt_client_init(&m);if(client){ESP_ERROR_CHECK(esp_mqtt_client_register_event(client,MQTT_EVENT_ANY,mqttEvent,nullptr));auto started=esp_mqtt_client_start(client);
   if(started==ESP_OK){bits=xEventGroupWaitBits(flags,MQTT_READY|FAIL,pdFALSE,pdFALSE,pdMS_TO_TICKS(5000));
    if((bits&MQTT_READY)&&!(bits&FAIL))for(unsigned attempt=0;attempt<6&&batch.count;attempt++){
     uint64_t sequence=headSequence(batch);const char* line=batch.rows[batch.head];int msg=esp_mqtt_client_publish(client,rowTopic,line,0,1,0);
     printf("ATTEMPT seq=%" PRIu64 " sdk_msg_id=%d; PUBACK is not application ACK\n",sequence,msg);if(msg<0)continue;
     int64_t began=esp_timer_get_time();while(esp_timer_get_time()-began<1500000){Ack ack{};if(xQueueReceive(acks,&ack,pdMS_TO_TICKS(20))==pdTRUE&&applicationAck(batch,ack.journal,ack.sequence,sequence)){printf("APP_ACK seq=%" PRIu64 " remaining=%" PRIu32 "\n",sequence,batch.count);break;}}
    }
    ESP_ERROR_CHECK(esp_mqtt_client_stop(client));
   }
   ESP_ERROR_CHECK(esp_mqtt_client_destroy(client));
  }
 }else puts("WIFI_TIMEOUT_OR_FAILURE: cache unchanged");
 ESP_ERROR_CHECK(esp_wifi_stop());ESP_ERROR_CHECK(esp_wifi_deinit());
}
extern "C" void app_main(){
 int64_t start=esp_timer_get_time();uint64_t jid=(uint64_t(esp_random())<<32)|esp_random();bool continued=resume(batch,esp_reset_reason()==ESP_RST_DEEPSLEEP&&esp_sleep_get_wakeup_cause()==ESP_SLEEP_WAKEUP_TIMER,jid);wakeAge(batch);seal(batch);
 uint8_t mac[6];ESP_ERROR_CHECK(esp_read_mac(mac,ESP_MAC_WIFI_STA));char device[13];snprintf(device,sizeof device,"%02x%02x%02x%02x%02x%02x",mac[0],mac[1],mac[2],mac[3],mac[4],mac[5]);snprintf(rowTopic,sizeof rowTopic,"hixinwu/l06/%s/row",device);snprintf(ackTopic,sizeof ackTopic,"hixinwu/l06/%s/ack",device);
 printf("L06 %s journal=%016" PRIx64 " generated=%" PRIu64 " queued=%" PRIu32 " drops=%" PRIu64 " threshold=%d; unknown prior-session loss on reset\n",continued?"CONTINUED":"NEW",batch.journal,batch.generated,batch.count,batch.dropped,CONFIG_HX_BATCH_SIZE);
 if(!continued)vTaskDelay(pdMS_TO_TICKS(5000));
 if(batch.generated<CONFIG_HX_SAMPLE_LIMIT){i2c_master_bus_handle_t bus=nullptr;i2c_master_dev_handle_t dev=nullptr;ShtSample sample{};auto e=createBus(&bus);if(e==ESP_OK)e=addDevice(bus,0x44,&dev);if(e==ESP_OK)e=takeSample(dev,sample);
  bool stored=append(batch,esp_timer_get_time()/1000,e==ESP_OK&&sample.valid,sample.t_mC,sample.rh_raw_mPct);printf("SAMPLE journal=%016" PRIx64 " seq=%" PRIu64 " generated=%" PRIu64 " stored=%u valid=%u drop_total=%" PRIu64 "\n",batch.journal,stored?batch.lastAllocated:0,batch.generated,stored,e==ESP_OK&&sample.valid,batch.dropped);
  if(dev){ESP_ERROR_CHECK(i2c_master_bus_rm_device(dev));}
  if(bus){ESP_ERROR_CHECK(i2c_del_master_bus(bus));}
 }
 if(uploadDue(batch,CONFIG_HX_BATCH_SIZE,batch.generated>=CONFIG_HX_SAMPLE_LIMIT)){upload();}
 seal(batch);
 printf("END active_app_us=%" PRId64 " queued=%" PRIu32 " drops=%" PRIu64 "; no measured current/latency\n",esp_timer_get_time()-start,batch.count,batch.dropped);
 if(batch.generated>=CONFIG_HX_SAMPLE_LIMIT&&!batch.count){puts("DONE: target generated, all stored rows APP_ACKed; check dropped/valid counts separately; stop measurement here");return;}
 ESP_ERROR_CHECK(esp_sleep_enable_timer_wakeup(60000000ULL));fflush(stdout);ESP_ERROR_CHECK(uart_wait_tx_idle_polling(UART_NUM_0));esp_deep_sleep_start();
}