以太网温湿度传感器批量采集实战：Modbus TCP与IP地址规划

192.168.A.BCD
      │ │││
      │ ││└─ 序号（1~9，单区域最多9台，够了）
      │ │└── 功能位：1=温湿度 2=漏水 3=配电 4=门禁
      │ └─── 区域位：1=机房A 2=机房B 3=配电室 4=档案库
      └───── 固定：动环 VLAN 110

# 机房A 列头1 接入端口
interface GigabitEthernet0/0/11
 port link-type access
 port default vlan 110
 poe enable
 mac-address binding ip 192.168.110.111 vlan 110 mac 00e0-4c11-0111
 port-security max-mac-num 1

# batch_collector.py

import time
import socket
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import struct

@dataclass
class SensorNode:
    node_id: str
    ip: str
    unit_id: int = 1
    interval: float = 5.0
    timeout: float = 2.0

@dataclass
class PollResult:
    node_id: str
    ip: str
    success: bool
    temp: Optional[float] = None
    hum: Optional[float] = None
    quality: str = "GOOD"     # GOOD / STALE / TIMEOUT
    rtt_ms: float = 0.0
    raw_hex: str = ""

class BatchCollector:
    def __init__(self, nodes: List[SensorNode], max_workers=8,
                 batch_interval=10.0):
        self.nodes = nodes
        self.max_workers = max_workers   # 并发数，别超过传感器连接池
        self.batch_interval = batch_interval
        self._results: Dict[str, PollResult] = {}
        self._conn_pool = ConnectionPool(max_idle=30, max_lifetime=300)

    def poll_single(self, node: SensorNode) -> PollResult:
        t0 = time.perf_counter()
        try:
            sock = self._conn_pool.get(node.ip, 502, node.timeout)
            if sock is None:
                return PollResult(node.node_id, node.ip, False, quality="TIMEOUT")

            # MBAP + 读保持寄存器(0x03) 40001~40002
            tid = int(time.perf_counter() * 1000) & 0xFFFF
            mbap = struct.pack('>HHHBB', tid, 0, 6, node.unit_id, 0x03)
            pdu = struct.pack('>HHH', 0x0000, 0x0002)
            sock.sendall(mbap + pdu)
            resp = sock.recv(256)
            rtt = (time.perf_counter() - t0) * 1000

            if len(resp) >= 11:
                temp_raw = struct.unpack('>H', resp[9:11])[0]
                hum_raw = struct.unpack('>H', resp[11:13])[0]
                # 有符号处理
                if temp_raw > 32767:
                    temp_raw -= 65536
                temp = temp_raw * 0.1
                hum = hum_raw * 0.1
                raw = resp[9:13].hex()

                # 合理性校验：异常值拦截
                if not (-40 <= temp <= 85) or not (0 <= hum <= 100):
                    return PollResult(node.node_id, node.ip, False,
                                      quality="BAD_VALUE", raw_hex=raw)

                self._conn_pool.release(node.ip, sock)
                return PollResult(node.node_id, node.ip, True,
                                  temp=temp, hum=hum,
                                  quality="GOOD", rtt_ms=rtt, raw_hex=raw)

            self._conn_pool.release(node.ip, sock, close=True)
            return PollResult(node.node_id, node.ip, False, quality="BAD_FRAME")

        except socket.timeout:
            return PollResult(node.node_id, node.ip, False,
                              quality="TIMEOUT", rtt_ms=rtt)
        except Exception:
            return PollResult(node.node_id, node.ip, False, quality="ERROR")

    def poll_batch(self) -> Dict[str, PollResult]:
        """一轮全量采集"""
        with ThreadPoolExecutor(max_workers=self.max_workers) as ex:
            futures = {ex.submit(self.poll_single, n): n for n in self.nodes}
            for f in as_completed(futures, timeout=self.batch_interval):
                r = f.result()
                self._results[r.node_id] = r
        return self._results

    def run_forever(self, on_batch=None):
        """持续采集循环"""
        while True:
            t0 = time.time()
            results = self.poll_batch()
            if on_batch:
                on_batch(results)
            elapsed = time.time() - t0
            sleep_for = max(0, self.batch_interval - elapsed)
            time.sleep(sleep_for)


class ConnectionPool:
    """简单TCP连接池，避免每台每次新建握手"""
    def __init__(self, max_idle=30, max_lifetime=300):
        self._pool: Dict[str, tuple] = {}   # ip -> (sock, last_use, created)
        self.max_idle = max_idle
        self.max_lifetime = max_lifetime

    def get(self, ip, port, timeout):
        now = time.time()
        if ip in self._pool:
            sock, last, created = self._pool[ip]
            if (now - last < self.max_idle) and (now - created < self.max_lifetime):
                return sock
            else:
                try: sock.close()
                except: pass
                del self._pool[ip]
        # 新建
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            sock.connect((ip, port))
            self._pool[ip] = (sock, now, now)
            return sock
        except Exception:
            return None

    def release(self, ip, sock, close=False):
        if close:
            try: sock.close()
            except: pass
            self._pool.pop(ip, None)

def push_batch(results: Dict[str, PollResult]):
    points = []
    for r in results.values():
        tags = {"node": r.node_id, "ip": r.ip}
        fields = {"rtt_ms": r.rtt_ms}
        if r.success:
            fields.update({"temp": r.temp, "hum": r.hum})
        fields["quality"] = r.quality
        points.append(Point("env_batch").tags(tags).fields(fields))
    write_api.write(bucket="env", record=points)