多个无线传感器汇总到 MQTT：做一个 ESP-NOW 网关

// ---- main/gateway_model.h ----
/* SPDX-License-Identifier: MIT */
#pragma once
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <string.h>
#define W4_BYTES 44
#define W4_HEX_BYTES 88
#define W4_CAPACITY 4
#define W4_DATA 1
#define W4_ACK 2
typedef struct {uint8_t node[6];uint64_t boot,mono_ms;uint32_t seq;int32_t t_milli;uint32_t rh_milli;bool valid;} W4Sample;
typedef struct {W4Sample rows[W4_CAPACITY];unsigned head,count;uint64_t dropped;} W4Queue;
typedef enum {W4_GOT_IP,W4_LOST_IP,W4_DISCONNECTED} W4LinkEvent;
typedef struct {bool online,lost_ip;} W4Link;
/* Event loop and owner call these only while holding the same short lock. */
static inline void w4_link_notice(W4Link*s,W4LinkEvent event){s->online=event==W4_GOT_IP;if(event==W4_LOST_IP)s->lost_ip=true;}
static inline W4Link w4_link_take(W4Link*s){W4Link out=*s;s->lost_ip=false;return out;}
static inline bool w4_link_needs_disconnect(W4Link s){return s.lost_ip&&!s.online;}

static inline bool w4_recent(bool seen,uint64_t callback_at,uint64_t now){return seen&&now>=callback_at&&now-callback_at<=15000;}
static inline uint32_t w4_crc(const uint8_t*p,size_t n){uint32_t c=UINT32_MAX;for(size_t i=0;i<n;i++){c^=p[i];for(unsigned j=0;j<8;j++)c=(c>>1)^((c&1)?UINT32_C(0xedb88320):0);}return ~c;}
static inline void w4_put(uint8_t*p,uint64_t v,unsigned n){for(unsigned i=0;i<n;i++)p[i]=(uint8_t)(v>>(8*(n-i-1)));}
static inline uint64_t w4_get(const uint8_t*p,unsigned n){uint64_t v=0;for(unsigned i=0;i<n;i++)v=(v<<8)|p[i];return v;}
static inline bool w4_mac_valid(const uint8_t m[6]){unsigned sum=0;for(unsigned i=0;i<6;i++)sum|=m[i];return sum&&!(m[0]&1);}
static inline bool w4_sample_valid(const W4Sample*s){return w4_mac_valid(s->node)&&s->boot&&s->seq&&s->mono_ms<=INT64_MAX&&(s->valid?(s->t_milli>=-40000&&s->t_milli<=125000&&s->rh_milli<=100000):(s->t_milli==0&&s->rh_milli==0));}
static inline void w4_encode(const W4Sample*s,uint8_t kind,uint8_t b[W4_BYTES]){memset(b,0,W4_BYTES);b[0]='W';b[1]='4';b[2]=1;b[3]=kind;b[4]=s->valid;memcpy(b+6,s->node,6);w4_put(b+12,s->boot,8);w4_put(b+20,s->seq,4);w4_put(b+24,s->mono_ms,8);w4_put(b+32,(uint32_t)s->t_milli,4);w4_put(b+36,s->rh_milli,4);w4_put(b+40,w4_crc(b,40),4);}
static inline bool w4_decode(const uint8_t*b,size_t n,W4Sample*out,uint8_t*kind){
 if(!b||!out||!kind||n!=W4_BYTES||b[0]!='W'||b[1]!='4'||b[2]!=1||(b[3]!=W4_DATA&&b[3]!=W4_ACK)||b[4]>1||b[5]||w4_get(b+40,4)!=w4_crc(b,40))return false;
 W4Sample s={.boot=w4_get(b+12,8),.seq=(uint32_t)w4_get(b+20,4),.mono_ms=w4_get(b+24,8),.rh_milli=(uint32_t)w4_get(b+36,4),.valid=b[4]};memcpy(s.node,b+6,6);int64_t t=(int64_t)w4_get(b+32,4);if(t>INT32_MAX)t-=INT64_C(4294967296);s.t_milli=(int32_t)t;if(!w4_sample_valid(&s))return false;*out=s;*kind=b[3];return true;
}
static inline bool w4_same_id(const W4Sample*a,const W4Sample*b){return !memcmp(a->node,b->node,6)&&a->boot==b->boot&&a->seq==b->seq;}
static inline bool w4_equal(const W4Sample*a,const W4Sample*b){return w4_same_id(a,b)&&a->mono_ms==b->mono_ms&&a->valid==b->valid&&a->t_milli==b->t_milli&&a->rh_milli==b->rh_milli;}
/* 1=new, 2=identical pending duplicate, -1=conflicting identity, 0=full/drop-new. */
static inline int w4_push(W4Queue*q,W4Sample s){for(unsigned i=0;i<q->count;i++){W4Sample*old=&q->rows[(q->head+i)%W4_CAPACITY];if(w4_same_id(old,&s))return w4_equal(old,&s)?2:-1;}if(q->count==W4_CAPACITY){q->dropped++;return 0;}q->rows[(q->head+q->count)%W4_CAPACITY]=s;q->count++;return 1;}
static inline const W4Sample*w4_head(const W4Queue*q){return q->count?&q->rows[q->head]:NULL;}
static inline bool w4_pop_ack(W4Queue*q,const W4Sample*ack){const W4Sample*h=w4_head(q);if(!h||!w4_equal(h,ack))return false;q->head=(q->head+1)%W4_CAPACITY;q->count--;return true;}
static inline void w4_hex(const uint8_t b[W4_BYTES],char out[W4_HEX_BYTES+1]){const char*h="0123456789abcdef";for(unsigned i=0;i<W4_BYTES;i++){out[2*i]=h[b[i]>>4];out[2*i+1]=h[b[i]&15];}out[W4_HEX_BYTES]=0;}
static inline int w4_nibble(char c){return c>='0'&&c<='9'?c-'0':c>='a'&&c<='f'?c-'a'+10:-1;}
static inline bool w4_unhex(const char*text,size_t n,uint8_t out[W4_BYTES]){if(n!=W4_HEX_BYTES)return false;for(unsigned i=0;i<W4_BYTES;i++){int a=w4_nibble(text[2*i]),b=w4_nibble(text[2*i+1]);if(a<0||b<0)return false;out[i]=(uint8_t)(a*16+b);}return true;}
typedef struct {char text[W4_HEX_BYTES];unsigned used;bool active;} W4Assembly;
/* -1=discard, 0=bounded partial, 1=complete. First piece must name the exact
   topic; continuations may omit topic. Reset state on MQTT disconnect too. */
