# 使用 Ziegler-Nichols 法进行 PID 参数整定：实践指南

PID（比例-积分-微分）控制器广泛应用于工业控制中。虽然其结构简单，但如何选择合适的比例增益 $ K_p $、积分时间 $ T_i $ 和微分时间 $ T_d $ 是实现良好控制性能的关键。本文将详细介绍一种经典且实用的 PID 参数整定方法——**Ziegler-Nichols 法**。



---

## 一、Ziegler-Nichols 方法简介

Ziegler-Nichols 方法是一种基于系统临界响应的实验整定方法，由 John G. Ziegler 和 Nathaniel B. Nichols 在 20 世纪 40 年代提出。该方法无需对系统进行复杂的数学建模，适用于大多数线性或近似线性的动态系统。

### 核心思想：

通过引入一个“临界状态”来获取系统的动态特性，并据此推荐一组初始 PID 参数。所谓“临界状态”，指的是使用纯比例控制时，系统刚刚开始出现等幅持续振荡的状态。

---

## 二、关键概念

### 1. 临界增益 $ K_u $

使系统产生等幅持续振荡的最小比例增益值。此时系统处于边界稳定状态。

### 2. 临界周期 $ T_u $

在临界增益 $ K_u $ 下，系统输出的等幅振荡周期。即相邻两个波峰之间的时间间隔。

这两个参数反映了系统的固有动态特性，是后续 PID 参数设计的基础。

---

## 三、Ziegler-Nichols 方法实施步骤

### 第一步：设定控制系统为纯比例控制模式

关闭积分和微分作用，只保留比例部分：

```python
output = Kp * error
```

### 第二步：逐步增加 $ K_p $

从一个小值开始，逐渐增大 $ K_p $，直到系统输出出现稳定的等幅振荡。

> 注意：此过程应在仿真或受控环境中进行，以避免设备损坏。


![image.png](/static/img/01e3c1130ecccc83b75ffafd2847033b.image.webp)


### 第三步：记录临界参数

- 记录下使系统恰好进入等幅振荡的增益值，记作 $ K_u $
- 测量等幅振荡的周期，记作 $ T_u $

例如：
- 系统在 $ K_p = 5.2 $ 时开始振荡
- 振荡周期约为 1.6 秒  
→ 则 $ K_u = 5.2 $, $ T_u = 1.6 \, \text{s} $

### 第四步：根据 ZN 表格设置 PID 参数

| 控制类型 | $ K_p $      | $ T_i $        | $ T_d $          |
|----------|----------------|------------------|--------------------|
| P        | $ 0.5 K_u $  | —                | —                  |
| PI       | $ 0.45 K_u $ | $ 0.83 T_u $   | —                  |
| PID      | $ 0.6 K_u $  | $ 0.5 T_u $    | $ 0.125 T_u $    |

代入示例数据：

- $ K_u = 5.2 $, $ T_u = 1.6 $
- 得到 PID 参数：
  - $ K_p = 0.6 × 5.2 = 3.12 $
  - $ T_i = 0.5 × 1.6 = 0.8 $
  - $ T_d = 0.125 × 1.6 = 0.2 $

---

## 四、注意事项与适用场景

### ✅ 优点：

- 不需要系统模型即可调参；
- 快速获得一组有效的 PID 参数；
- 实现简单，适合工程人员快速部署。

### ❌ 缺点：

- 对于非线性、强扰动或存在延迟的系统效果有限；
- 可能导致超调较大；
- 需要人为判断等幅振荡状态，存在主观误差。

### 📌 适用对象：

- 线性或弱非线性系统；
- 响应较快、无严重延迟的过程；
- 工业现场调试初期快速整定参数。

---

## 五、改进与扩展建议

- **积分限幅（Anti-Windup）**：防止积分项过大导致过冲；
- **带滤波的微分项**：增强抗噪声能力；
- **结合频域分析工具**：如 Bode 图、Nyquist 图，辅助参数选取；
- **模糊 PID / 自适应 PID**：用于复杂系统或多工况控制。

---

## 六、结语

Ziegler-Nichols 方法是一种经典的 PID 参数整定手段，尤其适用于缺乏精确系统模型或需快速部署控制算法的场景。虽然它不能保证最优性能，但提供了一个良好的起点。在实际应用中，建议结合具体系统响应进一步手动或自动优化 PID 参数，以达到更佳的控制效果。

使用 Ziegler-Nichols 法进行 PID 参数整定：实践指南

# 基于MQTT的设备监控与控制系统设计

## 引言

物联网(IoT)设备的远程监控与控制是现代智能系统的基础需求。本文将介绍一个基于MQTT协议的设备监控与控制系统，该系统由两部分组成：模拟单片机设备和PyQt客户端。我们将详细讨论系统的设计思路、代码实现以及实际应用场景。



