Python-PyQt6制作的MQTT桌面應(yīng)用解析Json數(shù)據(jù)并顯示（阿里云）

一 前言

繼續(xù)寫，這篇文章實(shí)現(xiàn)了桌面應(yīng)用的MQTT通信，以及我添加了一個(gè)界面，實(shí)現(xiàn)json數(shù)據(jù)的解析，這步主要時(shí)為了方便我們的調(diào)試和學(xué)習(xí)一下python的json數(shù)據(jù)解析功能，結(jié)果還是符合預(yù)期

本篇效果如下

上篇連接

Python -- PyQt6+paho.mqtt 制作的MQTT桌面收發(fā)器（阿里云示范）https://blog.csdn.net/herui_2/article/details/144512439?spm=1001.2014.3001.5501

?二 環(huán)境安裝

前面我們以及實(shí)現(xiàn)了桌面開發(fā)和mqtt庫(kù)函數(shù)的使用，這邊我們就直接引用一下

1. 編譯器

可以查看這篇文章

Python -- PyQt6 制作簡(jiǎn)易的桌面應(yīng)用（安裝-入門）https://herui.blog.csdn.net/article/details/144501509?spm=1001.2014.3001.5502

2. 環(huán)境庫(kù)下載

可以查看這篇文章

Python -- paho.mqtt 庫(kù)制作簡(jiǎn)易的MQTT通信（阿里云）https://herui.blog.csdn.net/article/details/144508263?spm=1001.2014.3001.5502

?三 代碼編寫

首先我們需要在阿里云里面建立我們的產(chǎn)品和設(shè)備，并且獲取到相關(guān)的mqtt連接參數(shù)

可以參考這個(gè)文章里面的云平臺(tái)部分

ESP32 -- 使用MQTT協(xié)議連接云平臺(tái)（帶圖文說明）https://herui.blog.csdn.net/article/details/135317019?spm=1001.2014.3001.5502

獲取到兩個(gè)設(shè)備的mqtt參數(shù)就好了

??

1. Python部分

這個(gè)代碼實(shí)現(xiàn)Mqtt連接，并且實(shí)現(xiàn)設(shè)備的參數(shù)的自定義，實(shí)現(xiàn)了設(shè)備的重連功能以及自定義發(fā)送的主題和內(nèi)容，實(shí)現(xiàn)了按鍵發(fā)送

把我們平臺(tái)上面獲取的內(nèi)容修改進(jìn)去

??

?也可以直接運(yùn)行之后修改上面的內(nèi)容

?

代碼如下

import sys
import json
from PyQt6.QtWidgets import QApplication, QWidget, QVBoxLayout, QLineEdit, QLabel, QPushButton, QGridLayout, QTextEdit, 
    QCheckBox, QTabWidget
import paho.mqtt.client as mqtt
import threading

# MQTT 服務(wù)器設(shè)置（初始值）
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"

# 全局變量來存儲(chǔ)更新的 MQTT 參數(shù)
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 回調(diào)函數(shù)：連接成功時(shí)觸發(fā)
def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    client.subscribe(mqtt_params["subscribe_topic"])  # 訂閱消息主題


# MQTT 回調(diào)函數(shù)：接收到消息時(shí)觸發(fā)
def on_message(client, userdata, msg):
    print(f"Received message on topic {msg.topic}: {msg.payload.decode()}")

    # 檢查消息是否是自己發(fā)送的
    if app_window.sent_message is not None and msg.payload.decode() == app_window.sent_message:
        return  # 如果是自己發(fā)送的消息，跳過處理

    # 繼續(xù)進(jìn)行JSON解析和顯示
    try:
        # 假設(shè)接收到的消息是 JSON 格式
        data = json.loads(msg.payload.decode())
        if "humi" in data and "tmep" in data:
            humidity = data["humi"]
            temperature = data["tmep"]
            # 將解析后的數(shù)據(jù)傳遞給顯示界面
            app_window.update_data(humidity, temperature)
            app_window.log(f"Received - Humidity: {humidity}, Temperature: {temperature}")
    except json.JSONDecodeError:
        print("Failed to decode JSON message.")


