mqtt協(xié)議調(diào)試時需要個客戶端工具，但網(wǎng)上找的體積包都很大，都不夠小巧和便攜。于是趁周末時間用python搞出來了個客戶端工具，使用python+tinker+paho.mqtt實現(xiàn)。源碼量很少但功能不弱，相當(dāng)?shù)妮p量級。分享給有需要的小伙伴，喜歡的可以點擊收藏。

前言

用python實現(xiàn)個跨平臺的mqtt客戶端工具，同時介紹下python的mqtt客戶端庫paho.mqtt的使用。界面這里選擇使用了python自帶的tkinter，雖不是很好用，但相當(dāng)?shù)妮p量級，對于造一個工具來說足夠啦。且配合ttkbootstrap這個包，界面可以美化，還更換皮膚，這點兒挺不錯。但是如果界面特別復(fù)雜的話推薦pyqt。

工具下載地址：https://download.csdn.net/download/qq8864/88351834

界面效果：

環(huán)境準(zhǔn)備

使用python3實現(xiàn)的mqtt客戶端工具, 相當(dāng)?shù)妮p量級。源碼不大僅一個文件。

需要安裝依賴包：

pip install ttkbootstrap
pip install -i https://pypi.doubanio.com/simple paho-mqtt

tkinter是python自帶的標(biāo)準(zhǔn)gui庫，對于我們自己日常做一些小程序出來給自己使用是非常不錯的。因為tkinter相比較其它強(qiáng)大的gui庫（PyQT，WxPython等等）而言要簡單、方便、學(xué)起來也容易得很多，所以用來造個小工具非常nice，但它做出來的界面不是很好看。

ttkbootstrap 介紹

ttkbootstrap 是一個基于 tkinter 和 ttk 的Python庫，它提供了一套現(xiàn)代化的主題和樣式，可以用于創(chuàng)建漂亮的圖形用戶界面（GUI）應(yīng)用程序。它是基于 Bootstrap 框架的設(shè)計風(fēng)格，為 tkinter 應(yīng)用程序提供了一致的外觀和用戶體驗。

官方文檔：ttkbootstrap - ttkbootstrap

ttkbootstrap 庫簡單示例，選擇主題：?

import ttkbootstrap as ttk
from ttkbootstrap.constants import *
root = ttk.Window()
root.geometry("500x400+500+150")  
style = ttk.Style()
theme_names = style.theme_names()#以列表的形式返回多個主題名
theme_selection = ttk.Frame(root, padding=(10, 10, 10, 0))
theme_selection.pack(fill=X, expand=YES)
lbl = ttk.Label(theme_selection, text="選擇主題:")
theme_cbo = ttk.Combobox(
        master=theme_selection,
        text=style.theme.name,
        values=theme_names,
)
theme_cbo.pack(padx=10, side=RIGHT)
theme_cbo.current(theme_names.index(style.theme.name))
lbl.pack(side=RIGHT)
def change_theme(event):
    theme_cbo_value = theme_cbo.get()
    style.theme_use(theme_cbo_value)
    theme_selected.configure(text=theme_cbo_value)
    theme_cbo.selection_clear()
theme_cbo.bind('<<ComboboxSelected>>', change_theme)
theme_selected = ttk.Label(
        master=theme_selection,
        text="litera",
        font="-size 24 -weight bold"
)
theme_selected.pack(side=LEFT)
root.mainloop()

ttkbootstrap簡單使用

import ttkbootstrap as ttk
#實例化創(chuàng)建應(yīng)用程序窗口
root = ttk.Window(
        title="窗口名字",        #設(shè)置窗口的標(biāo)題
        themename="litera",     #設(shè)置主題
        size=(1066,600),        #窗口的大小
        position=(100,100),     #窗口所在的位置
        minsize=(0,0),          #窗口的最小寬高
        maxsize=(1920,1080),    #窗口的最大寬高
        resizable=None,         #設(shè)置窗口是否可以更改大小
        alpha=1.0,              #設(shè)置窗口的透明度(0.0完全透明）
        )
# root.place_window_center()    #讓顯現(xiàn)出的窗口居中
# root.resizable(False,False)   #讓窗口不可更改大小
# root.wm_attributes('-topmost', 1)#讓窗口位置其它窗口之上
root.mainloop()

標(biāo)簽顯示：

import ttkbootstrap as ttk
from ttkbootstrap.constants import *
root = ttk.Window()
ttk.Label(root,text="標(biāo)簽1",bootstyle=INFO).pack(side=ttk.LEFT, padx=5, pady=10)
ttk.Label(root,text="標(biāo)簽2",bootstyle="inverse").pack(side=ttk.LEFT, padx=5, pady=10)
ttk.Label(root,text="標(biāo)簽3",bootstyle="inverse-danger").pack(side=ttk.LEFT, padx=5, pady=10)
ttk.Label(root, text="標(biāo)簽4", bootstyle=WARNING, font=("微軟雅黑", 15), background='#94a2a4').pack(side=LEFT, padx=5, pady=10)
root.mainloop()
'''
# bootstyle colors
PRIMARY = 'primary'
SECONDARY = 'secondary'
SUCCESS = 'success'
DANGER = 'danger'
WARNING = 'warning'
INFO = 'info'
LIGHT = 'light'
DARK = 'dark'

# bootstyle types
OUTLINE = 'outline'
LINK = 'link'
TOGGLE = 'toggle'
INVERSE = 'inverse'
STRIPED = 'striped'
TOOLBUTTON = 'toolbutton'
ROUND = 'round'
SQUARE = 'square'
'''

