# 仅捕获Modbus TCP流量
tcp port 502

# 捕获特定设备的通信
host 192.168.1.100 and tcp port 502

# 排除广播流量
not broadcast and not multicast

# 基本过滤
modbus                    # 所有Modbus流量
modbus.func_code == 0x03  # 仅读取保持寄存器
modbus.func_code == 0x10  # 仅写入多个寄存器

# 高级过滤
modbus.unit_id == 1       # 特定从站设备
modbus.trans_id == 1234   # 特定事务ID
tcp.analysis.flags        # TCP分析标志

# 成功响应 - 绿色
modbus && tcp.flags.ack == 1

# 错误响应 - 红色
modbus.excep_code != 0x00

# 超时重传 - 黄色
tcp.analysis.retransmission

1. 过滤规则: tcp.analysis.flags && !tcp.analysis.ack_rtt
2. 分析步骤:
   - 检查SYN-ACK往返时间
   - 识别重传报文
   - 分析网络延迟原因
3. 解决方案:
   - 调整TCP超时参数
   - 优化网络拓扑
   - 增加心跳机制

1. 过滤规则: modbus.func_code == 0x03 || modbus.func_code == 0x10
2. 验证方法:
   - 对比请求和响应数据
   - 检查寄存器地址连续性
   - 验证数据字节顺序
3. 常见问题:
   - 字节序错误 (Big-endian vs Little-endian)
   - 地址偏移计算错误
   - 数据类型转换问题

原始报文: 01 03 00 00 00 0A C5 CD
解析结果:
├── 设备地址: 01
├── 功能码: 03 (读取保持寄存器)
├── 起始地址: 0x0000
├── 寄存器数量: 10
└── CRC校验: 0xC5CD (正确)

# 设备发现扫描
modbus-scanner -t tcp -a 1-247 -r 1-100 192.168.1.0/24

# 功能码支持检测
modbus-scanner -t rtu -p /dev/ttyUSB0 -b 9600 -f all

# 寄存器映射扫描
modbus-scanner -m mapping -s 0 -e 65535 -d 1

[设备1] 温度传感器 - 192.168.1.101:502
├── 寄存器 40001: 25.3°C (实时更新)
├── 寄存器 40002: 60.2% RH
└── 线圈 00001: ON

[设备2] PLC控制器 - 192.168.1.102:502
├── 寄存器 40010: 1200 RPM
├── 寄存器 40011: 3.2 MPa
└── 离散输入 10001: TRIGGERED

# 自动导出配置
{
  "export": {
    "format": "csv",
    "interval": 1000,
    "fields": [
      "timestamp",
      "device_id", 
      "register_address",
      "value",
      "quality"
    ],
    "destination": "C:/logs/modbus_data_%Y%m%d.csv"
  }
}

// Modbus Poll脚本示例
function testSequence() {
    // 1. 初始化连接
    connect("192.168.1.100", 502);

    // 2. 基本功能测试
    testReadCoils(0, 16, "线圈读取测试");
    testReadRegisters(40001, 10, "寄存器读取测试");

    // 3. 写入测试
    testWriteCoil(0, true, "线圈写入测试");
    testWriteRegister(40001, 1234, "寄存器写入测试");

    // 4. 边界测试
    testBoundaryConditions();

    // 5. 性能测试
    testPerformance(1000, "1000次请求性能测试");

    // 生成测试报告
    generateReport("test_report.html");
}

设置断点条件:
- 特定寄存器值变化
- 错误响应发生
- 通信超时
- 数据越界

监视列表配置:
$temp = [40001]  // 温度寄存器
$speed = [40010] // 转速寄存器
$status = [00001] // 状态线圈

触发条件:
IF $temp > 100 THEN ALERT("温度过高!")
IF $speed < 500 THEN LOG("转速过低")

# 温度波动模拟
def temperature_simulation(cycle):
    import math
    base_temp = 25.0
    daily_variation = 8.0 * math.sin(2 * math.pi * cycle / 86400)
    random_noise = random.uniform(-0.5, 0.5)
    return base_temp + daily_variation + random_noise

# 压力趋势模拟  
def pressure_trend(cycle):
    start_pressure = 100.0
    trend = 0.01 * (cycle / 3600)  # 每小时上升0.01
    fluctuation = 0.5 * math.sin(2 * math.pi * cycle / 600)
    return start_pressure + trend + fluctuation

