工业传感开发实战：Python对接Modbus TCP以太网温湿度传感器实现采集

原创

HONSOR盛世宏博

发布于 2026-10-22 11:10:04

1130

工业传感开发实战：Python对接Modbus TCP以太网温湿度传感器实现采集

物联网 · 盛世宏博 · 工业传感开发 · Python · Modbus TCP · 以太网温湿度传感器
关键词：pymodbus、ModbusClient、读保持寄存器、INT16补码、FLOAT32字节序、连接池、指数退避、质量位、断网缓存、生产级采集标签：#物联网 #Modbus #TCP/IP #POE供电 #腾讯云 #Python #InfluxDB #以太网温湿度传感器 #网口温湿度变送器 #机房监控

一、先定调：Demo 和生产的差距在哪

网上“十行 pymodbus 读温湿度”的帖子很多，但离生产还差五件事：

异常不是抛异常就完了：0x83、超时、TCP 断连要分类处理
连接不能每次 new：频繁建连会打爆传感器连接池（多数只支持 1~4 个并发）
原始值要能验：先存十六进制，再算温度，排错时一眼看出是协议问题还是解析问题
断网不能丢数据：SQLite 环形缓存 + 采集时间戳，恢复后补传
批量要有调度：错峰、限并发、退避，别把 200 台传感器轮询打成一波 DDoS

本篇从一个能跑的 Demo，递进到能上线、能排错、能扛断网的生产级采集器。

二、最小可跑：单节点冒烟（先别急着封装）

# demo_smoke.py
from pymodbus.client import ModbusTcpClient

IP = "192.168.110.11"
UNIT = 1

with ModbusTcpClient(IP, port=502, timeout=2) as c:
    if not c.connect():
        print("connect fail")
        exit(1)

    # 40001 温度 / 40002 湿度 → 0-based 地址 0，读 2 个寄存器
    rr = c.read_holding_registers(0, 2, slave=UNIT)
    if rr.isError():
        print("modbus error:", rr)
        exit(1)

    t, h = rr.registers
    if t > 32767:          # INT16 补码
        t -= 65536
    print(f"T={t*0.1:.1f}℃  H={h*0.1:.1f}%RH  raw={[hex(x) for x in rr.registers]}")

跑通这一版，确认：Ping 通、502 通、地址基对、Unit ID 对、缩放对。

跑不通就别往下封装——先回到上一篇的抓包定位法。

三、生产级单节点采集器

3.1 数据结构

from dataclasses import dataclass
from typing import Optional

@dataclass
class Sample:
    node_id: str
    ip: str
    ts: float                 # 采集时刻（不是发送时刻）
    temp: Optional[float]
    hum: Optional[float]
    dew: Optional[float]
    quality: str              # GOOD / STALE / TIMEOUT / BAD_VALUE / PROTO_ERR
    rtt_ms: float
    raw: str                  # 原始寄存器十六进制，排错用

3.2 单节点采集（带分类异常处理）

# collector_node.py
import time, struct, socket
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse

class NodeCollector:
    def __init__(self, node_id, ip, unit=1, timeout=2.0):
        self.node_id = node_id
        self.ip = ip
        self.unit = unit
        self.timeout = timeout
        self._client = None

    def _ensure(self):
        if self._client is None or not self._client.is_socket_open():
            self._client = ModbusTcpClient(self.ip, port=502, timeout=self.timeout)
            self._client.connect()
        return self._client

    def poll(self) -> Sample:
        t0 = time.perf_counter()
        try:
            c = self._ensure()
            # 40001~40004：温度/湿度/状态/露点(浮点2寄存器)
            rr = c.read_holding_registers(0, 4, slave=self.unit)
            rtt = (time.perf_counter() - t0) * 1000

            if rr.isError():
                if isinstance(rr, ExceptionResponse):
                    return Sample(self.node_id, self.ip, time.time(),
                                  None, None, None, "PROTO_ERR", rtt, f"0x{rr.exception_code:02X}")
                raise ModbusException("error response")

            raw = rr.registers
            t = raw[0]
            if t > 32767:
                t -= 65536
            temp = t * 0.1
            hum = raw[1] * 0.1

            # FLOAT32 露点：按设备点表确认 ABCD / CDAB
            dew_bytes = struct.pack('>HH', raw[3], raw[2])   # CDAB 示例
            dew = struct.unpack('>f', dew_bytes)[0]

            if not (-40 <= temp <= 85) or not (0 <= hum <= 100):
                return Sample(self.node_id, self.ip, time.time(),
                              None, None, None, "BAD_VALUE", rtt, f"{raw}")

            return Sample(self.node_id, self.ip, time.time(),
                          temp, hum, dew, "GOOD", rtt, f"{[hex(x) for x in raw]}")

        except (socket.timeout, ConnectionRefusedError):
            return Sample(self.node_id, self.ip, time.time(),
                          None, None, None, "TIMEOUT", 0, "")
        except Exception as e:
            return Sample(self.node_id, self.ip, time.time(),
                          None, None, None, "PROTO_ERR", 0, str(e))