## 系统架构

该系统基于发布-订阅模式，采用MQTT协议实现设备与客户端之间的双向通信。系统结构如下：

1. **设备模拟器**：模拟单片机设备，定期发送GPS位置数据，并接收控制命令
2. **PyQt客户端**：显示设备GPS数据，并提供用户界面发送控制命令
3. **MQTT服务器**：充当消息代理，负责转发消息

我在之前已经介绍了如何使用EMQT，这里不再重复，有关信息查看这里：

https://www.dong-blog.fun/post/1963

https://www.dong-blog.fun/post/2049

## 通信协议设计

系统使用两个主题进行通信：

1. **设备数据主题**(`device/gps`)：设备向此主题发布GPS数据
   ```json
   {
     "device_id": "device_001",
     "timestamp": 1692345678,
     "latitude": 30.056789,
     "longitude": 120.123456
   }
   ```

2. **控制指令主题**(`device/control`)：客户端向此主题发布控制命令
   - 按钮控制命令：
     ```json
     {
       "command": "UP",
       "timestamp": 1692345700
     }
     ```
   - 自定义文本命令：
     ```json
     {
       "text": "自定义指令内容",
       "timestamp": 1692345720
     }
     ```

通过区分`command`和`text`键，系统能够清晰地区分不同类型的控制信号。

## 设备模拟器实现

设备模拟器(`device_simulator.py`)主要功能是：

1. 模拟GPS数据采集并发布到MQTT服务器
2. 接收并处理控制命令

```python
import paho.mqtt.client as mqtt
import json
import time
import random

# MQTT broker configuration
broker = "10.100.80.98"
port = 1883
topic = "device/gps"
control_topic = "device/control"

# Create MQTT client with Version 2 API
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

def on_connect(client, userdata, flags, rc, properties=None):
    print(f"Connected with result code {rc}")
    # Subscribe to control topic
    client.subscribe(control_topic)
    print(f"已订阅控制主题: {control_topic}")

def on_message(client, userdata, msg, properties=None):
    try:
        data = json.loads(msg.payload.decode())
        print(f"收到控制数据: {data}")
        
        # 处理不同类型的消息
        if "command" in data:
            command = data["command"]
            print(f"执行命令: {command}")
            
            # 这里可以添加对命令的具体响应逻辑
            if command == "UP":
                print("设备向上移动")
            elif command == "DOWN":
                print("设备向下移动")
            elif command == "LEFT":
                print("设备向左移动")
            elif command == "RIGHT":
                print("设备向右移动")
            elif command == "CONFIRM":
                print("设备确认操作")
            elif command == "CANCEL":
                print("设备取消操作")
        # 处理自定义文本消息
        elif "text" in data:
            text = data["text"]
            print(f"收到自定义文本: {text}")
            # 处理文本命令的逻辑
            print(f"设备执行文本命令: {text}")
    except Exception as e:
        print(f"处理消息时出错: {e}")

# Set callback functions
client.on_connect = on_connect
client.on_message = on_message

# Connect to broker
try:
    client.connect(broker, port, 60)
    # Start the network loop
    client.loop_start()
    print(f"已连接到MQTT服务器: {broker}:{port}")
except Exception as e:
    print(f"连接MQTT服务器失败: {e}")
    exit(1)

try:
    device_id = "device_001"
    print(f"设备 {device_id} 模拟器开始运行")
    print("按Ctrl+C停止程序")
    
    while True:
        # Generate random GPS coordinates around a specific location
        # Base coordinates: 30.0°N, 120.0°E
        latitude = 30.0 + random.uniform(-0.1, 0.1)
        longitude = 120.0 + random.uniform(-0.1, 0.1)
        
        # Create message payload
        message = {
            "device_id": device_id,
            "timestamp": int(time.time()),
            "latitude": round(latitude, 6),
            "longitude": round(longitude, 6)
        }
        
        # Publish message
        client.publish(topic, json.dumps(message))
        print(f"已发送GPS数据: 纬度={message['latitude']}, 经度={message['longitude']}")
        
        # Wait for 2 seconds before sending next message
        time.sleep(2)

except KeyboardInterrupt:
    print("\n接收到停止信号，正在停止设备模拟器...")
    client.loop_stop()
    client.disconnect()
    print("设备模拟器已停止")
except Exception as e:
    print(f"运行时错误: {e}")
    client.loop_stop()
    client.disconnect() 
```

设备模拟器使用随机生成的GPS坐标数据，模拟设备移动的场景，同时能够对不同类型的命令做出响应。

日志信息：

![image.png](/static/img/f16b0cd67edc4cfeece4924145b41fda.image.webp)

