从 RJ45 到 MQTT Broker：以太网温湿度传感器数据透传至云端的协议转换网关

┌─────────────────────────────────────────────────────────────┐
│                    协议转换网关                               │
│                                                             │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐ │
│  │  Modbus TCP  │    │  数据解析与   │    │   MQTT 客户端 │ │
│  │  采集引擎     │───▶│  点表映射     │───▶│  发布/订阅    │ │
│  │              │    │              │    │              │ │
│  │  · 轮询调度  │    │  · 寄存器解码 │    │  · TLS 加密   │ │
│  │  · 超时重试  │    │  · 字节序转换 │    │  · 自动重连   │ │
│  │  · 并发控制  │    │  · 质量位标记 │    │  · QoS 1/2   │ │
│  │  · 断线检测  │    │  · 变化过滤   │    │  · 遗嘱消息   │ │
│  └──────────────┘    └──────────────┘    └──────────────┘ │
│         │                    │                    │         │
│  ┌──────▼──────┐   ┌────────▼────────┐  ┌──────▼──────┐  │
│  │  本地缓存    │   │  连接管理器      │  │  配置热加载  │  │
│  │  · SQLite   │   │  · TCP Keepalive│  │  · 点表更新  │  │
│  │  · WAL      │   │  · MQTT 会话    │  │  · 参数调整  │  │
│  │  · 补传队列 │   │  · 指数退避     │  │  · 不重启    │  │
│  └─────────────┘   └─────────────────┘  └─────────────┘  │
└─────────────────────────────────────────────────────────────┘

import asyncio
import struct
from pymodbus.client import AsyncModbusTcpClient

class ModbusPoller:
    def __init__(self, devices, poll_interval=5):
        self.devices = devices  # [{ip, port, slave_id, registers: [{addr, count, type}]}]
        self.poll_interval = poll_interval
        self.clients = {}

    async def get_client(self, ip, port):
        key = f"{ip}:{port}"
        if key not in self.clients or not self.clients[key].connected:
            client = AsyncModbusTcpClient(ip, port=port)
            await client.connect()
            self.clients[key] = client
        return self.clients[key]

    async def poll_device(self, device):
        try:
            client = await self.get_client(device['ip'], device['port'])
            results = {}

            for reg_block in device['registers']:
                # 批量读取连续寄存器（一次请求读多个）
                resp = await client.read_holding_registers(
                    address=reg_block['addr'],
                    count=reg_block['count'],
                    slave=device['slave_id']
                )
                if resp.isError():
                    results[reg_block['name']] = {'value': None, 'quality': 2}  # bad
                else:
                    # 字节序转换
                    values = self.decode_registers(resp.registers, reg_block['type'])
                    results[reg_block['name']] = {'value': values, 'quality': 0}  # good

            return {
                'device_id': device['id'],
                'timestamp': asyncio.get_event_loop().time(),
                'data': results
            }
        except Exception as e:
            return {
                'device_id': device['id'],
                'timestamp': asyncio.get_event_loop().time(),
                'error': str(e),
                'quality': 2
            }

    async def poll_all(self):
        """并发轮询所有设备"""
        tasks = [self.poll_device(d) for d in self.devices]
        return await asyncio.gather(*tasks, return_exceptions=True)

    def decode_registers(self, registers, data_type):
        """根据数据类型解码寄存器"""
        if data_type == 'INT16':
            return registers[0]
        elif data_type == 'FLOAT32_BE':
            # 大端浮点：两个 16 位寄存器拼成 32 位浮点
            raw = (registers[0] << 16) | registers[1]
            return struct.unpack('>f', struct.pack('>I', raw))[0]
        elif data_type == 'FLOAT32_LE':
            raw = (registers[1] << 16) | registers[0]
            return struct.unpack('<f', struct.pack('<I', raw))[0]

class ChangeFilter:
    def __init__(self, deadband_temp=0.2, deadband_humid=0.5):
        self.last_values = {}
        self.deadband_temp = deadband_temp
        self.deadband_humid = deadband_humid

    def should_report(self, device_id, temp, humid):
        key = device_id
        if key not in self.last_values:
            self.last_values[key] = (temp, humid)
            return True

        last_temp, last_humid = self.last_values[key]
        temp_changed = abs(temp - last_temp) >= self.deadband_temp
        humid_changed = abs(humid - last_humid) >= self.deadband_humid

        if temp_changed or humid_changed:
            self.last_values[key] = (temp, humid)
            return True
        return False

{
  "device_type": "ethernet_th_sensor_v2",
  "poll_interval": 5,
  "registers": [
    {
      "name": "temperature",
      "address": 40001,
      "count": 1,
      "type": "INT16",
      "scale": 0.1,
      "unit": "°C",
      "target_field": "temp"
    },
    {
      "name": "humidity",
      "address": 40002,
      "count": 1,
      "type": "INT16",
      "scale": 0.1,
      "unit": "%RH",
      "target_field": "rh"
    },
    {
      "name": "pm25",
      "address": 40003,
      "count": 1,
      "type": "UINT16",
      "scale": 1,
      "unit": "μg/m³",
      "target_field": "pm25"
    },
    {
      "name": "uptime_hours",
      "address": 40004,
      "count": 2,
      "type": "UINT32_BE",
      "scale": 1,
      "unit": "h",
      "target_field": "uptime"
    },
    {
      "name": "status_word",
      "address": 40006,
      "count": 1,
      "type": "UINT16",
      "bitmap": {
        "0": "sensor_error",
        "1": "eeprom_error",
        "2": "communication_error",
        "3": "calibration_needed"
      },
      "target_field": "fault"
    }
  ]
}

