元器件车间仓储区：RJ45温湿度记录仪，TCP/IP本地存储与远端同步

┌─────────────────────────────────────────────────────────────┐
│                    设备侧（记录仪）                           │
│                                                             │
│  ┌──────────┐    ┌──────────┐    ┌──────────────────────┐  │
│  │ 传感器    │───→│ 采集任务  │───→│ 本地存储管理器        │  │
│  │ 采样1Hz   │    │ 5s/次    │    │ Flash循环缓冲区       │  │
│  └──────────┘    └──────────┘    │ 12字节/条 × 4096条    │  │
│                                   └──────────┬───────────┘  │
│                                              │              │
│                              ┌───────────────▼───────────┐  │
│                              │ TCP/IP 同步客户端           │  │
│                              │ 连接管理 + 补传协议         │  │
│                              └───────────────┬───────────┘  │
└──────────────────────────────────────────────┼──────────────┘
                                               │ TCP 502/9000
                    ┌──────────────────────────▼────────────┐
                    │          车间本地服务器                 │
                    │  ┌─────────────┐   ┌───────────────┐  │
                    │  │ 接收服务     │   │ 时序数据库     │  │
                    │  │ TCP 502      │──→│ InfluxDB      │  │
                    │  └─────────────┘   └───────┬───────┘  │
                    │                             │          │
                    │  ┌─────────────┐            │          │
                    │  │ 去重引擎     │←───────────┘          │
                    │  │ dev+ts主键   │                      │
                    │  └─────────────┘                      │
                    └────────────────────────────────────────┘
                                     │
                              ┌──────▼──────┐
                              │  云端备份    │
                              │  (异步同步)  │
                              └─────────────┘

30天 × 24小时 × 3600秒 ÷ 5秒 = 518,400条
518,400条 × 12字节/条 = 6,220,800字节 ≈ 6MB

// storage.h
#ifndef _STORAGE_H_
#define _STORAGE_H_

#include <stdint.h>
#include <stdbool.h>

#define FLASH_PAGE_SIZE      256
#define FLASH_SECTOR_SIZE     4096
#define RECORD_SIZE           12     // 每条记录12字节
#define RECORDS_PER_SECTOR    (FLASH_SECTOR_SIZE / RECORD_SIZE)  // 341条/扇区
#define DATA_SECTOR_COUNT     1984   // 7.9MB / 4KB
#define DATA_START_SECTOR     16      // 跳过系统区

// 单条记录结构（12字节，packed）
typedef struct __attribute__((packed)) {
    uint32_t collect_ts;      // 采集时间戳（Unix epoch）
    int16_t  temperature;     // 温度 ×100
    uint16_t humidity;        // 湿度 ×10
    uint16_t voltage;         // 供电电压 ×100
    uint8_t  status;         // 状态位
    uint8_t  reserved;        // 保留，对齐用
} env_record_t;               // 12字节

// 循环缓冲区元数据（存储在RAM，掉电不丢因为有Flash镜像）
typedef struct {
    uint32_t write_sector;    // 当前写入扇区号
    uint32_t write_offset;    // 扇区内偏移（记录数）
    uint32_t total_records;   // 总记录数（循环计数）
    uint32_t oldest_ts;       // 最旧记录的时间戳
    uint32_t newest_ts;       // 最新记录的时间戳
    uint16_t dev_addr;        // 设备地址
} ring_buffer_meta_t;

int storage_init(void);
int storage_write_record(const env_record_t *record);
int storage_read_record(uint32_t index, env_record_t *record);
int storage_read_range(uint32_t start_ts, uint32_t end_ts, 
                       env_record_t *out_buf, uint32_t max_count,
                       uint32_t *actual_count);
int storage_get_unsynced(uint32_t last_sync_ts, 
                         env_record_t *out_buf, uint32_t max_count,
                         uint32_t *actual_count);
void storage_format(void);
uint32_t storage_get_total_stored(void);

#endif

// storage.c 核心实现
#include "storage.h"
#include "w25qxx.h"
#include "crc16.h"

static ring_buffer_meta_t g_meta;
static env_record_t g_ram_cache[32];
static uint8_t g_cache_count = 0;
static uint32_t g_last_flush_ms = 0;