static inline int w4_ack_piece(W4Assembly*a,int offset,int total,const char*data,int len,bool topic_present,bool topic_right,bool retained,uint8_t out[W4_BYTES]){
 if(offset==0){a->active=false;a->used=0;}
 if(retained||offset<0||total!=W4_HEX_BYTES||len<=0||!data||offset>W4_HEX_BYTES||len>W4_HEX_BYTES-offset||(topic_present&&!topic_right)||(offset==0&&(!topic_present||!topic_right))){a->active=false;a->used=0;return -1;}
 if(offset==0)a->active=true;
 if(!a->active||(unsigned)offset!=a->used){a->active=false;a->used=0;return -1;}
 memcpy(a->text+a->used,data,(size_t)len);a->used+=(unsigned)len;
 if(a->used<W4_HEX_BYTES)return 0;
 a->active=false;a->used=0;return w4_unhex(a->text,W4_HEX_BYTES,out)?1:-1;
}
static inline uint8_t w4_sht_crc(const uint8_t*p){uint8_t c=0xff;for(unsigned i=0;i<2;i++){c^=p[i];for(unsigned j=0;j<8;j++)c=(uint8_t)((c<<1)^((c&0x80)?0x31:0));}return c;}
static inline bool w4_sht_decode(const uint8_t b[6],int32_t*t,uint32_t*rh){if(w4_sht_crc(b)!=b[2]||w4_sht_crc(b+3)!=b[5])return false;int64_t tv=-45000+(INT64_C(175000)*w4_get(b,2)+32767)/65535;int64_t hv=-6000+(INT64_C(125000)*w4_get(b+3,2)+32767)/65535;if(tv<-40000||tv>125000)return false;if(hv<0)hv=0;if(hv>100000)hv=100000;*t=(int32_t)tv;*rh=(uint32_t)hv;return true;}

