# 仅捕获Modbus TCP流Quantität
tcp port 502

# 捕获特定Ausrüstung的Kommunikation
host 192.168.1.100 and tcp port 502

# 排除Rundfunk流Quantität
not broadcast and not multicast

# 基本过滤
modbus                    # 所有Modbus流Quantität
modbus.func_code == 0x03  # 仅Lesen und halten Register
modbus.func_code == 0x10  # 仅Schreiben in mehrere Register

# 高级过滤
modbus.unit_id == 1       # 特定Von der StationAusrüstung
modbus.trans_id == 1234   # 特定事务ID
tcp.analysis.flags        # TCPUnterteilung析标志

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

# Fehler.Antwort - 红色
modbus.excep_code != 0x00

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

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

1. 过滤Regeln: modbus.func_code == 0x03 || modbus.func_code == 0x10
2. 验证方法:
   - 对比Bitte和AntwortDaten
   - 检查Register-Adresse连续性
   - 验证DatenDie Reihenfolge
3. Häufige Fragen:
   - BefehlseiteFehler. (Big-endian vs Little-endian)
   - Adresse偏移BerechnungFehler.
   - Datentyp转换问题

Original-Meldung: 01 03 00 00 00 0A C5 CD
Analyseergebnisse:
├── Geräte Adresse: 01
├── Funktionscode: 03 (Lesen und halten Register)
├── Ausgang Adresse: 0x0000
├── Anzahl der Register: 10
└── CRC-Test: 0xC5CD (Richtig.)

# Ausrüstung发现扫描
modbus-scanner -t tcp -a 1-247 -r 1-100 192.168.1.0/24

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

# Registerabbildung扫描
modbus-scanner -m mapping -s 0 -e 65535 -d 1

[Ausrüstung1] Temperatur-Sensor - 192.168.1.101:502
├── Register. 40001: 25.3°C (实时更新)
├── Register. 40002: 60.2% RH
└── 线圈 00001: ON

[Ausrüstung2] PLCDer Controller - 192.168.1.102:502
├── Register. 40010: 1200 RPM
├── Register. 40011: 3.2 MPa
└── 离散输入 10001: TRIGGERED

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