#define CACHE_FLUSH_INTERVAL_MS  1000   // 1秒强制刷盘
#define CACHE_THRESHOLD          32     // 满32条刷盘

int storage_init(void) {
    w25qxx_init();
    
    // 读取元数据（从Flash最后一个扇区恢复）
    uint8_t meta_buf[sizeof(ring_buffer_meta_t)];
    w25qxx_read_data((DATA_START_SECTOR + DATA_SECTOR_COUNT) * FLASH_SECTOR_SIZE,
                     meta_buf, sizeof(meta_buf));
    
    memcpy(&g_meta, meta_buf, sizeof(g_meta));
    
    // 校验元数据有效性
    uint16_t crc = crc16_modbus(meta_buf, sizeof(g_meta) - 2);
    if (crc != *(uint16_t *)(meta_buf + sizeof(g_meta) - 2)) {
        // 元数据损坏，重新格式化
        storage_format();
    }
    
    return 0;
}

int storage_write_record(const env_record_t *record) {
    // 写入RAM缓存
    memcpy(&g_ram_cache[g_cache_count], record, RECORD_SIZE);
    g_cache_count++;
    
    uint32_t now = sys_now();
    
    // 条件满足时刷入Flash
    if (g_cache_count >= CACHE_THRESHOLD || 
        (now - g_last_flush_ms) >= CACHE_FLUSH_INTERVAL_MS) {
        return flush_to_flash();
    }
    
    return 0;
}

static int flush_to_flash(void) {
    if (g_cache_count == 0) return 0;
    
    // 计算写入地址
    uint32_t addr = (g_meta.write_sector * FLASH_SECTOR_SIZE) + 
                    (g_meta.write_offset * RECORD_SIZE);
    
    // 如果当前扇区写满，擦除下一个扇区
    if (g_meta.write_offset >= RECORDS_PER_SECTOR) {
        uint32_t next_sector = (g_meta.write_sector - DATA_START_SECTOR + 1) % DATA_SECTOR_COUNT 
                               + DATA_START_SECTOR;
        
        // 擦除（Flash擦除是阻塞的，约30-50ms）
        w25qxx_erase_sector(next_sector * FLASH_SECTOR_SIZE);
        
        g_meta.write_sector = next_sector;
        g_meta.write_offset = 0;
        addr = next_sector * FLASH_SECTOR_SIZE;
    }
    
    // 批量写入
    uint8_t write_buf[RECORD_SIZE * CACHE_THRESHOLD];
    for (int i = 0; i < g_cache_count; i++) {
        memcpy(write_buf + i * RECORD_SIZE, &g_ram_cache[i], RECORD_SIZE);
    }
    
    w25qxx_write_data(addr, write_buf, g_cache_count * RECORD_SIZE);
    
    // 更新元数据
    g_meta.write_offset += g_cache_count;
    g_meta.total_records++;
    g_meta.newest_ts = g_ram_cache[g_cache_count - 1].collect_ts;
    if (g_meta.total_records <= DATA_SECTOR_COUNT * RECORDS_PER_SECTOR) {
        g_meta.oldest_ts = g_ram_cache[0].collect_ts;
    }
    
    // 持久化元数据到Flash最后一个扇区
    persist_meta();
    
    g_cache_count = 0;
    g_last_flush_ms = sys_now();
    
    return 0;
}

static void persist_meta(void) {
    uint8_t buf[sizeof(ring_buffer_meta_t)];
    memcpy(buf, &g_meta, sizeof(g_meta));
    uint16_t crc = crc16_modbus(buf, sizeof(g_meta) - 2);
    *(uint16_t *)(buf + sizeof(g_meta) - 2) = crc;
    
    w25qxx_erase_sector((DATA_START_SECTOR + DATA_SECTOR_COUNT) * FLASH_SECTOR_SIZE);
    w25qxx_write_data((DATA_START_SECTOR + DATA_SECTOR_COUNT) * FLASH_SECTOR_SIZE,
                      buf, sizeof(buf));
}

