从 RJ45 到 MQTT Broker：以太网温湿度变送器数据透传网关的协议转换实现

[以太网温湿度传感器]──PoE交换机──[协议转换网关]──TLS/MQTT──> [云端 MQTT Broker] ──> [InfluxDB] ──> [Grafana]
   Modbus TCP:502       局域网            MQTT 1883/8883       腾讯云 CVM            时序数据库        可视化

库房局域网（192.168.10.x）
┌────────────────────────────────────────────────────────────┐
│                                                            │
│  [PoE 交换机]                                              │
│    ├── 端口1: 温湿度变送器 A (192.168.10.101, Unit ID 1)  │
│    ├── 端口2: 温湿度变送器 B (192.168.10.102, Unit ID 1)  │
│    ├── 端口3: 温湿度变送器 C (192.168.10.103, Unit ID 1)  │
│    └── 端口8: 协议转换网关 (192.168.10.200)               │
│                                                            │
└───────────────┬────────────────────────────────────────────┘
                │ 互联网 / 4G / 专线
                ▼
┌────────────────────────────────────────────────────────────┐
│  腾讯云 CVM (MQTT Broker - EMQX / Mosquitto)              │
│  Topic: warehouse/+/env/#                                 │
└───────────────┬────────────────────────────────────────────┘
                ▼
┌────────────────────────────────────────────────────────────┐
│  InfluxDB (时序数据库) → Grafana (可视化)                  │
└────────────────────────────────────────────────────────────┘

#!/usr/bin/env python3
"""
modbus_collector.py - 以太网温湿度变送器 Modbus TCP 采集
"""

import time
import logging
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)

# 传感器配置表
SENSORS = [
    {
        "name": "BJ-IDC01-RACK07-INLET",
        "ip": "192.168.10.101",
        "port": 502,
        "unit_id": 1,
        "timeout": 3,
    },
    {
        "name": "BJ-IDC01-RACK07-OUTLET",
        "ip": "192.168.10.102",
        "port": 502,
        "unit_id": 1,
        "timeout": 3,
    },
    {
        "name": "BJ-IDC01-COLD-AISLE",
        "ip": "192.168.10.103",
        "port": 502,
        "unit_id": 1,
        "timeout": 3,
    },
]


def read_sensor(sensor: dict) -> dict | None:
    """
    读取单个以太网温湿度变送器的保持寄存器
    寄存器映射（常见厂商约定）：
    - 40001 (addr=0): 温度 ×10, INT16
    - 40002 (addr=1): 湿度 ×10, INT16
    - 40003 (addr=2): 露点 ×10, INT16 (部分型号支持)
    """
    client = ModbusTcpClient(
        sensor["ip"],
        port=sensor["port"],
        timeout=sensor["timeout"]
    )

    try:
        if not client.connect():
            logging.error(f"[{sensor['name']}] 连接失败: {sensor['ip']}:502")
            return None

        # 读取保持寄存器 40001~40003 (3 个寄存器)
        resp = client.read_holding_registers(
            address=0,
            count=3,
            slave=sensor["unit_id"]
        )

        if resp.is_error():
            logging.error(f"[{sensor['name']}] Modbus 异常响应: {resp}")
            return None

        raw_temp = resp.registers[0]
        raw_humi = resp.registers[1]
        raw_dew = resp.registers[2] if len(resp.registers) > 2 else 0

        # 处理有符号整数（温度可能为负）
        if raw_temp > 32767:
            raw_temp -= 65536
        if raw_dew > 32767:
            raw_dew -= 65536

        data = {
            "sensor_name": sensor["name"],
            "ip": sensor["ip"],
            "temperature": round(raw_temp / 10.0, 1),
            "humidity": round(raw_humi / 10.0, 1),
            "dew_point": round(raw_dew / 10.0, 1),
            "timestamp": int(time.time()),
            "status": "ok",
        }

        logging.info(
            f"[{sensor['name']}] T={data['temperature']}℃ "
            f"H={data['humidity']}%RH DP={data['dew_point']}℃"
        )
        return data

    except ModbusException as e:
        logging.error(f"[{sensor['name']}] Modbus 异常: {e}")
        return None
    except Exception as e:
        logging.error(f"[{sensor['name']}] 未知异常: {e}")
        return None
    finally:
        client.close()


