Python -- PyQt6+paho.mqtt 制作的MQTT桌面收发器（阿里云示范）
·
目录
一 前言
继续写，这篇文章实现了桌面应用的MQTT通信，我这里采用的时阿里云作为示范，前面上写了一篇桌面应用以及Mqtt部分的文章以及PyQt6的使用，里面提到了python的环境安装以及PyQt6环境配置，我现在就默认大家都装好了设备了
效果如下
二 环境安装
前面我们以及实现了桌面开发和mqtt库函数的使用，这边我们就直接引用一下
1. 编译器
可以查看这篇文章
https://herui.blog.csdn.net/article/details/144501509?spm=1001.2014.3001.55022. 环境库下载
可以查看这篇文章
三 代码编写
首先我们需要在阿里云里面建立我们的产品和设备，并且获取到相关的mqtt连接参数
可以参考这个文章里面的云平台部分
获取到两个设备的mqtt参数就好了
1. Python部分
这个代码实现Mqtt连接，并且实现设备的参数的自定义，实现了设备的重连功能以及自定义发送的主题和内容，实现了按键发送
把我们平台上面获取的内容修改进去
也可以直接运行之后修改上面的内容
import sys
from PyQt6.QtWidgets import QApplication, QWidget, QVBoxLayout, QLineEdit, QLabel, QPushButton, QGridLayout, QTextEdit, QCheckBox, QComboBox
import paho.mqtt.client as mqtt
import threading
import time
# MQTT 服务器设置（初始值）
MQTT_BROKER_IP = "iot-06z00axdhgfk24n.mqtt.iothub.aliyuncs.com"
MQTT_BROKER_PORT = 1883
MQTT_CLIENT_ID = "h9sjD6ci5EI.smartdevice|securemode=2,signmethod=hmacsha256,timestamp=1734329040945|"
MQTT_TOPIC_PUBLISH = "/broadcast/h9sjD6ci5EI/test1"
MQTT_TOPIC_SUBSCRIBE = "/broadcast/h9sjD6ci5EI/test2"
MQTT_USERNAME = "smartdevice&h9sjD6ci5EI"
MQTT_PASSWORD = "4d1a97eaee5c0c8bd5fdad2292f5a83239c2a21bcb280eb8bec8a28741549a9e"
# 全局变量来存储更新的 MQTT 参数
mqtt_params = {
"broker_ip": MQTT_BROKER_IP,
"broker_port": MQTT_BROKER_PORT,
"client_id": MQTT_CLIENT_ID,
"username": MQTT_USERNAME,
"password": MQTT_PASSWORD,
"publish_topic": MQTT_TOPIC_PUBLISH,
"subscribe_topic": MQTT_TOPIC_SUBSCRIBE,
}
# MQTT 回调函数：连接成功时触发
def on_connect(client, userdata, flags, rc):
print(f"Connected with result code {rc}")
client.subscribe(mqtt_params["subscribe_topic"]) # 订阅消息主题
# MQTT 回调函数：接收到消息时触发
def on_message(client, userdata, msg):
print(f"Received message on topic {msg.topic}: {msg.payload.decode()}")
# MQTT 回调函数：断开连接时触发
def on_disconnect(client, userdata, rc):
print(f"Disconnected with result code {rc}")
class MQTTApp(QWidget):
def __init__(self):
super().__init__()
self.initUI()
self.client = mqtt.Client(mqtt_params["client_id"]) # 创建 MQTT 客户端
self.setup_mqtt() # 设置 MQTT 回调函数
self.connected = False # 跟踪连接状态
def initUI(self):
"""初始化界面"""
self.setWindowTitle('MQTT Client with PyQt6') # 设置窗口标题
self.setGeometry(400, 400, 600, 600) # 设置窗口大小和位置
layout = QVBoxLayout() # 创建垂直布局
grid = QGridLayout() # 创建网格布局
self.broker_ip_input = QLineEdit(mqtt_params["broker_ip"]) # 服务器地址输入框
self.broker_port_input = QLineEdit(str(mqtt_params["broker_port"])) # 服务器端口输入框
self.client_id_input = QLineEdit(mqtt_params["client_id"]) # 客户端ID输入框
self.username_input = QLineEdit(mqtt_params["username"]) # 用户名输入框
self.password_input = QLineEdit(mqtt_params["password"]) # 密码输入框
self.publish_topic_input = QLineEdit(mqtt_params["publish_topic"]) # 发布主题输入框
self.subscribe_topic_input = QLineEdit(mqtt_params["subscribe_topic"]) # 订阅主题输入框
# 添加输入框和标签到网格布局
grid.addWidget(QLabel('Broker IP:'), 0, 0)
grid.addWidget(self.broker_ip_input, 0, 1)
grid.addWidget(QLabel('Broker Port:'), 1, 0)
grid.addWidget(self.broker_port_input, 1, 1)
grid.addWidget(QLabel('Client ID:'), 2, 0)
grid.addWidget(self.client_id_input, 2, 1)
grid.addWidget(QLabel('Username:'), 3, 0)
grid.addWidget(self.username_input, 3, 1)
grid.addWidget(QLabel('Password:'), 4, 0)
grid.addWidget(self.password_input, 4, 1)
grid.addWidget(QLabel('Publish Topic:'), 5, 0)
grid.addWidget(self.publish_topic_input, 5, 1)
grid.addWidget(QLabel('Subscribe Topic:'), 6, 0)
grid.addWidget(self.subscribe_topic_input, 6, 1)
layout.addLayout(grid) # 将网格布局添加到主布局
self.message_input = QLineEdit("Hello MQTT") # 消息输入框
layout.addWidget(self.message_input)
# 添加 Connect 和 Disconnect 按钮
self.connect_button = QPushButton('连接') # 连接按钮
self.connect_button.clicked.connect(self.connect_mqtt) # 连接按钮点击事件
layout.addWidget(self.connect_button)
self.disconnect_button = QPushButton('断开') # 断开按钮
self.disconnect_button.clicked.connect(self.disconnect_mqtt) # 断开按钮点击事件
layout.addWidget(self.disconnect_button)
self.send_button = QPushButton('发送数据') # 发送按钮
self.send_button.clicked.connect(self.send_message) # 发送按钮点击事件
self.send_button.setEnabled(False) # 初始禁用，直到连接成功
layout.addWidget(self.send_button)
# 添加日志显示框
self.log_display = QTextEdit()
self.log_display.setReadOnly(True) # 设置为只读
layout.addWidget(self.log_display)
# 添加自动换行复选框
self.auto_wrap_checkbox = QCheckBox('Auto-Wrap') # 自动换行复选框
self.auto_wrap_checkbox.setChecked(True) # 默认开启自动换行
layout.addWidget(self.auto_wrap_checkbox)
self.setLayout(layout)
def setup_mqtt(self):
"""设置 MQTT 客户端的回调函数"""
self.client.on_connect = on_connect # 设置连接回调
self.client.on_message = self.on_message # 设置消息回调
self.client.on_disconnect = on_disconnect # 设置断开回调
def update_mqtt_params(self):
"""更新 MQTT 参数"""
mqtt_params["broker_ip"] = self.broker_ip_input.text()
mqtt_params["broker_port"] = int(self.broker_port_input.text())
mqtt_params["client_id"] = self.client_id_input.text()
mqtt_params["username"] = self.username_input.text()
mqtt_params["password"] = self.password_input.text()
mqtt_params["publish_topic"] = self.publish_topic_input.text()
mqtt_params["subscribe_topic"] = self.subscribe_topic_input.text()
def connect_mqtt(self):
"""连接到 MQTT 服务器"""
if not self.connected:
self.client = mqtt.Client(mqtt_params["client_id"]) # 创建 MQTT 客户端
self.setup_mqtt() # 设置 MQTT 回调函数
self.update_mqtt_params() # 更新 MQTT 参数
self.client.username_pw_set(self.username_input.text(), self.password_input.text()) # 设置用户名和密码
self.client.connect(self.broker_ip_input.text(), int(self.broker_port_input.text()), 60) # 连接到服务器
self.client_thread = threading.Thread(target=self.client.loop_forever) # 启动 MQTT 客户端循环线程
self.client_thread.daemon = True
self.client_thread.start()
self.connected = True # 更新连接状态
self.connect_button.setEnabled(False) # 禁用连接按钮
self.disconnect_button.setEnabled(True) # 启用断开按钮
self.send_button.setEnabled(True) # 启用发送按钮
self.log("Connected to MQTT broker.") # 记录日志
def disconnect_mqtt(self):
"""断开 MQTT 连接"""
if self.connected:
self.client.loop_stop() # 停止 MQTT 客户端循环
self.client.disconnect() # 断开连接
self.connected = False # 更新连接状态
self.connect_button.setEnabled(True) # 启用连接按钮
self.disconnect_button.setEnabled(False) # 禁用断开按钮
self.send_button.setEnabled(False) # 禁用发送按钮
self.log("Disconnected from MQTT broker.") # 记录日志
def send_message(self):
"""发送消息"""
if self.connected:
message = self.message_input.text() # 获取输入的消息
qos = 0 # 获取当前选中的 QoS 等级（此处默认设置为 0）
self.client.publish(mqtt_params["publish_topic"], message.encode(), qos=qos) # 发布消息
print(f"Sent message: {message} with QoS {qos} to topic {mqtt_params['publish_topic']}")
def on_message(self, client, userdata, msg):
"""处理接收到的消息"""
self.log(f"topic {msg.topic}: {msg.payload.decode()}") # 记录收到的消息
def log(self, message):
"""更新日志显示框"""
self.log_display.append(message)
# 获取文本框的总内容高度和可视区域高度
total_height = self.log_display.document().size().height()
visible_height = self.log_display.height()
# 设置预留空间的高度，这里假设每行高度固定，实际情况可以根据实际字体和布局调整
reserved_space_per_line = 15
reserved_space_height = reserved_space_per_line * 5 # 假设预留5行空间
# 如果开启自动换行，则滚动到底部，但留出预留空间
if self.auto_wrap_checkbox.isChecked():
scrollbar = self.log_display.verticalScrollBar()
scrollbar.setValue(total_height - visible_height + reserved_space_height)
def closeEvent(self, event):
"""关闭窗口时断开 MQTT 连接"""
self.disconnect_mqtt()
event.accept()
def main():
"""主函数，启动应用程序"""
app = QApplication(sys.argv) # 创建应用程序对象
ex = MQTTApp() # 创建 MQTT 客户端窗口
ex.show() # 显示窗口
sys.exit(app.exec()) # 启动应用程序事件循环
if __name__ == "__main__":
main() # 执行主函数
运行之后即可 没有错误就会开始,需要先点击连接，连接成功之后，就可以进行数据通信了
2. MQTTX部分
需要安装MQTTX软件，是由EMQX公司提供的，下载安装即可
连接mqtt
打开软件添加对应的MQTT信息，点击连接即可
订阅主题
填写对应的python发布的Mqtt主题消息的名称，进行连接即可
发布主题
四 效果
点击发送就可以实现两个部分的相互通信了
云平台
更多推荐
8
0- 0
已为社区贡献2条内容
所有评论(0)