┌──────────────────────────────────────────────────────────┐
│  REPLAY_REQ (客户端→服务端)                              │
│  ┌──────┬────────┬──────────┬───────────────────────┐   │
│  │header│dev_addr│start_ts  │ end_ts (0=最新)        │   │
│  │0xAA01│1B      │4B        │ 4B                    │   │
│  └──────┴────────┴──────────┴───────────────────────┘   │
│  总计: 11字节                                             │
├──────────────────────────────────────────────────────────┤
│  REPLAY_GRANT (服务端→客户端)                            │
│  ┌──────┬────────┬──────────┬───────────────────────┐   │
│  │header│dev_addr│total_cnt │ max_seq               │   │
│  │0xAA02│1B      │4B        │ 4B (分片用)           │   │
│  └──────┴────────┴──────────┴───────────────────────┘   │
│  总计: 11字节                                             │
├──────────────────────────────────────────────────────────┤
│  REPLAY_DATA (客户端→服务端, 可多帧)                     │
│  ┌──────┬────────┬──────────┬────────┬──────────────┐   │
│  │header│dev_addr│seq_no    │count   │ records...    │   │
│  │0xAA03│1B      │4B        │1B      │ count×12B    │   │
│  └──────┴────────┴──────────┴────────┴──────────────┘   │
│  总计: 8 + count×12 字节                                  │
├──────────────────────────────────────────────────────────┤
│  REPLAY_ACK (服务端→客户端)                              │
│  ┌──────┬────────┬──────────┬───────────────────────┐   │
│  │header│dev_addr│seq_no    │ status (0=OK 1=ERR)   │   │
│  │0xAA04│1B      │4B        │ 1B                    │   │
│  └──────┴────────┴──────────┴───────────────────────┘   │
│  总计: 8字节                                              │
└──────────────────────────────────────────────────────────┘

// sync_client.c
typedef enum {
    SYNC_IDLE = 0,
    SYNC_CONNECTING,
    SYNC_WAIT_GRANT,
    SYNC_SENDING,
    SYNC_WAIT_ACK,
    SYNC_COMPLETE,
    SYNC_FAILED,
} sync_state_t;

static sync_state_t g_sync_state = SYNC_IDLE;
static uint32_t g_last_sync_ts = 0;      // 上次同步到的最后时间戳
static uint32_t g_last_connect_attempt = 0;
static uint8_t g_retry_count = 0;

#define SYNC_RETRY_INTERVAL_MS  30000     // 30秒重试
#define SYNC_TIMEOUT_MS         10000     // 10秒超时
#define SYNC_MAX_RETRIES        5

void sync_task(void) {
    uint32_t now = sys_now();
    
    switch (g_sync_state) {
        case SYNC_IDLE:
            // 定期检查网络是否可达
            if (network_is_up() && (now - g_last_connect_attempt) > SYNC_RETRY_INTERVAL_MS) {
                g_sync_state = SYNC_CONNECTING;
                g_last_connect_attempt = now;
                tcp_connect_sync_server();
            }
            break;
            
        case SYNC_CONNECTING:
            // TCP连接中，等待回调
            if (now - g_last_connect_attempt > SYNC_TIMEOUT_MS) {
                g_sync_state = SYNC_FAILED;
            }
            break;
            
        case SYNC_WAIT_GRANT:
            // 已发送REPLAY_REQ，等待GRANT
            if (now - g_grant_sent_time > SYNC_TIMEOUT_MS) {
                g_retry_count++;
                if (g_retry_count >= SYNC_MAX_RETRIES) {
                    g_sync_state = SYNC_FAILED;
                } else {
                    resend_replay_req();
                }
            }
            break;
            
        case SYNC_SENDING:
            // 发送DATA帧
            send_next_data_frame();
            g_sync_state = SYNC_WAIT_ACK;
            break;
            
        case SYNC_WAIT_ACK:
            if (now - g_data_sent_time > SYNC_TIMEOUT_MS) {
                // 重传当前帧
                resend_current_frame();
            }
            break;
            
        case SYNC_COMPLETE:
            g_retry_count = 0;
            g_last_sync_ts = g_current_sync_ts;
            save_last_sync_ts(g_last_sync_ts);  // 持久化到Flash
            tcp_close_sync_connection();
            g_sync_state = SYNC_IDLE;
            break;
            
        case SYNC_FAILED:
            g_retry_count = 0;
            tcp_close_sync_connection();
            g_sync_state = SYNC_IDLE;
            break;
    }
}