关键点：0x83 要能区分出来（ExceptionResponse），它是“设备活着但拒绝”，别当网络故障。

四、批量采集：连接池 + 退避 + 错峰

4.1 连接池（避免每次 new Socket）

# conn_pool.py
import time, socket
from threading import Lock

class TcpPool:
    def __init__(self, max_idle=30, max_life=300):
        self._pool = {}          # ip -> (sock, last, created)
        self._lock = Lock()
        self.max_idle = max_idle
        self.max_life = max_life

    def get(self, ip, port, timeout):
        with self._lock:
            now = time.time()
            if ip in self._pool:
                s, last, created = self._pool[ip]
                if (now - last < self.max_idle) and (now - created < self.max_life):
                    return s
                self._safe_close(s)
                del self._pool[ip]
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(timeout)
            s.connect((ip, port))
            self._pool[ip] = (s, now, now)
            return s

    def release(self, ip, s, close=False):
        with self._lock:
            if close:
                self._safe_close(s)
                self._pool.pop(ip, None)
            else:
                self._pool[ip] = (s, time.time(), self._pool[ip][2])

    @staticmethod
    def _safe_close(s):
        try: s.close()
        except: pass

4.2 带退避的批量调度

# batch.py
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

BACKOFF = [2, 4, 8, 15, 30]     # 秒，上限 30

class BatchCollector:
    def __init__(self, nodes, pool, max_workers=8, cycle=10.0):
        self.nodes = nodes
        self.pool = pool
        self.max_workers = max_workers
        self.cycle = cycle
        self._bo = {n.node_id: 0 for n in nodes}
        self._last_good = {n.node_id: 0.0 for n in nodes}

    def poll_one(self, n):
        # 退避：失败次数越高，跳过的轮次越多
        if self._bo[n.node_id] and (time.time() - self._last_good[n.node_id]) < BACKOFF[min(self._bo[n.node_id], len(BACKOFF)-1)]:
            return None
        col = NodeCollector(n.node_id, n.ip, n.unit)
        s = col.poll()
        if s and s.quality == "GOOD":
            self._bo[n.node_id] = 0
            self._last_good[n.node_id] = time.time()
        elif s:
            self._bo[n.node_id] += 1
        return s

    def cycle_once(self):
        t0 = time.time()
        with ThreadPoolExecutor(max_workers=self.max_workers) as ex:
            futs = [ex.submit(self.poll_one, n) for n in self.nodes]
            for f in as_completed(futs, timeout=self.cycle):
                s = f.result()
                if s:
                    yield s
        elapsed = time.time() - t0
        time.sleep(max(0, self.cycle - elapsed))

退避很重要：传感器断网时如果还每秒打一次，502 端口和 TCP 连接池会被打满，恢复也更慢。

五、断网不丢：SQLite 环形缓存 + 补传

# store.py
import sqlite3, json, time

DB = "env_cache.db"

def init_db():
    with sqlite3.connect(DB) as c:
        c.execute("""CREATE TABLE IF NOT EXISTS samples(
            node_id TEXT, ts REAL, payload TEXT,
            PRIMARY KEY(node_id, ts))""")
        c.execute("PRAGMA journal_mode=WAL")
        c.execute("PRAGMA synchronous=NORMAL")

def save(s: Sample):
    if s.quality != "GOOD":
        return
    with sqlite3.connect(DB) as c:
        c.execute("INSERT OR IGNORE INTO samples VALUES(?,?,?)",
                  (s.node_id, s.ts, json.dumps({
                      "t": s.temp, "h": s.hum, "d": s.dew, "rtt": s.rtt_ms})))

def flush_to_platform(post_fn, since_ts=0):
    """补传：按采集时间戳顺序，失败回滚"""
    with sqlite3.connect(DB) as c:
        rows = c.execute("SELECT node_id,ts,payload FROM samples WHERE ts>? ORDER BY ts",
                         (since_ts,)).fetchall()
        for node_id, ts, payload in rows:
            if not post_fn(node_id, ts, payload):
                return ts          # 断点，下次从此继续
            c.execute("DELETE FROM samples WHERE node_id=? AND ts=?", (node_id, ts))
    return None

