工业传感通讯避坑：TCP/UDP 以太网温湿度传感器现场测试全流程

阶段 0：到货开箱         ── 30 分钟
阶段 1：实验室预测试      ── 1 天
阶段 2：现场布线后上电    ── 随施工
阶段 3：TCP 协议测试      ── 2 小时
阶段 4：UDP 协议测试      ── 1 小时
阶段 5：综合压力与稳定性   ── 24–72 小时
阶段 6：验收与文档归档    ── 2 小时

□ 数量核对：实物数量 vs 合同/送货单
□ 外观检查：外壳无裂纹、屏显无碎裂、端子无变形
□ 铭牌记录：型号、序列号（SN）、固件版本、PoE 等级、MAC 地址
□ 配件清点：安装支架、说明书、合格证、校准证书（如有）
□ 随机抽测：从 N 台中抽 2–3 台，上电验证基本功能

测试项 1：PoE 供电
  方法：接 PoE 交换机，观察端口灯 + 传感器电源灯
  通过标准：3 秒内完成链路协商，电源灯常亮
  记录：协商速率（100M/1000M）、PoE 端口功率消耗

测试项 2：静态 IP 配置
  方法：用厂家配置工具修改 IP，设置子网掩码、网关、从站 ID
  通过标准：配置保存后重启，IP 保持不变
  记录：配置工具版本、配置保存耗时

测试项 3：Modbus TCP 基础通信
  方法：Modbus Poll 读取 40001-40002（温度、湿度）
  通过标准：读取值与标准温湿度计偏差 ≤0.5℃ / ≤3%RH
  记录：响应时间、数据格式（INT16 ×0.1）、字节序

测试项 4：SNMP 功能
  方法：snmpwalk 读取系统信息、温度 OID
  通过标准：返回正确数值，响应 < 100ms
  记录：MIB OID 树、社区名、SNMP 版本

测试项 5：UDP 告警功能（如有）
  方法：触发越限，用 Wireshark 抓 UDP 包
  通过标准：接收端收到告警报文，内容正确
  记录：UDP 端口、报文格式、触发条件

步骤 1：用厂家配置工具批量设置 IP
  → 避免现场一台台改 IP

步骤 2：用 Python 脚本批量验证连通性
  → 50 台一次跑完，自动输出通/断

步骤 3：记录所有设备的 MAC-IP-位置对应关系
  → 生成台账，现场按标签安装

import subprocess
import concurrent.futures
import time