class PumpSimulator:
    def __init__(self):
        self.running = False
        self.speed = 0
        self.pressure = 0

    def update(self, cycle):
        if self.running:
            # 运行状态模拟
            self.speed = 1500 + 100 * math.sin(cycle / 10)
            self.pressure = 3.2 + 0.1 * math.sin(cycle / 5)
        else:
            # 停止状态
            self.speed = 0
            self.pressure = 0

        return {
            'coils': {'running': self.running},
            'registers': {
                'speed': int(self.speed),
                'pressure': int(self.pressure * 100)
            }
        }

<test-scenario name="通信故障测试">
    <phase duration="30s" mode="normal" />
    <phase duration="10s" mode="timeout" error-rate="100%" />
    <phase duration="20s" mode="partial" error-rate="50%" />
    <phase duration="30s" mode="normal" />
</test-scenario>

load-test:
  clients: 50
  requests-per-second: 100
  duration: 300s
  ramp-up: 30s
  metrics:
    - response-time
    - throughput  
    - error-rate
    - cpu-usage
    - memory-usage

支持所有Modbus地址类型:
- 0xxxx: 线圈 (000001-065536)
- 1xxxx: 离散输入 (100001-165536)  
- 3xxxx: 输入寄存器 (300001-365536)
- 4xxxx: 保持寄存器 (400001-465536)

# 寄存器配置示例
address,type,value,description,min,max,unit
40001,float,25.3,温度传感器,0,100,°C
40003,int32,1200,电机转速,0,3000,RPM
40005,uint16,60,湿度百分比,0,100,%
40007,bool,1,设备状态,0,1,ON/OFF

-- Lua测试脚本示例
function test_sequence()
    -- 初始化测试
    print("开始Modbus设备测试...")

    -- 测试1: 基本通信
    local result1 = test_communication()
    assert(result1, "基本通信测试失败")

    -- 测试2: 数据读写
    local result2 = test_data_access()
    assert(result2, "数据访问测试失败")

    -- 测试3: 边界条件
    local result3 = test_boundary_conditions()
    assert(result3, "边界条件测试失败")

    -- 测试4: 性能测试
    local result4 = test_performance()
    assert(result4, "性能测试失败")

    print("所有测试通过!")
    return true
end

modpoll [选项] <主机名或IP地址>

# 扫描网络中的Modbus设备
modpoll -m tcp -a 1-247 -t 3 -r 1 -c 1 192.168.1.0/24

# 扫描串口设备
modpoll -m rtu -a 1-247 -p /dev/ttyUSB0 -b 9600 -t 3 -r 1

# 快速设备识别
modpoll -m tcp -a 1-10 -t 0x2B -r 0 -c 2 192.168.1.100

# 读取保持寄存器 (功能码03)
modpoll -m tcp -t 4 -r 40001 -c 10 192.168.1.100

# 读取输入寄存器 (功能码04)
modpoll -m tcp -t 3 -r 30001 -c 5 192.168.1.100

# 读取线圈状态 (功能码01)
modpoll -m tcp -t 0 -r 1 -c 16 192.168.1.100

# 读取离散输入 (功能码02)
modpoll -m tcp -t 1 -r 10001 -c 8 192.168.1.100

# 写入单个寄存器 (功能码06)
modpoll -m tcp -t 4:hex -r 40001 -c 1 192.168.1.100 0x1388

# 写入多个寄存器 (功能码16)
modpoll -m tcp -t 4:hex -r 40010 -c 3 192.168.1.100 0x0001 0x0002 0x0003

# 写入单个线圈 (功能码05)
modpoll -m tcp -t 0 -r 1 -c 1 192.168.1.100 ON

# 写入多个线圈 (功能码15)
modpoll -m tcp -t 0 -r 1 -c 8 192.168.1.100 ON OFF ON OFF ON OFF ON OFF

# 定时轮询显示
modpoll -m tcp -t 4 -r 40001 -c 5 -p 1000 -1 192.168.1.100

# 数据记录到文件
modpoll -m tcp -t 4 -r 40001 -c 3 -p 5000 -l modbus_log.csv 192.168.1.100

# 变化触发动作
modpoll -m tcp -t 4 -r 40001 -c 1 -p 1000 -T 25 -c "alert.sh" 192.168.1.100

# CSV格式输出
modpoll -m tcp -t 4 -r 40001 -c 10 -f c 192.168.1.100

# JSON格式输出
modpoll -m tcp -t 4 -r 40001 -c 5 -f j 192.168.1.100