// TCP连接成功回调
void on_sync_connected(void) {
    g_sync_state = SYNC_WAIT_GRANT;
    send_replay_req();
    g_grant_sent_time = sys_now();
}

// 收到REPLAY_GRANT回调
void on_replay_grant(uint32_t total_count, uint32_t max_seq) {
    g_total_to_send = total_count;
    g_total_seqs = max_seq;
    g_current_seq = 0;
    g_sync_state = SYNC_SENDING;
}

// 收到REPLAY_ACK回调
void on_replay_ack(uint32_t seq_no, uint8_t status) {
    if (status == 0 && seq_no == g_current_seq) {
        g_current_seq++;
        if (g_current_seq >= g_total_seqs) {
            g_sync_state = SYNC_COMPLETE;
        } else {
            g_sync_state = SYNC_SENDING;
        }
    } else {
        // 重传
        g_sync_state = SYNC_SENDING;
    }
}

#define RECORDS_PER_FRAME  32    // 每帧最多32条 = 384字节 + 8字节头 = 392字节

static void send_next_data_frame(void) {
    if (!g_sync_pcb || !g_sync_pcb->state == ESTABLISHED) return;
    
    uint8_t tx_buf[8 + RECORDS_PER_FRAME * RECORD_SIZE];
    
    // 帧头
    tx_buf[0] = 0xAA; tx_buf[1] = 0x03;                    // header
    tx_buf[2] = g_dev_addr;                                  // dev_addr
    tx_buf[3] = (g_current_seq >> 24) & 0xFF;               // seq MSB
    tx_buf[4] = (g_current_seq >> 16) & 0xFF;
    tx_buf[5] = (g_current_seq >> 8) & 0xFF;
    tx_buf[6] = g_current_seq & 0xFF;                        // seq LSB
    
    // 从Flash读取数据
    uint32_t records_in_frame = RECORDS_PER_FRAME;
    env_record_t records[RECORDS_PER_FRAME];
    uint32_t actual = read_records_for_sync(g_current_seq * RECORDS_PER_FRAME,
                                            records, records_in_frame);
    
    if (actual == 0) {
        g_sync_state = SYNC_COMPLETE;
        return;
    }
    
    tx_buf[7] = (uint8_t)actual;                             // count
    
    // 填充记录
    for (int i = 0; i < actual; i++) {
        memcpy(tx_buf + 8 + i * RECORD_SIZE, &records[i], RECORD_SIZE);
    }
    
    // TCP发送
    struct pbuf *p = pbuf_alloc(PBUF_TRANSPORT, 8 + actual * RECORD_SIZE, PBUF_RAM);
    if (p) {
        memcpy(p->payload, tx_buf, 8 + actual * RECORD_SIZE);
        tcp_write(g_sync_pcb, p->payload, p->len, TCP_WRITE_FLAG_COPY);
        tcp_output(g_sync_pcb);
        pbuf_free(p);
    }
    
    g_data_sent_time = sys_now();
}

static uint32_t read_records_for_sync(uint32_t index, env_record_t *out, uint32_t max_count) {
    // 从Flash按时间顺序读取未同步的记录
    // 使用g_last_sync_ts作为起点
    return storage_read_range(g_last_sync_ts + 1, 0xFFFFFFFF,
                              out, max_count, &actual);
}

#!/usr/bin/env python3
"""
TCP同步接收服务 - 元器件仓储温湿度记录仪
"""
import asyncio
import struct
import socket
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional
import asyncpg
from influxdb_client import InfluxDBClient, Point, WritePrecision


@dataclass
class SyncSession:
    dev_addr: int
    total_count: int = 0
    max_seq: int = 0
    received_seqs: set = field(default_factory=set)
    records: List[dict] = field(default_factory=list)
    state: str = "WAIT_REQ"