// Modbus PollDrehbuchBeispiel
function testSequence() {
    // 1. 初始化Verbindung
    connect("192.168.1.100", 502);

    // 2. 基本Funktion测试
    testReadCoils(0, 16, "线圈Lesen测试");
    testReadRegisters(40001, 10, "Register.Lesen测试");

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

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

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

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

设置断点条件:
- 特定Registerwert变化
- Fehler.Antwort发生
- Kommunikation超时
- Daten越界

监视列表配置:
$temp = [40001]  // TemperaturRegister.
$speed = [40010] // 转速Register.
$status = [00001] // Status线圈

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

# Temperatur波动模拟
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:
            # BetriebStatus模拟
            self.speed = 1500 + 100 * math.sin(cycle / 10)
            self.pressure = 3.2 + 0.1 * math.sin(cycle / 5)
        else:
            # Stopp!Status
            self.speed = 0
            self.pressure = 0

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

<test-scenario name="Kommunikation故障测试">
    <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

支持所有ModbusAdresseTypus:
- 0xxxx: 线圈 (000001-065536)
- 1xxxx: 离散输入 (100001-165536)  
- 3xxxx: 输入Register. (300001-365536)
- 4xxxx: 保持Register. (400001-465536)

# Register.配置Beispiel
address,type,value,description,min,max,unit
40001,float,25.3,Temperatur-Sensor,0,100,°C
40003,int32,1200,Elektrische Motoren转速,0,3000,RPM
40005,uint16,60,湿度百Unterteilung比,0,100,%
40007,bool,1,AusrüstungStatus,0,1,ON/OFF

-- Lua测试DrehbuchBeispiel
function test_sequence()
    -- 初始化测试
    print("开始ModbusAusrüstung测试...")

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

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

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

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

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

modpoll [选项] <主机名或IPAdresse>

# 扫描网络中的ModbusAusrüstung
modpoll -m tcp -a 1-247 -t 3 -r 1 -c 1 192.168.1.0/24

# 扫描Serielle Ausrüstung
modpoll -m rtu -a 1-247 -p /dev/ttyUSB0 -b 9600 -t 3 -r 1

# 快速Ausrüstung识别
modpoll -m tcp -a 1-10 -t 0x2B -r 0 -c 2 192.168.1.100

# Lesen und halten Register (Funktionscode03)
modpoll -m tcp -t 4 -r 40001 -c 10 192.168.1.100

# Eingangsregister lesen (Funktionscode04)
modpoll -m tcp -t 3 -r 30001 -c 5 192.168.1.100

# Lesen Sie den Zustand der Spule (Funktionscode01)
modpoll -m tcp -t 0 -r 1 -c 16 192.168.1.100

# Lesen von diskreten Eingaben (Funktionscode02)
modpoll -m tcp -t 1 -r 10001 -c 8 192.168.1.100

# Schreiben in ein einzelnes Register (Funktionscode06)
modpoll -m tcp -t 4:hex -r 40001 -c 1 192.168.1.100 0x1388

# Schreiben in mehrere Register (Funktionscode16)
modpoll -m tcp -t 4:hex -r 40010 -c 3 192.168.1.100 0x0001 0x0002 0x0003

# Schreiben Sie in eine einzelne Schleife (Funktionscode05)
modpoll -m tcp -t 0 -r 1 -c 1 192.168.1.100 ON

# Schreiben Sie mehrere Schleife (Funktionscode15)
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

# Daten aufzeichnenzu文件
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

# CSVDas Format输出
modpoll -m tcp -t 4 -r 40001 -c 10 -f c 192.168.1.100

# JSONDas Format输出
modpoll -m tcp -t 4 -r 40001 -c 5 -f j 192.168.1.100

# Custom definiertDas Format
modpoll -m tcp -t 4 -r 40001 -c 3 -f "Address: %a, Value: %v" 192.168.1.100

#!/bin/bash
# modpoll批Quantität测试Drehbuch

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

echo "开始ModbusAusrüstung测试..." | tee -a $LOG_FILE

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

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

# 测试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次Bitteabgeschlossen." | tee -a $LOG_FILE

# 生成测试报告
generate_report $LOG_FILE

# 增加超时ZeitÜberprüfenAntwort
modpoll -m tcp -t 4 -r 40001 -c 1 -o 5000 192.168.1.100

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

# 测试不同DatenQuantität
for size in 1 10 50 100 125; do
    echo "测试 $size ein Register...."
    modpoll -m tcp -t 4 -r 40001 -c $size -o 3000 192.168.1.100
done

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

# 写入后立即Lesen验证
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

Ausrüstung模型结构:
├── Kommunikationsparameter
│   ├── Vereinbarung: RTU/TCP/ASCII
│   ├── Porter Rate: 9600/19200/38400/115200
│   ├── Datenplatz: 8
│   ├── Stoppen Sie Platz: 1/2
│   └── Prüfstand.: None/Even/Odd
├── Daten点定义
│   ├── 线圈 (0xxxx)
│   ├── 离散输入 (1xxxx)
│   ├── 输入Register. (3xxxx)
│   └── 保持Register. (4xxxx)
└── Daten映射
    ├── 缩放因子
    ├── 偏移Quantität
    ├── Datentyp
    └── 工程单位

Echtzeitüberwachung界面:
─────────────────────────────────────
Ausrüstung: TemperaturDer Controller [192.168.1.101:502]
─────────────────────────────────────
Adresse      Name          值     单位   Status
40001     Temperatur        25.3    °C     ✅
40002     湿度        60.2    %      ✅  
40003     压力        101.3   kPa    ✅
00001     加热器       ON            ✅
10001     报警         OFF           ✅
─────────────────────────────────────
更新Zeit: 2026-02-24 17:15:32
KommunikationStatus: 正常 (延迟: 12ms)

{
  "dashboard": {
    "title": "生产线监控",
    "widgets": [
      {
        "type": "gauge",
        "title": "Temperatur监控",
        "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"
      }
    ]
  }
}

// 报警Regeln配置
alarmRules: [
  {
    name: "高温报警",
    condition: "temperature > 80",
    actions: [
      "sendEmail('operator@factory.com', '高温报警')",
      "sendSMS('+8613800138000', 'AusrüstungTemperatur过高')",
      "logEvent('high_temperature_alarm')"
    ],
    severity: "CRITICAL"
  },
  {
    name: "Kommunikation中断",
    condition: "deviceStatus == 'OFFLINE'",
    actions: [
      "notifyMaintenanceTeam()",
      "startBackupSystem()"
    ],
    severity: "HIGH"
  }
]

#!/usr/bin/env python3
"""
Modbus性能Unterteilung析Werkzeuge
"""

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):
        """Betrieb性能测试"""
        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:
                # 执行ModbusBitte
                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"Bitte {i+1} 失败: {e}")

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

            # 控制Bitte频率
            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 "没有测试Daten"

        rt = self.metrics['response_times']

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