## PyQt客户端实现

PyQt客户端(`gps_monitor.py`)提供了图形界面，主要功能包括：

1. 显示设备GPS数据
2. 提供方向控制按钮
3. 支持自定义文本命令输入和发送

客户端的核心在于MQTT与PyQt的集成，我们使用信号-槽机制来解决多线程问题：

```python
import sys
import json
import time
import paho.mqtt.client as mqtt
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, 
                            QHBoxLayout, QPushButton, QTextEdit, QLabel)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject

# 创建一个MQTT信号处理类
class MQTTSignals(QObject):
    message_received = pyqtSignal(dict)

class GPSMonitor(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("GPS Monitor and Controller")
        self.setGeometry(100, 100, 800, 600)
        
        # MQTT setup
        self.broker = "10.100.80.98"
        self.port = 1883
        self.gps_topic = "device/gps"
        self.control_topic = "device/control"
        
        # 创建信号对象
        self.mqtt_signals = MQTTSignals()
        self.mqtt_signals.message_received.connect(self.update_display)
        
        self.setup_mqtt()
        self.setup_ui()
        
    def setup_mqtt(self):
        # 使用MQTT V5客户端
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        
        try:
            self.client.connect(self.broker, self.port, 60)
            self.client.loop_start()
        except Exception as e:
            print(f"MQTT连接错误: {e}")
        
    def on_connect(self, client, userdata, flags, rc, properties=None):
        print(f"Connected with result code {rc}")
        self.client.subscribe(self.gps_topic)
        
    def on_message(self, client, userdata, msg, properties=None):
        try:
            data = json.loads(msg.payload.decode())
            # 使用信号发送数据，不直接更新UI
            self.mqtt_signals.message_received.emit(data)
        except Exception as e:
            print(f"Error processing message: {e}")
            
    def setup_ui(self):
        # Main widget and layout
        main_widget = QWidget()
        self.setCentralWidget(main_widget)
        layout = QVBoxLayout(main_widget)
        
        # GPS display area
        self.gps_display = QTextEdit()
        self.gps_display.setReadOnly(True)
        layout.addWidget(self.gps_display)
        
        # Control buttons layout
        button_layout = QHBoxLayout()
        
        # Direction buttons
        self.btn_up = QPushButton("上")
        self.btn_down = QPushButton("下")
        self.btn_left = QPushButton("左")
        self.btn_right = QPushButton("右")
        self.btn_confirm = QPushButton("确认")
        self.btn_cancel = QPushButton("取消")
        
        # Add buttons to layout
        button_layout.addWidget(self.btn_up)
        button_layout.addWidget(self.btn_down)
        button_layout.addWidget(self.btn_left)
        button_layout.addWidget(self.btn_right)
        button_layout.addWidget(self.btn_confirm)
        button_layout.addWidget(self.btn_cancel)
        
        # Command input
        self.command_input = QTextEdit()
        self.command_input.setMaximumHeight(100)
        layout.addWidget(QLabel("命令输入:"))
        layout.addWidget(self.command_input)
        
        # Send button
        self.btn_send = QPushButton("发送")
        layout.addWidget(self.btn_send)
        
        # Add button layout to main layout
        layout.addLayout(button_layout)
        
        # Connect button signals
        self.btn_up.clicked.connect(lambda: self.send_command("UP"))
        self.btn_down.clicked.connect(lambda: self.send_command("DOWN"))
        self.btn_left.clicked.connect(lambda: self.send_command("LEFT"))
        self.btn_right.clicked.connect(lambda: self.send_command("RIGHT"))
        self.btn_confirm.clicked.connect(lambda: self.send_command("CONFIRM"))
        self.btn_cancel.clicked.connect(lambda: self.send_command("CANCEL"))
        self.btn_send.clicked.connect(self.send_custom_command)
        
    def update_display(self, data):
        display_text = f"时间戳: {data['timestamp']}\n"
        display_text += f"设备ID: {data['device_id']}\n"
        display_text += f"纬度: {data['latitude']}\n"
        display_text += f"经度: {data['longitude']}\n"
        display_text += "-" * 50 + "\n"
        
        current_text = self.gps_display.toPlainText()
        self.gps_display.setText(display_text + current_text)
        
    def send_command(self, command):
        message = {
            "command": command,
            "timestamp": int(time.time())
        }
        self.client.publish(self.control_topic, json.dumps(message))
        print(f"已发送命令: {command}")
        
    def send_custom_command(self):
        command = self.command_input.toPlainText().strip()
        if command:
            # 使用"text"作为键名而不是"command"
            message = {
                "text": command,
                "timestamp": int(time.time())
            }
            self.client.publish(self.control_topic, json.dumps(message))
            self.command_input.clear()
            print(f"已发送自定义文本: {command}")
    
    def closeEvent(self, event):
        """窗口关闭时的处理"""
        self.client.loop_stop()
        self.client.disconnect()
        event.accept()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = GPSMonitor()
    window.show()
    sys.exit(app.exec_()) 
```

