pip install paho-mqtt --target={package-path}

pip install PyQt5 --target={package-path}

import sys
import json
import paho.mqtt.client as mqtt # 導入 paho-mqtt 庫
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QLabel, QLineEdit, QSpinBox, QPushButton, QGridLayout, QGroupBox, QPlainTextEdit, QSpacerItem, QSizePolicy, QMenuBar, QStatusBar
import time
class MQTTClientDebugger(QMainWindow):
def __init__(self):
super().__init__()
self.connected = False
self.setWindowTitle("MQTT 客户端調試助手")
self.setGeometry(100, 100, 1019, 772) # 設置窗口大小
self.centralWidget = QWidget(self)
self.setCentralWidget(self.centralWidget)
self.client = None # MQTT 客户端實例
# 主佈局
self.mainLayout = QVBoxLayout(self.centralWidget)
# 連接設置佈局
self.connectionLayout = QHBoxLayout()
self.host = "117.78.5.125"
self.clientId = ""
self.username = ""
self.passWord = ""
self.formLayout = QFormLayout()
self.hostLineEdit = QLineEdit()
self.hostLineEdit.setText(self.host) \# 默認服務器IP地址
self.formLayout.addRow(QLabel("服務器域名或者IP地址:"), self.hostLineEdit)
# 端口號
self.spinBoxPort = QSpinBox()
self.spinBoxPort.setMaximum(99999)
self.spinBoxPort.setValue(1883) # 默認端口號
self.formLayout.addRow(QLabel("服務器端口:"), self.spinBoxPort)
self.clientIdLineEdit = QLineEdit()
self.clientIdLineEdit.setText(self.clientId) #客户端ID
self.formLayout.addRow(QLabel("ClientId"), self.clientIdLineEdit)
self.usernameLineEdit = QLineEdit()
self.usernameLineEdit.setText(self.username) # 設備用户名
self.formLayout.addRow(QLabel("Username"), self.usernameLineEdit)
self.passwordLineEdit = QLineEdit()
self.passwordLineEdit.setText(self.passWord) # 默認密碼
self.formLayout.addRow(QLabel("Password"), self.passwordLineEdit)
self.connectionLayout.addLayout(self.formLayout)
# 連接按鈕
self.connectButton = QPushButton("Connect")
self.connectButton.clicked.connect(self.connect_to_server)
self.connectionLayout.addWidget(self.connectButton)
self.mainLayout.addLayout(self.connectionLayout)
# MQTT 主題和消息佈局
self.gridLayout = QGridLayout()
self.gridLayout.addWidget(QLabel("訂閲主題:"), 0, 0)
self.subscribeTopicLineEdit = QLineEdit("\$oc/devices/"+ self.username +"/sys/messages/down")
self.gridLayout.addWidget(self.subscribeTopicLineEdit, 0, 1)
self.gridLayout.addWidget(QPushButton("訂閲"), 0, 2)
self.gridLayout.addWidget(QLabel("發佈主題:"), 1, 0)
self.publishTopicLineEdit = QLineEdit("\$oc/devices/"+ self.username +"/sys/properties/report")
self.gridLayout.addWidget(self.publishTopicLineEdit, 1, 1)
self.gridLayout.addWidget(QLabel("主題消息:"), 2, 0)
self.messageLineEdit = QLineEdit('{"services": [{"service_id": "stm32","properties":{"DHT11_T":18.1,"DHT11_H":16.2,"SOIL":12.4,"BH1750":124.5,"MOTOR_SW":1,"SOIL_MAX":30,"run_mode":1}}]}')
self.gridLayout.addWidget(self.messageLineEdit, 2, 1)
self.publishButton = QPushButton("發佈")
self.publishButton.clicked.connect(self.publish_message)
self.gridLayout.addWidget(self.publishButton, 2, 2)
self.mainLayout.addLayout(self.gridLayout)
# 日誌區域
self.logGroupBox = QGroupBox("日誌消息:")
self.logLayout = QHBoxLayout()
self.logTextEdit = QPlainTextEdit()
self.logTextEdit.setReadOnly(True)
self.logLayout.addWidget(self.logTextEdit)
self.logGroupBox.setLayout(self.logLayout)
self.mainLayout.addWidget(self.logGroupBox)
# 底部按鈕佈局
self.bottomLayout = QHBoxLayout()
self.testButton = QPushButton("測試按鈕(一鍵填充MQTT信息)")
self.clearButton = QPushButton("一鍵清除MQTT信息")
self.clearLogButton = QPushButton("清除日誌消息")
self.viewTutorialButton = QPushButton("【查看物聯網項目開發教程】")
self.quitButton = QPushButton("退出軟件")
self.testButton.clicked.connect(self.fillMQTTInfo)
self.clearButton.clicked.connect(self.clearMQTTInfo)
self.clearLogButton.clicked.connect(self.clear_logs)
self.bottomLayout.addWidget(self.testButton)
self.bottomLayout.addWidget(self.clearButton)
self.bottomLayout.addWidget(self.clearLogButton)
self.bottomLayout.addWidget(self.viewTutorialButton)
self.bottomLayout.addWidget(self.quitButton)
self.mainLayout.addLayout(self.bottomLayout)
# 菜單欄
self.menuBar = self.menuBar()
self.fileMenu = self.menuBar.addMenu("File")
quitAction = self.fileMenu.addAction("Quit")
quitAction.triggered.connect(self.close)
# 狀態欄
self.statusBar = QStatusBar()
self.setStatusBar(self.statusBar)
# 連接到MQTT服務器
def connect_to_server(self):
host = self.hostLineEdit.text()
port = self.spinBoxPort.value()
client_id = self.clientIdLineEdit.text()
username = self.usernameLineEdit.text()
password = self.passwordLineEdit.text()
try:
self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,client_id)
self.client.username_pw_set(username, password) # 設置用户名和密碼
# 設置連接成功、消息接收、連接丟失等回調函數
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.on_disconnect = self.on_disconnect
# 連接到服務器
self.client.connect(host, port, 60)
# 啓動 MQTT 客户端
self.client.loop_start()
while not self.connected and not self._stop_event.is_set():
time.sleep(0.1)
self.connected = True
except Exception as e:
return f"{e}"
def on_connect(self, client, userdata, flags, rc):
"""當連接到MQTT服務器時調用"""
self.log(f"連接成功，返回碼：{rc}")
# 連接成功後訂閲主題
self.subscribeTopicLineEdit = QLineEdit("\$oc/devices/"+ self.usernameLineEdit.text() +"/sys/messages/down")
subscribe_topic = self.subscribeTopicLineEdit.text()
try:
client.subscribe(subscribe_topic)
except Exception as e:
self.log(f"訂閲失敗：{e}")
def on_message(self, client, userdata, msg):
"""當接收到MQTT消息時調用"""
self.log(f"接收到消息：{msg.topic} {msg.payload.decode()}")
def on_disconnect(self, client, userdata, rc):
"""當斷開連接時調用"""
self.log(f"MQTT服務器斷開連接，返回碼：{rc}")
# 發佈消息
def publish_message(self):
self.publishTopicLineEdit = QLineEdit("\$oc/devices/"+ self.usernameLineEdit.text() +"/sys/properties/report")
topic = self.publishTopicLineEdit.text()
message = self.messageLineEdit.text()
if self.client:
self.client.publish(topic, message)
self.log(f"發佈消息：{topic} {message}")
# 發送心跳包
def send_heartbeat(self):
if self.client:
self.client.ping()
self.log("發送心跳包")
# 日誌輸出
def log(self, message):
"""向日志框輸出信息"""
self.logTextEdit.appendPlainText(message)
# 填充MQTT信息
def fillMQTTInfo(self):
self.hostLineEdit.setText(self.host)
self.spinBoxPort.setValue(1883)
self.clientIdLineEdit.setText(self.clientId)
self.usernameLineEdit.setText(self.username)
self.passwordLineEdit.setText(self.passWord)
self.subscribeTopicLineEdit.setText("\$oc/devices/"+ self.username +"/sys/messages/down")
self.publishTopicLineEdit.setText("\$oc/devices/"+ self.username +"/sys/properties/report")
self.messageLineEdit.setText('{"services": [{"service_id": "stm32","properties":{"DHT11_T":18.1,"DHT11_H":16.2,"SOIL":12.4,"BH1750":124.5,"MOTOR_SW":1,"SOIL_MAX":30,"run_mode":1}}]}')
# 清除MQTT信息
def clearMQTTInfo(self):
self.hostLineEdit.clear()
self.spinBoxPort.clear()
self.clientIdLineEdit.clear()
self.usernameLineEdit.clear()
self.passwordLineEdit.clear()
self.subscribeTopicLineEdit.clear()
self.publishTopicLineEdit.clear()
# 清除日誌信息
def clear_logs(self):
self.logTextEdit.clear()
if \__name_\_ == '__main__':
app = QApplication(sys.argv)
mainWin = MQTTClientDebugger()
mainWin.show()
sys.exit(app.exec_())