两个不能省的原则：

缓存键是 (node_id, ts)，平台侧 INSERT ... ON CONFLICT DO NOTHING，天然去重
ts 是采集时刻，断网 5 分钟后补传，曲线也不会出现“竖线”

六、写阈值：必须回读确认

def set_threshold(col: NodeCollector, reg_addr, value, scale=0.1):
    c = col._ensure()
    c.write_register(reg_addr, int(value / scale), slave=col.unit)
    time.sleep(0.05)                       # 设备处理延时
    rb = c.read_holding_registers(reg_addr, 1, slave=col.unit)
    if rb.isError() or rb.registers[0] != int(value / scale):
        raise RuntimeError(f"{col.node_id} 回读不一致")

生产里所有写操作：记录操作人、源 IP、改前、改后、毫秒时间戳，进审计表。

七、平台对接：InfluxDB 写入（带质量位）

from influxdb_client import InfluxDBClient, Point, WriteOptions

def to_point(s: Sample) -> Point:
    p = Point("env").tag("node", s.node_id).tag("ip", s.ip)
    if s.quality == "GOOD":
        p = p.field("temp", s.temp).field("hum", s.hum).field("dew", s.dew)
    return p.field("quality", s.quality).field("rtt_ms", s.rtt_ms)

Grafana 里：WHERE quality='GOOD'，STALE/OFFLINE 不画图、只标状态，避免“平台上一根平直线”的假象。

八、部署清单（上线前逐条过）

项目	要求
pymodbus 版本	3.x（3.0 后 API 有变化，别混用教程）
超时	2s，别默认无限
并发	≤8，别超传感器连接池
采集周期	机房 5~10s，库房 30s
日志	存原始 hex + 质量位，不存密码
断网缓存	SQLite WAL ≥24h
系统服务	systemd 托管，崩溃自动拉起
防火墙	只放网关 IP → 传感器:502
监控	采集器自身 /metrics（Prometheus）暴露在线节点数、错误率

九、常见开发坑

每次 poll new 一个 Client：连接数暴涨，传感器 RST。→ 连接池/长连接
0x0000 写成 40001：pymodbus 填 0，混用读错值。→ 台账写清 0-based
UINT16 当 INT16：负温度变 64536。→ 先补码判断
×0.1 在驱动和 HMI 各乘一次：显示 2.3 实际 23。→ 只在采集层乘
浮点寄存器长度填 1：露点读成两个独立整数。→ FLOAT32 占 2 寄存器
抓异常就重连：0x83 也重连，白白消耗连接。→ ExceptionResponse 单独处理
平台时间戳用发送时刻：断网补传曲线挤成竖线。→ 采集时刻 ts
质量位不上传：断线期间旧值一直画，巡检员以为环境稳。→ quality 进时序库

十、收尾

Python 对接 Modbus TCP 温湿度传感器，门槛在“能读”，水平在“能稳、能查、能扛断网”：

能读：地址基、Unit ID、缩放、符号、字节序
能稳：连接池、限并发、退避、写后回读
能查：原始 hex 留日志、质量位进时序库、0x83 单独分类
能扛断网：SQLite WAL 环形缓存、采集时间戳、补传去重

做到这四点，这个采集器才从“脚本”变成机房里能跑三年的工业组件。

十一、一句话给开发岗

别在 Demo 里庆祝“读到 23.5℃”；把网线拔掉再插上，曲线没断、数据没丢、日志里有原始 hex——那才叫对接完成。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

项目实战

#以太网温湿度变送器

工业传感开发实战：Python对接Modbus TCP以太网温湿度传感器实现采集

工业传感开发实战：Python对接Modbus TCP以太网温湿度传感器实现采集

工业传感开发实战：Python对接Modbus TCP以太网温湿度传感器实现采集

一、先定调：Demo 和生产的差距在哪

二、最小可跑：单节点冒烟（先别急着封装）

三、生产级单节点采集器

3.1 数据结构

3.2 单节点采集（带分类异常处理）

四、批量采集：连接池 + 退避 + 错峰

4.1 连接池（避免每次 new Socket）

4.2 带退避的批量调度

五、断网不丢：SQLite 环形缓存 + 补传

六、写阈值：必须回读确认

七、平台对接：InfluxDB 写入（带质量位）

八、部署清单（上线前逐条过）

九、常见开发坑

十、收尾

十一、一句话给开发岗

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