这种设计解决了MQTT回调线程与GUI线程的冲突问题，确保了应用程序的稳定性。

PyQt运行界面：


![image.png](/static/img/fbc705192e8b8b6cfae3e7b4ed5b0c00.image.webp)


## 线程安全设计

MQTT客户端的回调函数在非GUI线程中执行，而PyQt应用程序要求UI更新必须在主线程中进行。为了解决这个问题，我们采用了Qt的信号-槽机制：

1. 创建一个`MQTTSignals`类，定义`message_received`信号
2. 在MQTT回调函数中发射信号
3. 将信号连接到GUI线程中的槽函数

这种方式确保了线程安全，避免了直接在回调函数中更新UI导致的崩溃问题。

## 系统扩展性

该系统具有良好的扩展性：

1. **多设备支持**：通过增加设备ID识别，可轻松扩展为多设备系统
2. **协议扩展**：可以增加新的消息类型和命令
3. **功能增强**：可以添加历史数据记录、轨迹显示、报警功能等


## 系统运行效果

系统运行后，设备模拟器会每2秒发送一次GPS数据，PyQt客户端会实时显示这些数据，并可通过按钮或自定义文本发送控制命令。设备模拟器收到命令后会打印相应的响应信息。



## 源码与使用方法

使用前需要安装必要的依赖：

```bash
pip install paho-mqtt PyQt5
```

要运行系统，请先启动设备模拟器，然后运行PyQt客户端：

```bash
python device_simulator.py
python gps_monitor.py
```

利用EMQX实现单片机和PyQt的数据MQTT互联

之前博客介绍了如何搭建EMQX：https://www.dong-blog.fun/post/1963


本博客在 EMQX 中配置规则将数据写入 MySQL，可以通过 规则引擎 + 数据桥接 实现。以下是详细步骤：




## 一、准备工作
1. 确保 MySQL 已就绪：

docker 启动一个mysql数据库：


```bash
mkdir mysql 
cd mysql

docker run -d \
  --name mysql_for_emqx \
  -p 3306:3306 \
  -e MYSQL_ROOT_PASSWORD=yourpassword \
  -e MYSQL_DATABASE=emqx_data \
  -v $PWD/mysql_data:/var/lib/mysql \
  --restart unless-stopped \
  docker.1ms.run/mysql:8.0 \
  --character-set-server=utf8mb4 \
  --collation-server=utf8mb4_unicode_ci
```

进入mysql数据库交互终端：


```
# 进入MySQL交互终端
docker exec -it mysql_for_emqx mysql -h 127.0.0.1 -uroot -pyourpassword
```

• 创建数据库和表（示例表结构）：

 ```sql
CREATE DATABASE emqx_data; # 这句不用执行，docker run的时候已经创建了此数据库
USE emqx_data;
SHOW TABLES;
CREATE TABLE mqtt_messages (
     id INT AUTO_INCREMENT PRIMARY KEY,
     topic VARCHAR(255),
     payload TEXT,
     qos TINYINT,
     client_id VARCHAR(255),
     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
 );
 SHOW TABLES;
 ```
看到这个就是表已经创建了：

```
mysql> SHOW TABLES;
+---------------------+
| Tables_in_emqx_data |
+---------------------+
| mqtt_messages       |
+---------------------+
1 row in set (0.00 sec)

```

2. 获取 MySQL 连接信息：
   • 主机：`mysql_host`（如 `127.0.0.1`）

   • 端口：`3306`

   • 用户名：`root`

   • 密码：`yourpassword`

   • 数据库名：`emqx_data`


---

## 二、通过 EMQX Dashboard 配置（推荐）
1. 登录 Dashboard：
   • 访问 `http://10.100.80.98:18083`，输入账号密码（默认 `admin/public`）。


2. 创建规则：


   • 进入 规则引擎 → 规则 → 创建。

   • SQL 示例（匹配所有test消息）：

 ```sql
SELECT 
    topic, 
    payload, 
    qos, 
    clientid AS client_id 
FROM 
    "test/#"
 ```
   • 点击 测试 验证规则是否能匹配到 MQTT 消息。

如何测试自己的SQL是不是写对了呢，这样搞，先把启用调试打开，然后设置好主题，然后点击运行调试，在下图这个设置里，相当于我用客户端发了消息到主题（test/haha）。可以看到右边有结果`{"client_id":"c_emqx","payload":"{\"msg\": \"hello\"}","qos":1,"topic":"test/haha"}` ，说明规则生效了。然后点保存即可。