# MQTT 回調(diào)函數(shù)：斷開連接時(shí)觸發(fā)
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"])  # 創(chuàng)建 MQTT 客戶端
        self.setup_mqtt()  # 設(shè)置 MQTT 回調(diào)函數(shù)
        self.connected = False  # 跟蹤連接狀態(tài)
        self.sent_message = None  # 用于記錄當(dāng)前發(fā)送的消息，防止打印自己的消息

    def initUI(self):
        """初始化界面"""
        self.setWindowTitle('MQTT Client with PyQt6')  # 設(shè)置窗口標(biāo)題
        self.setGeometry(400, 400, 600, 600)  # 設(shè)置窗口大小和位置

        # 創(chuàng)建選項(xiàng)卡（Tab）布局
        self.tabs = QTabWidget()
        self.tab1 = QWidget()
        self.tab2 = QWidget()

        # 第一個(gè)Tab界面：用于連接和發(fā)送數(shù)據(jù)
        layout1 = QVBoxLayout()
        grid = QGridLayout()
        self.broker_ip_input = QLineEdit(mqtt_params["broker_ip"])  # 服務(wù)器地址輸入框
        self.broker_port_input = QLineEdit(str(mqtt_params["broker_port"]))  # 服務(wù)器端口輸入框
        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"])  # 發(fā)布主題輸入框
        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)
        layout1.addLayout(grid)

        self.message_input = QLineEdit("Hello MQTT")  # 消息輸入框
        layout1.addWidget(self.message_input)

        self.connect_button = QPushButton('連接')  # 連接按鈕
        self.connect_button.clicked.connect(self.connect_mqtt)
        layout1.addWidget(self.connect_button)

        self.disconnect_button = QPushButton('斷開')  # 斷開按鈕
        self.disconnect_button.clicked.connect(self.disconnect_mqtt)
        layout1.addWidget(self.disconnect_button)

        self.send_button = QPushButton('發(fā)送數(shù)據(jù)')  # 發(fā)送按鈕
        self.send_button.clicked.connect(self.send_message)
        self.send_button.setEnabled(False)
        layout1.addWidget(self.send_button)

        self.log_display = QTextEdit()
        self.log_display.setReadOnly(True)  # 設(shè)置為只讀
        layout1.addWidget(self.log_display)

        self.auto_wrap_checkbox = QCheckBox('Auto-Wrap')
        self.auto_wrap_checkbox.setChecked(True)
        layout1.addWidget(self.auto_wrap_checkbox)

        self.tab1.setLayout(layout1)
        # 第二個(gè)Tab界面：用于顯示解析后的消息
        layout2 = QVBoxLayout()  # 主布局

        # 創(chuàng)建一個(gè) QGridLayout，用于放置濕度和溫度標(biāo)簽
        grid2 = QGridLayout()

        # 使用 QLineEdit 顯示濕度和溫度，并設(shè)置為只讀
        self.humidity_label = QLineEdit("xx")
        self.humidity_label.setReadOnly(True)  # 設(shè)置 QLineEdit 為只讀
        self.humidity_label.setStyleSheet("font-size: 18px; font-weight: bold;")  # 設(shè)置標(biāo)題樣式

        self.temperature_label = QLineEdit("xx")
        self.temperature_label.setReadOnly(True)  # 設(shè)置 QLineEdit 為只讀
        self.temperature_label.setStyleSheet("font-size: 18px; font-weight: bold;")  # 設(shè)置標(biāo)題樣式

        # 添加控件到 grid 布局
        grid2.addWidget(QLabel("Humidity:"), 1, 0)
        grid2.addWidget(self.humidity_label, 1, 1)
        grid2.addWidget(QLabel("Temperature:"), 2, 0)
        grid2.addWidget(self.temperature_label, 2, 1)

        # 將 grid2 布局添加到 layout2 中
        layout2.addLayout(grid2)

        # 設(shè)置 tab2 的布局
        self.tab2.setLayout(layout2)

        # 將兩個(gè)Tab添加到Tab控件
        self.tabs.addTab(self.tab1, "MQTT參數(shù)設(shè)置")
        self.tabs.addTab(self.tab2, "MQTT消息解析")

        main_layout = QVBoxLayout()
        main_layout.addWidget(self.tabs)
        self.setLayout(main_layout)

    def setup_mqtt(self):
        """設(shè)置 MQTT 客戶端的回調(diào)函數(shù)"""
        self.client.on_connect = on_connect
        self.client.on_message = on_message
        self.client.on_disconnect = on_disconnect

    def update_data(self, humidity, temperature):
        """更新第二個(gè)界面上的濕度和溫度數(shù)據(jù)"""
        self.humidity_label.setText(f"{humidity}")
        self.temperature_label.setText(f"{temperature}")

    def update_mqtt_params(self):

        """更新 MQTT 參數(shù)"""
        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()

        # 打印當(dāng)前參數(shù)，以確認(rèn)是否正確
        print(f"Broker IP: {mqtt_params['broker_ip']}")
        print(f"Broker Port: {mqtt_params['broker_port']}")
        print(f"Client ID: {mqtt_params['client_id']}")
        print(f"Username: {mqtt_params['username']}")
        print(f"Password: {mqtt_params['password']}")
        print(f"Publish Topic: {mqtt_params['publish_topic']}")
        print(f"Subscribe Topic: {mqtt_params['subscribe_topic']}")

    def connect_mqtt(self):
        """連接到 MQTT 服務(wù)器"""
        if not self.connected:
            self.update_mqtt_params()  # 更新 MQTT 參數(shù)
            self.client = mqtt.Client(mqtt_params["client_id"])  # 創(chuàng)建 MQTT 客戶端
            self.setup_mqtt()  # 設(shè)置 MQTT 回調(diào)函數(shù)
            self.client.username_pw_set(mqtt_params['username'],mqtt_params['password'])

            try:
                print(f"Connecting to {mqtt_params['broker_ip']}:{mqtt_params['broker_port']}")
                self.client.connect(mqtt_params['broker_ip'], int(mqtt_params['broker_port']), 60)
                self.client_thread = threading.Thread(target=self.client.loop_forever)
                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.")
            except Exception as e:
                self.log(f"Failed to connect: {e}")
                print(f"Failed to connect: {e}")

    def disconnect_mqtt(self):
        """斷開 MQTT 連接"""
        if self.connected:
            self.client.loop_stop()
            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):
        """發(fā)送消息"""
        if self.connected:
            message = self.message_input.text()
            self.sent_message = message  # 記錄發(fā)送的消息
            qos = 2
            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']}")
            self.log(f"Sent: {message}")

    def log(self, message):
        """更新日志顯示框"""
        if self.sent_message is None or message != f"Sent: {self.sent_message}":
            self.log_display.append(message)  # 僅顯示接收的數(shù)據(jù)

    def closeEvent(self, event):
        """關(guān)閉窗口時(shí)斷開 MQTT 連接"""
        self.disconnect_mqtt()
        event.accept()