// ---- main/main.c ----
/* SPDX-License-Identifier: MIT */
#include <stdio.h>
#include <inttypes.h>
#include <stdlib.h>
#include <string.h>
#include <stdatomic.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include "esp_event.h"
#include "esp_log.h"
#include "esp_mac.h"
#include "esp_netif.h"
#include "esp_now.h"
#include "esp_random.h"
#include "esp_timer.h"
#include "esp_wifi.h"
#include "nvs_flash.h"
#include "mqtt_client.h"
#include "driver/i2c_master.h"
#include "sdkconfig.h"
#include "gateway_model.h"

#define CHANNEL 6
static const char *TAG="W04";
static const bool gateway=CONFIG_W04_ROLE==0;
static uint8_t addresses[3][6];
static atomic_bool mqtt_online;
static portMUX_TYPE link_lock=portMUX_INITIALIZER_UNLOCKED;
static W4Link link_state;
static atomic_uint_fast32_t callback_drops;
static esp_mqtt_client_handle_t mqtt;
static char data_topic[64],ack_topic[64];
static W4Assembly assembly;
typedef enum {RADIO_RX,RADIO_TX,MQTT_ACK} EventKind;
typedef struct {EventKind kind;uint8_t mac[6],bytes[W4_BYTES],channel;int status;uint64_t at_ms;} Event;
static QueueHandle_t events;
static i2c_master_bus_handle_t bus;
static i2c_master_dev_handle_t sht;
static uint64_t ms(void){return (uint64_t)esp_timer_get_time()/1000;}
static void require(esp_err_t e,const char*operation){if(e!=ESP_OK){ESP_LOGE(TAG,"%s=%s",operation,esp_err_to_name(e));abort();}}
static bool parse_mac(const char*text,uint8_t out[6]){
    if(strlen(text)!=17)return false;
    for(unsigned i=0;i<6;i++){
        int a=w4_nibble(text[3*i]),b=w4_nibble(text[3*i+1]);
        if(a<0||b<0||(i<5&&text[3*i+2]!=':'))return false;
        out[i]=(uint8_t)(a*16+b);
    }
    return w4_mac_valid(out);
}
static void mac_text(const uint8_t m[6],char out[13]){snprintf(out,13,"%02x%02x%02x%02x%02x%02x",m[0],m[1],m[2],m[3],m[4],m[5]);}
static bool configuration(const uint8_t own[6]){
    if(!parse_mac(CONFIG_W04_GATEWAY_MAC,addresses[0])||!parse_mac(CONFIG_W04_NODE_A_MAC,addresses[1])||!parse_mac(CONFIG_W04_NODE_B_MAC,addresses[2]))return false;
    if(!memcmp(addresses[0],addresses[1],6)||!memcmp(addresses[0],addresses[2],6)||!memcmp(addresses[1],addresses[2],6)||memcmp(own,addresses[CONFIG_W04_ROLE],6))return false;
    if(gateway&&(strstr(CONFIG_W04_SSID,"REPLACE_")||strstr(CONFIG_W04_PASSWORD,"REPLACE_")||strstr(CONFIG_W04_BROKER,"REPLACE_")||!strlen(CONFIG_W04_SSID)||strlen(CONFIG_W04_SSID)>32||strlen(CONFIG_W04_PASSWORD)<8||strlen(CONFIG_W04_PASSWORD)>63||strncmp(CONFIG_W04_BROKER,"mqtt://",7)))return false;
    return true;
}
static void post(const Event*e){if(xQueueSend(events,e,0)!=pdTRUE)atomic_fetch_add(&callback_drops,1);}
static void link_notice(W4LinkEvent event){portENTER_CRITICAL(&link_lock);w4_link_notice(&link_state,event);portEXIT_CRITICAL(&link_lock);}
static W4Link link_take(void){portENTER_CRITICAL(&link_lock);W4Link out=w4_link_take(&link_state);portEXIT_CRITICAL(&link_lock);return out;}
static void wifi_event(void*unused,esp_event_base_t base,int32_t id,void*data){
    (void)unused;
    if(base==IP_EVENT&&id==IP_EVENT_STA_GOT_IP){link_notice(W4_GOT_IP);ESP_LOGI(TAG,"GOT_IP: not MQTT/app delivery");}
    else if(base==IP_EVENT&&id==IP_EVENT_STA_LOST_IP){link_notice(W4_LOST_IP);}
    else if(base==WIFI_EVENT&&id==WIFI_EVENT_STA_DISCONNECTED){link_notice(W4_DISCONNECTED);const wifi_event_sta_disconnected_t*d=data;ESP_LOGW(TAG,"AP_DISCONNECTED reason=%u",d->reason);}
}
static void recv_cb(const esp_now_recv_info_t*i,const uint8_t*data,int n){
    if(!i||!i->src_addr||!i->rx_ctrl||!data||n!=W4_BYTES)return;
    bool known=gateway?(!memcmp(i->src_addr,addresses[1],6)||!memcmp(i->src_addr,addresses[2],6)):!memcmp(i->src_addr,addresses[0],6);
    if(!known)return;
    Event e={.kind=RADIO_RX,.channel=i->rx_ctrl->channel,.at_ms=ms()};memcpy(e.mac,i->src_addr,6);memcpy(e.bytes,data,W4_BYTES);post(&e);
}
static void sent_cb(const esp_now_send_info_t*i,esp_now_send_status_t status){if(i&&i->des_addr){Event e={.kind=RADIO_TX,.status=status};memcpy(e.mac,i->des_addr,6);post(&e);}}
static void mqtt_event(void*unused,esp_event_base_t base,int32_t id,void*data){
    (void)unused;(void)base;esp_mqtt_event_handle_t e=data;
    if(id==MQTT_EVENT_CONNECTED){atomic_store(&mqtt_online,true);int mid=esp_mqtt_client_subscribe(mqtt,ack_topic,1);ESP_LOGI(TAG,"MQTT_CONNECTED subscribe_request=%d; validate actual ACK separately",mid);}
    else if(id==MQTT_EVENT_DISCONNECTED){atomic_store(&mqtt_online,false);assembly=(W4Assembly){0};}
    else if(id==MQTT_EVENT_DATA){
        Event event={.kind=MQTT_ACK};bool present=e->topic_len>0;bool right=present&&e->topic_len==(int)strlen(ack_topic)&&!memcmp(e->topic,ack_topic,(size_t)e->topic_len);
        if(w4_ack_piece(&assembly,e->current_data_offset,e->total_data_len,e->data,e->data_len,present,right,e->retain,event.bytes)==1)post(&event);
    }
}
static void start_network(void){
    require(esp_netif_init(),"netif_init");require(esp_event_loop_create_default(),"event_loop");
    if(gateway){if(!esp_netif_create_default_wifi_sta())abort();require(esp_event_handler_register(WIFI_EVENT,ESP_EVENT_ANY_ID,wifi_event,NULL),"wifi_handler");require(esp_event_handler_register(IP_EVENT,ESP_EVENT_ANY_ID,wifi_event,NULL),"ip_handler");}
    wifi_init_config_t init=WIFI_INIT_CONFIG_DEFAULT();require(esp_wifi_init(&init),"wifi_init");require(esp_wifi_set_storage(WIFI_STORAGE_RAM),"wifi_RAM");require(esp_wifi_set_mode(WIFI_MODE_STA),"wifi_STA");
    if(gateway){wifi_config_t cfg={0};memcpy(cfg.sta.ssid,CONFIG_W04_SSID,strlen(CONFIG_W04_SSID));memcpy(cfg.sta.password,CONFIG_W04_PASSWORD,strlen(CONFIG_W04_PASSWORD));cfg.sta.threshold.authmode=WIFI_AUTH_WPA2_PSK;require(esp_wifi_set_config(WIFI_IF_STA,&cfg),"wifi_config");}
    require(esp_wifi_start(),"wifi_start");require(esp_wifi_set_ps(WIFI_PS_NONE),"wifi_PS_NONE");
    if(!gateway){wifi_country_t country;require(esp_wifi_get_country(&country),"country");if(CHANNEL<country.schan||CHANNEL>=country.schan+country.nchan){ESP_LOGE(TAG,"channel6 outside actual country range; stopped");abort();}require(esp_wifi_set_channel(CHANNEL,WIFI_SECOND_CHAN_NONE),"node_channel");}
    require(esp_now_init(),"espnow_init");require(esp_now_register_recv_cb(recv_cb),"espnow_recv");require(esp_now_register_send_cb(sent_cb),"espnow_sent");
    for(unsigned i=gateway?1:0;i<(gateway?3:1);i++){esp_now_peer_info_t peer={.channel=0,.ifidx=WIFI_IF_STA,.encrypt=false};memcpy(peer.peer_addr,addresses[i],6);require(esp_now_add_peer(&peer),"add_peer");}
    if(gateway){
        esp_mqtt_client_config_t cfg={.broker.address.uri=CONFIG_W04_BROKER,.network.reconnect_timeout_ms=2000,.outbox.limit=4096};mqtt=esp_mqtt_client_init(&cfg);if(!mqtt)abort();require(esp_mqtt_client_register_event(mqtt,ESP_EVENT_ANY_ID,mqtt_event,NULL),"mqtt_event");require(esp_mqtt_client_start(mqtt),"mqtt_start");
    }
}
static void start_sensor(void){
    i2c_master_bus_config_t cfg={.i2c_port=I2C_NUM_0,.sda_io_num=8,.scl_io_num=9,.clk_source=I2C_CLK_SRC_DEFAULT,.glitch_ignore_cnt=7,.flags.enable_internal_pullup=false};require(i2c_new_master_bus(&cfg,&bus),"i2c_bus");i2c_device_config_t dev={.dev_addr_length=I2C_ADDR_BIT_LEN_7,.device_address=0x44,.scl_speed_hz=100000};require(i2c_master_bus_add_device(bus,&dev,&sht),"SHT40_device");
}
static W4Sample sample(const uint8_t own[6],uint64_t boot,uint32_t seq){
    W4Sample s={.boot=boot,.seq=seq,.mono_ms=ms()};memcpy(s.node,own,6);const uint8_t cmd=0xfd;uint8_t raw[6];esp_err_t e=i2c_master_transmit(sht,&cmd,1,100);if(e==ESP_OK){vTaskDelay(pdMS_TO_TICKS(20));e=i2c_master_receive(sht,raw,6,100);}
    if(e==ESP_OK)s.valid=w4_sht_decode(raw,&s.t_milli,&s.rh_milli);
    if(!s.valid)ESP_LOGW(TAG,"SHT_INVALID seq=%"PRIu32" api=%s; explicit invalid sample, not old value",seq,esp_err_to_name(e));
    return s;
}
static void log_id(const char*event,const W4Sample*s){char node[13];mac_text(s->node,node);ESP_LOGI(TAG,"%s node=%s boot=%016"PRIx64" seq=%"PRIu32" sample_valid=%d",event,node,s->boot,s->seq,s->valid);}
static bool radio_send(const uint8_t peer[6],const W4Sample*s,uint8_t kind){uint8_t raw[W4_BYTES];w4_encode(s,kind,raw);esp_err_t e=esp_now_send(peer,raw,W4_BYTES);if(e!=ESP_OK)ESP_LOGW(TAG,"ESP_NOW_SEND_REQUEST=%s; not application ACK",esp_err_to_name(e));return e==ESP_OK;}
void app_main(void){
    uint8_t own[6];require(esp_read_mac(own,ESP_MAC_WIFI_STA),"read_STA_MAC");ESP_LOGI(TAG,"ACTUAL_STA="MACSTR" role=%d; configure this board's matching role",MAC2STR(own),CONFIG_W04_ROLE);
    if(!configuration(own)){ESP_LOGW(TAG,"CONFIG_CLOSED: exact owned three STA MACs and gateway LAN config needed; no radio application initialized");return;}
    esp_err_t nvs=nvs_flash_init();if(nvs!=ESP_OK){ESP_LOGE(TAG,"NVS=%s; preserve/backup, no automatic erase",esp_err_to_name(nvs));return;}
    char gateway_id[13];mac_text(addresses[0],gateway_id);snprintf(data_topic,sizeof(data_topic),"hixinwu/w04/%s/row",gateway_id);snprintf(ack_topic,sizeof(ack_topic),"hixinwu/w04/%s/ack",gateway_id);
    events=xQueueCreate(16,sizeof(Event));if(!events)abort();start_network();if(!gateway)start_sensor();
    W4Queue pending[2]={0};W4Sample committed[2]={0};bool ack_pending[2]={false,false};uint64_t publish_at[2]={0,0};
    bool seen[2]={false,false},was_recent[2]={false,false};uint64_t seen_at[2]={0,0};
    uint64_t boot=0;esp_fill_random(&boot,sizeof(boot));if(!boot)boot=1;
    uint64_t next_sample=ms(),next_radio=0,next_connect=0,next_status=ms(),next_channel=0;uint32_t seq=0;unsigned radio_turn=0,last_drops=0;bool radio_busy=false,radio_stalled=false,channel_ok=!gateway;uint8_t busy_peer[6]={0};uint64_t busy_at=0,missed=0;
    for(;;){
        Event e;
        while(xQueueReceive(events,&e,0)==pdTRUE){
            if(e.kind==RADIO_TX){if(radio_busy&&!memcmp(busy_peer,e.mac,6)){radio_busy=false;ESP_LOGI(TAG,"MAC_CALLBACK=%d; not receiver SQLite/applicationACK",e.status);}continue;}
            W4Sample s;uint8_t kind;if(!w4_decode(e.bytes,W4_BYTES,&s,&kind))continue;
            if(e.kind==MQTT_ACK){
                if(!gateway||kind!=W4_DATA)continue;
                for(unsigned i=0;i<2;i++)if(w4_pop_ack(&pending[i],&s)){committed[i]=s;ack_pending[i]=true;publish_at[i]=0;log_id("APP_COMMIT_ACK",&s);}
            }else if(e.channel==CHANNEL){
                if(gateway&&kind==W4_DATA&&!memcmp(e.mac,s.node,6)){
                    unsigned i=!memcmp(e.mac,addresses[1],6)?0:1;seen[i]=true;seen_at[i]=e.at_ms;int result=w4_push(&pending[i],s);log_id(result==1?"GATEWAY_QUEUED":result==2?"GATEWAY_PENDING_DUPLICATE":result==0?"GATEWAY_FULL_DROP_NEW":"GATEWAY_ID_CONFLICT",&s);
                }else if(!gateway&&kind==W4_ACK&&w4_pop_ack(&pending[0],&s)){next_radio=0;log_id("NODE_END_TO_END_ACK",&s);}
            }
        }
        const uint64_t now=ms();
        if(radio_busy&&now-busy_at>=2000&&!radio_stalled){radio_stalled=true;ESP_LOGE(TAG,"MAC_CALLBACK_TIMEOUT: radio stopped until manual reset; delivery may already have happened");}
        if(gateway){
            for(unsigned i=0;i<2;i++){bool recent=w4_recent(seen[i],seen_at[i],now);if(recent!=was_recent[i]){was_recent[i]=recent;ESP_LOGI(TAG,"NODE_%c CALLBACK_TRAFFIC_%s; not fresh sample/physical presence proof",'A'+(int)i,recent?"RECENT":"STALE");}}
            W4Link link=link_take();bool online=link.online;
            if(w4_link_needs_disconnect(link)){(void)esp_wifi_disconnect();next_connect=now+2000;}
            if(!online&&now>=next_connect){next_connect=now+5000;esp_err_t rc=esp_wifi_connect();ESP_LOGI(TAG,"CONNECT_REQUEST=%s",esp_err_to_name(rc));}
            if(now>=next_channel){next_channel=now+1000;uint8_t c=0;wifi_second_chan_t secondary;channel_ok=online&&esp_wifi_get_channel(&c,&secondary)==ESP_OK&&c==CHANNEL;ESP_LOGI(TAG,"GATEWAY_CHANNEL actual=%u qualified=%d required=6; no set_channel while associated",c,channel_ok);}
            for(unsigned i=0;i<2;i++){
                const W4Sample*s=w4_head(&pending[i]);
                if(s&&online&&atomic_load(&mqtt_online)&&now>=publish_at[i]){uint8_t raw[W4_BYTES];char hex[W4_HEX_BYTES+1];w4_encode(s,W4_DATA,raw);w4_hex(raw,hex);int mid=esp_mqtt_client_enqueue(mqtt,data_topic,hex,W4_HEX_BYTES,1,0,true);publish_at[i]=now+2000;log_id(mid>=0?"MQTT_ENQUEUE_ACCEPTED":"MQTT_ENQUEUE_REJECTED",s);}
            }
            if(channel_ok&&!radio_busy&&!radio_stalled){
                for(unsigned n=0;n<2;n++){unsigned i=(radio_turn+n)%2;if(!ack_pending[i])continue;if(radio_send(addresses[i+1],&committed[i],W4_ACK)){radio_busy=true;busy_at=now;memcpy(busy_peer,addresses[i+1],6);ack_pending[i]=false;radio_turn=(i+1)%2;}break;}
            }
        }else{
            if(now>=next_sample){uint64_t due=(now-next_sample)/5000+1;next_sample+=due*5000;missed+=due-1;if(seq==UINT32_MAX)ESP_LOGE(TAG,"SEQ_EXHAUSTED: no wrap/new samples; preserve logs and reset deliberately");else{W4Sample s=sample(own,boot,++seq);log_id(w4_push(&pending[0],s)==1?"SAMPLED_QUEUED":"NODE_FULL_DROP_NEW",&s);}}
            const W4Sample*s=w4_head(&pending[0]);
            if(s&&!radio_busy&&!radio_stalled&&ms()>=next_radio){next_radio=ms()+2000;if(radio_send(addresses[0],s,W4_DATA)){radio_busy=true;busy_at=ms();memcpy(busy_peer,addresses[0],6);}}
        }
        if(now>=next_status){next_status=now+5000;unsigned drops=(unsigned)atomic_load(&callback_drops);if(drops!=last_drops){last_drops=drops;ESP_LOGW(TAG,"CALLBACK_DROP_NEW=%u",drops);}ESP_LOGI(TAG,"QUEUE countA=%u countB=%u dropA=%"PRIu64" dropB=%"PRIu64" skipped_plan=%"PRIu64" RAM_ONLY; reset/power loss may lose unACKed samples",pending[0].count,pending[1].count,pending[0].dropped,pending[1].dropped,missed);}
        vTaskDelay(pdMS_TO_TICKS(10));
    }
}