![image.png](/static/img/0465f02a27f1cfc4c9ccae64a3e6c733.image.webp)




3. 添加 MySQL 数据桥接：

本来应该这么做：
```
• 在规则详情页点击 添加动作 → 数据桥接 → MySQL。
• 填写 MySQL 连接信息：


 名称：mysql_bridge
 模式：异步（高性能）
 MySQL 服务器：mysql_host:3306
 数据库：emqx_data
 用户名：root
 密码：yourpassword
 SQL 模板：INSERT INTO mqtt_messages(topic, payload, qos, client_id) VALUES(${topic}, ${payload}, ${qos}, ${client_id})

• 关键参数说明：

 ◦ `${topic}`、`${payload}` 等是规则 SQL 输出的字段。

 ◦ 如果字段名冲突，可用 `${payload.key}` 提取 JSON 中的嵌套字段。
```


但是在免费的EMQX里面，没有数据桥接的功能，数据桥接的功能是企业版要收费的。

所以只能这么做》》》》

首先点击添加动作：

![image.png](/static/img/31d7c310963740bf2e47d2ff94dfe797.image.webp)

然后写个python flask代码，启动一个http api接口，用于接收消息并把数据写数据库里去，我在本地直接启动下面这个代码。这个代码会开启一个端口5000的http api接口服务，路由是 /mqtt 。当有请求携带数据POST访问 /mqtt 接口后，这个接口会解析传入的data里的数据`topic, payload, qos, client_id`，并写入到数据库里。


```python
# 安装依赖：pip install flask pymysql
from flask import Flask, request
import pymysql
from datetime import datetime

app = Flask(__name__)

@app.route('/mqtt', methods=['POST'])
def handle_mqtt():
    try:
        data = request.json
        print("Received data:", data)  # 调试用
        
        # 连接MySQL（参数需与你的MySQL容器匹配）
        conn = pymysql.connect(
            host='127.0.0.1',  # 如果是Docker容器间通信，改用容器名如'mysql_for_emqx'
            port=3306,
            user='root',
            password='yourpassword',
            database='emqx_data'
        )
        
        # 写入数据
        with conn.cursor() as cursor:
            sql = """
            INSERT INTO mqtt_messages (topic, payload, qos, client_id)
            VALUES (%s, %s, %s, %s)
            """
            cursor.execute(sql, (
                data['topic'],
                str(data['payload']),
                data['qos'],
                data['client_id']
            ))
        conn.commit()
        return {"status": "success"}, 200
        
    except Exception as e:
        print("Error:", e)
        return {"status": "error", "message": str(e)}, 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)
```
在本地测试这个flask服务是否成功：

```bash
curl -X POST http://127.0.0.1:5000/mqtt -H "Content-Type: application/json" -d '{
    "topic": "test/123",
    "payload": "hello",
    "qos": 1,
    "client_id": "client_1"
}'
```

看来是ok的：

![image.png](/static/img/1fff94cbff6ee1789686391f6e8d9d27.image.webp)

curl访问接口，并把数据传入了flask，flask代码里写入到了数据库。我们可以查看mysql的表里是不是多了数据：
```
# 进入MySQL交互终端
docker exec -it mysql_for_emqx mysql -h 127.0.0.1 -uroot -pyourpassword

USE emqx_data;

SELECT * FROM emqx_data.mqtt_messages;
```
看到这个信息就是可以了,可以看到表里面有个数据：

```bash
mysql> USE emqx_data;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> SELECT * FROM emqx_data.mqtt_messages;
+----+----------+---------+------+-----------+---------------------+
| id | topic    | payload | qos  | client_id | created_at          |
+----+----------+---------+------+-----------+---------------------+
|  1 | test/123 | hello   |    1 | client_1  | 2025-04-28 08:00:05 |
+----+----------+---------+------+-----------+---------------------+
1 row in set (0.00 sec)


```

添加 HTTP 的连接器，URL写flask的ip和接口，测试连接显示连接成功，则可以保存。在下图里连接器名称我写了mysql_server，这是随便自己写的。URL我写了`http://10.100.80.98:5000` ，这是我的ip和flask的服务端口。

![image.png](/static/img/3ed19ae125e6ea69efec2048ff47a7de.image.webp)

添加动作。名称可以随便写。URL路径写`/mqtt` ， 连接器选刚才创建的连接器。

请求体这么写：

```
{
    "topic": "${topic}",
    "payload": "${payload}",
    "qos": ${qos},
    "client_id": "${client_id}"
}
```

点测试连接，OK的就没问题，保存好（要么保存要么更新，别忘记保存，优先选更新）。如下图：