def main():
    """主函數(shù)，啟動(dòng)應(yīng)用程序"""
    global app_window
    app = QApplication(sys.argv)
    app_window = MQTTApp()
    app_window.show()
    sys.exit(app.exec())


if __name__ == "__main__":
    main()

?2. MQTTX部分

需要安裝MQTTX軟件，是由EMQX公司提供的，下載安裝即可

MQTTX：全功能 MQTT 客戶端工具M(jìn)QTTX 是一款強(qiáng)大的全功能 MQTT 5.0 客戶端工具，適用于桌面、命令行和 WebSocket。它使得開發(fā)和測(cè)試 MQTT 應(yīng)用更加簡(jiǎn)單高效。https://mqttx.app/zhhttps://mqttx.app/zhhttps://mqttx.app/zh

連接mqtt

打開軟件添加對(duì)應(yīng)的MQTT信息，點(diǎn)擊連接即可

?

訂閱主題

填寫對(duì)應(yīng)的python發(fā)布的Mqtt主題消息的名稱，進(jìn)行連接即可

?

發(fā)布主題

?

四 效果

點(diǎn)擊發(fā)送就可以實(shí)現(xiàn)兩個(gè)部分的相互通信了

實(shí)現(xiàn)Json數(shù)據(jù)的接收 解析 展示

?云平臺(tái)

?

五 結(jié)束

到目前為止，Python實(shí)現(xiàn)了MQTT通信，并且成功解析和顯示JSON數(shù)據(jù)。后續(xù)的工作主要是根據(jù)自己的需求調(diào)整界面效果，進(jìn)一步優(yōu)化和完善功能。整個(gè)過程我只是進(jìn)行了簡(jiǎn)單的接觸和實(shí)驗(yàn)，未來可以根據(jù)實(shí)際需求繼續(xù)擴(kuò)展和定制。

聯(lián)系方式 微信號(hào)：13648103287