def collect_all() -> list[dict]:
    """轮询所有传感器"""
    results = []
    for sensor in SENSORS:
        data = read_sensor(sensor)
        if data:
            results.append(data)
        time.sleep(0.1)  # 避免同一总线/交换机端口突发拥塞
    return results

"""
mqtt_publisher.py - 将采集到的温湿度数据发布到 MQTT Broker
"""

import json
import ssl
import paho.mqtt.client as mqtt

MQTT_CONFIG = {
    "broker": "mqtt.your-tencent-cloud.com",
    "port": 8883,  # TLS 端口
    "username": "warehouse_user",
    "password": "secure_password_here",
    "client_id": "gateway_bj_idc01",
    "base_topic": "warehouse/bj_idc01/env",
    "qos": 1,  # at least once
    "keepalive": 60,
}

# TLS 证书路径
CA_CERT = "/etc/mqtt/ca.crt"
CLIENT_CERT = "/etc/mqtt/client.crt"
CLIENT_KEY = "/etc/mqtt/client.key"


def on_connect(client, userdata, flags, rc):
    if rc == 0:
        logging.info("MQTT Broker 连接成功")
    else:
        logging.error(f"MQTT 连接失败, rc={rc}")


def on_disconnect(client, userdata, rc):
    logging.warning(f"MQTT 断开连接, rc={rc}")


def on_publish(client, userdata, mid):
    logging.debug(f"消息发布成功, mid={mid}")


def create_mqtt_client() -> mqtt.Client:
    """创建并配置 MQTT 客户端"""
    client = mqtt.Client(
        client_id=MQTT_CONFIG["client_id"],
        clean_session=False
    )

    # 设置认证
    client.username_pw_set(
        MQTT_CONFIG["username"],
        MQTT_CONFIG["password"]
    )

    # TLS 配置
    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    context.load_verify_locations(CA_CERT)
    context.load_cert_chain(CLIENT_CERT, CLIENT_KEY)
    client.tls_set_context(context)

    # 回调
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_publish = on_publish

    return client


def publish_data(client: mqtt.Client, data: dict):
    """发布单条传感器数据"""
    topic = f"{MQTT_CONFIG['base_topic']}/{data['sensor_name']}"

    payload = json.dumps({
        "temperature": data["temperature"],
        "humidity": data["humidity"],
        "dew_point": data["dew_point"],
        "timestamp": data["timestamp"],
        "ip": data["ip"],
    }, ensure_ascii=False)

    result = client.publish(topic, payload, qos=MQTT_CONFIG["qos"], retain=True)

    if result.rc != mqtt.MQTT_ERR_SUCCESS:
        logging.error(f"发布失败: {topic}, rc={result.rc}")
    else:
        logging.debug(f"发布成功: {topic}")


def publish_batch(client: mqtt.Client, batch: list[dict]):
    """批量发布"""
    for data in batch:
        publish_data(client, data)

"""
local_cache.py - 断网时本地缓存数据，恢复后补传
"""

import sqlite3
import json
import os

CACHE_DB = "/var/lib/gateway/sensor_cache.db"


def init_cache():
    """初始化本地缓存数据库"""
    os.makedirs(os.path.dirname(CACHE_DB), exist_ok=True)
    conn = sqlite3.connect(CACHE_DB)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS cache (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            topic TEXT NOT NULL,
            payload TEXT NOT NULL,
            timestamp INTEGER NOT NULL,
            uploaded INTEGER DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()


def save_to_cache(topic: str, payload: dict):
    """保存数据到本地缓存"""
    conn = sqlite3.connect(CACHE_DB)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO cache (topic, payload, timestamp) VALUES (?, ?, ?)",
        (topic, json.dumps(payload), int(time.time()))
    )
    conn.commit()
    conn.close()


def get_unsent_records(limit: int = 100) -> list[tuple]:
    """获取未上传的记录"""
    conn = sqlite3.connect(CACHE_DB)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT id, topic, payload FROM cache WHERE uploaded = 0 ORDER BY timestamp ASC LIMIT ?",
        (limit,)
    )
    records = cursor.fetchall()
    conn.close()
    return records