![image.png](/static/img/6e2b03133985d7a45e4f6d1bac28a9e9.image.webp)


## 三、调试

简单一点，直接模拟个mqtt客户端，往broker主题上发数据。装好工具：

```
sudo apt-get update
sudo apt-get install mosquitto-clients  # 只安装客户端工具（不安装服务端）
```

直接发一条数据：
```bash
mosquitto_pub \
-h 10.100.80.98 \
-p 1883 \
-t "test/abc" \
-m "xxxxxddddd" \
-q 1 
```

这个客户端往broker的主题`"test/abc"` 发了消息，消息是`"xxxxxddddd"` 。发了之后，由broker的规则接收到，并通过动作访问了flask，flask代码收到数据会存到mysql里。在emqx里可以进入这里看数据面板：


![image.png](/static/img/5792a2f987be1172e7f26c6aff78e6d9.image.webp)


也可以看看当前数据库是否有这条数据，有这条数据说明一切ok：

![image.png](/static/img/1746b367f6bb4642bc34c0ed3cb4efde.image.webp)



如果你不想装`mosquitto-clients` ，也可以用这个代码去直接发布消息，也会在数据库看到此消息。
```
import paho.mqtt.client as mqtt

# 定义MQTT服务器地址和端口
broker = "10.100.80.98"
port = 1883

# 创建MQTT客户端
client = mqtt.Client()

# 连接到MQTT服务器
client.connect(broker, port, 60)

# 发布消息
topic = "test/abc"
message = "Hello, EMQX!"
client.publish(topic, message)

# 断开连接
client.disconnect()
```

实现 EMQX 到 MySQL 的数据持久化

论文：https://arxiv.org/pdf/2407.17490
https://github.com/YuxiangChai/AMEX-codebase/tree/main/data_utils
https://huggingface.co/datasets/Yuxiang007/AMEX

![image.png](/static/img/13ad854c509e3840241e3af0b36a42ce.image.webp)



AMEX数据集包括三个层次的注释：
1. GUI交互元素定位 ：
  - 分类两类交互元素：可点击元素和可滚动元素。
2. GUI屏幕和元素功能描述 ：
  - 使用GPT生成屏幕和元素的功能描述，并进行人工检查以确保准确性。
3. 复杂自然语言指令与GUI动作链 ：
  - 每条指令平均包含12.8个步骤的动作链，显著高于现有数据集。





### 数据实操

下载数据后，合并并解压数据：

```
zip --fix screenshot.zip --out screenshot_merged.zip



unzip screenshot_merged.zip # 解压


# 或者
sudo apt install -y p7zip-full
7z x -mmt=on screenshot.zip  # 启用多线程解压
```



AMEX数据集包括三个层次的注释：
1. GUI交互元素定位 ：
  - 分类两类交互元素：可点击元素和可滚动元素。
2. GUI屏幕和元素功能描述 ：
  - 使用GPT生成屏幕和元素的功能描述，并进行人工检查以确保准确性。
3. 复杂自然语言指令与GUI动作链 ：
  - 每条指令平均包含12.8个步骤的动作链，显著高于现有数据集。

下载数据后，合并并解压数据：
zip --fix screenshot.zip --out screenshot_merged.zip
unzip screenshot_merged.zip
数据统计

Total JSON files processed: 3046
Total dictionaries (steps) found: 38709

Number of dictionaries per action type:
==================================================
PRESS_BACK: 135
PRESS_ENTER: 651
PRESS_HOME: 13
SWIPE: 7628
TAP: 24815
TASK_COMPLETE: 2828
TASK_IMPOSSIBLE: 220
TYPE: 2419


样本示例

总共有 8 种不同的动作类型：

1. SWIPE - 滑动操作，包含方向（UP/DOWN/LEFT/RIGHT）
2. TAP - 点击操作，包含坐标位置
3. TYPE - 输入文本操作
4. PRESS_BACK - 按下返回键
5. PRESS_HOME - 按下主页键
6. PRESS_ENTER - 按下回车键
7. TASK_COMPLETE - 任务完成
8. TASK_IMPOSSIBLE - 任务无法完成