{
  "message_id": "msg-a3-01-20260921-143052-001",
  "device_id": "TH-A3-01",
  "site_id": "site-tianjin-001",
  "room_id": "archive-3f-a3",
  "sample_time": "2026-09-21T14:30:52+08:00",
  "report_time": "2026-09-21T14:30:52.156+08:00",
  "schema_version": "1.0",
  "metrics": {
    "temp": 22.4,
    "rh": 55.8,
    "pm25": 12,
    "uptime": 8760
  },
  "fault": {
    "sensor_error": false,
    "eeprom_error": false,
    "communication_error": false,
    "calibration_needed": true
  },
  "quality": "good"
}

import paho.mqtt.client as mqtt
import ssl

def create_mqtt_client(config):
    client = mqtt.Client(
        client_id=config['client_id'],
        clean_session=False,       # 持久会话，断线重连后不丢消息
        protocol=mqtt.MQTTv311
    )

    # 认证
    if config.get('username'):
        client.username_pw_set(config['username'], config['password'])
    elif config.get('certfile'):
        client.tls_set(
            ca_certs=config['ca_certs'],
            certfile=config['certfile'],
            keyfile=config['keyfile'],
            cert_reqs=ssl.CERT_REQUIRED,
            tls_version=ssl.PROTOCOL_TLSv1_2
        )

    # 遗嘱消息（LWT）—— 网关异常断开时，云端知道
    client.will_set(
        topic=f"gateway/{config['gateway_id']}/status",
        payload='{"status":"offline","reason":"unexpected_disconnect"}',
        qos=1,
        retain=True
    )

    # 连接参数
    client.connect_async(
        host=config['broker_host'],
        port=config.get('port', 8883),
        keepalive=60
    )

    return client

import sqlite3
import json

class LocalCache:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS message_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                device_id TEXT,
                topic TEXT,
                payload TEXT,
                sample_time TEXT,
                created_at REAL,
                sent INTEGER DEFAULT 0,
                retry_count INTEGER DEFAULT 0
            )
        ''')
        self.conn.commit()

    def store(self, device_id, topic, payload, sample_time):
        self.conn.execute(
            'INSERT INTO message_queue (device_id, topic, payload, sample_time, created_at) VALUES (?, ?, ?, ?, ?)',
            (device_id, topic, json.dumps(payload), sample_time, time.time())
        )
        self.conn.commit()

    def get_unsent(self, limit=100):
        cursor = self.conn.execute(
            'SELECT id, topic, payload FROM message_queue WHERE sent = 0 ORDER BY sample_time ASC LIMIT ?',
            (limit,)
        )
        return [(row[0], row[1], json.loads(row[2])) for row in cursor.fetchall()]

    def mark_sent(self, msg_id):
        self.conn.execute('UPDATE message_queue SET sent = 1 WHERE id = ?', (msg_id,))
        self.conn.commit()

    def cleanup_sent(self, older_than_days=7):
        """清理 7 天前已发送的消息"""
        cutoff = time.time() - (older_than_days * 86400)
        self.conn.execute('DELETE FROM message_queue WHERE sent = 1 AND created_at < ?', (cutoff,))
        self.conn.commit()

class BackfillManager:
    def __init__(self, cache, mqtt_client):
        self.cache = cache
        self.mqtt_client = mqtt_client
        self.backfill_batch_size = 50

    def try_backfill(self):
        """尝试补传缓存的消息"""
        if not self.mqtt_client.is_connected():
            return  # 还没连上，不补传

        unsent = self.cache.get_unsent(self.backfill_batch_size)
        if not unsent:
            return

        for msg_id, topic, payload in unsent:
            try:
                result = self.mqtt_client.publish(
                    topic,
                    json.dumps(payload),
                    qos=1,
                    retain=False
                )
                if result.rc == mqtt.MQTT_ERR_SUCCESS:
                    self.cache.mark_sent(msg_id)
            except Exception:
                break  # 发送失败，停止补传，等下次重试

{
  "message_id": "msg-a3-01-20260921-143052-001",
  "sample_time": "2026-09-21T14:30:52+08:00",
  "report_time": "2026-09-21T18:45:12+08:00",
  "is_backfill": true,
  "metrics": { "temp": 22.4, "rh": 55.8 }
}

# systemd 服务，自动重启
[Unit]
Description=Env Gateway Protocol Converter
After=network.target

[Service]
Type=simple
User=envgw
WorkingDirectory=/opt/env-gateway
ExecStart=/opt/env-gateway/venv/bin/python main.py --config /etc/env-gateway/config.yaml
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

主网关（192.168.10.100）──┐
                           ├── 虚拟 IP（192.168.10.10，VRRP/Keepalived）
备网关（192.168.10.101）──┘

import signal

class ConfigManager:
    def __init__(self, config_path):
        self.config_path = config_path
        self.config = self.load()
        signal.signal(signal.SIGHUP, self.reload)  # kill -HUP pid

    def reload(self, signum, frame):
        logger.info("Reloading configuration...")
        new_config = self.load()
        if self.validate(new_config):
            self.config = new_config
            self.apply_changes()
            logger.info("Configuration reloaded successfully")
        else:
            logger.error("Invalid configuration, keeping old one")

    def load(self):
        with open(self.config_path) as f:
            return yaml.safe_load(f)