def ping_host(ip):
    """Ping 单个 IP，返回结果"""
    try:
        result = subprocess.run(
            ['ping', '-c', '3', '-W', '2', ip],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            # 解析丢包率
            output = result.stdout
            if '0% packet loss' in output:
                return (ip, 'OK', '0% loss')
            else:
                loss = output.split('% packet loss')[0].split()[-1]
                return (ip, 'WARN', f'{loss}% loss')
        else:
            return (ip, 'FAIL', 'no response')
    except subprocess.TimeoutExpired:
        return (ip, 'FAIL', 'timeout')

def batch_ping(ip_list, max_workers=20):
    """并发 Ping 多个 IP"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(ping_host, ip): ip for ip in ip_list}
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())

    # 按 IP 排序输出
    results.sort(key=lambda x: x[0])
    print(f"\n{'IP':<20} {'Status':<8} {'Detail'}")
    print("-" * 50)
    for ip, status, detail in results:
        mark = "✅" if status == "OK" else ("⚠️ " if status == "WARN" else "❌")
        print(f"{ip:<20} {mark} {status:<6} {detail}")

    ok_count = sum(1 for _, s, _ in results if s == "OK")
    print(f"\n总计: {ok_count}/{len(results)} 通过")
    return results

# 从台账文件读取 IP 列表
ips = [f"192.168.100.{i}" for i in range(50, 100)]
batch_ping(ips)

□ 网线 8 芯全通（测线仪逐根确认）
□ RJ45 水晶头：T568B 标准，屏蔽层接好
□ 屏蔽网线单点接地（机柜端）
□ PoE 交换机端口已使能、优先级设为 Critical
□ 端口 EEE（节能）已关闭
□ 超距点位已确认（>100m 需中继或光纤）
□ 传感器安装位置：距地 1.5–2m，避开直吹/热源
□ 安装牢固，水晶头无受力

1. 先给 PoE 交换机上电，等待启动完成（约 1–2 分钟）
2. 逐台插入传感器网线（不要一次性全插，便于定位问题）
3. 观察交换机端口灯：绿色/橙色常亮 = 链路 up
4. 等待 30 秒，让所有端口协商完成
5. 用批量 Ping 脚本验证所有设备在线

import asyncio
import time
from pymodbus.client import AsyncModbusTcpClient

class ModbusTester:
    def __init__(self, host, port=502, slave_id=1):
        self.host = host
        self.port = port
        self.slave_id = slave_id

    async def test_basic_read(self):
        """基础读取测试"""
        client = AsyncModbusTcpClient(self.host, port=self.port, timeout=2.0)
        await client.connect()
        
        if not client.connected:
            print(f"❌ {self.host} 连接失败")
            return False
        
        # 读取温度、湿度
        resp = await client.read_holding_registers(0, count=2, slave=self.slave_id)
        if resp.isError():
            print(f"❌ {self.host} 读取失败: {resp}")
            return False
        
        temp = resp.registers[0] * 0.1
        humid = resp.registers[1] * 0.1
        print(f"✅ {self.host}: 温度 {temp}℃, 湿度 {humid}%RH")
        await client.close()
        return True

    async def test_concurrent(self, threads=10):
        """并发读取测试"""
        tasks = []
        for i in range(threads):
            tasks.append(self.test_basic_read())
        results = await asyncio.gather(*tasks, return_exceptions=True)
        success = sum(1 for r in results if r is True)
        print(f"并发 {threads} 次读取: {success}/{threads} 成功")
        return success == threads

    async def test_stability(self, duration=3600):
        """稳定性测试（1 小时）"""
        client = AsyncModbusTcpClient(self.host, port=self.port, timeout=2.0)
        await client.connect()
        
        count = 0
        errors = 0
        start = time.time()
        
        while time.time() - start < duration:
            try:
                resp = await client.read_holding_registers(0, count=2, slave=self.slave_id)
                if resp.isError():
                    errors += 1
                else:
                    count += 1
            except Exception:
                errors += 1
            await asyncio.sleep(1)
        
        await client.close()
        print(f"稳定性测试: {count} 成功, {errors} 失败, 成功率 {count/(count+errors)*100:.1f}%")
        return errors == 0

    async def test_reconnect(self):
        """断线重连测试"""
        client = AsyncModbusTcpClient(self.host, port=self.port, timeout=2.0)
        await client.connect()
        
        # 第一次读取
        await client.read_holding_registers(0, count=2, slave=self.slave_id)
        print(f"✅ 首次连接成功")
        
        # 模拟断线（关闭连接）
        client.close()
        print(f"⚠️ 模拟断线")
        
        # 等待 5 秒
        await asyncio.sleep(5)
        
        # 重新连接
        await client.connect()
        if client.connected:
            resp = await client.read_holding_registers(0, count=2, slave=self.slave_id)
            if not resp.isError():
                print(f"✅ 重连后读取成功")
                return True
        
        print(f"❌ 重连后读取失败")
        return False

# 批量测试
async def batch_test(hosts):
    tester = ModbusTester(hosts[0])
    for host in hosts:
        tester.host = host
        print(f"\n{'='*50}")
        print(f"测试设备: {host}")
        await tester.test_basic_read()

hosts = [f"192.168.100.{i}" for i in range(50, 55)]
asyncio.run(batch_test(hosts))

抓包点：传感器所接交换机端口镜像，或采集端主机网卡

显示过滤器：
  ip.addr == 192.168.100.50 && tcp.port == 502

重点观察：
  □ 三次握手是否完整（SYN → SYN+ACK → ACK）
  □ 是否有重传（tcp.analysis.retransmission）
  □ 是否有重复 ACK（tcp.analysis.duplicate_ack）
  □ 是否有零窗口（tcp.zero_window）
  □ 是否有连接重置（tcp.flags.reset == 1）
  □ Modbus 响应时间（frame.time_delta_displayed）

import socket
import struct
import threading
import time

class UDPAlertReceiver:
    def __init__(self, port=8888):
        self.port = port
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind(('', port))
        self.sock.settimeout(1.0)
        self.received = {}
        self.running = True

    def start(self):
        """启动接收线程"""
        self.thread = threading.Thread(target=self._receive_loop)
        self.thread.start()

    def stop(self):
        self.running = False
        self.thread.join()

    def _receive_loop(self):
        while self.running:
            try:
                data, addr = self.sock.recvfrom(1024)
                seq = struct.unpack('!I', data[:4])[0]
                timestamp = struct.unpack('!I', data[4:8])[0]
                temp = struct.unpack('!f', data[8:12])[0]
                humid = struct.unpack('!f', data[12:16])[0]
                alarm_bits = struct.unpack('!H', data[16:18])[0]

                self.received[seq] = {
                    'addr': addr,
                    'timestamp': timestamp,
                    'temp': temp,
                    'humid': humid,
                    'alarm': alarm_bits,
                    'recv_time': time.time()
                }

                print(f"收到告警 #{seq} from {addr[0]}: "
                      f"temp={temp:.1f}℃, humid={humid:.1f}%RH, "
                      f"alarm_bits={alarm_bits:04b}")

            except socket.timeout:
                continue

    def get_stats(self, expected_count=None):
        total = len(self.received)
        if expected_count:
            loss_rate = (expected_count - total) / expected_count * 100
            print(f"接收: {total}/{expected_count}, 丢包率: {loss_rate:.2f}%")
        else:
            print(f"共接收 {total} 个告警报文")
        return total

# 使用示例
receiver = UDPAlertReceiver(port=8888)
receiver.start()

# 等待传感器触发告警（手动调节或模拟越限）
print("等待 UDP 告警...")
time.sleep(30)

receiver.stop()
receiver.get_stats(expected_count=30)  # 假设期望收到 30 个

# 从测试主机向传感器 UDP 端口发测试包（如果设备支持 UDP 配置/命令）
echo -n "test" | nc -u -w1 192.168.100.50 8888

# 或用 Python
python3 -c "
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.sendto(b'hello', ('192.168.100.50', 8888))
print('sent')
"

import asyncio
import time
import csv
from pymodbus.client import AsyncModbusTcpClient

class ComprehensiveTest:
    def __init__(self, hosts, duration_hours=24):
        self.hosts = hosts
        self.duration = duration_hours * 3600
        self.results = {host: [] for host in hosts}

    async def poll_all(self):
        """并发轮询所有设备"""
        tasks = []
        for host in self.hosts:
            tasks.append(self.poll_one(host))
        await asyncio.gather(*tasks, return_exceptions=True)

    async def poll_one(self, host):
        """轮询单台设备"""
        client = AsyncModbusTcpClient(host, port=502, timeout=2.0)
        await client.connect()
        if not client.connected:
            self.results[host].append({'time': time.time(), 'status': 'offline'})
            return

        try:
            resp = await client.read_holding_registers(0, count=2, slave=1)
            if resp.isError():
                self.results[host].append({'time': time.time(), 'status': 'error'})
            else:
                self.results[host].append({
                    'time': time.time(),
                    'status': 'ok',
                    'temp': resp.registers[0] * 0.1,
                    'humid': resp.registers[1] * 0.1
                })
        except Exception:
            self.results[host].append({'time': time.time(), 'status': 'timeout'})
        finally:
            client.close()

    async def run(self):
        """运行综合测试"""
        start = time.time()
        round_num = 0
        while time.time() - start < self.duration:
            round_num += 1
            print(f"Round {round_num}: {time.strftime('%Y-%m-%d %H:%M:%S')}")
            await self.poll_all()
            await asyncio.sleep(5)  # 5 秒采集周期

        # 输出统计
        self.print_stats()

    def print_stats(self):
        print("\n" + "="*60)
        print("综合测试结果")
        print("="*60)
        for host, data in self.results.items():
            total = len(data)
            ok = sum(1 for d in data if d['status'] == 'ok')
            offline = sum(1 for d in data if d['status'] == 'offline')
            error = sum(1 for d in data if d['status'] == 'error')
            timeout = sum(1 for d in data if d['status'] == 'timeout')
            online_rate = ok / total * 100 if total > 0 else 0

            print(f"\n{host}:")
            print(f"  总采集: {total}")
            print(f"  成功: {ok} ({online_rate:.1f}%)")
            print(f"  离线: {offline}")
            print(f"  错误: {error}")
            print(f"  超时: {timeout}")

        # 保存 CSV
        with open(f'test_results_{int(time.time())}.csv', 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['host', 'time', 'status', 'temp', 'humid'])
            for host, data in self.results.items():
                for d in data:
                    writer.writerow([
                        host, d['time'], d['status'],
                        d.get('temp', ''), d.get('humid', '')
                    ])

hosts = [f"192.168.100.{i}" for i in range(50, 100)]
test = ComprehensiveTest(hosts, duration_hours=24)
asyncio.run(test.run())

□ 全部设备在线（Ping + Modbus TCP 双验证）
□ 数据准确（与标准器具对比，偏差在允许范围内）
□ TCP 502 端口可达
□ UDP 告警功能正常（如有）
□ SNMP 可读取（如有）
□ 断网恢复测试通过
□ 断电恢复测试通过
□ 24h 稳定性测试通过
□ 数据按时入库（InfluxDB/时序库）
□ 联动功能正常（越限告警、设备联动）
□ 台账完整（MAC/IP/位置/SN 对应关系）
□ 配置备份（传感器配置、交换机配置、组态工程）
□ 文档齐全（点位图、布线记录、测试报告）