暂时无法在飞书文档外展示此内容
让我为您解释这个样本的含义：
这是一个记录用户在宜家(IKEA)安卓应用上购买游戏椅的交互过程的数据样本。让我们从几个关键部分来分析：
1. 任务描述：
- instruction: "我想买一把游戏椅。将三个价格最低的产品保存到愿望清单，如果可能的话选择黑色。"
- episode_id: 一个唯一的标识符，用于追踪这个特定的交互会话
2. 设备信息：
- 设备屏幕尺寸为 1440 x 2960 像素
- 应用包名为 hk.ikea.android（香港宜家安卓应用）
3. 交互步骤：
整个流程包含16个步骤，主要动作包括：
- 滑动(SWIPE)：用于浏览页面内容
- 点击(TAP)：选择特定项目
- 输入(TYPE)：搜索"gaming chair"
- 按回车(PRESS_ENTER)：确认搜索
- 任务完成(TASK_COMPLETE)：标记任务结束
4. 交互细节：
- 每个步骤都记录了精确的触摸坐标(touch_coord)和抬起坐标(lift_coord)
- 每个步骤都保存了对应的屏幕截图(image_path)
- 每个步骤都标记了是否需要人工检查(need_human_check)
5. 流程分析：
用户的操作流程大致是：
1. 打开应用
2. 搜索"gaming chair"
3. 浏览搜索结果
4. 选择三个价格最低的产品
5. 将选中的产品添加到愿望清单
6. 完成任务
动作类型示例
Action Type: SWIPE
==================================================

Sample 1:
{
  "step_id": 1,
  "action": "SWIPE",
  "touch_coord": [
    637,
    2567
  ],
  "lift_coord": [
    564,
    1390
  ],
  "device_dim": [
    1440,
    3040
  ],
  "package_name": "com.android.launcher3",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_28_15_40_b31f339a2d6441bb994b4a30f184a267-1.png"
}

Action Type: TAP
==================================================

Sample 1:
{
  "step_id": 3,
  "action": "TAP",
  "touch_coord": [
    457,
    1633
  ],
  "lift_coord": [
    457,
    1633
  ],
  "device_dim": [
    1440,
    2960
  ],
  "package_name": "com.android.launcher3",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_22_18_56_6623dffea11a48f2955bafde23a3f1c7-3.png"
}
Action Type: TYPE
==================================================

Sample 1:
{
  "step_id": 4,
  "action": "TYPE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    2960
  ],
  "package_name": "com.amazon.mShop.android.shopping",
  "type_text": "Nike Basketball Shoes",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_5_14_15_40_ca5bacfe2a574da4ac05174b973cf321-4.png"
}

Action Type: PRESS_BACK
==================================================

Sample 1:
{
  "step_id": 4,
  "action": "PRESS_BACK",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    2960
  ],
  "package_name": "com.discord",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_5_13_16_3_61ec5c908f0f4969b460fb47bf9eb054-4.png"
}
Action Type: PRESS_ENTER
==================================================

Sample 1:
{
  "step_id": 4,
  "action": "PRESS_ENTER",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3120
  ],
  "package_name": "com.microsoft.office.outlook",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_5_8_15_56_d950b959ebad431a84368d7b63da755d-4.png"
}

Action Type: TASK_IMPOSSIBLE
==================================================

Sample 1:
{
  "step_id": 5,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3120
  ],
  "package_name": "com.espn.score_center",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_26_0_42_87159d38462b4ffdb7fee96370eea8cf-5.png"
}

Sample 2:
{
  "step_id": 6,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3040
  ],
  "package_name": "musclebooster.workout.home.gym.abs.loseweight",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_22_16_39_72e6b2de98254c44b9295e165567ac83-6.png"
}

Sample 3:
{
  "step_id": 10,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3120
  ],
  "package_name": "com.podcast.podcasts",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_26_15_24_95f480ce21604f1792190e5bfef54549-10.png"
}

Sample 4:
{
  "step_id": 15,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3040
  ],
  "package_name": "com.agoda.mobile.consumer",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_28_15_41_f7039f20e9944e92bdada777525a2268-15.png"
}

Sample 5:
{
  "step_id": 12,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3040
  ],
  "package_name": "musclebooster.workout.home.gym.abs.loseweight",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_22_16_27_3d79d989a7fd4f22a7cf0ce44ec8d47e-12.png"
}

Action Type: PRESS_HOME
==================================================

Sample 1:
{
  "step_id": 13,
  "action": "PRESS_HOME",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    2960
  ],
  "package_name": "com.seatgeek.android",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_3_20_14_36_21b637d11bea46b8adb3c2efc9f03501-13.png"
}

Action Type: TASK_COMPLETE
==================================================

Sample 1:
{
  "step_id": 18,
  "action": "TASK_COMPLETE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3120
  ],
  "package_name": "com.microsoft.teams",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_24_18_25_f100d55037f04908900699457c6f5676-18.png"
}

Action Type: TASK_IMPOSSIBLE
==================================================