# ---- host/wire.py ----
"""SPDX-License-Identifier: MIT. W04 exact44-byte DATA in lowercase88-byte hex."""
from dataclasses import dataclass
import binascii,re,struct
FORMAT=struct.Struct('!2sBBBB6sQIQiII')
MAC=re.compile(r'[0-9a-f]{12}\Z')
def mac_valid(node):
 return bool(MAC.fullmatch(node)) and int(node,16)!=0 and not(int(node[:2],16)&1)
@dataclass(frozen=True)
class Sample:
 node:str
 boot:str
 seq:int
 mono_ms:int
 valid:bool
 t_milli:int
 rh_milli:int
 payload:str
def decode(payload):
 if not isinstance(payload,bytes)or len(payload)!=88 or not re.fullmatch(rb'[0-9a-f]{88}',payload):raise ValueError('exact lowercase88 ASCII bytes required')
 raw=bytes.fromhex(payload.decode('ascii'))
 magic,version,kind,valid,reserved,node,boot,seq,mono,t,rh,crc=FORMAT.unpack(raw)
 if magic!=b'W4'or version!=1 or kind!=1 or valid>1 or reserved or binascii.crc32(raw[:40])!=crc:raise ValueError('frame header/CRC')
 node=node.hex()
 if not mac_valid(node)or not boot or not seq or mono>9223372036854775807:raise ValueError('identity/time range')
 if (valid and not(-40000<=t<=125000 and rh<=100000))or(not valid and(t!=0 or rh!=0)):raise ValueError('sample range/invalid marker')
 return Sample(node,f'{boot:016x}',seq,mono,bool(valid),t,rh,payload.decode('ascii'))