# 自定义格式
modpoll -m tcp -t 4 -r 40001 -c 3 -f "Address: %a, Value: %v" 192.168.1.100

#!/bin/bash
# modpoll批量测试脚本

DEVICE="192.168.1.100"
LOG_FILE="test_$(date +%Y%m%d_%H%M%S).log"

echo "开始Modbus设备测试..." | tee -a $LOG_FILE

# 测试1: 基本连接
echo "测试1: 基本连接测试" | tee -a $LOG_FILE
modpoll -m tcp -a 1 -t 3 -r 30001 -c 1 $DEVICE >> $LOG_FILE 2>&1
check_result $? "基本连接"

# 测试2: 寄存器读写
echo "测试2: 寄存器读写测试" | tee -a $LOG_FILE
modpoll -m tcp -a 1 -t 4 -r 40001 -c 5 $DEVICE >> $LOG_FILE 2>&1
check_result $? "寄存器读取"

# 测试3: 性能测试
echo "测试3: 性能测试" | tee -a $LOG_FILE
for i in {1..100}; do
    modpoll -m tcp -a 1 -t 4 -r 40001 -c 1 $DEVICE > /dev/null 2>&1
done
echo "100次请求完成" | tee -a $LOG_FILE

# 生成测试报告
generate_report $LOG_FILE

# 增加超时时间查看响应
modpoll -m tcp -t 4 -r 40001 -c 1 -o 5000 192.168.1.100

# 启用详细日志
modpoll -m tcp -t 4 -r 40001 -c 1 -v 192.168.1.100

# 测试不同数据量
for size in 1 10 50 100 125; do
    echo "测试 $size 个寄存器..."
    modpoll -m tcp -t 4 -r 40001 -c $size -o 3000 192.168.1.100
done

# 多次读取验证稳定性
for i in {1..10}; do
    echo "第 $i 次读取:"
    modpoll -m tcp -t 4 -r 40001 -c 3 -f c 192.168.1.100
    sleep 1
done

# 写入后立即读取验证
modpoll -m tcp -t 4:hex -r 40010 -c 1 192.168.1.100 0x55AA
sleep 0.5
modpoll -m tcp -t 4 -r 40010 -c 1 192.168.1.100

设备模型结构:
├── 通信参数
│   ├── 协议: RTU/TCP/ASCII
│   ├── 波特率: 9600/19200/38400/115200
│   ├── 数据位: 8
│   ├── 停止位: 1/2
│   └── 校验位: None/Even/Odd
├── 数据点定义
│   ├── 线圈 (0xxxx)
│   ├── 离散输入 (1xxxx)
│   ├── 输入寄存器 (3xxxx)
│   └── 保持寄存器 (4xxxx)
└── 数据映射
    ├── 缩放因子
    ├── 偏移量
    ├── 数据类型
    └── 工程单位

实时监控界面:
─────────────────────────────────────
设备: 温度控制器 [192.168.1.101:502]
─────────────────────────────────────
地址      名称          值     单位   状态
40001     温度        25.3    °C     ✅
40002     湿度        60.2    %      ✅  
40003     压力        101.3   kPa    ✅
00001     加热器       ON            ✅
10001     报警         OFF           ✅
─────────────────────────────────────
更新时间: 2026-02-24 17:15:32
通信状态: 正常 (延迟: 12ms)

{
  "dashboard": {
    "title": "生产线监控",
    "widgets": [
      {
        "type": "gauge",
        "title": "温度监控",
        "dataSource": "modbus://192.168.1.101/40001",
        "min": 0,
        "max": 100,
        "warning": 80,
        "critical": 90
      },
      {
        "type": "chart",
        "title": "历史趋势",
        "dataSources": [
          "modbus://192.168.1.101/40001",
          "modbus://192.168.1.101/40002"
        ],
        "period": "24h"
      }
    ]
  }
}

// 报警规则配置
alarmRules: [
  {
    name: "高温报警",
    condition: "temperature > 80",
    actions: [
      "sendEmail('operator@factory.com', '高温报警')",
      "sendSMS('+8613800138000', '设备温度过高')",
      "logEvent('high_temperature_alarm')"
    ],
    severity: "CRITICAL"
  },
  {
    name: "通信中断",
    condition: "deviceStatus == 'OFFLINE'",
    actions: [
      "notifyMaintenanceTeam()",
      "startBackupSystem()"
    ],
    severity: "HIGH"
  }
]

