以太网温湿度传感器可靠性设计：断线重连、CRC校验与数据丢包优化

# reconnect.py
import socket
import time
from enum import Enum

class ConnState(Enum):
    IDLE = 0
    CONNECTING = 1
    ESTABLISHED = 2
    RECONNECTING = 3
    FAILED = 4

class SmartReconnect:
    def __init__(self, ip, port=502, timeout=2.0):
        self.ip = ip
        self.port = port
        self.timeout = timeout
        self.state = ConnState.IDLE
        self._sock = None
        self._fail_count = 0
        self._last_connect_ts = 0
        self._backoff = [1, 2, 4, 8, 15, 30, 60]   # 指数退避，上限 60s

    def _backoff_delay(self):
        idx = min(self._fail_count, len(self._backoff) - 1)
        return self._backoff[idx]

    def connect(self, force=False):
        now = time.time()
        if not force and self.state == ConnState.RECONNECTING:
            elapsed = now - self._last_connect_ts
            if elapsed < self._backoff_delay():
                return False          # 退避期内不重试

        self.state = ConnState.CONNECTING
        self._last_connect_ts = now

        try:
            if self._sock:
                self._safe_close()
            self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._sock.settimeout(self.timeout)
            self._sock.connect((self.ip, self.port))

            # TCP Keepalive 内核级配置
            self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            # Linux: 空闲 10s → 探测 5s → 重试 3 次
            self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)
            self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 5)
            self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)

            self.state = ConnState.ESTABLISHED
            self._fail_count = 0
            return True

        except (socket.timeout, ConnectionRefusedError, OSError) as e:
            self._fail_count += 1
            self.state = ConnState.RECONNECTING
            return False

    def _safe_close(self):
        try:
            self._sock.shutdown(socket.SHUT_RDWR)
        except:
            pass
        try:
            self._sock.close()
        except:
            pass
        self._sock = None

    def is_alive(self):
        """应用层心跳：发一个轻量 MBAP 头探测"""
        if not self._sock:
            return False
        try:
            # 构造最小 Modbus TCP 读请求（读 1 个寄存器）
            mbap = b'\x00\x01\x00\x00\x00\x06\x01\x03\x00\x00\x00\x01'
            self._sock.send(mbap)
            data = self._sock.recv(12)
            return len(data) >= 9
        except:
            self.state = ConnState.RECONNECTING
            return False

# app_crc.py
import struct
import zlib

class AppChecksum:
    """应用层数据完整性校验"""

    @staticmethod
    def checksum_registers(registers):
        """对寄存器列表计算 CRC32，用于跨网络传输校验"""
        raw = struct.pack(f'>{len(registers)}H', *registers)
        return zlib.crc32(raw) & 0xFFFFFFFF

    @staticmethod
    def verify_response(expected_crc, registers):
        actual = AppChecksum.checksum_registers(registers)
        return expected_crc == actual

# 传感器侧（伪代码，固件层面）：
# 在 Modbus TCP 响应中，利用保留寄存器字段携带 CRC32
# 例如：读 4 个寄存器（温度/湿度/状态/露点），额外读第 5 个寄存器 = CRC32 低 16 位

def validate_sample(temp, hum, dew, raw_registers):
    """多维度交叉验证"""
    issues = []

    # 1. 范围校验
    if not (-40 <= temp <= 85):
        issues.append(f"TEMP_OUT_OF_RANGE:{temp}")
    if not (0 <= hum <= 100):
        issues.append(f"HUM_OUT_OF_RANGE:{hum}")

    # 2. 物理一致性：露点 ≤ 温度
    if dew and dew > temp + 0.5:
        issues.append(f"DEW_EXCEEDS_TEMP:dew={dew},temp={temp}")

    # 3. 突变检测：与前值比较
    if hasattr(validate_sample, '_last_temp'):
        dt = abs(temp - validate_sample._last_temp)
        if dt > 5.0:          # 1 次采样变化 >5℃ → 可疑
            issues.append(f"TEMP_SUDDEN_JUMP:{dt}℃")

    validate_sample._last_temp = temp

    # 4. 寄存器位模式校验
    for i, reg in enumerate(raw_registers):
        if reg == 0xFFFF:
            issues.append(f"REG_{i}_IS_0xFFFF")

    return len(issues) == 0, issues