def mark_as_uploaded(record_ids: list[int]):
    """标记记录为已上传"""
    conn = sqlite3.connect(CACHE_DB)
    cursor = conn.cursor()
    placeholders = ",".join("?" * len(record_ids))
    cursor.execute(
        f"UPDATE cache SET uploaded = 1 WHERE id IN ({placeholders})",
        record_ids
    )
    conn.commit()
    conn.close()

"""
main.py - 网关主程序
"""

import time
import logging

from modbus_collector import collect_all
from mqtt_publisher import create_mqtt_client, publish_batch
from local_cache import init_cache, save_to_cache, get_unsent_records, mark_as_uploaded

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)

POLL_INTERVAL = 30  # 轮询间隔（秒）
MQTT_RECONNECT_DELAY = 5  # MQTT 重连延迟


def main():
    init_cache()
    mqtt_client = create_mqtt_client()

    # 尝试连接 MQTT Broker
    try:
        mqtt_client.connect(MQTT_CONFIG["broker"], MQTT_CONFIG["port"], MQTT_CONFIG["keepalive"])
        mqtt_client.loop_start()
    except Exception as e:
        logging.error(f"MQTT 初始连接失败: {e}")

    while True:
        try:
            # 1. 采集所有传感器数据
            batch = collect_all()

            if not batch:
                logging.warning("本轮未采集到任何数据")
                time.sleep(POLL_INTERVAL)
                continue

            # 2. 检查 MQTT 连接状态
            if mqtt_client.is_connected():
                # 3a. 先补传缓存中未发送的数据
                unsent = get_unsent_records(limit=50)
                if unsent:
                    logging.info(f"补传缓存数据: {len(unsent)} 条")
                    for record_id, topic, payload_str in unsent:
                        try:
                            mqtt_client.publish(
                                topic,
                                payload_str,
                                qos=MQTT_CONFIG["qos"],
                                retain=True
                            )
                        except Exception:
                            break  # 发送失败，停止补传
                    else:
                        mark_as_uploaded([r[0] for r in unsent])

                # 3b. 发布本轮新数据
                for data in batch:
                    topic = f"{MQTT_CONFIG['base_topic']}/{data['sensor_name']}"
                    payload = json.dumps({
                        "temperature": data["temperature"],
                        "humidity": data["humidity"],
                        "dew_point": data["dew_point"],
                        "timestamp": data["timestamp"],
                        "ip": data["ip"],
                    }, ensure_ascii=False)
                    try:
                        mqtt_client.publish(topic, payload, qos=MQTT_CONFIG["qos"], retain=True)
                    except Exception as e:
                        logging.error(f"发布失败，写入缓存: {e}")
                        save_to_cache(topic, payload)
            else:
                # MQTT 未连接，全部写入缓存
                logging.warning("MQTT 未连接，数据写入本地缓存")
                for data in batch:
                    topic = f"{MQTT_CONFIG['base_topic']}/{data['sensor_name']}"
                    payload = json.dumps({
                        "temperature": data["temperature"],
                        "humidity": data["humidity"],
                        "dew_point": data["dew_point"],
                        "timestamp": data["timestamp"],
                        "ip": data["ip"],
                    }, ensure_ascii=False)
                    save_to_cache(topic, payload)

                # 尝试重连
                try:
                    mqtt_client.reconnect()
                except Exception:
                    pass

        except Exception as e:
            logging.error(f"主循环异常: {e}")

        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()

# 过滤 Modbus TCP 流量
tcp.port == 502

# 过滤特定传感器的通信
ip.addr == 192.168.10.101 && tcp.port == 502

# 过滤 Modbus 读保持寄存器请求
modbus.func_code == 3

# 过滤异常响应
modbus.exception_code != 0

TCP Payload:
00 01   -- Transaction ID (事务标识符)
00 00   -- Protocol ID (协议标识符, 0=Modbus)
00 06   -- Length (后续字节数 = 6)
01      -- Unit ID (从站地址)
03      -- Function Code (03 = 读保持寄存器)
00 00   -- Starting Address (起始地址 = 0, 对应 40001)
00 03   -- Quantity of Registers (读取 3 个寄存器)