#!/usr/bin/env python3
"""
Modbus性能分析工具
"""

import time
import statistics
from datetime import datetime
import matplotlib.pyplot as plt
from pymodbus.client import ModbusTcpClient

class ModbusPerformanceAnalyzer:
    def __init__(self, host, port=502):
        self.client = ModbusTcpClient(host, port)
        self.metrics = {
            'response_times': [],
            'success_count': 0,
            'error_count': 0,
            'start_time': None,
            'end_time': None
        }

    def run_test(self, duration=60, requests_per_second=10):
        """运行性能测试"""
        self.metrics['start_time'] = datetime.now()

        total_requests = duration * requests_per_second
        interval = 1.0 / requests_per_second

        for i in range(total_requests):
            start = time.time()

            try:
                # 执行Modbus请求
                result = self.client.read_holding_registers(40001, 10)

                if result.isError():
                    self.metrics['error_count'] += 1
                else:
                    self.metrics['success_count'] += 1

            except Exception as e:
                self.metrics['error_count'] += 1
                print(f"请求 {i+1} 失败: {e}")

            response_time = (time.time() - start) * 1000  # 转换为毫秒
            self.metrics['response_times'].append(response_time)

            # 控制请求频率
            elapsed = time.time() - start
            if elapsed < interval:
                time.sleep(interval - elapsed)

        self.metrics['end_time'] = datetime.now()
        return self.metrics

    def generate_report(self):
        """生成性能报告"""
        if not self.metrics['response_times']:
            return "没有测试数据"

        rt = self.metrics['response_times']

        report = f"""
Modbus性能测试报告
==================
测试时间: {self.metrics['start_time']} - {self.metrics['end_time']}
总请求数: {len(rt)}
成功请求: {self.metrics['success_count']}
失败请求: {self.metrics['error_count']}
成功率: {(self.metrics['success_count'] / len(rt)) * 100:.2f}%

响应时间统计 (毫秒):
- 平均值: {statistics.mean(rt):.2f}
- 中位数: {statistics.median(rt):.2f}
- 最小值: {min(rt):.2f}
- 最大值: {max(rt):.2f}
- 标准差: {statistics.stdev(rt):.2f}
- 95百分位: {sorted(rt)[int(len(rt) * 0.95)]:.2f}
- 99百分位: {sorted(rt)[int(len(rt) * 0.99)]:.2f}

建议:
"""

        avg_rt = statistics.mean(rt)
        if avg_rt < 10:
            report += "- 性能优秀，无需优化n"
        elif avg_rt < 50:
            report += "- 性能良好，可考虑小规模优化n"
        elif avg_rt < 100:
            report += "- 性能一般，建议进行优化n"
        else:
            report += "- 性能较差，需要重点优化n"

        return report

    def plot_results(self, save_path=None):
        """绘制性能图表"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 8))

        # 响应时间趋势
        axes[0, 0].plot(self.metrics['response_times'])
        axes[0, 0].set_title('响应时间趋势')
        axes[0, 0].set_xlabel('请求序号')
        axes[0, 0].set_ylabel('响应时间 (ms)')
        axes[0, 0].grid(True)

        # 响应时间分布
        axes[0, 1].hist(self.metrics['response_times'], bins=50, edgecolor='black')
        axes[0, 1].set_title('响应时间分布')
        axes[0, 1].set_xlabel('响应时间 (ms)')
        axes[0, 1].set_ylabel('频次')
        axes[0, 1].grid(True)

        # 成功失败统计
        labels = ['成功', '失败']
        sizes = [self.metrics['success_count'], self.metrics['error_count']]
        axes[1, 0].pie(sizes, labels=labels, autopct='%1.1f%%')
        axes[1, 0].set_title('请求成功率')

        # 累积分布函数
        sorted_rt = sorted(self.metrics['response_times'])
        cdf = [i/len(sorted_rt) for i in range(len(sorted_rt))]
        axes[1, 1].plot(sorted_rt, cdf)
        axes[1, 1].set_title('响应时间CDF')
        axes[1, 1].set_xlabel('响应时间 (ms)')
        axes[1, 1].set_ylabel('累积概率')
        axes[1, 1].grid(True)

        plt.tight_layout()

        if save_path:
            plt.savefig(save_path, dpi=300)
            print(f"图表已保存到: {save_path}")

        plt.show()

# 使用示例
if __name__ == "__main__":
    analyzer = ModbusPerformanceAnalyzer("192.168.1.100")
    metrics = analyzer.run_test(duration=30, requests_per_second=20)
    print(analyzer.generate_report())
    analyzer.plot_results("modbus_performance.png")

tcp_optimization:
  tcp_nodelay: true      # 禁用Nagle算法
  tcp_keepalive: true    # 启用TCP保活
  keepalive_time: 60     # 保活时间(秒)
  keepalive_intvl: 10    # 保活间隔(秒)
  keepalive_probes: 3    # 保活探测次数

socket_options:
  so_reuseaddr: true     # 地址重用
  so_rcvbuf: 65536       # 接收缓冲区大小
  so_sndbuf: 65536       # 发送缓冲区大小

# 批量读取优化
def optimized_read(client, address_ranges):
    """
    优化读取策略：
    1. 合并相邻地址的读取请求
    2. 避免读取不必要的数据
    3. 使用合适的超时时间
    """
    results = {}

    for start, count in address_ranges:
        # 检查是否可合并
        if can_merge_with_previous(start, count):
            continue

        try:
            # 使用合适的超时
            client.timeout = calculate_timeout(count)

            # 执行读取
            result = client.read_holding_registers(start, count)

            if not result.isError():
                results[(start, count)] = result.registers
            else:
                # 失败时尝试分片读取
                results.update(fallback_read(client, start, count))

        except Exception as e:
            log_error(f"读取失败 {start}-{start+count}: {e}")

    return results

class ModbusConnectionPool:
    """Modbus连接池管理"""

    def __init__(self, host, port=502, max_connections=10):
        self.host = host
        self.port = port
        self.max_connections = max_connections
        self.pool = []
        self.in_use = set()

    def get_connection(self):
        """获取连接"""
        # 1. 检查空闲连接
        for conn in self.pool:
            if conn not in self.in_use and conn.is_socket_open():
                self.in_use.add(conn)
                return conn

        # 2. 创建新连接
        if len(self.pool) < self.max_connections:
            conn = ModbusTcpClient(self.host, self.port)
            self.pool.append(conn)
            self.in_use.add(conn)
            return conn

        # 3. 等待连接释放
        raise NoAvailableConnection("连接池已满")

    def release_connection(self, conn):
        """释放连接"""
        if conn in self.in_use:
            self.in_use.remove(conn)

    def close_all(self):
        """关闭所有连接"""
        for conn in self.pool:
            conn.close()
        self.pool.clear()
        self.in_use.clear()

1. 现象收集
   - 通信完全中断
   - 数据不一致
   - 响应缓慢
   - 间歇性故障

2. 基本信息收集
   - 网络拓扑
   - 设备型号和固件版本
   - 配置参数
   - 错误日志

# 使用modpoll进行快速测试
# 1. 测试基本连接
modpoll -m tcp -a 1 -t 3 -r 30001 -c 1 192.168.1.100

# 2. 测试不同功能码
for func in 1 2 3 4; do
    echo "测试功能码 0x0$func..."
    modpoll -m tcp -a 1 -t $func -r 1 -c 1 192.168.1.100
done

# 3. 测试不同数据量
for count in 1 10 50 100; do
    echo "测试 $count 个寄存器..."
    modpoll -m tcp -a 1 -t 4 -r 40001 -c $count 192.168.1.100
done

1. Wireshark抓包分析
   - 捕获完整通信过程
   - 分析报文时序
   - 检查协议合规性

2. 设备模拟测试
   - 使用ModbusPal模拟正常设备
   - 对比实际设备行为
   - 隔离问题范围

3. 性能压力测试
   - 测试不同负载下的表现
   - 识别性能瓶颈
   - 验证系统稳定性

1. 配置优化
   - 调整超时参数
   - 优化扫描频率
   - 配置重试机制

2. 代码修复
   - 修复协议实现错误
   - 优化数据处理逻辑
   - 增强错误处理

3. 监控部署
   - 部署持续监控
   - 设置报警规则
   - 建立维护流程

标准排查流程:
1. 快速检查 (5分钟)
   - 网络连通性
   - 设备电源状态
   - 基本通信测试

2. 详细分析 (30分钟)
   - 协议分析
   - 性能测试
   - 日志分析

3. 深度诊断 (2小时)
   - 模拟测试
   - 对比测试
   - 环境隔离

4. 解决方案 (根据复杂度)
   - 配置调整
   - 软件更新
   - 硬件更换