AntwortZeit统计 (毫秒):
- 平均值: {statistics.mean(rt):.2f}
- 中位数: {statistics.median(rt):.2f}
- Der Mindestwert: {min(rt):.2f}
- Maximaler Wert: {max(rt):.2f}
- 标准schlecht: {statistics.stdev(rt):.2f}
- 95百Unterteilung位: {sorted(rt)[int(len(rt) * 0.95)]:.2f}
- 99百Unterteilung位: {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 += "- 性能较schlecht，需要重点优化\n"

        return report

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

        # AntwortZeit趋势
        axes[0, 0].plot(self.metrics['response_times'])
        axes[0, 0].set_title('AntwortZeit趋势')
        axes[0, 0].set_xlabel('BitteSeriennummer.')
        axes[0, 0].set_ylabel('AntwortZeit (ms)')
        axes[0, 0].grid(True)

        # AntwortZeitUnterteilung布
        axes[0, 1].hist(self.metrics['response_times'], bins=50, edgecolor='black')
        axes[0, 1].set_title('AntwortZeitUnterteilung布')
        axes[0, 1].set_xlabel('AntwortZeit (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('Bitte成功率')

        # 累积Unterteilung布函数
        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('AntwortZeitCDF')
        axes[1, 1].set_xlabel('AntwortZeit (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"图表已保存zu: {save_path}")

        plt.show()

# 使用Beispiel
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    # aktiviert.TCP保活
  keepalive_time: 60     # 保活Zeit(秒)
  keepalive_intvl: 10    # 保活间隔(秒)
  keepalive_probes: 3    # 保活探测次数

socket_options:
  so_reuseaddr: true     # Adresse重用
  so_rcvbuf: 65536       # Empfang缓冲区大小
  so_sndbuf: 65536       # Senden Sie缓冲区大小

# 批QuantitätLesen优化
def optimized_read(client, address_ranges):
    """
    优化Lesen策略：
    1. 合并相邻Adresse的LesenBitte
    2. 避免Lesen不必要的Daten
    3. 使用合适的超时Zeit
    """
    results = {}

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

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

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

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

        except Exception as e:
            log_error(f"Lesen gescheitert {start}-{start+count}: {e}")

    return results

class ModbusConnectionPool:
    """ModbusVerbindung池管理"""

    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):
        """获取Verbindung"""
        # 1. 检查空闲Verbindung
        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. 创建新Verbindung
        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. 等待Verbindung释放
        raise NoAvailableConnection("Verbindung池已满")

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

    def close_all(self):
        """Schließen所有Verbindung"""
        for conn in self.pool:
            conn.close()
        self.pool.clear()
        self.in_use.clear()

1. 现象收集
   - Kommunikation完全中断
   - Daten不一致
   - Antwort缓慢
   - 间歇性故障

2. 基本信息收集
   - 网络拓扑
   - Ausrüstung型Nr.和固件版本
   - 配置Parameter
   - Fehler.日志

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

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

# 3. 测试不同DatenQuantität
for count in 1 10 50 100; do
    echo "测试 $count ein Register...."
    modpoll -m tcp -a 1 -t 4 -r 40001 -c $count 192.168.1.100
done

1. Wireshark抓包Unterteilung析
   - 捕获vollständigKommunikation过程
   - Unterteilung析Meldung时序
   - 检查Vereinbarung合规性

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

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

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

2. 代码修复
   - 修复Vereinbarung实现Fehler.
   - 优化Daten处理逻辑
   - 增强Fehler.处理

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

标准排查流程:
1. 快速检查 (5Unterteilung钟)
   - 网络连通性
   - Ausrüstung电源Status
   - 基本Kommunikation测试

2. 详细Unterteilung析 (30Unterteilung钟)
   - VereinbarungUnterteilung析
   - 性能测试
   - 日志Unterteilung析

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

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