Sample 1:
{
  "step_id": 5,
  "action": "TASK_IMPOSSIBLE",
  "touch_coord": [
    0,
    0
  ],
  "lift_coord": [
    0,
    0
  ],
  "device_dim": [
    1440,
    3120
  ],
  "package_name": "com.espn.score_center",
  "type_text": "",
  "need_human_check": false,
  "interest_region": [
    [
      0,
      0
    ],
    [
      0,
      0
    ]
  ],
  "image_path": "2024_4_26_0_42_87159d38462b4ffdb7fee96370eea8cf-5.png"
}

AMEX vivo UI agents 数据集介绍

在 Ubuntu 22.04 上，我们经常需要查看当前的网络带宽占用情况，尤其是下载大文件时，了解实时的下载速度可以帮助我们判断网络状况或管理下载任务。本文将介绍几种简单有效的方法来监控 Ubuntu 22.04 的下载带宽，包括终端工具和图形界面方案。



---

**1. 使用 `nload`（推荐）**
`nload` 是一个轻量级的命令行工具，可以直观地显示网络流量，包括下载和上传速度。

**安装 `nload`**
```bash
sudo apt update && sudo apt install nload
```

**运行 `nload`**
```bash
nload
```
• 界面说明：

  • Incoming (下载速度)：右侧显示当前下载带宽（默认单位 `Mb/s`）。

  • Outgoing (上传速度)：左侧显示上传带宽。

  • 按 `↑`/`↓` 切换网卡（如 `eth0`、`wlan0`）。

  • 按 `q` 退出。


**指定单位（如 MB/s）**
```bash
nload -u M
```
• `-u M` 表示以 `MB/s` 显示（1 MB = 8 Mb）。


---

**2. 使用 `iftop`（按连接监控）**
`iftop` 可以显示每个网络连接的带宽占用情况，适合查看哪些 IP/端口占用了带宽。

**安装 `iftop`**
```bash
sudo apt install iftop
```

**运行 `iftop`**
```bash
sudo iftop -i wlan0  # 替换为你的网卡名（如 eth0）
```
• 界面说明：

  • 顶部显示实时带宽（下载/上传）。

  • 底部列出占用带宽的 IP 和端口。

  • 按 `P` 切换端口显示，`q` 退出。


---

**3. 使用 `vnstat`（长期统计）**
`vnstat` 可以记录历史网络使用情况，并支持实时监控。

**安装 `vnstat`**
```bash
sudo apt install vnstat
```

**实时监控**
```bash
vnstat -l
```
• 按 `Ctrl+C` 退出。


---

**4. 使用系统自带工具**
**终端命令（`/proc/net/dev`）**
```bash
watch -n 1 "cat /proc/net/dev | grep wlan0"
```
• 每 1 秒刷新一次，显示 `wlan0`（无线网卡）的流量统计。

• 计算下载速度：观察 `RX bytes`（接收字节）的变化，手动计算差值。


**图形界面（GNOME 系统监视器）**
1. 打开 “系统监视器”（`gnome-system-monitor`）。
2. 切换到 “资源” 标签页。
3. 查看 “网络” 部分的实时流量图表。

---

**5. 使用 `bmon`（高级监控）**
`bmon` 提供更详细的网络统计信息，支持图形化显示。

**安装 `bmon`**
```bash
sudo apt install bmon
```

**运行 `bmon`**
```bash
bmon
```
• 按 `d` 切换视图，`q` 退出。


---

**总结**
| 工具 | 特点 | 适用场景 |
|------|------|---------|
| `nload` | 简单直观，显示实时带宽 | 快速查看下载/上传速度 |
| `iftop` | 按连接显示带宽占用 | 排查哪个 IP/进程占用带宽 |
| `vnstat` | 长期统计网络使用 | 记录历史流量 |
| `/proc/net/dev` | 系统原生方法 | 手动计算流量 |
| `bmon` | 高级网络监控 | 详细统计和图形化显示 |
| GNOME 系统监视器 | 图形界面 | 适合桌面用户 |

---

**常见问题**
**Q1: 如何知道我的网卡名称？**
```bash
ip a
```
• 查找类似 `eth0`（有线）或 `wlan0`（无线）的网卡。


**Q2: 为什么 `nload` 显示的单位是 Mb/s，而下载工具显示 MB/s？**
• `1 MB/s = 8 Mb/s`（1 字节 = 8 比特）。

• 使用 `nload -u M` 可以切换为 `MB/s`。


---

**结语**
在 Ubuntu 22.04 上，监控下载带宽有多种方法，从简单的 `nload` 到详细的 `iftop` 和 `bmon`，你可以根据自己的需求选择合适的工具。希望这篇指南能帮助你更好地管理网络流量！🚀

​​Ubuntu 22.04 如何实时监控下载带宽（文件下载速度）​

我是一个专注技术分享的博主,尤其爱搞深度学习和单片机电子。你现在看到的是我的个人博客网站。