?paho-mqtt庫介紹

paho-mqtt 是目前 Python 中使用較多的 MQTT 客戶端庫，它在 Python 2.7 或 3.x 上為客戶端類提供了對 MQTT v3.1 和 v3.1.1 的支持。它還提供了一些幫助程序功能，使將消息發(fā)布到 MQTT 服務(wù)器變得非常簡單。?

paho-mqtt簡單使用:

# -*- coding: utf-8 -*-# -*- coding: utf-8 -*-
 
 
import paho.mqtt.client as mqtt
import time
 
 
def on_connect(client, userdata, flags, rc):
    print "鏈接"
    print("Connected with result code: " + str(rc))
 
 
def on_message(client, userdata, msg):
    print "消息內(nèi)容"
    print(msg.topic + " " + str(msg.payload))
 
 
#   訂閱回調(diào)
def on_subscribe(client, userdata, mid, granted_qos):
    print "訂閱"
    print("On Subscribed: qos = %d" % granted_qos)
    pass
 
 
#   取消訂閱回調(diào)
def on_unsubscribe(client, userdata, mid, granted_qos):
    print "取消訂閱"
    print("On unSubscribed: qos = %d" % granted_qos)
    pass
 
 
#   發(fā)布消息回調(diào)
def on_publish(client, userdata, mid):
    print "發(fā)布消息"
    print("On onPublish: qos = %d" % mid)
    pass
 
 
#   斷開鏈接回調(diào)
def on_disconnect(client, userdata, rc):
    print "斷開鏈接"
    print("Unexpected disconnection rc = " + str(rc))
    pass
 
 
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.on_publish = on_publish
client.on_disconnect = on_disconnect
client.on_unsubscribe = on_unsubscribe
client.on_subscribe = on_subscribe
client.connect('127.0.0.1', 1883, 600) # 600為keepalive的時間間隔
while True:
    client.publish(topic='mqtt11', payload='amazing', qos=0, retain=False)
    time.sleep(2)

參數(shù)解釋:

• keepalive => ? 心跳間隔，單位是秒，如果 broker 和 client 在這段時間內(nèi)沒有任何通訊，client 會給 broker 發(fā)送一個 ping 消息
• retain ?=> ?如果設(shè)為 Ture ，這條消息會被設(shè)為保留消息?
• payload ?=> 消息內(nèi)容，字符串類型，如果設(shè)為 None ，會發(fā)送一條長度為 0 消息。如果設(shè)置了 int 或者 3. float 類型的值，會當(dāng)做字符串發(fā)送，如果你想發(fā)送真正的 int 或者 float 值，需要用 struct.pack() 生成消息, mqtt的publish 只支持None, string, int, float 類型的數(shù)據(jù), ?如果需要發(fā)送json類型數(shù)據(jù)可以通過json.dumps()將數(shù)據(jù)進(jìn)行轉(zhuǎn)換后在發(fā)送, 接收端在on_message()回調(diào)函數(shù)中通過json.loads() 將數(shù)據(jù)解析就可以了
• topic => ?這條消息所屬的話題
• qos ?=> 消息的安全等級 Qos詳細(xì)介紹
• qos=0 ? ?QoS0，At most once，至多一次；
• QoS0 代表，Sender 發(fā)送的一條消息，Receiver 最多能收到一次，也就是說 Sender 盡力向 Receiver 發(fā)送消息，如果發(fā)送失敗，也就算了；
• qos=1 ? ?QoS1，At least once，至少一次；
• QoS1 代表，Sender 發(fā)送的一條消息，Receiver 至少能收到一次，也就是說 Sender 向 Receiver 發(fā)送消息，如果發(fā)送失敗，會繼續(xù)重試，直到 Receiver 收到消息為止，但是因為重傳的原因，Receiver 有可能會收到重復(fù)的消息；
• qos=2 ? ?QoS2，Exactly once，確保只有一次
• QoS2 代表，Sender 發(fā)送的一條消息，Receiver 確保能收到而且只收到一次，也就是說 Sender 盡力向 Receiver 發(fā)送消息，如果發(fā)送失敗，會繼續(xù)重試，直到 Receiver 收到消息為止，同時保證 Receiver 不會因為消息重傳而收到重復(fù)的消息。

源碼實現(xiàn)

?源碼實現(xiàn)比較簡單，主要是里面有一些注意事項。比如圖片的顯示，網(wǎng)上關(guān)于tkinter不能顯示圖片的問題有很多。這里給出絕對可行的做法。

label的圖片顯示

from PIL import Image, ImageTk       
#.....
        img = Image.open("me.jpg")  # 替換為你的圖片路徑
        img = img.resize((80,80))
        #self._img = ImageTk.PhotoImage(file = "me.jpg")    
        self._img = ImageTk.PhotoImage(img)     
        self.about = Label(self.fr1) 
        self.about.image = self._img
        self.about.configure(image=self._img)
        self.about.place(x=65,y=0,width=80,height=80)

判斷是否為16進(jìn)制

def ISHEX(data):        #判斷輸入字符串是否為十六進(jìn)制
    if len(data)%2:
        return False
    for item in data:
        if item not in '0123456789ABCDEFabcdef': #循環(huán)判斷數(shù)字和字符
            return False
    return True

保存文件