# ring_cache.py
import sqlite3
import time
import json
from collections import deque

class RingCache:
    """内存 + 磁盘双层缓存，防止进程崩溃丢数据"""

    def __init__(self, max_memory=500, db_path="sensor_cache.db"):
        self.max_memory = max_memory
        self.memory = deque(maxlen=max_memory)
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as c:
            c.execute("""CREATE TABLE IF NOT EXISTS cache(
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                node_id TEXT, collect_ts REAL, recv_ts REAL,
                payload TEXT, uploaded INTEGER DEFAULT 0)""")
            c.execute("CREATE INDEX IF NOT EXISTS idx_uploaded ON cache(uploaded, collect_ts)")
            c.execute("PRAGMA journal_mode=WAL")

    def push(self, node_id, temp, hum, quality, raw):
        """采集到数据立即入缓存"""
        now = time.time()
        item = {
            "node_id": node_id,
            "collect_ts": now,        # 采集时刻（关键！）
            "recv_ts": now,           # 接收时刻
            "temp": temp,
            "hum": hum,
            "quality": quality,
            "raw": raw
        }
        # 内存层
        self.memory.append(item)
        # 磁盘层（异步写入）
        with sqlite3.connect(self.db_path) as c:
            c.execute("INSERT INTO cache(node_id, collect_ts, recv_ts, payload, uploaded) VALUES(?,?,?,?,0)",
                      (node_id, now, now, json.dumps(item)))

    def pop_pending(self, limit=100):
        """取出未上传的数据，按采集时间排序"""
        with sqlite3.connect(self.db_path) as c:
            rows = c.execute(
                "SELECT id, payload FROM cache WHERE uploaded=0 ORDER BY collect_ts LIMIT ?",
                (limit,)).fetchall()
            return rows

    def mark_uploaded(self, ids):
        """标记已成功上传"""
        with sqlite3.connect(self.db_path) as c:
            for rid in ids:
                c.execute("UPDATE cache SET uploaded=1 WHERE id=?", (rid,))

    def cleanup(self, retain_hours=168):
        """清理已上传的旧数据（保留 7 天）"""
        cutoff = time.time() - retain_hours * 3600
        with sqlite3.connect(self.db_path) as c:
            c.execute("DELETE FROM cache WHERE uploaded=1 AND collect_ts<?", (cutoff,))

# reorder.py
class SequenceTracker:
    """处理网络乱序和重复数据"""

    def __init__(self, window_size=100):
        self.window = {}           # node_id -> {seq_no: sample}
        self.acked = {}            # node_id -> last_seq
        self.window_size = window_size

    def on_receive(self, node_id, seq_no, sample):
        if node_id not in self.acked:
            self.acked[node_id] = seq_no - 1

        last = self.acked[node_id]

        if seq_no <= last:
            return None            # 重复包，丢弃

        if seq_no > last + self.window_size:
            # 跳号太多，可能丢包，触发重传请求
            return {"action": "request_retransmit", "from": last + 1, "to": seq_no - 1}

        self.window.setdefault(node_id, {})[seq_no] = sample

        # 尝试交付连续序列
        deliverable = []
        while (last + 1) in self.window.get(node_id, {}):
            last += 1
            deliverable.append(self.window[node_id].pop(last))

        self.acked[node_id] = last
        return {"action": "deliver", "samples": deliverable}

# 交换机端口配置示例（华为/华三思路）
interface GigabitEthernet0/0/1
 description Sensor-Port
 port link-type access
 port default vlan 100
 storm-control broadcast min-cir 1000     # 广播风暴抑制
 storm-control multicast min-cir 2000     # 组播抑制
 loopback-detect enable                    # 环路检测
 poe priority high                         # PoE 供电优先级
 error-down auto-recovery cause crc-error interval 300  # CRC 错自动恢复

# 定期检查端口 CRC 错误计数
for port in /sys/class/net/eth*/statistics/; do
    crc=$(cat $port/rx_crc_errors 2>/dev/null)
    if [ "$crc" -gt 0 ]; then
        echo "$(basename $port): CRC errors = $crc"
    fi
done

传感器 MCU → [值域校验] → 网络传输 → [TCP校验+应用CRC] → 采集进程
    → [突变检测] → 环形缓存 → [去重+排序] → 时序数据库
    → [质量位标记] → Grafana 展示