class TCPSyncServer:
    """TCP同步服务器"""
    
    HEADER_REPLAY_REQ = 0xAA01
    HEADER_REPLAY_GRANT = 0xAA02
    HEADER_REPLAY_DATA = 0xAA03
    HEADER_REPLAY_ACK = 0xAA04
    
    def __init__(self, host: str, port: int, influx_url: str, influx_token: str, 
                 influx_org: str, influx_bucket: str, pg_dsn: str):
        self.host = host
        self.port = port
        self.sessions: Dict[tuple, SyncSession] = {}
        
        # InfluxDB
        self.influx_client = InfluxDBClient(url=influx_url, token=influx_token, org=influx_org)
        self.influx_write = self.influx_client.write_api()
        self.influx_bucket = influx_bucket
        self.influx_org = influx_org
        
        # PostgreSQL（用于去重索引）
        self.pg_dsn = pg_dsn
        self.pg_pool = None
    
    async def start(self):
        self.pg_pool = await asyncpg.create_pool(self.pg_dsn)
        # 初始化去重表
        async with self.pg_pool.acquire() as conn:
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS env_record_dedup (
                    dev_addr    SMALLINT,
                    collect_ts  BIGINT,
                    sync_time   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (dev_addr, collect_ts)
                );
                CREATE INDEX IF NOT EXISTS idx_dedup_dev_ts 
                ON env_record_dedup(dev_addr, collect_ts);
            """)
        
        server = await asyncio.start_server(self.handle_client, self.host, self.port)
        print(f"TCP同步服务器监听 {self.host}:{self.port}")
        async with server:
            await server.serve_forever()
    
    async def handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        addr = writer.get_extra_info('peername')
        print(f"客户端连接: {addr}")
        
        session = SyncSession(dev_addr=0)
        self.sessions[addr] = session
        
        try:
            while True:
                data = await reader.read(4096)
                if not data:
                    break
                
                await self.process_frame(data, session, writer)
                
        except asyncio.IncompleteReadError:
            pass
        except Exception as e:
            print(f"客户端{addr} 异常: {e}")
        finally:
            writer.close()
            await writer.wait_closed()
            self.sessions.pop(addr, None)
            print(f"客户端断开: {addr}")
    
    async def process_frame(self, data: bytes, session: SyncSession, writer: asyncio.StreamWriter):
        if len(data) < 3:
            return
        
        header = struct.unpack(">H", data[:2])[0]
        dev_addr = data[2]
        session.dev_addr = dev_addr
        
        if header == self.HEADER_REPLAY_REQ:
            # 解析请求
            start_ts, end_ts = struct.unpack(">II", data[3:11])
            
            # 查询待同步记录数
            async with self.pg_pool.acquire() as conn:
                # 获取未同步的记录数（简化：从InfluxDB查最后同步时间）
                # 实际工程用PostgreSQL记录同步状态
                total_count = await conn.fetchval("""
                    SELECT COUNT(*) FROM env_records 
                    WHERE dev_addr = $1 AND collect_ts > $2
                """, dev_addr, start_ts)
                
                max_seq = (total_count + 31) // 32  # 向上取整
            
            # 发送GRANT
            grant_frame = struct.pack(">HBII", self.HEADER_REPLAY_GRANT, 
                                       dev_addr, total_count, max_seq)
            writer.write(grant_frame)
            await writer.drain()
            
            session.total_count = total_count
            session.max_seq = max_seq
            session.state = "WAIT_DATA"
            
        elif header == self.HEADER_REPLAY_DATA:
            seq_no = struct.unpack(">I", data[3:7])[0]
            count = data[7]
            
            if len(data) < 8 + count * 12:
                return
            
            # 解析记录
            records = []
            for i in range(count):
                offset = 8 + i * 12
                ts, temp_raw, hum_raw, volt_raw, status = \
                    struct.unpack(">IhHHBB", data[offset:offset + 12])
                
                records.append({
                    'dev_addr': dev_addr,
                    'collect_ts': ts,
                    'temperature': temp_raw / 100.0,
                    'humidity': hum_raw / 10.0,
                    'voltage': volt_raw / 100.0,
                    'status': status,
                })
            
            # 去重写入
            await self.write_with_dedup(records)
            
            # 发送ACK
            ack_frame = struct.pack(">HBIB", self.HEADER_REPLAY_ACK,
                                     dev_addr, seq_no, 0)
            writer.write(ack_frame)
            await writer.drain()
            
            session.received_seqs.add(seq_no)
            
            if len(session.received_seqs) >= session.max_seq:
                session.state = "COMPLETE"
    
    async def write_with_dedup(self, records: List[dict]):
        """去重写入：先写PostgreSQL去重索引，再写InfluxDB"""
        if not records:
            return
        
        async with self.pg_pool.acquire() as conn:
            async with conn.transaction():
                for r in records:
                    # 尝试插入去重表，冲突则跳过
                    inserted = await conn.execute("""
                        INSERT INTO env_record_dedup (dev_addr, collect_ts)
                        VALUES ($1, $2)
                        ON CONFLICT (dev_addr, collect_ts) DO NOTHING
                    """, r['dev_addr'], r['collect_ts'])
                    
                    if inserted == 'INSERT 0 1':
                        # 新记录，写入InfluxDB
                        point = Point("env_sensor") \
                            .tag("device", f"sensor_{r['dev_addr']:03d}") \
                            .tag("source", "sync_replay") \
                            .field("temperature", r['temperature']) \
                            .field("humidity", r['humidity']) \
                            .field("voltage", r['voltage']) \
                            .field("status", r['status']) \
                            .time(datetime.fromtimestamp(r['collect_ts'], tz=timezone.utc),
                                  WritePrecision.S)
                        
                        self.influx_write.write(bucket=self.influx_bucket,
                                                org=self.influx_org,
                                                record=point)


async def main():
    server = TCPSyncServer(
        host="0.0.0.0",
        port=502,
        influx_url="http://localhost:8086",
        influx_token="your-token",
        influx_org="your-org",
        influx_bucket="env_monitor",
        pg_dsn="postgresql://user:pass@localhost/env_monitor",
    )
    await server.start()


if __name__ == "__main__":
    asyncio.run(main())

// NTP客户端（LWIP SNTP）
#include "lwip/apps/sntp.h"

static uint32_t g_ntp_offset = 0;     // RTC与NTP的偏移量
static uint32_t g_last_ntp_sync = 0;

#define NTP_SYNC_INTERVAL_MS  3600000  // 1小时同步一次

void ntp_task(void) {
    uint32_t now = sys_now();
    
    if (now - g_last_ntp_sync > NTP_SYNC_INTERVAL_MS) {
        sntp_setservername(0, "pool.ntp.org");
        sntp_init();
        
        // 等待NTP响应
        for (int i = 0; i < 10; i++) {
            uint32_t ntp_time = sntp_get_current_timestamp();
            if (ntp_time > 0) {
                uint32_t rtc_time = get_rtc_timestamp();
                g_ntp_offset = ntp_time - rtc_time;
                g_last_ntp_sync = now;
                
                // 校正RTC
                set_rtc_offset(g_ntp_offset);
                break;
            }
            sys_msleep(1000);
        }
        
        sntp_stop();
    }
}

// 获取校正后的时间戳
uint32_t get_corrected_timestamp(void) {
    return get_rtc_timestamp() + g_ntp_offset;
}

// link_monitor.c
static uint8_t g_link_up = 0;
static uint32_t g_last_server_contact = 0;
static uint8_t g_offline_mode = 0;

#define OFFLINE_THRESHOLD_MS  60000   // 60秒无服务端响应进入离线模式

void link_monitor_task(void) {
    uint32_t now = sys_now();
    
    // 检测物理链路
    if (ethernet_link_up()) {
        g_link_up = 1;
    } else {
        g_link_up = 0;
        g_offline_mode = 1;
        return;
    }
    
    // 检测TCP连接状态
    if (g_sync_state == SYNC_COMPLETE || g_sync_state == SYNC_IDLE) {
        g_last_server_contact = now;
        g_offline_mode = 0;
    } else if (g_sync_state == SYNC_FAILED) {
        if (now - g_last_server_contact > OFFLINE_THRESHOLD_MS) {
            g_offline_mode = 1;
        }
    }
    
    // 离线模式下，采集任务仍然写入Flash
    if (g_offline_mode) {
        // 提高采集频率？不，保持原频率，让Flash攒更多数据
        // 但可以降低UDP Trap发送频率，省电
        set_trap_interval(30000);  // 30秒
    } else {
        set_trap_interval(5000);   // 5秒
    }
}