def savefiles(self):   #保存日志TXT文本
        try:
            with open('log.txt','a') as file:       #a方式打開 文本追加模式
                file.write(self.txt_rx.get(0.0,'end'))
                messagebox.showinfo('提示', '保存成功')
        except:
            messagebox.showinfo('錯誤', '保存日志文件失?。?)

?自動滾動到末尾

def appendTxt(self,msg):
        current_t = datetime.now()
        current_ = current_t.strftime("%Y-%m-%d %H:%M:%S ")
        self.txt_rx.insert(END,current_)
        self.txt_rx.insert(END,msg)
        self.txt_rx.insert(END,"\n")  
        #滾動到末尾
        self.txt_rx.see(END)
        self.txt_rx.update_idletasks()      

mqtt的使用

def connect(self,addr,port,alive=60):
        self.client.connect(addr, port,alive)
        self.client.loop_start()

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()
        print("disconnect!")

    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker ok!\n")
            self.appendTxt("Connected to MQTT Broker ok!\n")
            self.var_bt1.set("斷開")
            self.isConnect = True
        else:
            print("Failed to connect, return code %d\n", rc)
            self.appendTxt(f"Failed to connect, return code: {rc}\n")
            self.isConnect = False

    def on_message(self, client, userdata, msg):
        self.tx_rx_cnt(1,0)
        print("Received message: " + msg.payload.decode())
        self.appendTxt(f"Received message:\n[topic]:{msg.topic}\n{msg.payload.decode()}\n")

    def subscribe(self, topic):
        #item = Entry(self.subitem).get()
        if topic in self.subitem.get(0, END):
            print("item already exists.")
        else:
            self.appendTxt(f"[訂閱topic]:{topic}\n")
            self.client.subscribe(topic)
            self.subitem.insert(END, topic)

    def publish(self, topic, message):
        self.client.publish(topic, message)
        self.appendTxt(f"[發(fā)布topic]:{topic}\n{message}\n")

打包為exe程序

python腳本打包為exe有很多方法。常用的為pyinstaller，使用簡單，但是有不少缺點，如打包體積大，啟動速度慢等。所以這里并不推薦它，推薦使用nuitka。

pyinstaller的打包

#pyinstaller方式打包
pyinstaller -F -w -i chengzi.ico py_word.py 打包指定exe圖標(biāo)打包

注：以上僅是簡單的打包，如果帶圖片或文件資源，打包后運(yùn)行也是失敗的。?

帶圖片資源的打包方法：

pyi-makespec -F .\mqttclienttool.py

執(zhí)行上面這條命令，會生成?mqttclienttool.spec文件，這個文件可以編輯。打開并編輯這個文件，

data選項中增加：datas=[('me.jpg', '.')], 小括號內(nèi)的寫法，第一個為文件名，第二個為路徑。還有個需要注意的地方是，代碼里引用路徑需要采用絕對路徑的方式如：

img_path = os.path.join(os.path.dirname(__file__), 'me.jpg')
img = Image.open(img_path) 

# -*- mode: python ; coding: utf-8 -*-


block_cipher = None


a = Analysis(
    ['mqttclienttool.py'],
    pathex=[],
    binaries=[],
    datas=[('me.jpg', '.')],
    hiddenimports=[],
    hookspath=[],
    hooksconfig={},
    runtime_hooks=[],
    excludes=[],
    win_no_prefer_redirects=False,
    win_private_assemblies=False,
    cipher=block_cipher,
    noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)

最后執(zhí)行以下命令即可。

 pyinstaller  .\mqttclienttool.spec

nuitka的安裝

• 直接利用pip即可安裝：pip install Nuitka

• 需具備(MSVS)或者M(jìn)inGW64等C++的編譯器。

nuitka使用過程

對于第三方依賴包較多的項目（比如需要import torch,tensorflow,cv2,numpy,pandas,geopy等等）而言，這里最好打包的方式是只將屬于自己的代碼轉(zhuǎn)成C++，不管這些大型的第三方包！

使用舉例：

nuitka --standalone --show-memory --show-progress --nofollow-imports --plugin-enable=qt-plugins --follow-import-to=utils,src --output-dir=out --windows-icon-from-ico=./logo.ico ./demo.py

簡單介紹下上面的nuitka的命令：

• --standalone：方便移植到其他機(jī)器，不用再安裝python

• --show-memory --show-progress：展示整個安裝的進(jìn)度過程

• --nofollow-imports：不編譯代碼中所有的import，比如keras，numpy之類的。

• --plugin-enable=qt-plugins：如果用到pyqt5來做界面，這里nuitka有其對應(yīng)的插件。

• --plugin-enable=qt-plugins

• --follow-import-to=utils,src：需要編譯成C++代碼的指定的2個包含源碼的文件夾，這里用,來進(jìn)行分隔。

• --output-dir=out：指定輸出的結(jié)果路徑為out。

• --windows-icon-from-ico=./logo.ico：指定生成的exe的圖標(biāo)為logo.ico這個圖標(biāo)，這里推薦一個將圖片轉(zhuǎn)成ico格式文件的網(wǎng)站（比特蟲）。

• --windows-disable-console：運(yùn)行exe取消彈框。這里沒有放上去是因為我們還需要調(diào)試，讓可以彈出DOS窗口便于查看print的日志。

完整實現(xiàn)

# -*- coding: utf-8 -*-
# @Time : 2023/09/17 12:49
# @Author : yangyongzhen
# @Email : 534117529@qq.com
# @File : mqttclienttool.py
# @Project : study
import time
import os
from tkinter.ttk import *
from tkinter import *
from datetime import datetime
import time
import threading
from tkinter import messagebox
from ttkbootstrap import Style
import paho.mqtt.client as mqtt
from PIL import Image, ImageTk

global gui           #全局型式保存GUI句柄

tx_cnt=0 #發(fā)送條數(shù)統(tǒng)計
rx_cnt=0 #接收條數(shù)統(tǒng)計

def ISHEX(data):        #判斷輸入字符串是否為十六進(jìn)制
    if len(data)%2:
        return False
    for item in data:
        if item not in '0123456789ABCDEFabcdef': #循環(huán)判斷數(shù)字和字符
            return False
    return True

'''GUI'''''''''''''''''''''''''''''''''''''''''''''''''''''''''
class GUI:
    def __init__(self):
        self.root = Tk()
        self.root.title('MQTT調(diào)試助手-author:blog.csdn.net/qq8864')             #窗口名稱
        self.root.geometry("820x560+500+150")         #尺寸位置
        self.root.resizable(False, False)
        self.interface()
        Style(theme='pulse') #主題修改 可選['cyborg', 'journal', 'darkly', 'flatly' 'solar', 'minty', 'litera', 'united', 'pulse', 'cosmo', 'lumen', 'yeti', 'superhero','sandstone']
        #self.client.on_log = self.log_callback
        self.isConnect = False
        self._img = None
    def interface(self):
        """"界面編寫位置"""
        #--------------------------------操作區(qū)域-----------------------------#
        self.fr1=Frame(self.root)
        self.fr1.place(x=0,y=0,width=220,height=600)     #區(qū)域1位置尺寸
        img_path = os.path.join(os.path.dirname(__file__), 'me.jpg')
        img = Image.open(img_path)  # 替換為你的圖片路徑
        img = img.resize((80,80))
        #self._img = ImageTk.PhotoImage(file = "me.jpg")    
        self._img = ImageTk.PhotoImage(img)     
        self.about = Label(self.fr1) 
        self.about.image = self._img
        self.about.configure(image=self._img)
        self.about.place(x=65,y=0,width=80,height=80)
        pos = 80
        self.lb_server =Label(self.fr1, text='地址:',anchor="e",fg='red')  #點擊可刷新
        self.lb_server.place(x=0,y=pos,width=50,height=35)
        self.txt_server = Text(self.fr1)
        self.txt_server.place(x=65,y=pos,width=155,height=26)
        self.txt_server.insert("1.0", "127.0.0.1")
        
        self.lb1 =Label(self.fr1, text='端口:',anchor="e",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=pos+40,width=50,height=35)
        self.txt_port = Text(self.fr1)
        self.txt_port.place(x=65,y=pos+40,width=155,height=26)
        self.txt_port.insert("1.0", 1883)
        
        self.lb1 =Label(self.fr1, text='clientID:',anchor="e",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=pos+80,width=50,height=35)
        self.txt_id = Text(self.fr1)
        self.txt_id.place(x=65,y=pos+80,width=155,height=26)
        self.txt_id.insert("1.0", "mqtt-client")
        
        self.lb1 =Label(self.fr1, text='用戶名:',anchor="e",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=pos+120,width=50,height=35)
        self.txt_name = Text(self.fr1)
        self.txt_name.place(x=65,y=pos+120,width=155,height=26)


        self.lb1 =Label(self.fr1, text='密碼 :',anchor="e",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=pos+160,width=50,height=35)
        self.txt_pwd = Text(self.fr1)
        self.txt_pwd.place(x=65,y=pos+160,width=155,height=26)
        
        self.lb1 =Label(self.fr1, text='心跳 :',anchor="e",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=pos+200,width=50,height=35)
        self.txt_heart = Text(self.fr1)
        self.txt_heart.place(x=65,y=pos+200,width=155,height=26)
        self.txt_heart.insert("1.0", 60)

        self.var_bt1 = StringVar()
        self.var_bt1.set("連接")
        self.btn1 = Button(self.fr1,textvariable=self.var_bt1,command=self.btn_connect) #綁定 btn_connect 方法
        self.btn1.place(x=170,y=pos+240,width=50,height=30)


        self.lb_s =Label(self.fr1, text='訂閱主題',bg="yellow",anchor='w') #字節(jié)統(tǒng)計
        self.lb_s.place(x=5,y=340,width=90,height=28)
        
        self.txt_sub = Text(self.fr1)
        self.txt_sub.place(x=5,y=368,width=155,height=28)
        self.btn5 = Button(self.fr1, text='訂閱',command=self.btn_sub) #測試用
        self.btn5.place(x=170,y=368,width=50,height=28)
    

        self.subitem = Listbox(self.fr1)
        self.subitem.place(x=5,y=402,width=215,height=85)
        #self.subitem.insert(END, "This is a read-only Text widget.")
        self.subitem.bind("<Button-3>", self.on_right_click)
        

        #-------------------------------文本區(qū)域-----------------------------#
        self.fr2=Frame(self.root)          #區(qū)域1 容器  relief   groove=凹  ridge=凸
        self.fr2.place(x=220,y=0,width=620,height=560)     #區(qū)域1位置尺寸

        self.txt_rx = Text(self.fr2)
        self.txt_rx.place(relheight=0.6,relwidth=0.9,relx=0.05,rely=0.01) #比例計算控件尺寸和位置
        
        self.scrollbar = Scrollbar(self.txt_rx)
        self.scrollbar.pack(side=RIGHT, fill=Y)
        self.txt_rx.config(yscrollcommand=self.scrollbar.set)
        self.scrollbar.config(command=self.txt_rx.yview)
        self.txt_rx.bind("<Configure>", self.check_scrollbar)

        self.lb_t =Label(self.fr2, text='發(fā)布主題',bg="yellow",anchor='w') #字節(jié)統(tǒng)計
        self.lb_t.place(relheight=0.04,relwidth=0.2,relx=0.05,rely=0.62)
        
        self.lb_qos =Label(self.fr2, text='QoS:',bg="yellow",anchor='w') #字節(jié)統(tǒng)計
        self.lb_qos.place(relheight=0.04,relwidth=0.15,relx=0.15,rely=0.62)
        
        self.var_cb1 = IntVar()
        self.comb1 = Combobox(self.fr2,textvariable=self.var_cb1)
        self.comb1['values'] = [0,1,2] #列出可用等級
        self.comb1.current(0)  # 設(shè)置默認(rèn)選項 0開始
        self.comb1.place(relheight=0.04,relwidth=0.08,relx=0.22,rely=0.615)
        
        self.txt_topic = Text(self.fr2)
        self.txt_topic.place(relheight=0.05,relwidth=0.9,relx=0.05,rely=0.66) #比例計算控件尺寸位置
        
        self.txt_tx = Text(self.fr2)
        self.txt_tx.place(relheight=0.15,relwidth=0.9,relx=0.05,rely=0.72) #比例計算控件尺寸位置

        self.btn6 = Button(self.fr2, text='發(fā)送',command=self.btn_send)  #綁定發(fā)送方法
        self.btn6.place(relheight=0.06,relwidth=0.11,relx=0.84,rely=0.88)
        
        self.btn3 = Button(self.fr2, text='清空',command = self.txt_clr) #綁定清空方法
        self.btn4 = Button(self.fr2, text='保存',command=self.savefiles) #綁定保存方法
        self.btn3.place(relheight=0.06,relwidth=0.11,relx=0.05,rely=0.88)
        self.btn4.place(relheight=0.06,relwidth=0.11,relx=0.18,rely=0.88)
        
        self.lb3 =Label(self.fr2, text='接收:0    發(fā)送:0',bg="yellow",anchor='w') #字節(jié)統(tǒng)計
        self.lb3.place(relheight=0.05,relwidth=0.3,relx=0.045,rely=0.945)

        self.lb4 = Label(self.fr2, text=' ', anchor='w',relief=GROOVE)  #時鐘
        self.lb4.place(relheight=0.05, relwidth=0.11, relx=0.84, rely=0.945)
#------------------------------------------方法-----------------------------------------------
    def check_scrollbar(self,*args):
        if self.txt_rx.yview() == (0.0, 1.0):
            self.scrollbar.pack_forget()
        else:
            self.scrollbar.place(RIGHT, fill=Y)
            
    def on_right_click(self,w):
        idx = self.subitem.curselection()
        print("Right-Clicked idx:", idx)
        if idx == ():
            return
        selected_item = self.subitem.get(idx)
        print("Right-Clicked on:", selected_item,idx)
        ret = messagebox.askyesno('取消訂閱', "取消訂閱:\n"+selected_item)
        if ret:
            self.subitem.delete(idx)
            self.client.unsubscribe(selected_item)
            self.appendTxt("取消訂閱:"+selected_item)
        
    def gettim(self):#獲取時間 未用
            timestr = time.strftime("%H:%M:%S")  # 獲取當(dāng)前的時間并轉(zhuǎn)化為字符串
            self.lb4.configure(text=timestr)  # 重新設(shè)置標(biāo)簽文本
            # tim_str = str(datetime.datetime.now()) + '\n'
            # self.lb4['text'] = tim_str
            #self.lb3['text'] = '接收：'+str(rx_cnt),'發(fā)送：'+str(tx_cnt)
            self.txt_rx.after(1000, self.gettim)     # 每隔1s調(diào)用函數(shù) gettime 自身獲取時間 GUI自帶的定時函數(shù)

    def txt_clr(self):#清空顯示
        self.txt_rx.delete(0.0, 'end')  # 清空文本框
        self.txt_tx.delete(0.0, 'end')  # 清空文本框

    def ascii_hex_get(self):#獲取單選框狀態(tài)
        if(self.var_cs.get()):
            return False
        else:
            return True

    def tx_rx_cnt(self,rx=0,tx=0):  #發(fā)送接收統(tǒng)計
        global tx_cnt
        global rx_cnt

        rx_cnt += rx
        tx_cnt += tx
        self.lb3['text'] = '接收：'+str(rx_cnt),'發(fā)送：'+str(tx_cnt)

    def savefiles(self):   #保存日志TXT文本
        try:
            with open('log.txt','a') as file:       #a方式打開 文本追加模式
                file.write(self.txt_rx.get(0.0,'end'))
                messagebox.showinfo('提示', '保存成功')
        except:
            messagebox.showinfo('錯誤', '保存日志文件失?。?)
    
    def log_callback(self,client, userdata, level, buf):
        print(buf)
        
    def appendTxt(self,msg,flag = None):
        current_t = datetime.now()
        current_ = current_t.strftime("%Y-%m-%d %H:%M:%S ")
        self.txt_rx.insert(END,current_)
        self.txt_rx.insert(END,msg)
        self.txt_rx.insert(END,"\n")  
        #滾動到末尾
        self.txt_rx.see(END)
        self.txt_rx.update_idletasks()      
                
    def connect(self,addr,port,alive=60):
        self.client.connect(addr, port,alive)
        self.client.loop_start()

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()
        print("disconnect!")

    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker ok!\n")
            self.appendTxt("Connected to MQTT Broker ok!\n")
            self.var_bt1.set("斷開")
            self.isConnect = True
        else:
            print("Failed to connect, return code %d\n", rc)
            self.appendTxt(f"Failed to connect, return code: {rc}\n")
            self.isConnect = False

    def on_message(self, client, userdata, msg):
        self.tx_rx_cnt(1,0)
        print("Received message: " + msg.payload.decode())
        self.appendTxt(f"Received message:\n[topic]:{msg.topic}\n{msg.payload.decode()}\n","RECV")

    def subscribe(self, topic):
        #item = Entry(self.subitem).get()
        if topic in self.subitem.get(0, END):
            print("item already exists.")
        else:
            self.appendTxt(f"[訂閱topic]:{topic}\n")
            self.client.subscribe(topic)
            self.subitem.insert(END, topic)

    def publish(self, topic, message,qos=0):
        self.client.publish(topic, message,qos)
        self.appendTxt(f"[發(fā)布topic]:{topic}\n{message}\n")
        
    def btn_connect(self):#連接
        global isConnect
        if self.var_bt1.get() == '連接':
            server = self.txt_server.get("1.0",END).strip()
            port = self.txt_port.get("1.0",END).strip()
            alive = self.txt_heart.get("1.0",END).strip()
            user = self.txt_name.get("1.0",END).strip()
            psd = self.txt_pwd.get("1.0",END).strip()
            cid = self.txt_id.get("1.0",END).strip()
            #用戶名密碼設(shè)置
            if len(user) !=0 :
                self.client.username_pw_set(user, psd)
                
            self.client = mqtt.Client(cid) #MQTT
            self.client.on_connect = self.on_connect
            self.client.on_message = self.on_message
            print("btn connect click: "+server+","+port+",QoS:"+self.comb1.get())
            self.appendTxt(f"連接 {server},port:{port}\n")
            self.connect(server,int(port),int(alive))
        else:
            self.disconnect()
            self.var_bt1.set("連接")
            self.isConnect = False
            self.appendTxt(f"斷開連接!\n")
        
    def btn_sub(self):#訂閱
        if self.isConnect:
            sub = self.txt_sub.get("1.0",END).strip()
            print("btn sub click,topic: "+sub)
            self.subscribe(sub)
        else:
            messagebox.showinfo('提示', '服務(wù)器未連接!')
        
    def btn_send(self):#發(fā)布
        if self.isConnect:
            pub_topic = self.txt_topic.get("1.0",END).strip()
            payload = self.txt_tx.get("1.0",END).strip()
            print("btn pub click,topic: "+pub_topic)
            self.publish(pub_topic,payload,int(self.comb1.get()))
            self.tx_rx_cnt(0,1)
        else:
            messagebox.showinfo('提示', '請連接服務(wù)器!')

if __name__ == '__main__':
    print('Start...')
    gui = GUI()
    gui.gettim()  #開啟時鐘
    gui.root.mainloop()
    print('End...')

最后，再附帶個python實現(xiàn)的串口調(diào)試助手源碼：

import time

from tkinter.ttk import *
from tkinter import *
import datetime
import serial  # 導(dǎo)入模塊
import serial.tools.list_ports
import threading
from tkinter import messagebox
from ttkbootstrap import Style



global UART          #全局型式保存串口句柄
global RX_THREAD     #全局型式保存串口接收函數(shù)
global gui           #全局型式保存GUI句柄

tx_cnt=0 #發(fā)送字符數(shù)統(tǒng)計
rx_cnt=0 #接收字符數(shù)統(tǒng)計


def ISHEX(data):        #判斷輸入字符串是否為十六進(jìn)制
    if len(data)%2:
        return False
    for item in data:
        if item not in '0123456789ABCDEFabcdef': #循環(huán)判斷數(shù)字和字符
            return False
    return True


def uart_open_close(fun,com,bund):  #串口打開關(guān)閉控制
    global UART
    global RX_THREAD

    if fun==1:#打開串口
        try:
           UART = serial.Serial(com, bund, timeout=0.2)  # 提取串口號和波特率并打開串口
           if UART.isOpen(): #判斷是否打開成功
               lock = threading.Lock()
               RX_THREAD = UART_RX_TREAD('URX1',lock)  #開啟數(shù)據(jù)接收進(jìn)程
               RX_THREAD.setDaemon(True)               #開啟守護(hù)進(jìn)程 主進(jìn)程結(jié)束后接收進(jìn)程也關(guān)閉 會報警告 不知道咋回事
               RX_THREAD.start()
               RX_THREAD.resume()
               return True
        except:
            return False
        return False
    else:                   #關(guān)閉串口
        print("關(guān)閉串口")
        RX_THREAD.pause()
        UART.close()

def uart_tx(data,isHex=False):          #串口發(fā)送數(shù)據(jù)
    global UART

    try:
        if  UART.isOpen():  #發(fā)送前判斷串口狀態(tài) 避免錯誤
            print("uart_send=" + data)
            gui.tx_rx_cnt(tx=len(data)) #發(fā)送計數(shù)
            if isHex:   #十六進(jìn)制發(fā)送
                data_bytes = bytes.fromhex(data)
                return UART.write(bytes(data_bytes))
            else:      #字符發(fā)送
                return UART.write(data.encode('gb2312'))
    except:#錯誤返回
        messagebox.showinfo('錯誤', '發(fā)送失敗')


class UART_RX_TREAD(threading.Thread):          #數(shù)據(jù)接收進(jìn)程 部分重構(gòu)
    global gui

    def __init__(self, name, lock):
        threading.Thread.__init__(self)
        self.mName = name
        self.mLock = lock
        self.mEvent = threading.Event()

    def run(self): #主函數(shù)
        print('開啟數(shù)據(jù)接收\r')
        while True:
            self.mEvent.wait()
            self.mLock.acquire()
            if UART.isOpen():
                rx_buf =  UART.read()
                if len(rx_buf) >0:
                    rx_buf += UART.readall()  #有延遲但不易出錯
                    gui.tx_rx_cnt(rx=len(rx_buf))
                    if gui.ascii_hex_get() == False:
                        print('收到hex數(shù)據(jù)', rx_buf.hex().upper())
                        gui.txt_rx.insert(END,  rx_buf.hex().upper())
                    else:
                        str_data = str(rx_buf, encoding='gb2312')
                        print("串口收到消息:", len(rx_buf), str_data)
                        gui.txt_rx.insert(END,str_data)
                        # self.txt_rx.insert(END,str_data)
            self.mLock.release()
           #time.sleep(3)
    def pause(self): #暫停
        self.mEvent.clear()

    def resume(self):#恢復(fù)
        self.mEvent.set()


'''GUI'''''''''''''''''''''''''''''''''''''''''''''''''''''''''
class GUI:
    def __init__(self):
        self.root = Tk()
        self.root.title('梵德覓串口調(diào)試助手')             #窗口名稱
        self.root.geometry("800x360+500+150")         #尺寸位置
        self.interface()
        Style(theme='pulse') #主題修改 可選['cyborg', 'journal', 'darkly', 'flatly' 'solar', 'minty', 'litera', 'united', 'pulse', 'cosmo', 'lumen', 'yeti', 'superhero','sandstone']


    def interface(self):
        """"界面編寫位置"""
        #--------------------------------操作區(qū)域-----------------------------#
        self.fr1=Frame(self.root)
        self.fr1.place(x=0,y=0,width=180,height=360)     #區(qū)域1位置尺寸

        self.lb1 =Label(self.fr1, text='端口號：',font="微軟雅黑",fg='red')  #點擊可刷新
        self.lb1.place(x=0,y=5,width=100,height=35)

        self.var_cb1 = StringVar()
        self.comb1 = Combobox(self.fr1,textvariable=self.var_cb1)
        self.comb1['values'] = list(serial.tools.list_ports.comports()) #列出可用串口
        self.comb1.current(0)  # 設(shè)置默認(rèn)選項 0開始
        self.comb1.place(x=10,y=40,width=150,height=30)
        com=list(serial.tools.list_ports.comports())

        print('**********可用串口***********')
        for i in range(0, len(com)):
            print(com[i])
        print('***************************')

        self.lb2 = Label(self.fr1, text='波特率：')
        self.comb2 = Combobox(self.fr1,values=['2400','9600','57600','115200'])
        self.comb2.current(3)                               #設(shè)置默認(rèn)選項 115200
        self.lb2.place(x=5,y=75,width=60,height=20)
        self.comb2.place(x=10,y=100,width=100,height=25)

        self.var_bt1 = StringVar()
        self.var_bt1.set("打開串口")
        self.btn1 = Button(self.fr1,textvariable=self.var_bt1,command=self.uart_opn_close) #綁定 uart_opn_close 方法
        self.btn1.place(x=10,y=140,width=60,height=30)



        self.var_cs = IntVar()  #定義返回類型
        self.rd1 = Radiobutton(self.fr1,text="Ascii",variable=self.var_cs,value=0,command = self.txt_clr) #選擇后清除顯示內(nèi)容
        self.rd2 = Radiobutton(self.fr1,text="Hex",variable=self.var_cs,value=1,command = self.txt_clr)
        self.rd1.place(x=5,y=180,width=60,height=30)
        self.rd2.place(x=5,y=210,width=60,height=30)


        self.btn3 = Button(self.fr1, text='清空',command = self.txt_clr) #綁定清空方法
        self.btn4 = Button(self.fr1, text='保存',command=self.savefiles) #綁定保存方法
        self.btn3.place(x=10,y=260,width=60,height=30)
        self.btn4.place(x=100,y=260,width=60,height=30)

        self.btn5 = Button(self.fr1, text='功能',command=self.ascii_hex_get) #測試用
        self.btn6 = Button(self.fr1, text='發(fā)送',command=self.uart_send)  #綁定發(fā)送方法
        self.btn5.place(x=10,y=315,width=60,height=30)
        self.btn6.place(x=100,y=315,width=60,height=30)

        #-------------------------------文本區(qū)域-----------------------------#
        self.fr2=Frame(self.root)          #區(qū)域1 容器  relief   groove=凹  ridge=凸
        self.fr2.place(x=180,y=0,width=620,height=360)     #區(qū)域1位置尺寸

        self.txt_rx = Text(self.fr2)
        self.txt_rx.place(relheight=0.6,relwidth=0.9,relx=0.05,rely=0.01) #比例計算控件尺寸和位置

        self.txt_tx = Text(self.fr2)
        self.txt_tx.place(relheight=0.25,relwidth=0.9,relx=0.05,rely=0.66) #比例計算控件尺寸位置

        self.lb3 =Label(self.fr2, text='接收:0    發(fā)送:0',bg="yellow",anchor='w') #字節(jié)統(tǒng)計
        self.lb3.place(relheight=0.05,relwidth=0.3,relx=0.045,rely=0.925)

        self.lb4 = Label(self.fr2, text=' ', anchor='w',relief=GROOVE)  #時鐘
        self.lb4.place(relheight=0.05, relwidth=0.1, relx=0.85, rely=0.935)
#------------------------------------------方法-----------------------------------------------
    def gettim(self):#獲取時間 未用
            timestr = time.strftime("%H:%M:%S")  # 獲取當(dāng)前的時間并轉(zhuǎn)化為字符串
            self.lb4.configure(text=timestr)  # 重新設(shè)置標(biāo)簽文本
            # tim_str = str(datetime.datetime.now()) + '\n'
            # self.lb4['text'] = tim_str
            self.txt_rx.after(1000, self.gettim)     # 每隔1s調(diào)用函數(shù) gettime 自身獲取時間 GUI自帶的定時函數(shù)

    def txt_clr(self):#清空顯示
        self.txt_rx.delete(0.0, 'end')  # 清空文本框
        self.txt_tx.delete(0.0, 'end')  # 清空文本框

    def ascii_hex_get(self):#獲取單選框狀態(tài)
        if(self.var_cs.get()):
            return False
        else:
            return True

    def uart_opn_close(self):#打開關(guān)閉串口
        if(self.var_bt1.get() == '打開串口'):
          if(uart_open_close(1,str(self.comb1.get())[0:5],self.comb2.get())==True): #傳遞下拉框選擇的參數(shù) COM號+波特率  【0:5】表示只提取COM號字符
             self.var_bt1.set('關(guān)閉串口')                             #改變按鍵內(nèi)容
             self.txt_rx.insert(0.0, self.comb1.get() + ' 打開成功\r\n')  # 開頭插入
          else:
             print("串口打開失敗")
             messagebox.showinfo('錯誤','串口打開失敗')
        else:
            uart_open_close(0, 'COM1', 115200) #關(guān)閉時參數(shù)無效
            self.var_bt1.set('打開串口')

    def uart_send(self): #發(fā)送數(shù)據(jù)
        send_data = self.txt_tx.get(0.0, 'end').strip()
        if self.ascii_hex_get():    #字符發(fā)送
            uart_tx(send_data)
        else:
            send_data = send_data.replace(" ", "").replace("\n", "0A").replace("\r", "0D") #替換空格和回車換行
            if(ISHEX(send_data)==False):
                messagebox.showinfo('錯誤', '請輸入十六進(jìn)制數(shù)')
                return
            uart_tx(send_data,True)

    def tx_rx_cnt(self,rx=0,tx=0):  #發(fā)送接收統(tǒng)計
        global tx_cnt
        global rx_cnt

        rx_cnt += rx
        tx_cnt += tx
        self.lb3['text'] = '接收：'+str(rx_cnt),'發(fā)送：'+str(tx_cnt)

    def savefiles(self):   #保存日志TXT文本
        try:
            with open('log.txt','a') as file:       #a方式打開 文本追加模式
                file.write(self.txt_rx.get(0.0,'end'))
                messagebox.showinfo('提示', '保存成功')
        except:
            messagebox.showinfo('錯誤', '保存日志文件失??！')


if __name__ == '__main__':
    print('Star...')
    gui = GUI()
    gui.gettim()  #開啟時鐘
    gui.root.mainloop()
    UART.close()   #結(jié)束關(guān)閉 避免下次打開錯誤
    print('End...')

其他資源

[Python]tkinter的美顏-ttkbootstrap:讓tkinter更加美觀 - 知乎

在 Python 中使用 MQTT的方法_python mqtt_liming89的博客-CSDN博客

Python GUI之tkinter的皮膚(ttkbootstrap)打造出你的窗口之美_tkinter漂亮gui界面模板_清＆輕的博客-CSDN博客

使用python編寫mqtt客戶端向EMQX服務(wù)器發(fā)送數(shù)據(jù)_python mqtt客戶端_TMS320VC5257H的博客-CSDN博客

MQTT在Python中的使用mqtt-paho(簡單實例, 回調(diào)函數(shù),回調(diào)參數(shù)，qos安全等級)詳解及回調(diào)函數(shù)的正確用法_mqtt python-CSDN博客

MQTT在Python中的使用mqtt-paho(簡單實例, 回調(diào)函數(shù),回調(diào)參數(shù)，qos安全等級)詳解及回調(diào)函數(shù)的正確用法_mqtt python-CSDN博客

Nuitka打包教程_蒼穹之躍的博客-CSDN博客

使用nuitka打包python代碼為exe可執(zhí)行程序_ha_lee的博客-CSDN博客

Python 打包工具 Nuitka 入門指南_半勺蜂蜜~的博客-CSDN博客

Nuitka常見問題解決集錦-獨(dú)孤九劍之破Bug式 - 知乎

Python打包exe的王炸-Nuitka - 知乎

WinLibs - GCC+MinGW-w64 compiler for Windows

Nuitka打包一、安裝依賴-阿里云開發(fā)者社區(qū)

利用Nuitka打包py文件_https://github.com/ccache/ccache/releases/download-CSDN博客文章來源地址http://www.zghlxwxcb.cn/news/detail-729893.html

到了這里，關(guān)于Python實現(xiàn)的mqtt客戶端工具分享，小巧且超輕量級(python+tkinter+paho.mqtt)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！