# ---- host/receiver.py ----
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""Own LAN receiver: SQLite FULL commit before explicit application ACK."""
from pathlib import Path
import argparse,sqlite3,time,json,uuid
from wire import decode,mac_valid
class Store:
 def __init__(self,path):
  self.db=sqlite3.connect(path);self.db.execute('PRAGMA journal_mode=WAL');self.db.execute('PRAGMA synchronous=FULL')
  self.db.execute('CREATE TABLE IF NOT EXISTS samples(node TEXT,boot TEXT,seq INTEGER,mono_ms INTEGER,valid INTEGER,t_milli INTEGER,rh_milli INTEGER,payload TEXT,receiver_session TEXT,received_wall REAL,PRIMARY KEY(node,boot,seq))')
  self.db.execute('CREATE TABLE IF NOT EXISTS counters(name TEXT PRIMARY KEY,n INTEGER NOT NULL)');self.db.commit();self.session=uuid.uuid4().hex
 def counter(self,name):
  self.db.execute('INSERT INTO counters VALUES(?,1) ON CONFLICT(name)DO UPDATE SET n=n+1',(name,));self.db.commit()
 def ingest(self,payload,allowed):
  try:s=decode(payload)
  except ValueError:self.counter('invalid');return 'INVALID',None
  if s.node not in allowed:self.counter('unknown_node');return 'UNKNOWN_NODE',None
  old=self.db.execute('SELECT payload FROM samples WHERE node=? AND boot=? AND seq=?',(s.node,s.boot,s.seq)).fetchone()
  if old is not None:
   result='DUPLICATE'if old[0]==s.payload else'CONFLICT';self.counter(result.lower());return result,s if result=='DUPLICATE'else None
  self.db.execute('INSERT INTO samples VALUES(?,?,?,?,?,?,?,?,?,?)',(s.node,s.boot,s.seq,s.mono_ms,int(s.valid),s.t_milli,s.rh_milli,s.payload,self.session,time.time()));self.db.commit();return 'COMMITTED',s
 def close(self):self.db.close()
def main():
 import paho.mqtt.client as mqtt
 p=argparse.ArgumentParser();p.add_argument('--host',required=True);p.add_argument('--port',type=int,default=1883);p.add_argument('--gateway',required=True);p.add_argument('--node',action='append',required=True);p.add_argument('--db',required=True);a=p.parse_args()
 if not mac_valid(a.gateway)or len(a.node)!=2 or len(set(a.node))!=2 or any(not mac_valid(n)for n in a.node)or a.gateway in a.node:p.error('exactly two distinct owned node STA MACs and one different gateway STA MAC, lowercase12hex')
 topic='hixinwu/w04/'+a.gateway;store=Store(Path(a.db));client=mqtt.Client(client_id='w04-receiver-'+store.session[:12],clean_session=True)
 def connected(c,u,f,rc):
  if rc!=0:raise RuntimeError('broker connection refused')
  result,mid=c.subscribe(topic+'/row',1)
  if result!=mqtt.MQTT_ERR_SUCCESS:raise RuntimeError('subscription request rejected')
 def subscribed(c,u,mid,qos):
  if not qos or any(q==128 for q in qos):raise RuntimeError('subscription denied')
  print('OBSERVER_READY subscribed to one owned gateway topic',flush=True)
 def message(c,u,msg):
  if msg.topic!=topic+'/row':return
  try:
   if msg.retain:store.counter('retained_rejected');print('RETAINED_REJECTED no applicationACK',flush=True);return
   result,s=store.ingest(msg.payload,set(a.node))
  except sqlite3.Error:c.disconnect();raise
  print(json.dumps({'event':result,'id':None if s is None else[s.node,s.boot,s.seq],'sample_valid':None if s is None else s.valid}),flush=True)
  if s is not None:
   # ACK is the exact DATA text, never an SDK message ID. Duplicate also ACKs.
   info=c.publish(topic+'/ack',s.payload,qos=1,retain=False)
   if info.rc!=mqtt.MQTT_ERR_SUCCESS:print('APP_ACK_NOT_ACCEPTED_BY_CLIENT; node/gateway retry same sample',flush=True)
 client.on_connect=connected;client.on_subscribe=subscribed;client.on_message=message;client.connect(a.host,a.port,30)
 try:client.loop_forever()
 except KeyboardInterrupt:pass
 finally:client.disconnect();store.close()
if __name__=='__main__':main()