连续室内环境站：把有效读数、CSV和断网补传接成闭环

. /PATH/TO/esp-idf-v5.5.5/export.sh
idf.py --version
idf.py set-target esp32s3
idf.py menuconfig
idf.py build
idf.py -p /dev/cu.REPLACE_WITH_BRIDGE flash monitor

mosquitto -c broker-loopback.conf
python3.12 host/receiver.py receive --db ./receiver.sqlite --host 127.0.0.1 --port 18889 --client-dir /PATH/TO/CLIENTS
python3.12 host/receiver.py export --db ./receiver.sqlite --output /NEW/receiver-export.csv

python3.12 host/repair_csv.py /BACKUP/room.csv
python3.12 host/repair_csv.py /BACKUP/room.csv --output /NEW/PREFIX-room.csv

/* SPDX-License-Identifier: MIT */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <inttypes.h>
#include <errno.h>
#include <time.h>
#include <stdatomic.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include "freertos/event_groups.h"
#include "driver/spi_common.h"
#include "driver/sdspi_host.h"
#include "sdmmc_cmd.h"
#include "esp_vfs_fat.h"
#include "esp_log.h"
#include "esp_timer.h"
#include "esp_random.h"
#include "esp_mac.h"
#include "esp_netif_sntp.h"
#include "mqtt_client.h"
#include "nvs_flash.h"
#include "nvs.h"
#include "sdkconfig.h"
#include "net.h"
#include "sensirion_i2c_hal.h"
#include "sensirion_hal_idf.h"
#include "sht4x_i2c.h"
#include "station_model.h"
#include "station_journal.h"
#include "station_display.h"
static const char *TAG="S10",*PATH="/sd/room.csv";
static QueueHandle_t acks;static EventGroupHandle_t mqtt_flags;
static char row_topic[64],ack_topic[64],availability_topic[64],client_id[32];static int subscription_id;
static atomic_bool time_synced=false;
static atomic_uint_fast64_t last_sync_ms=0;
static station_scan_t scan;static station_pump_t pump;
static bool storage_ready,write_stopped,read_stopped,nvs_ready,checkpoint_fault;
static long next_offset=-1,pending_end=-1;static char pending_line[STATION_LINE_MAX];
static uint64_t lost_samples=0;static unsigned uncheckpointed=0;
typedef struct{uint64_t journal,sequence;} app_ack_t;
static void sync_notice(struct timeval *time){(void)time;atomic_store(&last_sync_ms,(uint64_t)esp_timer_get_time()/1000ULL);atomic_store(&time_synced,true);}
static esp_err_t mount_sd(void){
 sdmmc_host_t host=SDSPI_HOST_DEFAULT();host.max_freq_khz=10000;
 spi_bus_config_t bus={.mosi_io_num=11,.miso_io_num=13,.sclk_io_num=12,.quadwp_io_num=-1,.quadhd_io_num=-1,.max_transfer_sz=4096};
 esp_err_t e=spi_bus_initialize(host.slot,&bus,SDSPI_DEFAULT_DMA);if(e!=ESP_OK)return e;
 sdspi_device_config_t dev=SDSPI_DEVICE_CONFIG_DEFAULT();dev.gpio_cs=10;dev.host_id=host.slot;
 esp_vfs_fat_sdmmc_mount_config_t cfg={.format_if_mount_failed=false,.max_files=4,.allocation_unit_size=16*1024};
 sdmmc_card_t *card=NULL;e=esp_vfs_fat_sdspi_mount("/sd",&host,&dev,&cfg,&card);if(e!=ESP_OK)spi_bus_free(host.slot);return e;
}
static void load_checkpoint(void){
 uint64_t journal=0,sequence=0;nvs_handle_t h;if(!nvs_ready||nvs_open("room_station",NVS_READONLY,&h)!=ESP_OK)return;
 uint8_t bytes[28];size_t size=sizeof(bytes);esp_err_t e=nvs_get_blob(h,"ack",bytes,&size);nvs_close(h);
 if(e!=ESP_OK||!station_checkpoint_decode(bytes,size,&journal,&sequence)||journal!=pump.journal)return;
 if(sequence>scan.last_sequence){read_stopped=true;write_stopped=true;ESP_LOGE(TAG,"CHECKPOINT_AHEAD: possible lost media tail; preserve card and database; no silent reset");return;}
 pump.acked=sequence;
}
static void checkpoint(void){
 if(!nvs_ready){checkpoint_fault=true;return;}uint8_t bytes[28];station_checkpoint_encode(pump.journal,pump.acked,bytes);
 nvs_handle_t h;esp_err_t e=nvs_open("room_station",NVS_READWRITE,&h);
 if(e==ESP_OK){e=nvs_set_blob(h,"ack",bytes,sizeof(bytes));if(e==ESP_OK)e=nvs_commit(h);nvs_close(h);}
 checkpoint_fault=e!=ESP_OK;if(!checkpoint_fault)uncheckpointed=0;
 ESP_LOGI(TAG,"CHECKPOINT ack=%"PRIu64" result=%s; a failed checkpoint may replay duplicates",pump.acked,esp_err_to_name(e));
}
static void inspect_media(void){
 FILE *file=fopen(PATH,"r");station_result_t result=STATION_OK;
 if(!file){if(errno!=ENOENT)result=STATION_IO_ERROR;}
 else{result=station_scan(file,&scan);if(fclose(file)!=0)result=STATION_IO_ERROR;}
 write_stopped=result!=STATION_OK;read_stopped=result==STATION_IO_ERROR;
 pump.journal=scan.journal;load_checkpoint();
 if(scan.good_bytes)next_offset=station_offset_after(PATH,pump.acked);
 ESP_LOGI(TAG,"SCAN result=%d validated_prefix_bytes=%u last_seq=%"PRIu64" ack=%"PRIu64"; no automatic truncate/format",result,(unsigned)scan.good_bytes,scan.last_sequence,pump.acked);
}
static void mqtt_event(void *arg,esp_event_base_t base,int32_t id,void *data){
 (void)arg;(void)base;esp_mqtt_event_handle_t e=data;
 if(id==MQTT_EVENT_CONNECTED){subscription_id=esp_mqtt_client_subscribe(e->client,ack_topic,1);ESP_LOGI(TAG,"MQTT connected; waiting for ACK topic SUBACK");}
 else if(id==MQTT_EVENT_SUBSCRIBED&&e->msg_id==subscription_id){xEventGroupSetBits(mqtt_flags,BIT0);esp_mqtt_client_publish(e->client,availability_topic,"online",0,1,1);}
 else if(id==MQTT_EVENT_DISCONNECTED){xEventGroupClearBits(mqtt_flags,BIT0);}
 else if(id==MQTT_EVENT_DATA&&station_ack_frame(e->topic,e->topic_len,ack_topic,e->current_data_offset,e->total_data_len,e->data_len,e->retain)){
  app_ack_t ack;if(station_ack_decode(e->data,e->data_len,&ack.journal,&ack.sequence)&&xQueueSend(acks,&ack,0)!=pdTRUE)ESP_LOGW(TAG,"ACK queue full: retry/dedup will recover; no cursor advanced");
 }
}
static void pump_rows(esp_mqtt_client_handle_t client,uint64_t now){
 app_ack_t ack;
 while(xQueueReceive(acks,&ack,0)==pdTRUE)if(station_pump_ack(&pump,ack.journal,ack.sequence)){
  next_offset=pending_end;pending_line[0]=0;if(++uncheckpointed>=16&&!checkpoint_fault)checkpoint();
 }
 if(!storage_ready||read_stopped||pump.acked>=scan.last_sequence||!(xEventGroupGetBits(mqtt_flags)&BIT0)||!station_pump_due(&pump,now))return;
 if(!pump.waiting){
  if(next_offset<0)next_offset=station_offset_after(PATH,pump.acked);
  if(next_offset<0||!station_next(PATH,next_offset,pending_line,&pending_end)){read_stopped=true;ESP_LOGE(TAG,"BACKFILL_READ: preserve card; no cursor advanced");return;}
  station_record_t row;if(!station_record_decode(pending_line,&row)||row.journal!=pump.journal||row.sequence!=pump.acked+1){read_stopped=true;return;}
 }
 station_pump_attempt(&pump,pump.acked+1,now);
 int queued=esp_mqtt_client_enqueue(client,row_topic,pending_line,0,1,0,true);
 ESP_LOGI(TAG,"BACKFILL seq=%"PRIu64" queued=%d; broker QoS1 is not receiver database ACK",pump.inflight,queued);
}
void app_main(void){
 esp_err_t e=nvs_flash_init();nvs_ready=e==ESP_OK;if(!nvs_ready)ESP_LOGW(TAG,"NVS init=%s; no automatic erase",esp_err_to_name(e));
 e=mount_sd();storage_ready=e==ESP_OK;if(storage_ready)inspect_media();else ESP_LOGE(TAG,"SD mount=%s; sample/display continue without durable rows",esp_err_to_name(e));
 sensirion_i2c_hal_init();sht4x_init(0x44);bool screen=station_display_init();
 if(!screen)ESP_LOGW(TAG,"OLED init path failed; serial/SD continue, no pixel claim");
 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(client_id,sizeof(client_id),"hixinwu-s10-%s",device);snprintf(row_topic,sizeof(row_topic),"hixinwu/s10/%s/row",device);snprintf(ack_topic,sizeof(ack_topic),"hixinwu/s10/%s/ack",device);snprintf(availability_topic,sizeof(availability_topic),"hixinwu/s10/%s/availability",device);
 acks=xQueueCreate(8,sizeof(app_ack_t));mqtt_flags=xEventGroupCreate();if(!acks||!mqtt_flags){ESP_LOGE(TAG,"queue/event allocation failed");return;}
 net_start();
 esp_mqtt_client_config_t cfg={.broker.address.uri=CONFIG_DEMO_BROKER_URL,.credentials.client_id=client_id,.session.last_will={.topic=availability_topic,.msg="offline",.qos=1,.retain=1},.network={.timeout_ms=3000,.reconnect_timeout_ms=3000},.outbox.limit=2048};
 esp_mqtt_client_handle_t client=esp_mqtt_client_init(&cfg);if(!client){ESP_LOGE(TAG,"MQTT init failed");return;}ESP_ERROR_CHECK(esp_mqtt_client_register_event(client,ESP_EVENT_ANY_ID,mqtt_event,NULL));
 bool mqtt_started=false,sntp_started=false,last_valid=false;uint64_t sample_at=0,display_at=0,last_good=0;int32_t last_t=0,last_rh=0;unsigned failures=0;
 ESP_LOGI(TAG,"S10 START device=%s; SD512KiB cap, sample60s, full64 time; no physical/24h run claimed",device);
 for(;;){
  uint64_t now=(uint64_t)esp_timer_get_time()/1000ULL;
  if(net_online()&&!mqtt_started){e=esp_mqtt_client_start(client);mqtt_started=e==ESP_OK;}
  if(net_online()&&!sntp_started){esp_sntp_config_t time_cfg=ESP_NETIF_SNTP_DEFAULT_CONFIG(CONFIG_STATION_NTP_SERVER);time_cfg.sync_cb=sync_notice;time_cfg.wait_for_sync=false;e=esp_netif_sntp_init(&time_cfg);sntp_started=e==ESP_OK;}
  if(now>=sample_at){
   sample_at=now+60000;int32_t t=0,rh=0;int16_t error=demo_i2c_ready()?sht4x_measure_high_precision(&t,&rh):-1;last_valid=error==0;
   if(last_valid){last_t=t;last_rh=rh;last_good=now;failures=0;}else{if(++failures%3==0&&demo_i2c_ready())sht4x_soft_reset();ESP_LOGW(TAG,"SHT40 error=%d; invalid row fields, no old value as current",error);}
   station_record_t row={.journal=pump.journal,.sequence=scan.last_sequence+1,.uptime=(uint64_t)esp_timer_get_time()/1000ULL,.unix_s=(int64_t)time(NULL),.sample_valid=last_valid,.t_milli=last_valid?t:0,.rh_milli=last_valid?rh:0};
   row.time_valid=station_clock_recent(atomic_load(&time_synced),atomic_load(&last_sync_ms),row.uptime,row.unix_s);if(!row.time_valid)row.unix_s=0;
   if(!row.journal){row.journal=((uint64_t)esp_random()<<32)|esp_random();if(!row.journal)row.journal=1;}
   char line[STATION_LINE_MAX];station_result_t result=STATION_IO_ERROR;
   if(storage_ready&&!write_stopped&&scan.last_sequence<UINT64_MAX&&station_record_encode(&row,line,sizeof(line)))result=station_append(PATH,&scan,line,&station_default_io);
   if(result==STATION_OK){if(!scan.good_bytes)scan.good_bytes=strlen(STATION_CSV_HEADER);scan.good_bytes+=strlen(line);scan.last_sequence=row.sequence;scan.journal=pump.journal=row.journal;ESP_LOGI(TAG,"APPEND_API_SUCCESS %s",line);}
   else{++lost_samples;if(storage_ready)write_stopped=true;ESP_LOGW(TAG,"NOT_RECORDED result=%d lost_this_boot=%"PRIu64"; cap and true I/O errors are separate",result,lost_samples);}
  }
  if(mqtt_started)pump_rows(client,(uint64_t)esp_timer_get_time()/1000ULL);
  now=(uint64_t)esp_timer_get_time()/1000ULL;
  if(now>=display_at){
   display_at=now+1000;char text[8][33];const char *lines[8];for(unsigned i=0;i<8;i++)lines[i]=text[i];
   snprintf(text[0],33,"S10 ROOM STATION");
   if(last_valid&&now>=last_good&&now-last_good<=70000)snprintf(text[1],33,"T %.1f H %.1f",last_t/1000.0,(last_rh<0?0:last_rh>100000?100000:last_rh)/1000.0);else snprintf(text[1],33,"SENSOR INVALID");
   time_t utc=time(NULL);uint64_t sync_at=atomic_load(&last_sync_ms);bool clock_ok=station_clock_recent(atomic_load(&time_synced),sync_at,now,utc);snprintf(text[2],33,"UTC %s",clock_ok?(now-sync_at<=3600000ULL?"FRESH":"HOLDOVER"):"UNKNOWN");
   snprintf(text[3],33,"SD %s S %"PRIu64,!storage_ready?"MISSING":write_stopped?"STOP":"OK",scan.last_sequence);
   snprintf(text[4],33,"MQTT %s",xEventGroupGetBits(mqtt_flags)&BIT0?"ACK READY":"OFFLINE");
   snprintf(text[5],33,"ACK %"PRIu64,pump.acked);snprintf(text[6],33,"LOST %"PRIu64" RAM",lost_samples);snprintf(text[7],33,"CP %s READ %s",checkpoint_fault?"ERR":"OK",read_stopped?"STOP":"OK");
   if(screen&&!station_display_lines(lines))ESP_LOGW(TAG,"OLED transaction failed; pixel state unverified");
   ESP_LOGI(TAG,"STATUS seq=%"PRIu64" ack=%"PRIu64" clock=%s SD=%u write_stop=%u read_stop=%u lost=%"PRIu64,scan.last_sequence,pump.acked,text[2],storage_ready,write_stopped,read_stopped,lost_samples);
  }
  vTaskDelay(pdMS_TO_TICKS(100));
 }
}