TCP Payload:
00 01   -- Transaction ID (与请求对应)
00 00   -- Protocol ID
00 09   -- Length (后续字节数 = 9)
01      -- Unit ID
03      -- Function Code
06      -- Byte Count (后续 6 字节 = 3 寄存器 × 2)
01 2C   -- 寄存器 0: 0x012C = 300 → 温度 30.0℃
01 5E   -- 寄存器 1: 0x015E = 350 → 湿度 35.0%
00 64   -- 寄存器 2: 0x0064 = 100 → 露点 10.0℃

-- Measurement: env_sensor
-- Tags: sensor_name, location, ip
-- Fields: temperature, humidity, dew_point

-- 写入示例
INSERT env_sensor,sensor_name=BJ-IDC01-RACK07-INLET,location=BJ-IDC01,ip=192.168.10.101 temperature=23.6,humidity=45.2,dew_point=12.5

"""
influx_writer.py - 将 MQTT 接收的数据写入 InfluxDB
"""

from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS

INFLUX_CONFIG = {
    "url": "http://localhost:8086",
    "token": "your-influx-token",
    "org": "warehouse-org",
    "bucket": "env_data",
}

client = InfluxDBClient(
    url=INFLUX_CONFIG["url"],
    token=INFLUX_CONFIG["token"],
    org=INFLUX_CONFIG["org"]
)
write_api = client.write_api(write_options=SYNCHRONOUS)


def write_point(sensor_name: str, location: str, ip: str, temperature: float, humidity: float, dew_point: float):
    point = Point("env_sensor") \
        .tag("sensor_name", sensor_name) \
        .tag("location", location) \
        .tag("ip", ip) \
        .field("temperature", temperature) \
        .field("humidity", humidity) \
        .field("dew_point", dew_point) \
        .time(time.time_ns(), WritePrecision.NS)

    write_api.write(bucket=INFLUX_CONFIG["bucket"], record=point)

-- 最近 1 小时温度曲线
from(bucket: "env_data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "env_sensor")
  |> filter(fn: (r) => r._field == "temperature")
  |> filter(fn: (r) => r.location == "BJ-IDC01")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)

# /etc/systemd/system/env-gateway.service
[Unit]
Description=Environment Sensor Gateway (Modbus TCP -> MQTT)
After=network.target
Wants=network-online.target

[Service]
Type=simple
User=gateway
Group=gateway
WorkingDirectory=/opt/env-gateway
ExecStart=/usr/bin/python3 /opt/env-gateway/main.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

# 环境变量
Environment="PYTHONUNBUFFERED=1"

[Install]
WantedBy=multi-user.target

# 启用自启动
systemctl daemon-reload
systemctl enable env-gateway
systemctl start env-gateway

# 查看日志
journalctl -u env-gateway -f

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://mirrors.tencentyun.com/pypi/simple

COPY . .

CMD ["python", "main.py"]

# docker-compose.yml
version: "3.8"
services:
  gateway:
    build: .
    container_name: env-gateway
    restart: always
    network_mode: host
    volumes:
      - ./cache:/var/lib/gateway
      - ./certs:/etc/mqtt
    environment:
      - TZ=Asia/Shanghai

from threading import Lock

class ModbusConnectionPool:
    def __init__(self, max_connections=5):
        self.pool = {}
        self.lock = Lock()
        self.max_connections = max_connections

    def get_client(self, ip: str, port: int = 502) -> ModbusTcpClient:
        key = f"{ip}:{port}"
        with self.lock:
            if key in self.pool and self.pool[key].is_socket_open():
                return self.pool[key]

            if len(self.pool) >= self.max_connections:
                # 淘汰最久未使用的连接
                oldest_key = next(iter(self.pool))
                self.pool[oldest_key].close()
                del self.pool[oldest_key]

            client = ModbusTcpClient(ip, port=port, timeout=3)
            client.connect()
            self.pool[key] = client
            return client

# 一次读取 10 个寄存器（温度、湿度、露点、最高/最低温等）
resp = client.read_holding_registers(address=0, count=10, slave=unit_id)

以太网温湿度变送器 (Modbus TCP)
        ↓
协议转换网关 (Python: pymodbus → paho-mqtt)
        ↓ TLS
MQTT Broker (EMQX / Mosquitto on 腾讯云)
        ↓
InfluxDB (时序存储)
        ↓
Grafana (可视化 + 告警)